feat(scribe): archive, trash and look up a mailbox entry by message number (T-13/T-9)

The user wants a way to get rid of a message that is no longer useful. A
message is stored once and shared by every recipient; `history` is how
agents rebuild context (D-66), so deleting the message itself would take
it from every other reader. Deletion is therefore per mailbox entry
(D-70): `archive` (kept, out of the default view) and `trash` (out of the
mailbox, restorable). True erasure of content stays `redact` — not this
round.

Migration 009 (down_revision="008") adds
scribe_mailbox_entries.archived_at and trashed_at — independent nullable
timestamps, mirroring book_records.archived_at/trashed_at (migration
005). No data step: every existing entry is neither.

PATCH /scribe/mailbox/{entry_id} takes `archived` and `trashed` booleans
beside `read` (true sets the timestamp if not already set, false clears
it); the response carries both. GET /scribe/mailbox takes `view`: `inbox`
(default, neither archived nor trashed), `archive` (archived and not
trashed), `trash` (trashed, regardless of archived_at — trashing hides an
entry from the archive view too), `all` (no filter on either column).
`find` now excludes trashed entries (archived ones still match).

T-9: GET /scribe/mailbox?message_id=N is a direct lookup, not a filtered
list — the unique constraint on (address, message_id) guarantees zero or
one, in any view, ignoring unread/tag/before_id entirely. That is the
point: a client that already knows the message number never pages the
unread-first/id-descending listing to find it.

Judgement call: the listing's own cursor (`before_id`, a single `id`
comparison) still does not match its two-key sort (unread-first, then id
descending) — the pre-existing gap the dossier flagged as optional to
fix. Left unchanged: T-9's message_id lookup is the actual answer to "a
client pages to find one," and reworking the general cursor is a larger,
separate concern (which field(s) a keyset cursor needs) that risks the
already-passing test_mailbox_lists_unread_first for no requirement this
round asked for.

Judgement call: archived_at and trashed_at are independent timestamps,
never cross-clearing each other. Trashing an already-archived entry does
not clear archived_at, so restoring it (trashed=false) returns it to the
archive view, not the inbox — the view filter's own trash-takes-priority
rule is what keeps it out of view=archive while trashed, not a write-time
side effect.

Tests: 12 new (6 archive/trash flow + 4 message_id lookup + 2 migration
round-trip). Full suite 554 -> 566 passed.

Mutation checks (Edit, reverted after each):
- view=inbox dropped its trashed_at filter ->
  test_trash_hides_from_inbox_archive_and_find_shows_in_trash failed
  (assert 1 == 0).
- message_id lookup dropped the caller-address filter ->
  test_message_id_answers_nothing_for_a_message_not_in_their_mailbox
  failed (assert 1 == 0).
- trash updated every mailbox entry for the message instead of the
  caller's own row ->
  test_trashing_one_recipients_entry_leaves_the_others_untouched failed
  (assert 0 == 1).
All three reverted; full suite re-verified green (566 passed).

Migration 008 -> 009 -> 008 -> 009 round-tripped against core_api_test,
both manually (alembic CLI) and in
TestMigration009RoundTrip. Production stays at 008 — every alembic
invocation derived POSTGRES_* from TEST_DATABASE_URL.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

alembic/versions/20261019_1800_009_add_archive_and_trash_to_mailbox_entries.py

@@ -0,0 +1,45 @@
+"""Archive and trash a message in your own mailbox (core-api T-13)
+
+Revision ID: 009
+Revises: 008
+Create Date: 2026-10-19
+
+Deletion is per mailbox entry, never per message (workspace D-66/D-70): a
+message is retained and shared by every recipient, so only the entry that
+represents one recipient's copy can be archived or trashed. `archived_at`
+and `trashed_at` are independent nullable timestamps on
+`scribe_mailbox_entries`, mirroring `book_records.archived_at`/
+`trashed_at` (migration 005) — the service layer's `view=trash` takes
+priority over `view=archive` when both are set, so the column pair alone
+does not encode that precedence.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "009"
+down_revision: Union[str, None] = "008"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add the two nullable timestamp columns. No data step: every existing
+    entry is neither archived nor trashed."""
+    op.add_column(
+        "scribe_mailbox_entries",
+        sa.Column("archived_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.add_column(
+        "scribe_mailbox_entries",
+        sa.Column("trashed_at", sa.DateTime(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    """Drop both columns."""
+    op.drop_column("scribe_mailbox_entries", "trashed_at")
+    op.drop_column("scribe_mailbox_entries", "archived_at")

src/domains/scribe/controller.py

@@ -2,10 +2,11 @@
 Scribe Controller
 
 The Scribe's messaging verbs (dossier core-api T-5): register, list
-(addresses), send, mailbox, history, find, mark, tag/untag. Every route is
+(addresses), send, mailbox, history, find, mark, tag/untag, archive/trash
-behind an Office or user caller (T-1/T-4); every query filters on the
+a mailbox entry and look one up by message number (core-api T-13/T-9).
-caller's tenant (D-73) — an address or message in another tenant answers
+Every route is behind an Office or user caller (T-1/T-4); every query
-exactly like one that does not exist.
+filters on the caller's tenant (D-73) — an address or message in another
+tenant answers exactly like one that does not exist.
 """
 from typing import Optional
 
@@ -20,6 +21,7 @@ from src.domains.scribe.schemas import (
     MailboxEntryResponse,
     MailboxListResponse,
     MailboxPatchRequest,
+    MailboxView,
     MessageResponse,
     MessageSendRequest,
 )
@@ -54,6 +56,8 @@ def _entry_response(entry, message) -> MailboxEntryResponse:
         id=entry.id,
         address=entry.address,
         read_at=entry.read_at,
+        archived_at=entry.archived_at,
+        trashed_at=entry.trashed_at,
         tags=entry.tags,
         created_at=entry.created_at,
         message=_message_response(message),
@@ -171,16 +175,31 @@ class ScribeController(BaseController):
             summary="The caller identity's mailbox, unread first",
         )
         async def get_mailbox(
+            view: MailboxView = Query(
+                "inbox", description="inbox (default, neither archived nor trashed), archive, trash, all"
+            ),
             unread: bool = Query(False),
             tag: Optional[str] = Query(None),
             limit: int = Query(50, ge=1, le=200),
             before_id: Optional[int] = Query(None),
+            message_id: Optional[int] = Query(
+                None,
+                description="look up the caller's own entry for this message number directly, "
+                "in any view, ignoring unread/tag/before_id (core-api T-9)",
+            ),
             caller: Caller = Depends(require_scribe_reader),
             session: AsyncSession = Depends(get_async_session),
         ):
             try:
                 results, total = await service.get_mailbox(
-                    session, caller, unread=unread, tag=tag, limit=limit, before_id=before_id
+                    session,
+                    caller,
+                    view=view,
+                    unread=unread,
+                    tag=tag,
+                    limit=limit,
+                    before_id=before_id,
+                    message_id=message_id,
                 )
             except ScribeError as error:
                 raise HTTPException(

src/domains/scribe/models.py

@@ -134,6 +134,12 @@ class ScribeMailboxEntry(Base):
         Integer, ForeignKey("scribe_messages.id"), nullable=False
     )
     read_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    # Migration 009 (core-api T-13): deletion is per mailbox entry, never
+    # per message (D-66/D-70) — independent nullable timestamps, mirroring
+    # book_records.archived_at/trashed_at. The service layer's view=trash
+    # takes priority over view=archive when both are set.
+    archived_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    trashed_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
     tags: Mapped[List[str]] = mapped_column(ARRAY(Text), nullable=False, default=list)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),

src/domains/scribe/schemas.py

@@ -3,17 +3,20 @@ Scribe Domain Schemas
 
 Pydantic request/response payloads for the Scribe (D-66, D-68, D-73). This
 round ships `register`, `list` (addresses), `send`, `mailbox`, `history`,
-`find`, `mark`, `tag`/`untag` — groups, subscriptions, the notification
+`find`, `mark`, `tag`/`untag`, archive/trash a mailbox entry and look one
-matrix and delivery are later work (dossier core-api T-5).
+up by message number (core-api T-13/T-9) — groups, subscriptions, the
+notification matrix and delivery are later work (dossier core-api T-5).
 """
 from datetime import datetime
-from typing import List, Optional
+from typing import List, Literal, Optional
 
 from pydantic import Field
 
 from src.domains.scribe.models import ADDRESS_NAME_PATTERN
 from src.shared.base import BaseSchema
 
+MailboxView = Literal["inbox", "archive", "trash", "all"]
+
 
 class AddressRegisterRequest(BaseSchema):
     """`register` — the name only; owner (tenant, consumer) comes from the caller."""
@@ -61,6 +64,8 @@ class MailboxEntryResponse(BaseSchema):
     id: int
     address: str
     read_at: Optional[datetime]
+    archived_at: Optional[datetime]
+    trashed_at: Optional[datetime]
     tags: List[str]
     created_at: datetime
     message: MessageResponse
@@ -78,8 +83,13 @@ class HistoryResponse(BaseSchema):
 
 
 class MailboxPatchRequest(BaseSchema):
-    """`mark` (`read`) and `tag`/`untag` (`add_tags`/`remove_tags`) share one route."""
+    """`mark` (`read`), `archive`/`unarchive`, `trash`/`restore` and
+    `tag`/`untag` (`add_tags`/`remove_tags`) share one route. `archived`/
+    `trashed` are independent: `true` sets the timestamp (if not already
+    set), `false` clears it (workspace core-api T-13)."""
 
     read: Optional[bool] = None
+    archived: Optional[bool] = None
+    trashed: Optional[bool] = None
     add_tags: Optional[List[str]] = None
     remove_tags: Optional[List[str]] = None

src/domains/scribe/service.py

@@ -199,19 +199,57 @@ class ScribeService:
         self,
         session: AsyncSession,
         caller: Caller,
+        view: str = "inbox",
         unread: Optional[bool] = None,
         tag: Optional[str] = None,
         limit: int = 50,
         before_id: Optional[int] = None,
+        message_id: Optional[int] = None,
     ) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int]:
-        """`mailbox` — the caller's identity's entries, unread first."""
+        """`mailbox` — the caller's identity's entries, unread first.
+
+        `message_id`, when given, is a direct lookup rather than a filtered
+        list: zero or one entry (the unique constraint on
+        (address, message_id) guarantees at most one), in any view — it
+        ignores `view`/`unread`/`tag`/`before_id` entirely, which is the
+        point (core-api T-9): a client that already knows the message
+        number never pages the unread-first/id-descending listing to find
+        it.
+
+        Otherwise `view` selects which of the independent `archived_at`/
+        `trashed_at` states to show: `inbox` (default) is neither,
+        `archive` is archived and not trashed, `trash` is trashed
+        (regardless of archived_at — trashing hides an entry from the
+        archive view too), `all` applies no filter on either column.
+        """
         tenant = await self._require_tenant(session, caller)
         await self._require_owned_identity(session, caller, tenant)
 
+        if message_id is not None:
+            lookup_query = select(ScribeMailboxEntry).where(
+                ScribeMailboxEntry.owner_tenant == tenant,
+                ScribeMailboxEntry.address == caller.identity,
+                ScribeMailboxEntry.message_id == message_id,
+            )
+            entries = list((await session.execute(lookup_query)).scalars().all())
+            messages_by_id = await self._messages_for_entries(session, entries)
+            results = [(entry, messages_by_id[entry.message_id]) for entry in entries]
+            return results, len(results)
+
         conditions = [
             ScribeMailboxEntry.owner_tenant == tenant,
             ScribeMailboxEntry.address == caller.identity,
         ]
+        if view == "archive":
+            conditions.append(ScribeMailboxEntry.archived_at.isnot(None))
+            conditions.append(ScribeMailboxEntry.trashed_at.is_(None))
+        elif view == "trash":
+            conditions.append(ScribeMailboxEntry.trashed_at.isnot(None))
+        elif view == "all":
+            pass
+        else:  # "inbox" (default)
+            conditions.append(ScribeMailboxEntry.archived_at.is_(None))
+            conditions.append(ScribeMailboxEntry.trashed_at.is_(None))
         if unread:
             conditions.append(ScribeMailboxEntry.read_at.is_(None))
         if tag is not None:
@@ -236,7 +274,10 @@ class ScribeService:
     async def find_messages(
         self, session: AsyncSession, caller: Caller, q: str, limit: int = 50
     ) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int]:
-        """`find` — full text over messages in the caller identity's mailbox."""
+        """`find` — full text over messages in the caller identity's
+        mailbox, every entry that is not trashed (core-api T-13: trash
+        takes an entry out of `find` the same way it takes it out of
+        `inbox` and `archive`); archived entries still match."""
         tenant = await self._require_tenant(session, caller)
         await self._require_owned_identity(session, caller, tenant)
 
@@ -247,6 +288,7 @@ class ScribeService:
             .where(
                 ScribeMailboxEntry.owner_tenant == tenant,
                 ScribeMailboxEntry.address == caller.identity,
+                ScribeMailboxEntry.trashed_at.is_(None),
                 ScribeMessage.search.op("@@")(tsquery),
             )
         )
@@ -302,11 +344,16 @@ class ScribeService:
     async def patch_mailbox_entry(
         self, session: AsyncSession, entry_id: int, data: MailboxPatchRequest, caller: Caller
     ) -> Tuple[ScribeMailboxEntry, ScribeMessage]:
-        """`mark` (`read`) and `tag`/`untag` (`add_tags`/`remove_tags`) share
+        """`mark` (`read`), `archive`/`unarchive`, `trash`/`restore` and
-        this route. An entry outside the caller's tenant, or addressed to
+        `tag`/`untag` (`add_tags`/`remove_tags`) share this route. An entry
-        an identity the caller is not reading as, answers 404 not_found —
+        outside the caller's tenant, or addressed to an identity the caller
-        indistinguishable from a nonexistent entry, so an id never confirms
+        is not reading as, answers 404 not_found — indistinguishable from a
-        someone else's mail exists."""
+        nonexistent entry, so an id never confirms someone else's mail
+        exists. `archived`/`trashed` are independent timestamps (core-api
+        T-13): `true` sets the one not already set, `false` clears it —
+        trashing does not clear `archived_at`, so restoring a message that
+        was archived before it was trashed returns it to the archive view,
+        not the inbox."""
         tenant = await self._require_tenant(session, caller)
 
         entry = await session.get(ScribeMailboxEntry, entry_id)
@@ -318,6 +365,16 @@ class ScribeService:
         elif data.read is False:
             entry.read_at = None
 
+        if data.archived is True and entry.archived_at is None:
+            entry.archived_at = datetime.now(timezone.utc)
+        elif data.archived is False:
+            entry.archived_at = None
+
+        if data.trashed is True and entry.trashed_at is None:
+            entry.trashed_at = datetime.now(timezone.utc)
+        elif data.trashed is False:
+            entry.trashed_at = None
+
         if data.add_tags:
             current = set(entry.tags)
             entry.tags = list(entry.tags) + [t for t in data.add_tags if t not in current]

tests/test_scribe.py

@@ -618,3 +618,300 @@ class TestTenantRegistryEnforcement:
             "/scribe/addresses", headers=office_headers("jeroen", "office-claude", tenant=TENANT)
         )
         assert response.status_code == 200
+
+
+# =============================================================================
+# Archive and trash a mailbox entry (core-api T-13 / D-66, D-70): deletion
+# is per mailbox entry, never per message — the message stays in `history`
+# and another recipient's entry is untouched throughout.
+# =============================================================================
+
+
+def _send_and_get_entry_id(client, body: str, to: str = "tower-control") -> tuple:
+    """Register the usual pair if needed by the caller, send one message,
+    and return (message, entry_id) for the recipient's own mailbox entry."""
+    sent = client.post(
+        "/scribe/messages",
+        json={"to": [to], "body": body},
+        headers=office_headers("jeroen", "office-claude"),
+    ).json()
+    entry_id = client.get(
+        "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+    ).json()["entries"][0]["id"]
+    return sent, entry_id
+
+
+class TestArchiveAndTrash:
+    def test_archive_hides_from_inbox_and_shows_in_archive_view(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        _, entry_id = _send_and_get_entry_id(client, "archive me")
+
+        archived = client.patch(
+            f"/scribe/mailbox/{entry_id}",
+            json={"archived": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+        assert archived.status_code == 200
+        assert archived.json()["archived_at"] is not None
+        assert archived.json()["trashed_at"] is None
+
+        inbox = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()
+        assert inbox["total"] == 0
+
+        archive_view = client.get(
+            "/scribe/mailbox",
+            params={"view": "archive"},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert archive_view["total"] == 1
+        assert archive_view["entries"][0]["id"] == entry_id
+
+    def test_unarchive_brings_it_back_to_inbox(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        _, entry_id = _send_and_get_entry_id(client, "archive then unarchive")
+        client.patch(
+            f"/scribe/mailbox/{entry_id}",
+            json={"archived": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+
+        unarchived = client.patch(
+            f"/scribe/mailbox/{entry_id}",
+            json={"archived": False},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+        assert unarchived.json()["archived_at"] is None
+
+        inbox = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()
+        assert inbox["total"] == 1
+        assert inbox["entries"][0]["id"] == entry_id
+
+    def test_trash_hides_from_inbox_archive_and_find_shows_in_trash(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        _, entry_id = _send_and_get_entry_id(client, "trash-findable-term")
+
+        trashed = client.patch(
+            f"/scribe/mailbox/{entry_id}",
+            json={"trashed": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+        assert trashed.status_code == 200
+        assert trashed.json()["trashed_at"] is not None
+
+        inbox = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()
+        assert inbox["total"] == 0
+
+        archive_view = client.get(
+            "/scribe/mailbox",
+            params={"view": "archive"},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert archive_view["total"] == 0
+
+        found = client.get(
+            "/scribe/messages",
+            params={"q": "trash-findable-term"},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert found["total"] == 0
+
+        trash_view = client.get(
+            "/scribe/mailbox",
+            params={"view": "trash"},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert trash_view["total"] == 1
+        assert trash_view["entries"][0]["id"] == entry_id
+
+    def test_restore_brings_it_back_to_inbox(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        _, entry_id = _send_and_get_entry_id(client, "trash then restore")
+        client.patch(
+            f"/scribe/mailbox/{entry_id}",
+            json={"trashed": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+
+        restored = client.patch(
+            f"/scribe/mailbox/{entry_id}",
+            json={"trashed": False},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+        assert restored.json()["trashed_at"] is None
+
+        inbox = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()
+        assert inbox["total"] == 1
+        assert inbox["entries"][0]["id"] == entry_id
+
+    def test_message_stays_in_history_throughout_archive_and_trash(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        message, entry_id = _send_and_get_entry_id(client, "history survives")
+
+        for payload in ({"archived": True}, {"trashed": True}):
+            client.patch(
+                f"/scribe/mailbox/{entry_id}",
+                json=payload,
+                headers=office_headers("tower-control", "office-gemini"),
+            )
+            history = client.get(
+                "/scribe/addresses/tower-control/history",
+                headers=office_headers("tower-control", "office-gemini"),
+            ).json()
+            assert [m["id"] for m in history["messages"]] == [message["id"]]
+
+    def test_trashing_one_recipients_entry_leaves_the_others_untouched(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        register(client, "jeroen-2", "office-claude")
+
+        client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control", "jeroen-2"], "body": "broadcast"},
+            headers=office_headers("jeroen", "office-claude"),
+        )
+        tower_entry_id = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()["entries"][0]["id"]
+
+        client.patch(
+            f"/scribe/mailbox/{tower_entry_id}",
+            json={"trashed": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+
+        other_inbox = client.get(
+            "/scribe/mailbox", headers=office_headers("jeroen-2", "office-claude")
+        ).json()
+        assert other_inbox["total"] == 1
+        assert other_inbox["entries"][0]["trashed_at"] is None
+
+
+# =============================================================================
+# Look an entry up by message number (core-api T-9): zero or one, in any
+# view, no paging required.
+# =============================================================================
+
+
+class TestMailboxMessageIdLookup:
+    def test_message_id_finds_the_entry_in_any_view(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        message, entry_id = _send_and_get_entry_id(client, "locate me")
+        client.patch(
+            f"/scribe/mailbox/{entry_id}",
+            json={"trashed": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+
+        found = client.get(
+            "/scribe/mailbox",
+            params={"message_id": message["id"]},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert found["total"] == 1
+        assert found["entries"][0]["id"] == entry_id
+        assert found["entries"][0]["trashed_at"] is not None
+
+    def test_message_id_answers_nothing_for_a_message_not_in_their_mailbox(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        sent = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "not jeroen's"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+
+        found = client.get(
+            "/scribe/mailbox",
+            params={"message_id": sent["id"]},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+        assert found["total"] == 0
+        assert found["entries"] == []
+
+    def test_message_id_answers_nothing_across_tenants(self, client):
+        register(client, "jeroen", "office-claude", tenant=TENANT)
+        register(client, "tower-control", "office-gemini", tenant=TENANT)
+        sent = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "tenant-scoped"},
+            headers=office_headers("jeroen", "office-claude", tenant=TENANT),
+        ).json()
+
+        register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT)
+        found = client.get(
+            "/scribe/mailbox",
+            params={"message_id": sent["id"]},
+            headers=office_headers("tower-control", "office-gemini", OTHER_TENANT),
+        ).json()
+        assert found["total"] == 0
+
+    def test_message_id_finds_a_read_entry_past_the_first_page_in_one_call(self, client):
+        """dossier verification #4: today's listing sorts unread-first,
+        then by id descending, behind a single before_id cursor. Seed one
+        old entry that stays unread (so it sorts first regardless of its
+        low id) ahead of 51 read ones (one more than the default page size
+        of 50) — the oldest read entry (lowest id among them) sorts last,
+        past the first page. `message_id` answers it in one call, with no
+        `before_id` paging."""
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+
+        old_unread = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "old and unread"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+
+        read_message_ids = []
+        for i in range(51):
+            sent = client.post(
+                "/scribe/messages",
+                json={"to": ["tower-control"], "body": f"read {i}"},
+                headers=office_headers("jeroen", "office-claude"),
+            ).json()
+            read_message_ids.append(sent["id"])
+
+        all_entries = client.get(
+            "/scribe/mailbox",
+            params={"view": "all", "limit": 200},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()["entries"]
+        for entry in all_entries:
+            if entry["message"]["id"] != old_unread["id"]:
+                client.patch(
+                    f"/scribe/mailbox/{entry['id']}",
+                    json={"read": True},
+                    headers=office_headers("tower-control", "office-gemini"),
+                )
+
+        # The oldest (lowest-id) read message: last among the reads when
+        # sorted id-descending, so page 1 (the default limit of 50) never
+        # reaches it — plain listing would need a second before_id call.
+        target_message_id = read_message_ids[0]
+
+        page_one = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()
+        assert target_message_id not in {e["message"]["id"] for e in page_one["entries"]}
+
+        found = client.get(
+            "/scribe/mailbox",
+            params={"message_id": target_message_id},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert found["total"] == 1
+        assert found["entries"][0]["message"]["id"] == target_message_id

tests/test_scribe_migration.py

@@ -1,5 +1,5 @@
 """
-Tests for Scribe migration 006.
+Tests for Scribe migrations 006 and 009.
 
 Runs against `core_api_test` only (see conftest.py — skip when
 TEST_DATABASE_URL is unset, hard failure if it resolves to `core_api`).
@@ -27,6 +27,23 @@ async def _table_exists(url: str, table: str) -> bool:
         await engine.dispose()
 
 
+async def _columns(url: str, table: str) -> set:
+    asyncpg_url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
+    engine = create_async_engine(asyncpg_url, poolclass=NullPool)
+    try:
+        async with engine.connect() as conn:
+            result = await conn.execute(
+                text(
+                    "SELECT column_name FROM information_schema.columns "
+                    "WHERE table_name = :table"
+                ),
+                {"table": table},
+            )
+            return {row[0] for row in result.all()}
+    finally:
+        await engine.dispose()
+
+
 class TestMigration006RoundTrip:
     """Migration 006 must upgrade and downgrade cleanly (dossier verification #1)."""
 
@@ -58,3 +75,36 @@ class TestMigration006RoundTrip:
         assert await _table_exists(test_database_url, "scribe_addresses")
         assert await _table_exists(test_database_url, "scribe_messages")
         assert await _table_exists(test_database_url, "scribe_mailbox_entries")
+
+
+class TestMigration009RoundTrip:
+    """Migration 009 (core-api T-13) must upgrade and downgrade cleanly —
+    dossier verification #2: 008 -> 009 -> 008 -> 009."""
+
+    @pytest.mark.asyncio
+    async def test_upgrade_adds_archived_and_trashed_at(self, test_database_url):
+        result = run_alembic(test_database_url, "upgrade", "head")
+        assert result.returncode == 0, result.stderr
+
+        columns = await _columns(test_database_url, "scribe_mailbox_entries")
+        assert {"archived_at", "trashed_at"} <= columns
+
+    @pytest.mark.asyncio
+    async def test_downgrade_then_upgrade_round_trips_cleanly(self, test_database_url):
+        baseline = run_alembic(test_database_url, "upgrade", "head")
+        assert baseline.returncode == 0, baseline.stderr
+
+        down = run_alembic(test_database_url, "downgrade", "008")
+        assert down.returncode == 0, down.stderr
+        columns_at_008 = await _columns(test_database_url, "scribe_mailbox_entries")
+        assert "archived_at" not in columns_at_008
+        assert "trashed_at" not in columns_at_008
+        # The table itself, created by 006, must survive a downgrade of
+        # 009 — proves 009 only adds/drops its own two columns.
+        assert await _table_exists(test_database_url, "scribe_mailbox_entries")
+
+        # Leave the schema at head (009 is head) for the rest of the suite.
+        up = run_alembic(test_database_url, "upgrade", "head")
+        assert up.returncode == 0, up.stderr
+        columns_at_head = await _columns(test_database_url, "scribe_mailbox_entries")
+        assert {"archived_at", "trashed_at"} <= columns_at_head