feat(storage): add content-addressed attachment columns (additive)

Additive schema for the attachment cutover: workstream_attachments gains refcount +
origin, conversations gains the attachments ref-list column (migration 060 + _schema in
lockstep). Columns sit unused until the cutover, which fills them and retires the
message_id/reserved_* upload-lifecycle in favour of a content-addressed, refcounted blob
store keyed by the conversations ref-list.

Also registers the coordinator test's backend via init_storage: the attachment handlers
resolve storage through the global registry, so a bare SQLiteBackend left get_attachment
hitting a stale default db — latent until the new column made the schema drift bite.

tests/test_coordinator_endpoints.py

@@ -73,7 +73,6 @@ from turnstone.core.session_routes import (
     make_saved_handler,
     make_send_handler,
 )
-from turnstone.core.storage._sqlite import SQLiteBackend
 from turnstone.core.workstream import WorkstreamKind
 
 # ---------------------------------------------------------------------------
@@ -130,7 +129,16 @@ _coord_endpoint_config = SessionEndpointConfig(
 
 @pytest.fixture
 def storage(tmp_path):
-    return SQLiteBackend(str(tmp_path / "coord.db"))
+    # The attachment handlers resolve storage via ``memory.get_storage()`` (the
+    # registry singleton), not the instance passed to ``_make_client``/``_build_mgr``.
+    # Register the test backend so ``get_attachment`` & co. hit this fresh db rather
+    # than a stale default — otherwise schema drift (e.g. a new column) surfaces here.
+    from turnstone.core.storage import init_storage, reset_storage
+
+    reset_storage()
+    backend = init_storage("sqlite", path=str(tmp_path / "coord.db"), run_migrations=False)
+    yield backend
+    reset_storage()
 
 
 def _make_client(

tests/test_migration_060.py

@@ -166,6 +166,20 @@ class TestMigration060:
         finally:
             engine.dispose()
 
+    def test_content_addressed_attachment_columns_added(self, tmp_path: Path) -> None:
+        db_path = tmp_path / "060-ca-cols.db"
+        cfg = _alembic_cfg(db_path)
+        command.upgrade(cfg, "060")
+        engine = sa.create_engine(f"sqlite:///{db_path}")
+        try:
+            insp = sa.inspect(engine)
+            conv_cols = {c["name"] for c in insp.get_columns("conversations")}
+            att_cols = {c["name"] for c in insp.get_columns("workstream_attachments")}
+            assert "attachments" in conv_cols
+            assert {"refcount", "origin"} <= att_cols
+        finally:
+            engine.dispose()
+
     def test_ampersand_decoded_but_wrapper_tags_left_escaped(self, tmp_path: Path) -> None:
         """The un-wrap reverses only ``&amp;`` → ``&``.  Wrapper-tag entities are
         left escaped on purpose: re-activating ``&lt;system-reminder&gt;`` into a

turnstone/core/storage/_schema.py

@@ -61,6 +61,10 @@ conversations = sa.Table(
     # (history rendering + the Anthropic ``is_error`` result block) instead of
     # re-deriving it from a text heuristic.  Non-tool rows are always False.
     sa.Column("is_error", sa.Boolean, nullable=False, server_default=sa.false()),
+    # Ordered list of content-addressed attachment_id references for this turn
+    # (JSON; NULL for turns with no attachments) — the sole message->blob link in
+    # the content-addressed model; bytes resolve from workstream_attachments by id.
+    sa.Column("attachments", sa.Text, nullable=True),
 )
 
 sa.Index("idx_conversations_timestamp", conversations.c.timestamp)
@@ -554,6 +558,12 @@ workstream_attachments = sa.Table(
     # have actually been held longer than the threshold.
     sa.Column("reserved_at", sa.Text, nullable=True),
     sa.Column("created", sa.Text, nullable=False),
+    # Content-addressed blob store (canonical-trajectory cut): a deduped blob's
+    # live-reference count (pruned at 0) and its origin ('upload' | 'tool').  The
+    # cutover retires the message_id / reserved_* upload-lifecycle columns above in
+    # favour of refcount + the conversations.attachments ref-list.
+    sa.Column("refcount", sa.Integer, nullable=False, server_default=sa.text("0")),
+    sa.Column("origin", sa.Text, nullable=False, server_default=sa.text("'upload'")),
 )
 
 sa.Index("idx_ws_attachments_ws_id", workstream_attachments.c.ws_id)

turnstone/core/storage/migrations/versions/060_unwrap_tool_envelopes.py

@@ -253,6 +253,16 @@ def upgrade() -> None:
         batch_op.add_column(
             sa.Column("is_error", sa.Boolean, nullable=False, server_default=sa.false())
         )
+        # Content-addressed attachment ref-list (canonical-trajectory cut); the cutover
+        # fills it and retires the message_id/reserved_* link.
+        batch_op.add_column(sa.Column("attachments", sa.Text, nullable=True))
+    with op.batch_alter_table("workstream_attachments") as batch_op:
+        batch_op.add_column(
+            sa.Column("refcount", sa.Integer, nullable=False, server_default=sa.text("0"))
+        )
+        batch_op.add_column(
+            sa.Column("origin", sa.Text, nullable=False, server_default=sa.text("'upload'"))
+        )
 
 
 def downgrade() -> None:
@@ -263,3 +273,7 @@ def downgrade() -> None:
     with op.batch_alter_table("conversations") as batch_op:
         batch_op.add_column(sa.Column("_reminders", sa.Text, nullable=True))
         batch_op.drop_column("is_error")
+        batch_op.drop_column("attachments")
+    with op.batch_alter_table("workstream_attachments") as batch_op:
+        batch_op.drop_column("refcount")
+        batch_op.drop_column("origin")