mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
feat(attachments): AttachmentRef as the canonical by-reference content
Non-text content (user uploads, reloaded tool images) rides as AttachmentRef(id,kind) in the canonical Turn — session.messages carries ids, never bytes. Each output materializes it to inline data-URI/document parts by point-lookup on the content- addressed store: the wire (ChatSession._lower_messages_to_wire, in _full_messages), /history + export (reconstruct_messages resolves), and the per-turn token estimate (a by-ref image costs one fixed image budget; the doc char budget lands at send). reconstruct splits: reconstruct_turns = unresolved row→Turn (load_message_turns, the resume path); reconstruct_messages = resolved dict facade. RawContentBlock is demoted to the transient carrier for a resolved inline part on the dict↔Turn bridge.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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 ---------------------------------------------------
|
||||
|
||||
|
||||
|
||||
+46
-34
@@ -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", ""))
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user