feat(scribe): mailbox takes since_message_id, answers newest_message_id (T-16)

The Office's one polling call, glance, answers "has anything changed
since I last looked?" with counts; live core-api ignored the
parameters it needed, so glance could not ship.

GET /scribe/mailbox now takes since_message_id=N: only entries whose
message number is above N (strictly — not >=), combined with
view/unread/tag the same way they combine with each other today.
message_id, which already bypasses those filters, keeps bypassing
since_message_id too.

The response gains newest_message_id: the caller's newest message
number across its whole mailbox, every view, regardless of any filter
the call applied; null when the mailbox is empty. It exists because
the unread-first sort can return an older unread entry on a limit=1
page while a newer message is already read — the exact shape glance
hits. Computed from a scoped MAX(message_id) query (tenant and
address), never from the page itself, so it cannot read stale or leak
across the tenant/address boundary (D-73).

total's meaning is unchanged: it already counted every filter match
independent of limit (the count query runs against the pre-limit
subquery), so since_message_id just becomes one more filter it counts
against.

Three mutations run and reverted to confirm the new tests actually
cover the behavior: message_id > vs >= broke the boundary test;
computing newest_message_id from the sorted page instead of the whole
mailbox broke the older-unread test; dropping the address filter from
the MAX query broke the address-boundary test.

583 -> 592 tests passing; test database left at head (010).

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

CHANGELOG.md

@@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **`GET /scribe/mailbox` takes `since_message_id`, and answers with `newest_message_id`**
+  (core-api T-16) — the Office's `glance` call needs "has anything changed since I last
+  looked?" with counts. `since_message_id=N` keeps only entries whose message number is
+  above `N`, combined with `view`/`unread`/`tag` as today (`message_id`, which already
+  bypasses the other filters, keeps bypassing `since_message_id` too). `total` already
+  counted every match independent of `limit`; that is unchanged, `since_message_id` is
+  just one more filter it counts against. `newest_message_id` is the caller's newest
+  message number across its whole mailbox — every view, regardless of any filter applied
+  — null when the mailbox is empty; it exists because the unread-first sort can return an
+  older unread entry on a `limit=1` page while a newer message is already read.
+
 ### Changed
 
 - **Alembic refuses to run without a named migration target** (core-api T-3) —

src/domains/scribe/controller.py

@@ -196,13 +196,18 @@ class ScribeController(BaseController):
             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)",
+                "in any view, ignoring unread/tag/before_id/since_message_id (core-api T-9)",
+            ),
+            since_message_id: Optional[int] = Query(
+                None,
+                description="only entries whose message number is above this one, combined "
+                "with view/unread/tag as usual (ignored when message_id is given) (core-api T-16)",
             ),
             caller: Caller = Depends(require_scribe_reader),
             session: AsyncSession = Depends(get_async_session),
         ):
             try:
-                results, total = await service.get_mailbox(
+                results, total, newest_message_id = await service.get_mailbox(
                     session,
                     caller,
                     view=view,
@@ -211,6 +216,7 @@ class ScribeController(BaseController):
                     limit=limit,
                     before_id=before_id,
                     message_id=message_id,
+                    since_message_id=since_message_id,
                 )
             except ScribeError as error:
                 raise HTTPException(
@@ -220,6 +226,7 @@ class ScribeController(BaseController):
                 entries=[_entry_response(entry, message) for entry, message in results],
                 total=total,
                 limit=limit,
+                newest_message_id=newest_message_id,
             )
 
         @router.patch(

src/domains/scribe/schemas.py

@@ -78,6 +78,14 @@ class MailboxListResponse(BaseSchema):
     entries: List[MailboxEntryResponse]
     total: int
     limit: int
+    # Only `GET /scribe/mailbox` sets this (core-api T-16); `GET /scribe/messages`
+    # (find) shares this schema and leaves it at the default.
+    newest_message_id: Optional[int] = Field(
+        None,
+        description="the caller's newest message number across the whole mailbox "
+        "(every view), regardless of the filters this call applied; null when the "
+        "mailbox is empty",
+    )
 
 
 class HistoryResponse(BaseSchema):

src/domains/scribe/service.py

@@ -103,6 +103,22 @@ class ScribeService:
         messages = (await session.execute(stmt)).scalars().all()
         return {message.id: message for message in messages}
 
+    async def _newest_message_id(
+        self, session: AsyncSession, tenant: str, identity: str
+    ) -> Optional[int]:
+        """The caller's newest message number across its whole mailbox —
+        every view, independent of whatever filters the current call
+        applies (core-api T-16) — scoped to tenant *and* address so it can
+        never answer with another identity's or tenant's number (D-73).
+        `func.max` over no rows is SQL NULL, which the driver returns as
+        `None`: the empty-mailbox case falls out of the query rather than
+        needing a separate branch."""
+        stmt = select(func.max(ScribeMailboxEntry.message_id)).where(
+            ScribeMailboxEntry.owner_tenant == tenant,
+            ScribeMailboxEntry.address == identity,
+        )
+        return (await session.execute(stmt)).scalar()
+
     # =========================================================================
     # register / list addresses
     # =========================================================================
@@ -241,26 +257,40 @@ class ScribeService:
         limit: int = 50,
         before_id: Optional[int] = None,
         message_id: Optional[int] = None,
-    ) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int]:
+        since_message_id: Optional[int] = None,
+    ) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int, Optional[int]]:
         """`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.
+        ignores `view`/`unread`/`tag`/`before_id`/`since_message_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.
+        `since_message_id`, when given, keeps only entries whose message
+        number is above it (strictly — the boundary message itself is not
+        "since" itself), combined with `view`/`unread`/`tag` the same way
+        they combine with each other (core-api T-16).
+
+        Returns `(results, total, newest_message_id)`. `total` counts
+        whatever matched the filters above, independent of `limit` — true
+        before this round too, since the count query already runs against
+        the pre-`limit` subquery; `since_message_id` becomes one more
+        filter it counts against, nothing more. `newest_message_id` is
+        **not** scoped by any of those filters — see `_newest_message_id`.
         """
         tenant = await self._require_tenant(session, caller)
         await self._require_owned_identity(session, caller, tenant)
 
+        newest_message_id = await self._newest_message_id(session, tenant, caller.identity)
+
         if message_id is not None:
             lookup_query = select(ScribeMailboxEntry).where(
                 ScribeMailboxEntry.owner_tenant == tenant,
@@ -270,7 +300,7 @@ class ScribeService:
             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)
+            return results, len(results), newest_message_id
 
         conditions = [
             ScribeMailboxEntry.owner_tenant == tenant,
@@ -292,6 +322,8 @@ class ScribeService:
             conditions.append(ScribeMailboxEntry.tags.contains([tag]))
         if before_id is not None:
             conditions.append(ScribeMailboxEntry.id < before_id)
+        if since_message_id is not None:
+            conditions.append(ScribeMailboxEntry.message_id > since_message_id)
 
         base_query = select(ScribeMailboxEntry).where(*conditions)
         count_query = select(func.count()).select_from(base_query.subquery())
@@ -305,7 +337,7 @@ class ScribeService:
         entries = list((await session.execute(query)).scalars().all())
         messages_by_id = await self._messages_for_entries(session, entries)
 
-        return [(entry, messages_by_id[entry.message_id]) for entry in entries], total
+        return [(entry, messages_by_id[entry.message_id]) for entry in entries], total, newest_message_id
 
     async def find_messages(
         self, session: AsyncSession, caller: Caller, q: str, limit: int = 50

tests/test_scribe.py

@@ -1043,3 +1043,223 @@ class TestMailboxMessageIdLookup:
         ).json()
         assert found["total"] == 1
         assert found["entries"][0]["message"]["id"] == target_message_id
+
+
+# =============================================================================
+# since_message_id and newest_message_id (core-api T-16): the Office's one
+# polling call, `glance`, answers "has anything changed since I last
+# looked?" with these.
+# =============================================================================
+
+
+class TestSinceMessageId:
+    def test_since_message_id_returns_only_entries_above_it(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        ids = []
+        for body in ("one", "two", "three"):
+            sent = client.post(
+                "/scribe/messages",
+                json={"to": ["tower-control"], "body": body},
+                headers=office_headers("jeroen", "office-claude"),
+            ).json()
+            ids.append(sent["id"])
+
+        mailbox = client.get(
+            "/scribe/mailbox",
+            params={"since_message_id": ids[0]},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        returned_ids = {e["message"]["id"] for e in mailbox["entries"]}
+        assert returned_ids == {ids[1], ids[2]}
+        assert mailbox["total"] == 2
+
+    def test_since_message_id_boundary_excludes_the_named_message(self, client):
+        """dossier mutation check #1: `>=` would include message N itself;
+        `since_message_id=N` means strictly above N."""
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        sent = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "the boundary message"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+
+        mailbox = client.get(
+            "/scribe/mailbox",
+            params={"since_message_id": sent["id"]},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert mailbox["entries"] == []
+        assert mailbox["total"] == 0
+
+    def test_since_message_id_combines_with_unread_and_a_view(self, client):
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        first = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "first, will be read"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+        second = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "second, stays unread"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+        first_entry_id = client.get(
+            "/scribe/mailbox",
+            params={"message_id": first["id"]},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()["entries"][0]["id"]
+        client.patch(
+            f"/scribe/mailbox/{first_entry_id}",
+            json={"read": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+
+        mailbox = client.get(
+            "/scribe/mailbox",
+            params={"since_message_id": 0, "unread": True, "view": "inbox"},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        returned_ids = {e["message"]["id"] for e in mailbox["entries"]}
+        assert returned_ids == {second["id"]}
+
+    def test_total_counts_every_match_independent_of_limit(self, client):
+        """dossier verification #2: `unread=true&limit=1` gives the unread
+        count in one row via `total`."""
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        for body in ("a", "b", "c"):
+            client.post(
+                "/scribe/messages",
+                json={"to": ["tower-control"], "body": body},
+                headers=office_headers("jeroen", "office-claude"),
+            )
+
+        mailbox = client.get(
+            "/scribe/mailbox",
+            params={"unread": True, "limit": 1},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert len(mailbox["entries"]) == 1
+        assert mailbox["total"] == 3
+
+    def test_message_id_lookup_ignores_since_message_id(self, client):
+        """`message_id` already bypasses view/unread/tag; `since_message_id`
+        keeps that behavior (dossier build item #1)."""
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        sent = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "direct lookup"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+
+        found = client.get(
+            "/scribe/mailbox",
+            params={"message_id": sent["id"], "since_message_id": sent["id"]},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        assert found["total"] == 1
+        assert found["entries"][0]["message"]["id"] == sent["id"]
+
+
+class TestNewestMessageId:
+    def test_newest_message_id_is_true_newest_when_an_older_message_is_unread(self, client):
+        """dossier core-api T-16, the exact case the Office hit: the
+        unread-first sort makes `limit=1` return an older unread message
+        when a newer one is already read. `newest_message_id` must answer
+        with the newer one regardless (dossier mutation check #2)."""
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+
+        older = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "older, stays unread"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+        newer = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "newer, gets read"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+
+        newer_entry_id = client.get(
+            "/scribe/mailbox",
+            params={"message_id": newer["id"]},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()["entries"][0]["id"]
+        client.patch(
+            f"/scribe/mailbox/{newer_entry_id}",
+            json={"read": True},
+            headers=office_headers("tower-control", "office-gemini"),
+        )
+
+        page = client.get(
+            "/scribe/mailbox",
+            params={"limit": 1},
+            headers=office_headers("tower-control", "office-gemini"),
+        ).json()
+        # unread-first sort: the older, still-unread message leads page one —
+        # the condition newest_message_id exists to see past.
+        assert page["entries"][0]["message"]["id"] == older["id"]
+        assert page["newest_message_id"] == newer["id"]
+
+    def test_newest_message_id_is_null_for_an_empty_mailbox(self, client):
+        register(client, "tower-control", "office-gemini")
+        mailbox = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()
+        assert mailbox["entries"] == []
+        assert mailbox["newest_message_id"] is None
+
+    def test_newest_message_id_never_crosses_an_address_boundary(self, client):
+        """dossier mutation check #3: dropping the address filter would let
+        a later message delivered to a different address in the same
+        tenant — a higher message number that never reaches this
+        identity's mailbox — leak into its newest_message_id."""
+        register(client, "jeroen", "office-claude")
+        register(client, "tower-control", "office-gemini")
+        register(client, "ops", "office-claude")
+
+        mine = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "mine"},
+            headers=office_headers("jeroen", "office-claude"),
+        ).json()
+        client.post(
+            "/scribe/messages",
+            json={"to": ["ops"], "body": "not mine, later and higher id"},
+            headers=office_headers("jeroen", "office-claude"),
+        )
+
+        mailbox = client.get(
+            "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini")
+        ).json()
+        assert mailbox["newest_message_id"] == mine["id"]
+
+    def test_newest_message_id_never_crosses_a_tenant_boundary(self, client):
+        """Same concern as the address boundary above, across tenants
+        (D-73's hard boundary)."""
+        register(client, "jeroen", "office-claude", tenant=TENANT)
+        register(client, "tower-control", "office-gemini", tenant=TENANT)
+        mine = client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "for the TENANT tower-control"},
+            headers=office_headers("jeroen", "office-claude", tenant=TENANT),
+        ).json()
+
+        register(client, "jeroen", "office-claude", tenant=OTHER_TENANT)
+        register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT)
+        client.post(
+            "/scribe/messages",
+            json={"to": ["tower-control"], "body": "for the OTHER_TENANT tower-control, later/higher id"},
+            headers=office_headers("jeroen", "office-claude", tenant=OTHER_TENANT),
+        )
+
+        mailbox = client.get(
+            "/scribe/mailbox",
+            headers=office_headers("tower-control", "office-gemini", TENANT),
+        ).json()
+        assert mailbox["newest_message_id"] == mine["id"]