fix(engine): advance DM unread cursors on participant rows

CHANGELOG.md

@@ -16,7 +16,10 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 Packages without a separate changelog are covered by the cross-package notes below.
 
-## [Unreleased]
+## [Unreleased - Patch]
+
+- A DM participant's inbox unread count drops after they mark a message read, and stays dropped across later inbox reads.
+- Inbox channel unread counts stay correct when message ids gain a digit.
 
 ## [12.2.0] - 2026-10-29
 

README.md

@@ -717,7 +717,9 @@ GET    /workspace/events
 `GET /inbox` excludes archived channels from unread counts, and classifies mentions with the same
 exact `@handle` token contract used for delivery: escaped (`\@x`), email-address (`user@x`), prefix,
 and superstring text are not mentions. Mention results are limited to live channels the agent has
-joined or DMs the agent participates in.
+joined or DMs the agent participates in. A DM unread count includes messages from other participants
+after that participant's read cursor. Marking a message read, or acknowledging its delivery, moves
+the cursor forward and the count drops.
 
 Activity feed channel-message items include `channel_id` and `channel_name`; DM items include
 `conversation_id`.

openapi.yaml

@@ -4299,11 +4299,14 @@ paths:
       summary: Get inbox
       description: >-
         Get the authenticated agent's unified inbox (unread counts, mentions, DMs,
-        recent reactions). Unread counts exclude archived channels. Mentions are
-        matched with the exact `@handle` token contract used for delivery: escaped
-        (`\@x`), email-address (`user@x`), prefix, and superstring text are not
-        mentions, and only messages in live channels the agent has joined or DMs
-        the agent participates in are returned.
+        recent reactions). Unread counts exclude archived channels. A DM unread
+        count includes messages from other participants after that participant's
+        read cursor. Marking a message read, or acknowledging its delivery, moves
+        the cursor forward and the count drops. Mentions are matched with the
+        exact `@handle` token contract used for delivery: escaped (`\@x`),
+        email-address (`user@x`), prefix, and superstring text are not mentions,
+        and only messages in live channels the agent has joined or DMs the agent
+        participates in are returned.
       tags:
         - Inbox
       security:

packages/engine/CHANGELOG.md

@@ -7,7 +7,12 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [Unreleased - Patch]
+
+### Fixed
+
+- A DM participant's inbox unread count drops after they mark a message read or acknowledge its delivery. Migration `0078_dm_participant_read_cursor.sql` seeds each cursor from existing read receipts.
+- Channel unread counts in the inbox no longer drop newer messages whose ids have more digits than the read cursor.
 
 ## [12.2.0] - 2026-10-29
 

packages/engine/src/__tests__/atomicity.test.ts

@@ -17,10 +17,12 @@ import {
   channelMembers,
   channels,
   deliveries,
+  dmParticipants,
   files,
   messageAttachments,
   messageLogs,
   messages,
+  nodes,
   readReceipts,
   pendingEvents,
   sessionEvents,
@@ -31,6 +33,7 @@ import { sendDm } from '../engine/dm.js';
 import { createGroupDm, postGroupMessage } from '../engine/groupDm.js';
 import { postReply } from '../engine/thread.js';
 import { markRead } from '../engine/receipt.js';
+import { ackDelivery, ackDeliveriesUpToSeq, DELIVERY_EXPIRY_BATCH_SIZE } from '../engine/delivery.js';
 import { rotateAgentIdentity } from '../engine/agentIdentity.js';
 import {
   applyStatusEventEffect,
@@ -196,6 +199,135 @@ describe('atomic write paths', () => {
       expect(await db.select().from(readReceipts)).toHaveLength(1);
     });
 
+    it('does not advance the DM cursor when the delivery is already dead-lettered', async () => {
+      const { ws, alice, bob, db } = await seed();
+      const sent = await sendDm(db, ws.workspaceId, alice.agentId, { to: 'bob', text: 'psst' });
+      const [delivery] = await db.select().from(deliveries).where(eq(deliveries.messageId, sent.id));
+      await db.update(deliveries).set({ status: 'dead_lettered' }).where(eq(deliveries.id, delivery.id));
+
+      const acked = await ackDelivery(db, ws.workspaceId, bob.agentId, delivery.id);
+
+      expect(acked?.changed).toBe(false);
+      expect(acked?.delivery.status).toBe('dead_lettered');
+      const [cursor] = await db.select().from(dmParticipants).where(eq(dmParticipants.agentId, bob.agentId));
+      expect(cursor.lastReadId).toBeNull();
+      expect(await db.select().from(readReceipts)).toHaveLength(0);
+    });
+
+    it('acks a cumulative backlog in chunks and moves the cursor only on the last one', async () => {
+      const { ws, alice, bob, channelId, db } = await seed();
+      await db.insert(nodes).values({
+        id: 'node_ack_chunk',
+        workspaceId: ws.workspaceId,
+        name: 'ack-chunk',
+        tokenHash: 'ack-chunk-token',
+      });
+      await db.update(agents).set({
+        locationType: 'via_node',
+        locationNodeId: 'node_ack_chunk',
+        providerName: 'default',
+      }).where(eq(agents.id, bob.agentId));
+
+      const count = DELIVERY_EXPIRY_BATCH_SIZE + 1;
+      for (let i = 0; i < count; i++) {
+        await postMessage(db, ws.workspaceId, channelId, alice.agentId, { text: `n${i}` });
+      }
+      const bobRows = await db.select().from(deliveries).where(eq(deliveries.agentId, bob.agentId));
+      expect(bobRows).toHaveLength(count);
+      const upToSeq = Math.max(...bobRows.map((row) => row.seq));
+
+      const restore = injectUpdateFailure(db, dmParticipants, 'injected chunk cursor failure', {
+        skip: DELIVERY_EXPIRY_BATCH_SIZE,
+        times: 1,
+      });
+      await expect(ackDeliveriesUpToSeq(
+        db, ws.workspaceId, 'node_ack_chunk', 'default', 'bob', upToSeq,
+      )).rejects.toThrow('injected chunk cursor failure');
+      restore();
+
+      const mid = await db.select().from(deliveries).where(eq(deliveries.agentId, bob.agentId));
+      expect(mid.filter((row) => row.status === 'acked')).toHaveLength(DELIVERY_EXPIRY_BATCH_SIZE);
+      expect(mid.filter((row) => row.status === 'queued')).toHaveLength(1);
+      const [paused] = await db.select({ seq: agents.deliveryAckSeq }).from(agents).where(eq(agents.id, bob.agentId));
+      expect(paused.seq).toBe(0);
+
+      const done = await ackDeliveriesUpToSeq(
+        db, ws.workspaceId, 'node_ack_chunk', 'default', 'bob', upToSeq,
+      );
+      expect(done?.acked).toBe(1);
+      const after = await db.select().from(deliveries).where(eq(deliveries.agentId, bob.agentId));
+      expect(after.every((row) => row.status === 'acked')).toBe(true);
+      const [finished] = await db.select({ seq: agents.deliveryAckSeq }).from(agents).where(eq(agents.id, bob.agentId));
+      expect(finished.seq).toBe(upToSeq);
+    });
+
+    it('keeps the DM read cursor at the last acked message when a later chunk fails', async () => {
+      const { ws, alice, bob, db } = await seed();
+      await db.insert(nodes).values({
+        id: 'node_ack_dm_chunk',
+        workspaceId: ws.workspaceId,
+        name: 'ack-dm-chunk',
+        tokenHash: 'ack-dm-chunk-token',
+      });
+      await db.update(agents).set({
+        locationType: 'via_node',
+        locationNodeId: 'node_ack_dm_chunk',
+        providerName: 'default',
+      }).where(eq(agents.id, bob.agentId));
+
+      const count = DELIVERY_EXPIRY_BATCH_SIZE + 1;
+      for (let i = 0; i < count; i++) {
+        await sendDm(db, ws.workspaceId, alice.agentId, { to: 'bob', text: `dm${i}` });
+      }
+      const bobRows = (await db.select().from(deliveries).where(eq(deliveries.agentId, bob.agentId)))
+        .sort((a, b) => a.seq - b.seq);
+      expect(bobRows).toHaveLength(count);
+      const upToSeq = bobRows[count - 1].seq;
+      const endOfChunkOne = bobRows[DELIVERY_EXPIRY_BATCH_SIZE - 1].messageId;
+      const endOfBacklog = bobRows[count - 1].messageId;
+
+      const restore = injectUpdateFailure(db, dmParticipants, 'injected dm chunk cursor failure', {
+        skip: DELIVERY_EXPIRY_BATCH_SIZE,
+        times: 1,
+      });
+      await expect(ackDeliveriesUpToSeq(
+        db, ws.workspaceId, 'node_ack_dm_chunk', 'default', 'bob', upToSeq,
+      )).rejects.toThrow('injected dm chunk cursor failure');
+      restore();
+
+      const [paused] = await db.select().from(dmParticipants).where(eq(dmParticipants.agentId, bob.agentId));
+      expect(paused.lastReadId).toBe(endOfChunkOne);
+
+      const done = await ackDeliveriesUpToSeq(
+        db, ws.workspaceId, 'node_ack_dm_chunk', 'default', 'bob', upToSeq,
+      );
+      expect(done?.acked).toBe(1);
+      const [finished] = await db.select().from(dmParticipants).where(eq(dmParticipants.agentId, bob.agentId));
+      expect(finished.lastReadId).toBe(endOfBacklog);
+    });
+
+    it('rolls back a delivery ack when the DM read cursor update fails', async () => {
+      const { ws, alice, bob, db } = await seed();
+      const sent = await sendDm(db, ws.workspaceId, alice.agentId, { to: 'bob', text: 'psst' });
+      const [delivery] = await db.select().from(deliveries).where(eq(deliveries.messageId, sent.id));
+
+      const restore = injectUpdateFailure(db, dmParticipants, 'injected dm cursor failure');
+      await expect(ackDelivery(db, ws.workspaceId, bob.agentId, delivery.id)).rejects.toThrow(
+        'injected dm cursor failure',
+      );
+      restore();
+
+      const [after] = await db.select().from(deliveries).where(eq(deliveries.id, delivery.id));
+      expect(after.status).toBe('queued');
+      const [participant] = await db.select().from(dmParticipants).where(eq(dmParticipants.agentId, bob.agentId));
+      expect(participant.lastReadId).toBeNull();
+
+      const acked = await ackDelivery(db, ws.workspaceId, bob.agentId, delivery.id);
+      expect(acked?.changed).toBe(true);
+      const [cursor] = await db.select().from(dmParticipants).where(eq(dmParticipants.agentId, bob.agentId));
+      expect(cursor.lastReadId).toBe(sent.id);
+    });
+
     it('commits concurrent transactional sends without interleaving', async () => {
       const { ws, alice, bob, channelId, db } = await seed();
 
@@ -378,24 +510,41 @@ describe('atomic write paths', () => {
       expect(delivery.agentId).toBe(alice.agentId);
     });
 
-    it('markRead issues exactly one batch: receipt + delivery transition + lastReadId', async () => {
+    it('markRead issues exactly one batch: receipt + delivery transition + read cursors', async () => {
       const { ws, alice, bob, channelId, db } = await seed();
       const sent = await postMessage(db, ws.workspaceId, channelId, alice.agentId, { text: 'hello' });
       const batches = attachFakeBatch(db);
 
       const receipt = await markRead(db, ws.workspaceId, sent.id, bob.agentId);
 
       expect(batches).toHaveLength(1);
-      expect(batches[0]).toHaveLength(3);
+      expect(batches[0]).toHaveLength(4);
       expectStatementOn(batches[0], 'insert', 'read_receipts');
       expectStatementOn(batches[0], 'update', 'deliveries');
       expectStatementOn(batches[0], 'update', 'channel_members');
+      expectStatementOn(batches[0], 'update', 'dm_participants');
 
       expect(receipt?.message_id).toBe(sent.id);
       const [delivery] = await db.select().from(deliveries);
       expect(delivery.status).toBe('acked');
     });
 
+    it('acks a DM delivery and its read cursor in one batch', async () => {
+      const { ws, alice, bob, db } = await seed();
+      const sent = await sendDm(db, ws.workspaceId, alice.agentId, { to: 'bob', text: 'psst' });
+      const [delivery] = await db.select().from(deliveries).where(eq(deliveries.messageId, sent.id));
+      const batches = attachFakeBatch(db);
+
+      const acked = await ackDelivery(db, ws.workspaceId, bob.agentId, delivery.id);
+
+      expect(acked?.changed).toBe(true);
+      expect(batches).toHaveLength(1);
+      expectStatementOn(batches[0], 'update', 'deliveries');
+      expectStatementOn(batches[0], 'update', 'dm_participants');
+      const [cursor] = await db.select().from(dmParticipants).where(eq(dmParticipants.agentId, bob.agentId));
+      expect(cursor.lastReadId).toBe(sent.id);
+    });
+
     it('a failed batch applies nothing — no orphan message rows', async () => {
       const { ws, alice, channelId, db } = await seed();
       const batches = attachFakeBatch(db);