feat: add per-owner read/unread status for inbox messages

spec/openapi.yaml

@@ -2861,6 +2861,10 @@ components:
         delivered_by:
           type: string
           title: Delivered By
+        is_read:
+          type: boolean
+          title: Is Read
+          default: false
       type: object
       required:
       - message_id

src/mail/protocol/src/mail_protocol/core/inbox.py

@@ -25,6 +25,12 @@ class MAILInboxEntrySummary(BaseModel):
     body_size: int
     received_at: datetime
     delivered_by: Annotated[str, AfterValidator(validate_mail_address)]
+    # Per-owner read state. A message is delivered ``unread`` and flipped to
+    # ``read`` when its owner opens it via ``GET /inbox/{message_id}``. Because
+    # one message fans out to many recipients who share a single inbox entry,
+    # this value is supplied per owner at list time, not stored on the shared
+    # entry.
+    is_read: bool = False
 
 
 class MAILInboxEntry(BaseModel):

src/mail/server/src/mail_server/backends/memory/api.py

@@ -58,6 +58,7 @@
     load_messages,
     load_outbox_entries,
     load_outboxes,
+    load_read_inbox,
     load_refresh_tokens,
     load_swarms,
     load_trash_entries,
@@ -73,6 +74,7 @@
     save_messages,
     save_outbox_entries,
     save_outboxes,
+    save_read_inbox,
     save_refresh_tokens,
     save_swarms,
     save_trash_entries,
@@ -144,6 +146,9 @@ def _snapshot_persistence_state(self) -> dict[str, Any]:
             "messages": dict(self.messages),
             "inbox_entries": dict(self.inbox_entries),
             "inboxes": {address: list(ids) for address, ids in self.inboxes.items()},
+            "read_inbox": {
+                address: set(ids) for address, ids in self.read_inbox.items()
+            },
             "outbox_entries": dict(self.outbox_entries),
             "outboxes": {address: list(ids) for address, ids in self.outboxes.items()},
             "draft_entries": dict(self.draft_entries),
@@ -171,6 +176,7 @@ async def persist(self, *, reason: str = "manual") -> None:
             await save_messages(snapshot["messages"])
             await save_inbox_entries(snapshot["inbox_entries"])
             await save_inboxes(snapshot["inboxes"])
+            await save_read_inbox(snapshot["read_inbox"])
             await save_outbox_entries(snapshot["outbox_entries"])
             await save_outboxes(snapshot["outboxes"])
             await save_draft_entries(snapshot["draft_entries"])
@@ -289,6 +295,15 @@ async def on_server_startup(self, **kwargs: Any) -> None:
         Values: list of inbox entry message IDs
         """
 
+        self.read_inbox: dict[str, set[str]] = await load_read_inbox()
+        """
+        Per-owner inbox read state (the in-memory analogue of
+        ``mailbox_items.is_read``). A message is unread unless its id is present
+        in the owner's set.
+        Keys: user-agent addresses
+        Values: set of read inbox message IDs
+        """
+
         self.outbox_entries: dict[
             str, MAILOutboxEntrySummary
         ] = await load_outbox_entries()
@@ -353,9 +368,7 @@ async def on_server_startup(self, **kwargs: Any) -> None:
         Values: MAILListInBackend instances
         """
 
-        self.refresh_tokens: dict[str, RefreshTokenRecord] = (
-            await load_refresh_tokens()
-        )
+        self.refresh_tokens: dict[str, RefreshTokenRecord] = await load_refresh_tokens()
         """
         A dict of all stored refresh tokens on this server.
         Keys: token hashes (sha256 hex)
@@ -579,12 +592,17 @@ async def get_inbox(
         if inbox_msg_ids is None:
             raise ValueError(f"no inbox found for address {ua_address}")
 
+        read = self.read_inbox.get(ua_address, set())
         inbox_entries: list[MAILInboxEntrySummary] = []
         for msg_id in inbox_msg_ids:
             inbox_entry = self.inbox_entries.get(msg_id)
             if inbox_entry is None:
                 raise ValueError(f"no inbox entry found for message ID {msg_id}")
-            inbox_entries.append(inbox_entry)
+            # ``inbox_entries`` is shared across recipients; copy so this owner's
+            # read state never leaks onto the shared entry.
+            inbox_entries.append(
+                inbox_entry.model_copy(update={"is_read": msg_id in read})
+            )
 
         return _paginate_box(
             inbox_entries, filters, self._box_sort_key(filters, "received_at")
@@ -613,6 +631,9 @@ async def get_inbox_message(
         if message is None:
             raise ValueError(f"message with ID {message_id} not found in messages")
 
+        # Opening a message marks it read for this owner.
+        self.read_inbox.setdefault(ua_address, set()).add(message_id)
+
         return MAILInboxEntry(
             message=message,
             received_at=inbox_entry.received_at,
@@ -1143,6 +1164,8 @@ async def admin_delete_agent(
 
         # remove inbox from self.inboxes
         self.inboxes.pop(full_address)
+        # drop any per-owner read state alongside the inbox
+        self.read_inbox.pop(full_address, None)
         # remove outbox from self.outboxes
         self.outboxes.pop(full_address)
         # remove drafts box from self.drafts
@@ -1250,6 +1273,8 @@ async def admin_delete_daemon(
 
         # remove inbox from self.inboxes
         self.inboxes.pop(full_address)
+        # drop any per-owner read state alongside the inbox
+        self.read_inbox.pop(full_address, None)
         # remove outbox from self.outboxes
         self.outboxes.pop(full_address)
         # remove drafts box from self.drafts
@@ -1355,6 +1380,8 @@ async def admin_delete_user(self, admin: MAILAdmin, user_id: str) -> MAILUser:
 
         # remove inbox from self.inboxes
         self.inboxes.pop(full_address)
+        # drop any per-owner read state alongside the inbox
+        self.read_inbox.pop(full_address, None)
         # remove outbox from self.outboxes
         self.outboxes.pop(full_address)
         # remove drafts box from self.drafts

src/mail/server/src/mail_server/backends/memory/fs.py

@@ -283,6 +283,69 @@ async def load_inboxes() -> dict[str, list[str]]:
     return inboxes
 
 
+async def load_read_inbox() -> dict[str, set[str]]:
+    """
+    Load saved per-owner inbox read state from the local filesystem.
+
+    Mirrors ``load_inboxes``: one file per owner, one read message id per line.
+    A missing ``read_inbox`` directory means no read state has been persisted
+    yet (e.g. a deployment created before read tracking existed), which is
+    treated as "everything unread".
+    """
+
+    read_inbox_path = DEPLOYMENT_PATH.joinpath("read_inbox")
+    logger.info(f"loading read_inbox: {read_inbox_path}...")
+    read_inbox: dict[str, set[str]] = {}
+    if not read_inbox_path.is_dir():
+        logger.info("no read_inbox directory found; treating all messages as unread")
+        return read_inbox
+    with scandir(read_inbox_path) as entries:
+        for entry in entries:
+            if entry.is_file():
+                try:
+                    validate_mail_address(entry.name)
+                except ValueError as e:
+                    logger.warning(f"MAIL address validation failed: {e}")
+                    continue
+
+                with open(entry) as read_file:
+                    content = read_file.readlines()
+                    msg_ids: set[str] = set()
+                    for ln in content:
+                        msg_id = ln.strip()
+                        if not msg_id:
+                            continue
+                        try:
+                            validate_uuid(msg_id)
+                        except ValueError as e:
+                            logger.warning(f"Message ID validation failed: {e}")
+                            continue
+
+                        msg_ids.add(msg_id)
+
+                    read_inbox.update({entry.name: msg_ids})
+
+    logger.info(f"found read state for {len(read_inbox)} inboxes")
+
+    return read_inbox
+
+
+async def save_read_inbox(read_inbox: dict[str, set[str]]) -> None:
+    """
+    Save per-owner inbox read state from memory to the local filesystem.
+    """
+
+    logger.info(f"saving read state for {len(read_inbox)} inboxes...")
+
+    _save_directory_snapshot(
+        DEPLOYMENT_PATH.joinpath("read_inbox"),
+        {
+            address: "".join(f"{msg_id}\n" for msg_id in sorted(msg_ids))
+            for address, msg_ids in read_inbox.items()
+        },
+    )
+
+
 async def load_outbox_entries() -> dict[str, MAILOutboxEntrySummary]:
     """
     Load saved outbox entries from the local filesystem.
@@ -803,10 +866,7 @@ async def save_lists(lists: dict[str, MAILListInBackend]) -> None:
 
     _save_directory_snapshot(
         DEPLOYMENT_PATH.joinpath("lists"),
-        {
-            address: mail_list.model_dump_json()
-            for address, mail_list in lists.items()
-        },
+        {address: mail_list.model_dump_json() for address, mail_list in lists.items()},
     )
 
 

src/mail/server/src/mail_server/backends/memory/init.py

@@ -82,6 +82,11 @@ def init_memory_backend(
     INBOXES_PATH.mkdir(exist_ok=True)
     print(f"ensured deployment inboxes: {INBOXES_PATH}")
 
+    # ~/.mail-swarms/deployments/{deployment}/read_inbox
+    READ_INBOX_PATH = DEPLOYMENT_PATH.joinpath("read_inbox")
+    READ_INBOX_PATH.mkdir(exist_ok=True)
+    print(f"ensured deployment read_inbox: {READ_INBOX_PATH}")
+
     # ~/.mail-swarms/deployments/{deployment}/outbox_entries
     OUTBOX_ENTRIES_PATH = DEPLOYMENT_PATH.joinpath("outbox_entries")
     # print(f"ensuring deployment outbox_entries: {OUTBOX_ENTRIES_PATH}")

src/mail/server/src/mail_server/backends/sqlite/api.py

@@ -279,6 +279,8 @@ async def get_inbox_message(
             message = await store.messages.get(message_id)
             if message is None:
                 raise ValueError(f"message with ID {message_id} not found in messages")
+            # Opening a message marks it read for this owner.
+            await store.boxes.mark_read(ua_address, BOX_INBOX, message_id)
         return MAILInboxEntry(
             message=message,
             received_at=inbox_entry.received_at,

src/mail/server/src/mail_server/backends/sqlite/database.py

@@ -27,7 +27,7 @@
 from pathlib import Path
 from typing import Protocol
 
-from sqlalchemy import event
+from sqlalchemy import event, inspect, text
 from sqlalchemy.engine import Connection, make_url
 from sqlalchemy.ext.asyncio import (
     AsyncEngine,
@@ -116,11 +116,24 @@ def _ensure_schema_columns(connection: Connection) -> None:
     This is the forward-compatibility hook mirroring chorus' approach: when a
     queryable column is added to a ``*Row`` table in a later release, add an
     idempotent ``ALTER TABLE ... ADD COLUMN`` here so existing databases pick it
-    up without a migration framework. There are no such additions yet, so this
-    is currently a no-op.
+    up without a migration framework.
     """
 
-    del connection  # no additive columns yet; hook retained for forward-compat
+    inspector = inspect(connection)
+
+    def _columns(table: str) -> set[str]:
+        return {col["name"] for col in inspector.get_columns(table)}
+
+    # ``mailbox_items.is_read``: per-owner inbox read state (added in v2). SQLite
+    # backfills existing rows with the ``DEFAULT 0`` (unread), which is the
+    # correct legacy state for already-delivered messages.
+    if "is_read" not in _columns("mailbox_items"):
+        connection.execute(
+            text(
+                "ALTER TABLE mailbox_items "
+                "ADD COLUMN is_read BOOLEAN NOT NULL DEFAULT 0"
+            )
+        )
 
 
 def _ensure_sqlite_parent(url: str) -> None:

src/mail/server/src/mail_server/backends/sqlite/repositories.py

@@ -278,6 +278,36 @@ async def remove_membership(self, owner: str, box: str, item_id: str) -> bool:
         await self.session.flush()
         return True
 
+    async def mark_read(self, owner: str, box: str, item_id: str) -> None:
+        """Flip one membership row to ``is_read=True`` (idempotent)."""
+
+        await self.session.execute(
+            update(MailboxItemRow)
+            .where(
+                MailboxItemRow.owner_address == owner,
+                MailboxItemRow.box == box,
+                MailboxItemRow.item_id == item_id,
+            )
+            .values(is_read=True)
+        )
+        await self.session.flush()
+
+    async def read_states(
+        self, owner: str, box: str, item_ids: list[str]
+    ) -> dict[str, bool]:
+        """Map each requested ``item_id`` to its per-owner read flag."""
+
+        if not item_ids:
+            return {}
+        rows = await self.session.execute(
+            select(MailboxItemRow.item_id, MailboxItemRow.is_read).where(
+                MailboxItemRow.owner_address == owner,
+                MailboxItemRow.box == box,
+                MailboxItemRow.item_id.in_(item_ids),
+            )
+        )
+        return {item_id: is_read for item_id, is_read in rows}
+
     async def list_item_ids(self, owner: str, box: str) -> list[str]:
         """Item ids in a box, in insertion order (used by ``clear_trash``)."""
 
@@ -377,7 +407,15 @@ async def list_inbox(
             filters=filters,
             allow_message_sort=True,
         )
-        return [ser.inbox_entry_from_row(row) for row in rows], total
+        summaries = [ser.inbox_entry_from_row(row) for row in rows]
+        # ``is_read`` is per-owner, so it lives on ``mailbox_items``, not on the
+        # shared inbox entry; stitch it onto this owner's page.
+        read = await self.read_states(
+            owner, BOX_INBOX, [s.message_id for s in summaries]
+        )
+        for summary in summaries:
+            summary.is_read = read.get(summary.message_id, False)
+        return summaries, total
 
     async def list_outbox(
         self, owner: str, filters: BoxFilterParams

src/mail/server/src/mail_server/backends/sqlite/schema.py

@@ -202,6 +202,9 @@ class MailboxItemRow(Base):
     box: Mapped[str] = mapped_column(String(8))
     item_id: Mapped[str] = mapped_column(String(64))
     entered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
+    # Per-owner read state. Only meaningful for ``box == "inbox"`` (other boxes
+    # leave it at the default). Set ``True`` when the owner opens the message.
+    is_read: Mapped[bool] = mapped_column(default=False)
 
 
 class MessageBufferRow(Base):

tests/conftest.py

@@ -58,6 +58,7 @@ def deployment_dir(
         "messages",
         "inbox_entries",
         "inboxes",
+        "read_inbox",
         "outbox_entries",
         "outboxes",
         "draft_entries",