diff --git a/tests/test_session_attachments.py b/tests/test_session_attachments.py
index 9e23b53e..4b6ed832 100644
--- a/tests/test_session_attachments.py
+++ b/tests/test_session_attachments.py
@@ -76,7 +76,7 @@ class TestMultipartBuild:
             content=PNG_1x1,
         )
         _run_send(s, "what is this?", attachments=[att])
-        msg = turn_to_dict(s.messages[-1])
+        msg = s._lower_messages_to_wire(s.messages)[-1]
         assert msg["role"] == "user"
         assert isinstance(msg["content"], list)
         assert msg["content"][0] == {"type": "text", "text": "what is this?"}
@@ -94,7 +94,7 @@ class TestMultipartBuild:
             content=b"# hi\n",
         )
         _run_send(s, "summarize", attachments=[att])
-        msg = turn_to_dict(s.messages[-1])
+        msg = s._lower_messages_to_wire(s.messages)[-1]
         doc = msg["content"][1]
         assert doc == {
             "type": "document",
@@ -113,7 +113,7 @@ class TestMultipartBuild:
             Attachment("a3", "second.md", "text/markdown", "text", b"B"),
         ]
         _run_send(s, "look", attachments=atts)
-        msg = turn_to_dict(s.messages[-1])
+        msg = s._lower_messages_to_wire(s.messages)[-1]
         types = [p["type"] for p in msg["content"]]
         assert types == ["text", "image_url", "document", "document"]
         docs = [p for p in msg["content"] if p["type"] == "document"]
@@ -124,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 = turn_to_dict(s.messages[-1])["content"]
+        parts = s._lower_messages_to_wire(s.messages)[-1]["content"]
         assert any(
             p.get("type") == "text" and p.get("text") == "[unreadable attachment: bad.bin]"
             for p in parts
@@ -227,7 +227,9 @@ class TestProviderIntegration:
         ]
         _run_send(s, "look at both", attachments=atts)
 
-        _, converted = AnthropicProvider()._convert_messages(dicts_from_turns([s.messages[-1]]))
+        _, converted = AnthropicProvider()._convert_messages(
+            s._lower_messages_to_wire([s.messages[-1]])
+        )
         assert len(converted) == 1
         content = converted[0]["content"]
         types = [p["type"] for p in content]
@@ -279,7 +281,7 @@ class TestProviderIntegration:
         ]
         _run_send(s, "review", attachments=atts)
 
-        out = sanitize_messages(dicts_from_turns([s.messages[-1]]))
+        out = sanitize_messages(s._lower_messages_to_wire([s.messages[-1]]))
         parts = out[0]["content"]
         types = [p["type"] for p in parts]
         assert types == ["text", "text"]
@@ -333,13 +335,18 @@ class TestTokenAccounting:
     def test_text_doc_adds_text_char_budget(self, tmp_db, mock_openai_client):
         baseline = _make_session(mock_openai_client)
         _run_send(baseline, "hi")
-        plain_tokens = baseline._msg_tokens[-1]
+        plain_chars = baseline._msg_char_count(
+            baseline._lower_messages_to_wire(baseline.messages)[-1]
+        )
 
         big = "x" * 4000
         with_doc = _make_session(mock_openai_client)
         att = Attachment("a1", "big.md", "text/markdown", "text", big.encode())
         _run_send(with_doc, "hi", attachments=[att])
-        doc_tokens = with_doc._msg_tokens[-1]
+        doc_chars = with_doc._msg_char_count(
+            with_doc._lower_messages_to_wire(with_doc.messages)[-1]
+        )
 
-        # ~4000 chars / 4 chars_per_token ≈ ~1000 tokens added
-        assert doc_tokens - plain_tokens >= 900
+        # The ~4000-char doc lands at the resolved boundary (the per-turn
+        # placeholder no longer carries the bytes), well above the budget floor.
+        assert doc_chars - plain_chars >= 900
diff --git a/tests/test_trajectory.py b/tests/test_trajectory.py
index efb9bed8..0969a47b 100644
--- a/tests/test_trajectory.py
+++ b/tests/test_trajectory.py
@@ -47,6 +47,15 @@ _ROUNDTRIP: list[dict[str, Any]] = [
             {"type": "text", "text": "[unreadable attachment: bad.bin]"},
         ],
     },
+    {
+        # By-reference attachments: the canonical content form (id, never bytes).
+        "role": "user",
+        "content": [
+            {"type": "text", "text": "what's this?"},
+            {"type": "image", "attachment_id": "sha256:abc"},
+            {"type": "document", "attachment_id": "sha256:def"},
+        ],
+    },
     {"role": "assistant", "content": "hi there"},
     {"role": "assistant", "content": ""},  # empty assistant (no text, no tools)
     {
@@ -179,7 +188,44 @@ def test_empty_content_roundtrips_to_empty_string() -> None:
     assert turn_to_dict(Turn(Role.ASSISTANT)) == {"role": "assistant", "content": ""}
 
 
-def test_attachment_ref_emits_defensive_part() -> None:
-    # AttachmentRef isn't produced pre-by-ref-wiring, but the emit path is defined.
+def test_attachment_ref_is_the_canonical_non_text_form() -> None:
+    # By-reference placeholders ``{type: image|document, attachment_id}`` are the
+    # canonical content; a real inline ``image_url`` (no id) stays a RawContentBlock.
+    t = turn_from_dict(
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "q"},
+                {"type": "image", "attachment_id": "sha256:abc"},
+                {"type": "document", "attachment_id": "sha256:def"},
+                {"type": "image_url", "image_url": {"url": "data:..."}},
+            ],
+        }
+    )
+    assert isinstance(t.content[1], AttachmentRef) and t.content[1].attachment_id == "sha256:abc"
+    assert isinstance(t.content[2], AttachmentRef) and t.content[2].kind == "document"
+    assert isinstance(t.content[3], RawContentBlock)  # resolved inline part, no id
+
+
+def test_attachment_ref_emits_placeholder() -> None:
     t = Turn(Role.USER, (AttachmentRef(attachment_id="abc", kind="image"),))
     assert turn_to_dict(t)["content"] == [{"type": "image", "attachment_id": "abc"}]
+
+
+def test_resolve_attachment_refs_materializes_and_drops_missing() -> None:
+    from turnstone.core.trajectory import resolve_attachment_refs
+
+    turns = [
+        Turn(
+            Role.USER,
+            (
+                TextBlock("look"),
+                AttachmentRef(attachment_id="a1", kind="image"),
+                AttachmentRef(attachment_id="gone", kind="image"),
+            ),
+        )
+    ]
+    part = {"type": "image_url", "image_url": {"url": "data:img"}}
+    out = turn_to_dict(resolve_attachment_refs(turns, {"a1": part})[0])
+    # a1 → inline part; the pruned ref is dropped; text kept.
+    assert out["content"] == [{"type": "text", "text": "look"}, part]
diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py
index efe7833b..288e2908 100644
--- a/turnstone/core/memory.py
+++ b/turnstone/core/memory.py
@@ -22,6 +22,8 @@ from turnstone.core.workstream import WorkstreamKind
 if TYPE_CHECKING:
     from collections.abc import Callable
 
+    from turnstone.core.trajectory import Turn
+
 log = get_logger(__name__)
 
 
@@ -96,6 +98,19 @@ def load_messages(ws_id: str, *, repair: bool = True) -> list[dict[str, Any]]:
         return []
 
 
+def load_message_turns(ws_id: str) -> list[Turn]:
+    """Load a workstream's history as canonical ``Turn``s (by-reference content).
+
+    The resume path — see :meth:`StorageBackend.load_message_turns`.  Returns an
+    empty list on any storage error (a failed resume must not crash the session).
+    """
+    try:
+        return get_storage().load_message_turns(ws_id)
+    except Exception:
+        log.warning("Failed to load message turns for ws=%s", ws_id, exc_info=True)
+        return []
+
+
 # -- Workstream attachments ---------------------------------------------------
 
 
diff --git a/turnstone/core/session.py b/turnstone/core/session.py
index e6824eaf..1956bec5 100644
--- a/turnstone/core/session.py
+++ b/turnstone/core/session.py
@@ -62,6 +62,7 @@ from turnstone.core.memory import (
     delete_messages_after,
     delete_structured_memory_by_id,
     delete_workstream,
+    get_attachments,
     get_skill_by_name,
     get_structured_memory_by_name,
     get_workstream_display_name,
@@ -70,7 +71,7 @@ from turnstone.core.memory import (
     list_structured_memories,
     list_visible_structured_memories,
     list_workstreams_with_history,
-    load_messages,
+    load_message_turns,
     load_workstream_config,
     normalize_key,
     resolve_workstream,
@@ -108,7 +109,11 @@ from turnstone.core.safety import is_command_blocked, sanitize_command
 from turnstone.core.skill_field_validation import SKILL_RUNTIME_CONFIG_FIELDS
 from turnstone.core.skill_parser import MAX_SKILL_DESCRIPTION_LEN
 from turnstone.core.storage._registry import get_storage
-from turnstone.core.storage._utils import normalize_search_terms, strip_orphan_client_tool_blocks
+from turnstone.core.storage._utils import (
+    attachment_to_content_part,
+    normalize_search_terms,
+    strip_orphan_client_tool_blocks,
+)
 from turnstone.core.tool_advisory import (
     make_system_turn,
     render_user_interjection,
@@ -124,9 +129,11 @@ from turnstone.core.tools import (
     merge_mcp_tools,
 )
 from turnstone.core.trajectory import (
+    AttachmentRef,
     Role,
     Turn,
     dicts_from_turns,
+    resolve_attachment_refs,
     turn_from_dict,
     turn_to_dict,
     turns_from_dicts,
@@ -2482,7 +2489,7 @@ class ChatSession:
         so the resumed/forked workstream behaves identically to the
         original.  Returns True on success.
         """
-        turns = turns_from_dicts(load_messages(ws_id))
+        turns = load_message_turns(ws_id)
         if not turns:
             return False
         if not fork:
@@ -2884,8 +2891,30 @@ class ChatSession:
 
         ``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)
+        boundary moves inward when those layers migrate).  By-reference
+        attachments are materialized to inline parts at this output boundary."""
+        return self.system_messages + self._lower_messages_to_wire(self.messages)
+
+    def _lower_messages_to_wire(self, turns: list[Turn]) -> list[dict[str, Any]]:
+        """Lower canonical Turns to wire dicts, resolving AttachmentRef → bytes.
+
+        Attachment content lives by reference in ``self.messages`` (ids, not
+        bytes); at this output boundary the referenced blobs are batch-fetched
+        from the content-addressed store and expanded to inline content parts —
+        the same ``data:…``/document parts the providers have always received.
+        Allocation-free when no turn carries an attachment.
+        """
+        ids = sorted(
+            {b.attachment_id for t in turns for b in t.content if isinstance(b, AttachmentRef)}
+        )
+        if not ids:
+            return dicts_from_turns(turns)
+        parts_by_id = {
+            str(att["attachment_id"]): part
+            for att in get_attachments(ids)
+            if (part := attachment_to_content_part(att)) is not None
+        }
+        return dicts_from_turns(resolve_attachment_refs(turns, parts_by_id))
 
     def _prepare_wire_messages(
         self,
@@ -3512,37 +3541,16 @@ class ChatSession:
         self._invalidate_memory_cache()
         user_content: str | list[dict[str, Any]]
         if attachments:
+            # Attachments ride by reference (``{type: kind, attachment_id}`` →
+            # AttachmentRef); the bytes — already committed content-addressed
+            # below — materialize at each output (wire / display), where the
+            # bad-UTF-8 / unrenderable handling now lives.
             parts: list[dict[str, Any]] = [{"type": "text", "text": user_input}]
             for att in attachments:
                 if att.is_image:
-                    parts.append(
-                        {
-                            "type": "image_url",
-                            "image_url": {
-                                "url": _encode_image_data_uri(att.content, att.mime_type),
-                            },
-                        }
-                    )
+                    parts.append({"type": "image", "attachment_id": att.attachment_id})
                 elif att.is_text:
-                    try:
-                        text = att.content.decode("utf-8")
-                    except UnicodeDecodeError:
-                        log.warning(
-                            "attachment id=%s is not valid UTF-8; injecting placeholder",
-                            att.attachment_id,
-                        )
-                        parts.append(unreadable_placeholder(att.filename))
-                        continue
-                    parts.append(
-                        {
-                            "type": "document",
-                            "document": {
-                                "name": att.filename,
-                                "media_type": att.mime_type,
-                                "data": text,
-                            },
-                        }
-                    )
+                    parts.append({"type": "document", "attachment_id": att.attachment_id})
                 else:
                     log.warning(
                         "attachment id=%s has unknown kind=%r; injecting placeholder",
@@ -4792,9 +4800,13 @@ class ChatSession:
                 ptype = p.get("type")
                 if ptype == "text":
                     n += len(p.get("text", ""))
-                elif ptype == "image_url":
+                elif ptype == "image_url" or (ptype == "image" and p.get("attachment_id")):
+                    # Resolved inline image, or the by-reference image placeholder
+                    # — both cost one fixed image budget.
                     images += 1
-                elif ptype == "document":
+                elif ptype == "document" and not p.get("attachment_id"):
+                    # Resolved inline document; the by-reference placeholder carries
+                    # no bytes, so its char budget lands at send-time calibration.
                     d = p.get("document", {})
                     doc_chars += len(d.get("data", ""))
                     doc_chars += len(d.get("name", ""))
diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py
index 43efef2c..a16940cf 100644
--- a/turnstone/core/storage/_postgresql.py
+++ b/turnstone/core/storage/_postgresql.py
@@ -15,6 +15,7 @@ if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator, Sequence
 
     from turnstone.core.storage._notify import Notify, NotifyStream
+    from turnstone.core.trajectory import Turn
 
 import sqlalchemy as sa
 
@@ -124,6 +125,12 @@ from turnstone.core.storage._utils import prepare_provider_data_for_save, saniti
 from turnstone.core.storage._utils import (
     reconstruct_messages as _reconstruct_messages,
 )
+from turnstone.core.storage._utils import (
+    reconstruct_turns as _reconstruct_turns,
+)
+from turnstone.core.storage._utils import (
+    recover_trajectory as _recover_trajectory,
+)
 from turnstone.core.storage._utils import (
     row_to_dict as _row_to_dict,
 )
@@ -369,13 +376,13 @@ class PostgreSQLBackend:
                 )
             conn.commit()
 
-    def load_messages(
-        self, ws_id: str, *, limit: int | None = None, repair: bool = True
-    ) -> list[dict[str, Any]]:
-        # The trailing ``attachments`` column carries the per-row
-        # content-addressed ref-list; it is split off below to resolve blobs
-        # and is NOT part of the positional tuple ``reconstruct_messages``
-        # unpacks (id..is_error).
+    def _conversation_rows(
+        self, ws_id: str, limit: int | None
+    ) -> tuple[list[tuple[Any, ...]], dict[int, list[dict[str, Any]]] | None]:
+        """Fetch a ws's conversation rows + resolved attachment map (shared by
+        :meth:`load_messages` and :meth:`load_message_turns`).  The trailing
+        ``attachments`` ref-list column is split off and is NOT part of the
+        positional tuple ``reconstruct_*`` unpacks (id..is_error)."""
         _cols = (
             conversations.c.id,
             conversations.c.role,
@@ -405,10 +412,20 @@ class PostgreSQLBackend:
                     .order_by(conversations.c.id)
                 ).fetchall()
         attachments = self._resolve_row_attachments(rows)
-        # Strip the trailing ref-list column so the tuple shape stays exactly
-        # what ``reconstruct_messages`` expects (id..is_error).
         msg_rows = [tuple(r)[:10] for r in rows]
-        return _reconstruct_messages(msg_rows, ws_id, attachments or None, repair=repair)
+        return msg_rows, (attachments or None)
+
+    def load_messages(
+        self, ws_id: str, *, limit: int | None = None, repair: bool = True
+    ) -> list[dict[str, Any]]:
+        msg_rows, attachments = self._conversation_rows(ws_id, limit)
+        return _reconstruct_messages(msg_rows, ws_id, attachments, repair=repair)
+
+    def load_message_turns(self, ws_id: str) -> list[Turn]:
+        """Load the conversation as canonical ``Turn``s (unresolved AttachmentRef)
+        for resume; bytes materialize at each output, never here."""
+        msg_rows, attachments = self._conversation_rows(ws_id, None)
+        return _recover_trajectory(_reconstruct_turns(msg_rows, ws_id, attachments))
 
     def _resolve_row_attachments(self, rows: Sequence[Any]) -> dict[int, list[dict[str, Any]]]:
         """Build the ``reconstruct_messages`` attachment map from row ref-lists.
diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py
index b86e2b79..e319643a 100644
--- a/turnstone/core/storage/_protocol.py
+++ b/turnstone/core/storage/_protocol.py
@@ -9,6 +9,7 @@ if TYPE_CHECKING:
     from contextlib import AbstractContextManager
 
     from turnstone.core.storage._notify import NotifyStream
+    from turnstone.core.trajectory import Turn
     from turnstone.core.workstream import WorkstreamKind
 
 
@@ -212,6 +213,21 @@ class StorageBackend(Protocol):
         ``repair=False`` so the user sees the actual partial state
         instead of having the trailing turn silently stripped during
         live tool execution.
+
+        Attachments are resolved to inline content parts (the materialized
+        bytes a display/export consumer needs); :meth:`load_message_turns` is
+        the unresolved, by-reference counterpart for resume.
+        """
+        ...
+
+    def load_message_turns(self, ws_id: str) -> list[Turn]:
+        """Load a workstream's full history as canonical ``Turn``s for resume.
+
+        Unlike :meth:`load_messages` this keeps attachments *by reference*
+        (:class:`AttachmentRef`) — ``session.messages`` is the canonical Turn
+        trajectory and materializes bytes only at each output (wire / display).
+        The trailing-incomplete-tool-call strip (``recover_trajectory``) is
+        applied; mid-conversation orphans are left for the send-time repair.
         """
         ...
 
diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py
index 0cd03379..8fc857a1 100644
--- a/turnstone/core/storage/_sqlite.py
+++ b/turnstone/core/storage/_sqlite.py
@@ -17,6 +17,7 @@ if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator, Sequence
 
     from turnstone.core.storage._notify import Notify, NotifyStream
+    from turnstone.core.trajectory import Turn
 
 from turnstone.core.log import get_logger
 from turnstone.core.storage._protocol import (
@@ -124,6 +125,12 @@ from turnstone.core.storage._utils import prepare_provider_data_for_save, saniti
 from turnstone.core.storage._utils import (
     reconstruct_messages as _reconstruct_messages,
 )
+from turnstone.core.storage._utils import (
+    reconstruct_turns as _reconstruct_turns,
+)
+from turnstone.core.storage._utils import (
+    recover_trajectory as _recover_trajectory,
+)
 from turnstone.core.storage._utils import (
     row_to_dict as _row_to_dict,
 )
@@ -427,13 +434,16 @@ class SQLiteBackend:
                     self._fts5_available = False
             conn.commit()
 
-    def load_messages(
-        self, ws_id: str, *, limit: int | None = None, repair: bool = True
-    ) -> list[dict[str, Any]]:
-        # The trailing ``attachments`` column carries the per-row
-        # content-addressed ref-list; it is split off below to resolve blobs
-        # and is NOT part of the positional tuple ``reconstruct_messages``
-        # unpacks (id..is_error).
+    def _conversation_rows(
+        self, ws_id: str, limit: int | None
+    ) -> tuple[list[tuple[Any, ...]], dict[int, list[dict[str, Any]]] | None]:
+        """Fetch a ws's conversation rows + resolved attachment map.
+
+        Shared by :meth:`load_messages` (→ dicts, resolved for display) and
+        :meth:`load_message_turns` (→ canonical Turns for resume).  The trailing
+        ``attachments`` ref-list column is split off to resolve blobs and is NOT
+        part of the positional tuple ``reconstruct_*`` unpacks (id..is_error).
+        """
         _cols = (
             conversations.c.id,
             conversations.c.role,
@@ -467,10 +477,23 @@ class SQLiteBackend:
                 ).fetchall()
 
         attachments = self._resolve_row_attachments(rows)
-        # Strip the trailing ref-list column so the tuple shape stays exactly
-        # what ``reconstruct_messages`` expects (id..is_error).
         msg_rows = [tuple(r)[:10] for r in rows]
-        return _reconstruct_messages(msg_rows, ws_id, attachments or None, repair=repair)
+        return msg_rows, (attachments or None)
+
+    def load_messages(
+        self, ws_id: str, *, limit: int | None = None, repair: bool = True
+    ) -> list[dict[str, Any]]:
+        msg_rows, attachments = self._conversation_rows(ws_id, limit)
+        return _reconstruct_messages(msg_rows, ws_id, attachments, repair=repair)
+
+    def load_message_turns(self, ws_id: str) -> list[Turn]:
+        """Load the conversation as canonical ``Turn``s (unresolved AttachmentRef).
+
+        The resume path: ``session.messages`` holds the by-reference content;
+        bytes are materialized at each output (wire / display), never here.
+        """
+        msg_rows, attachments = self._conversation_rows(ws_id, None)
+        return _recover_trajectory(_reconstruct_turns(msg_rows, ws_id, attachments))
 
     def _resolve_row_attachments(self, rows: Sequence[Any]) -> dict[int, list[dict[str, Any]]]:
         """Build the ``reconstruct_messages`` attachment map from row ref-lists.
diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py
index ff2a5bc3..b5b426f2 100644
--- a/turnstone/core/storage/_utils.py
+++ b/turnstone/core/storage/_utils.py
@@ -10,15 +10,16 @@ from typing import Any
 from turnstone.core.attachments import unreadable_placeholder
 from turnstone.core.log import get_logger
 from turnstone.core.trajectory import (
+    AttachmentRef,
     ContentBlock,
     ProviderNative,
-    RawContentBlock,
     Role,
     TextBlock,
     ToolCall,
     Turn,
     TurnMeta,
     dicts_from_turns,
+    resolve_attachment_refs,
 )
 
 log = get_logger(__name__)
@@ -131,7 +132,7 @@ def prepare_provider_data_for_save(
     )
 
 
-def _attachment_to_content_part(att: dict[str, Any]) -> dict[str, Any] | None:
+def attachment_to_content_part(att: dict[str, Any]) -> dict[str, Any] | None:
     """Convert a stored attachment row into an OpenAI-style content part.
 
     Returns ``None`` if the attachment's ``kind`` / ``content`` cannot be
@@ -206,30 +207,36 @@ def build_attachments_by_msg(
     return grouped
 
 
-def _reconstruct_attachment_parts(
+def _reconstruct_attachment_refs(
     attachments_by_msg: dict[int, list[dict[str, Any]]] | None,
     row_id: int | None,
-) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
-    """Build ``(content_parts, attachments_meta)`` for a row from its ref-list.
+) -> tuple[list[AttachmentRef], list[dict[str, Any]]]:
+    """Build ``(attachment_refs, attachments_meta)`` for a row from its ref-list.
 
     ``attachments_by_msg`` maps a conversations row id to the ordered list of
-    content-addressed attachment rows referenced by that row's
-    ``attachments`` column (resolved to include ``content`` bytes).  Returns
-    the OpenAI-style content parts (image_url / document, in ref-list order)
+    content-addressed attachment rows referenced by that row's ``attachments``
+    column.  Returns the by-reference content blocks (:class:`AttachmentRef`, in
+    ref-list order — bytes resolve at the consumer, never carried in the Turn)
     and the display-oriented ``_attachments_meta`` siblings (kind / filename /
-    mime_type) — the latter is tracked even when a part can't be reconstructed
-    so history replay keeps filenames available (e.g. image pills).  Shared by
-    the user- and tool-row reconstruction so both surfaces stay byte-identical
-    in part shape.
+    mime_type) so history replay keeps filenames available (e.g. image pills).
+    Shared by the user- and tool-row reconstruction.
     """
-    parts: list[dict[str, Any]] = []
+    refs: list[AttachmentRef] = []
     meta: list[dict[str, Any]] = []
     if not attachments_by_msg or row_id is None:
-        return parts, meta
+        return refs, meta
     for att in attachments_by_msg.get(row_id, []):
-        part = _attachment_to_content_part(att)
-        if part is not None:
-            parts.append(part)
+        # AttachmentRef.kind is the by-reference content kind ('image' |
+        # 'document'); the stored blob kind ('image' | 'text') drives the actual
+        # resolution.  'document' (not 'text') keeps the placeholder type from
+        # colliding with a real text content part on the dict round-trip.
+        ref_kind = "image" if str(att.get("kind") or "") == "image" else "document"
+        refs.append(
+            AttachmentRef(
+                attachment_id=str(att.get("attachment_id") or ""),
+                kind=ref_kind,
+            )
+        )
         meta.append(
             {
                 "kind": str(att.get("kind") or ""),
@@ -237,7 +244,7 @@ def _reconstruct_attachment_parts(
                 "mime_type": str(att.get("mime_type") or ""),
             }
         )
-    return parts, meta
+    return refs, meta
 
 
 # ---------------------------------------------------------------------------
@@ -562,18 +569,31 @@ def reconstruct_messages(
     turns = reconstruct_turns(rows, ws_id, attachments_by_msg)
     if repair:
         turns = recover_trajectory(turns)
+    # Dict consumers (display, export) want materialized content, so resolve the
+    # by-reference attachments to inline parts using the blob rows already in
+    # hand.  ``load_message_turns`` is the unresolved canonical path for resume.
+    if attachments_by_msg:
+        parts_by_id = {
+            str(att.get("attachment_id") or ""): part
+            for atts in attachments_by_msg.values()
+            for att in atts
+            if (part := attachment_to_content_part(att)) is not None
+        }
+        if parts_by_id:
+            turns = resolve_attachment_refs(turns, parts_by_id)
     return dicts_from_turns(turns)
 
 
-def _content_blocks(text: str | None, parts: list[dict[str, Any]]) -> tuple[ContentBlock, ...]:
-    """Build typed content blocks from a row's text column + attachment parts.
+def _content_blocks(text: str | None, refs: list[AttachmentRef]) -> tuple[ContentBlock, ...]:
+    """Build typed content blocks from a row's text column + attachment refs.
 
-    A row with attachment parts becomes a leading text block plus one raw part
-    per attachment (``read_file`` vision output, user uploads); a text-only row
-    is a single text block, or empty.
+    A row with attachments becomes a leading text block plus one
+    :class:`AttachmentRef` per attachment (``read_file`` vision output, user
+    uploads — bytes resolve at the consumer); a text-only row is a single text
+    block, or empty.
     """
-    if parts:
-        return (TextBlock(text or ""), *(RawContentBlock(p) for p in parts))
+    if refs:
+        return (TextBlock(text or ""), *refs)
     if text:
         return (TextBlock(text),)
     return ()
@@ -646,10 +666,10 @@ def reconstruct_turns(
         src = str(source) if source else None
 
         if role == "user":
-            parts, am = _reconstruct_attachment_parts(attachments_by_msg, row_id)
+            refs, am = _reconstruct_attachment_refs(attachments_by_msg, row_id)
             if am:
                 meta.extra["attachments_meta"] = am
-            turns.append(Turn(Role.USER, _content_blocks(content, parts), source=src, meta=meta))
+            turns.append(Turn(Role.USER, _content_blocks(content, refs), source=src, meta=meta))
         elif role == "assistant":
             turns.append(
                 Turn(
@@ -661,11 +681,11 @@ def reconstruct_turns(
                 )
             )
         elif role == "tool":
-            tparts, _tmeta = _reconstruct_attachment_parts(attachments_by_msg, row_id)
+            trefs, _tmeta = _reconstruct_attachment_refs(attachments_by_msg, row_id)
             turns.append(
                 Turn(
                     Role.TOOL,
-                    _content_blocks(content, tparts),
+                    _content_blocks(content, trefs),
                     tool_call_id=tc_id or "",
                     is_error=is_error,
                     meta=meta,
diff --git a/turnstone/core/trajectory.py b/turnstone/core/trajectory.py
index f6ac1f74..73b0688c 100644
--- a/turnstone/core/trajectory.py
+++ b/turnstone/core/trajectory.py
@@ -9,15 +9,17 @@ canonical would break cross-provider resume.
 
 Field set and rationale: ``docs/design/canonical-trajectory-ideal-target.md`` §2.
 
-NOTE: ``AttachmentRef`` references a content-addressed blob in ``workstream_attachments``
-(the by-reference attachment model).  Wiring attachment content through ``Turn`` lands
-with that storage cut; until then the model is defined here but the dict↔Turn adapters
-cover text / tool_calls / native / tool turns.
+NOTE: non-text content rides as ``AttachmentRef`` — a reference to a content-addressed
+blob in ``workstream_attachments``.  ``Turn``s never carry bytes; each output boundary
+(the provider wire, the ``/history`` display, export) materializes the reference to an
+inline part by point-lookup against the blob store.  ``RawContentBlock`` is the transient
+carrier for an already-resolved inline part as it rides the dict↔Turn bridge — never a
+persisted or canonical form.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from enum import StrEnum
 from typing import Any
 
@@ -55,15 +57,15 @@ class AttachmentRef:
 
 @dataclass(slots=True)
 class RawContentBlock:
-    """Transitional carrier for a verbatim wire content-part dict (``image_url`` /
-    ``document``).
+    """Transient carrier for an already-resolved wire content-part dict (``image_url``
+    / ``document``).
 
-    The by-reference model (§2/§6) is :class:`AttachmentRef` — id + kind, with the
-    translator resolving bytes at wire time.  Until that wiring lands (it relocates
-    byte-resolution out of ``reconstruct``'s inline data-URL path), the dict↔Turn
-    adapters carry an attachment part *verbatim* here so multipart turns round-trip
-    byte-identically.  Contributes nothing to :attr:`Turn.text` (you cannot
-    full-text-search an image).  Removed when :class:`AttachmentRef` is wired through.
+    The canonical non-text content form is :class:`AttachmentRef` (by reference);
+    at an output boundary the reference is materialized to an inline part, and that
+    inline part rides the dict↔Turn bridge here so multipart turns round-trip
+    byte-identically.  Never persisted, never in ``session.messages`` — only
+    transiently between :func:`resolve_attachment_refs` and the translator.
+    Contributes nothing to :attr:`Turn.text` (you cannot full-text-search an image).
     """
 
     part: dict[str, Any]
@@ -188,9 +190,19 @@ def _content_from_raw(raw: Any) -> tuple[ContentBlock, ...]:
     if isinstance(raw, list):
         blocks: list[ContentBlock] = []
         for part in raw:
-            if isinstance(part, dict) and part.get("type") == "text":
+            if not isinstance(part, dict):
+                blocks.append(RawContentBlock(part))
+            elif part.get("type") == "text":
                 blocks.append(TextBlock(part.get("text", "")))
+            elif part.get("attachment_id"):
+                # The by-reference placeholder ``{type: kind, attachment_id}`` —
+                # the canonical form for non-text content (bytes resolve at send).
+                blocks.append(
+                    AttachmentRef(attachment_id=part["attachment_id"], kind=part.get("type", ""))
+                )
             else:
+                # A resolved inline part (image_url / document, carrying bytes) —
+                # transient wire-prep form that rides the dict↔Turn bridge.
                 blocks.append(RawContentBlock(part))
         return tuple(blocks)
     return (RawContentBlock(raw),)  # defensive — unexpected scalar
@@ -215,7 +227,7 @@ def _content_to_raw(content: tuple[ContentBlock, ...]) -> str | list[dict[str, A
             parts.append({"type": "text", "text": b.text})
         elif isinstance(b, RawContentBlock):
             parts.append(b.part)
-        elif isinstance(b, AttachmentRef):  # not produced pre-by-ref-wiring; defensive
+        elif isinstance(b, AttachmentRef):  # by-reference placeholder (unresolved)
             parts.append({"type": b.kind, "attachment_id": b.attachment_id})
     return parts
 
@@ -294,3 +306,33 @@ def turns_from_dicts(msgs: list[dict[str, Any]]) -> list[Turn]:
 
 def dicts_from_turns(turns: list[Turn]) -> list[dict[str, Any]]:
     return [turn_to_dict(t) for t in turns]
+
+
+def resolve_attachment_refs(
+    turns: list[Turn], parts_by_id: dict[str, dict[str, Any]]
+) -> list[Turn]:
+    """Replace each :class:`AttachmentRef` with its resolved inline content part.
+
+    *parts_by_id* maps an ``attachment_id`` to the wire content part (image_url /
+    document) built from the content-addressed blob — the bytes a translator
+    needs.  This is the send-time materialization of the by-reference content
+    lane; a ref whose blob is missing (pruned) is dropped, so the wire never
+    carries an unresolved reference.  Identity-preserving for turns that hold no
+    ``AttachmentRef``; never mutates the input turns.
+    """
+    out: list[Turn] = []
+    for t in turns:
+        if not any(isinstance(b, AttachmentRef) for b in t.content):
+            out.append(t)
+            continue
+        new_content: list[ContentBlock] = []
+        for b in t.content:
+            if isinstance(b, AttachmentRef):
+                part = parts_by_id.get(b.attachment_id)
+                if part is not None:
+                    new_content.append(RawContentBlock(part))
+                # else: the blob is gone (pruned) — drop the ref.
+            else:
+                new_content.append(b)
+        out.append(replace(t, content=tuple(new_content)))
+    return out