From 5f89717d0f5067ed3e0d0072d96c6d5033a34d98 Mon Sep 17 00:00:00 2001
From: Will Washburn <will.washburn@gmail.com>
Date: Wed, 28 Oct 2026 17:37:15 -0700
Subject: [PATCH 1/8] fix(engine): advance DM unread cursors on participant
 rows

DM inbox unread counts messages past dm_participants.last_read_id.
markRead and delivery ack move that cursor forward.
---
 CHANGELOG.md                                  |   4 +-
 packages/engine/CHANGELOG.md                  |   6 +-
 .../engine/src/__tests__/atomicity.test.ts    |   5 +-
 .../conformance/dmUnreadCursor.test.ts        | 129 ++++++++++++++++++
 .../db/__tests__/compactMigrations.test.ts    |   4 +
 .../src/db/__tests__/taskMigration.test.ts    |   1 +
 .../0078_dm_participant_read_cursor.sql       |   3 +
 packages/engine/src/db/schema.ts              |   3 +
 packages/engine/src/engine/delivery.ts        |   3 +
 packages/engine/src/engine/inbox.ts           |   5 +-
 packages/engine/src/engine/receipt.ts         |  24 ++++
 11 files changed, 179 insertions(+), 8 deletions(-)
 create mode 100644 packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
 create mode 100644 packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9895fbd22..57ee42ba0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,7 +16,9 @@ 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.
 
 ## [12.2.0] - 2026-10-29
 
diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md
index 296c075fb..eb4077880 100644
--- a/packages/engine/CHANGELOG.md
+++ b/packages/engine/CHANGELOG.md
@@ -7,7 +7,11 @@ 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
+
+- DM inbox unread counts follow `dm_participants.last_read_id` (migration `0078_dm_participant_read_cursor.sql`). `markRead` and delivery acknowledgement advance that cursor. A cursor on `channel_members` does not, because a DM participant has no channel membership row.
 
 ## [12.2.0] - 2026-10-29
 
diff --git a/packages/engine/src/__tests__/atomicity.test.ts b/packages/engine/src/__tests__/atomicity.test.ts
index a0c0f7599..760af438d 100644
--- a/packages/engine/src/__tests__/atomicity.test.ts
+++ b/packages/engine/src/__tests__/atomicity.test.ts
@@ -378,7 +378,7 @@ 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);
@@ -386,10 +386,11 @@ describe('atomic write paths', () => {
       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);
diff --git a/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts b/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
new file mode 100644
index 000000000..790ecc4f0
--- /dev/null
+++ b/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
@@ -0,0 +1,129 @@
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { createWorkspace, makeNodeStack, registerAgent, type TestStack } from './harness.js';
+
+describe('DM unread cursor', () => {
+  let stack: TestStack;
+  beforeEach(() => { stack = makeNodeStack(); });
+  afterEach(async () => { await stack.close(); });
+
+  const authed = (token: string) => ({
+    post: (path: string, body?: unknown) =>
+      stack.app.request(path, {
+        method: 'POST',
+        headers: {
+          authorization: `Bearer ${token}`,
+          ...(body === undefined ? {} : { 'content-type': 'application/json' }),
+        },
+        ...(body === undefined ? {} : { body: JSON.stringify(body) }),
+      }),
+    get: (path: string) => stack.app.request(path, { headers: { authorization: `Bearer ${token}` } }),
+  });
+
+  async function unreadCount(token: string, conversationId: string): Promise<number> {
+    const res = await authed(token).get('/v1/inbox');
+    expect(res.status).toBe(200);
+    const inbox = (await res.json()).data as {
+      unread_dms: Array<{ conversation_id: string; unread_count: number }>;
+    };
+    return inbox.unread_dms.find((row) => row.conversation_id === conversationId)?.unread_count ?? 0;
+  }
+
+  it('drops a 1:1 DM unread count as markRead advances the participant cursor', async () => {
+    const ws = await createWorkspace(stack.app, 'dm-unread');
+    const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice');
+    const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob');
+
+    const firstRes = await authed(alice.token).post('/v1/dm', { to: 'bob', text: 'one' });
+    expect(firstRes.status).toBeLessThan(300);
+    const first = (await firstRes.json()).data as { id: string; conversation_id: string };
+    const secondRes = await authed(alice.token).post('/v1/dm', { to: 'bob', text: 'two' });
+    expect(secondRes.status).toBeLessThan(300);
+    const second = (await secondRes.json()).data as { id: string };
+
+    expect(await unreadCount(bob.token, first.conversation_id)).toBe(2);
+
+    const readFirst = await authed(bob.token).post(`/v1/messages/${first.id}/read`);
+    expect(readFirst.status).toBe(200);
+    expect(await unreadCount(bob.token, first.conversation_id)).toBe(1);
+
+    const readSecond = await authed(bob.token).post(`/v1/messages/${second.id}/read`);
+    expect(readSecond.status).toBe(200);
+    expect(await unreadCount(bob.token, first.conversation_id)).toBe(0);
+
+    // An older id does not pull the cursor backward.
+    const readOlder = await authed(bob.token).post(`/v1/messages/${first.id}/read`);
+    expect(readOlder.status).toBe(200);
+    expect(await unreadCount(bob.token, first.conversation_id)).toBe(0);
+
+    const thirdRes = await authed(alice.token).post('/v1/dm', { to: 'bob', text: 'three' });
+    expect(thirdRes.status).toBeLessThan(300);
+    expect(await unreadCount(bob.token, first.conversation_id)).toBe(1);
+
+    // Alice's own sends are not unread for Alice.
+    expect(await unreadCount(alice.token, first.conversation_id)).toBe(0);
+  });
+
+  it('drops a group DM unread count from the same participant cursor', async () => {
+    const ws = await createWorkspace(stack.app, 'group-dm-unread');
+    const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice');
+    const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob');
+
+    const create = await authed(alice.token).post('/v1/dm/group', { participants: ['bob'] });
+    expect(create.status).toBe(201);
+    const conversationId = ((await create.json()) as { data: { id: string } }).data.id;
+
+    const msg = await authed(alice.token).post(`/v1/dm/${conversationId}/messages`, { text: 'hello group' });
+    expect(msg.status).toBe(201);
+    const messageId = ((await msg.json()) as { data: { id: string } }).data.id;
+
+    expect(await unreadCount(bob.token, conversationId)).toBe(1);
+    const read = await authed(bob.token).post(`/v1/messages/${messageId}/read`);
+    expect(read.status).toBe(200);
+    expect(await unreadCount(bob.token, conversationId)).toBe(0);
+  });
+
+  it('drops DM unread when the recipient acks the delivery', async () => {
+    const ws = await createWorkspace(stack.app, 'dm-unread-ack');
+    const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice');
+    const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob');
+
+    const sentRes = await authed(alice.token).post('/v1/dm', { to: 'bob', text: 'ack me' });
+    expect(sentRes.status).toBeLessThan(300);
+    const sent = (await sentRes.json()).data as { conversation_id: string };
+    expect(await unreadCount(bob.token, sent.conversation_id)).toBe(1);
+
+    const queued = await authed(bob.token).get('/v1/deliveries');
+    expect(queued.status).toBe(200);
+    const deliveryId = ((await queued.json()).data as Array<{ id: string }>)[0]?.id;
+    expect(deliveryId).toBeTruthy();
+
+    const ack = await authed(bob.token).post(`/v1/deliveries/${deliveryId}/ack`);
+    expect(ack.status).toBe(200);
+    expect(await unreadCount(bob.token, sent.conversation_id)).toBe(0);
+  });
+
+  it('still clears channel unread from the channel membership cursor', async () => {
+    const ws = await createWorkspace(stack.app, 'channel-unread');
+    const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice');
+    const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob');
+
+    const post = await authed(alice.token).post('/v1/channels/general/messages', { text: 'hello channel' });
+    expect(post.status).toBeLessThan(300);
+    const messageId = ((await post.json()) as { data: { id: string } }).data.id;
+
+    const before = await authed(bob.token).get('/v1/inbox');
+    const beforeInbox = (await before.json()).data as {
+      unread_channels: Array<{ channel_name: string; unread_count: number }>;
+    };
+    expect(beforeInbox.unread_channels.find((row) => row.channel_name === 'general')?.unread_count).toBe(1);
+
+    const read = await authed(bob.token).post(`/v1/messages/${messageId}/read`);
+    expect(read.status).toBe(200);
+
+    const after = await authed(bob.token).get('/v1/inbox');
+    const afterInbox = (await after.json()).data as {
+      unread_channels: Array<{ channel_name: string; unread_count: number }>;
+    };
+    expect(afterInbox.unread_channels.find((row) => row.channel_name === 'general')).toBeUndefined();
+  });
+});
diff --git a/packages/engine/src/db/__tests__/compactMigrations.test.ts b/packages/engine/src/db/__tests__/compactMigrations.test.ts
index f1d6b8a76..2b73985c2 100644
--- a/packages/engine/src/db/__tests__/compactMigrations.test.ts
+++ b/packages/engine/src/db/__tests__/compactMigrations.test.ts
@@ -176,6 +176,10 @@ function expectConstraintsPreserved(before: ReturnType<typeof constraints>, afte
       // 0063 appends four nullable recovery fields. Remove only their exact
       // additive DDL so changes to every pre-existing node constraint remain
       // visible to this preservation assertion.
+      // 0078 appends the DM unread cursor. Remove only that additive column.
+      if (table.name === 'dm_participants') {
+        table = { ...table, sql: table.sql.replace(', last_read_id TEXT', '') };
+      }
       if (table.name === 'nodes') {
         table = {
           ...table,
diff --git a/packages/engine/src/db/__tests__/taskMigration.test.ts b/packages/engine/src/db/__tests__/taskMigration.test.ts
index 3b38864d5..a085c2577 100644
--- a/packages/engine/src/db/__tests__/taskMigration.test.ts
+++ b/packages/engine/src/db/__tests__/taskMigration.test.ts
@@ -41,6 +41,7 @@ it('upgrades existing actions and results additively, preserving rows and every
       '0075_message_mention_fts.sql',
       '0076_deliveries_agent_status_created_index.sql',
       '0077_drop_workspace_plan.sql',
+      '0078_dm_participant_read_cursor.sql',
     ]);
     for (const table of before) {
       const columns = db.pragma(`table_info(${table.name})`) as { name: string }[];
diff --git a/packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql b/packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql
new file mode 100644
index 000000000..9187c4685
--- /dev/null
+++ b/packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql
@@ -0,0 +1,3 @@
+-- DM unread counts read a per-participant cursor. DM membership lives in
+-- dm_participants, so a cursor stored only on channel_members never moves.
+ALTER TABLE dm_participants ADD COLUMN last_read_id TEXT;
diff --git a/packages/engine/src/db/schema.ts b/packages/engine/src/db/schema.ts
index 870398442..d753fae88 100644
--- a/packages/engine/src/db/schema.ts
+++ b/packages/engine/src/db/schema.ts
@@ -900,6 +900,9 @@ export const dmParticipants = sqliteTable(
       .references(() => agents.id, { onDelete: 'cascade' }),
     joinedAt: integer('joined_at', { mode: 'timestamp' }).notNull().default(sql`(unixepoch())`),
     leftAt: integer('left_at', { mode: 'timestamp' }),
+    // Inbox unread for this participant. Channel cursors live on
+    // channel_members; a DM has no channel membership row.
+    lastReadId: text('last_read_id'),
   },
   (table) => [
     primaryKey({ columns: [table.conversationId, table.agentId] }),
diff --git a/packages/engine/src/engine/delivery.ts b/packages/engine/src/engine/delivery.ts
index a7c542c91..c77dc6fc5 100644
--- a/packages/engine/src/engine/delivery.ts
+++ b/packages/engine/src/engine/delivery.ts
@@ -22,6 +22,7 @@ import {
 } from './workspaceDeliveryPolicy.js';
 import { queryInChunks } from '../lib/queryChunks.js';
 import { withEngineOperation } from '../lib/engineDiagnostics.js';
+import { advanceDmParticipantReadCursor } from './receipt.js';
 
 type Db = ReturnType<typeof getDb>;
 
@@ -370,6 +371,8 @@ async function markRowsRead(tx: Db, agentId: string, rows: DeliveryWithChannel[]
         eq(channelMembers.agentId, agentId),
         sql`(${channelMembers.lastReadId} IS NULL OR CAST(${channelMembers.lastReadId} AS BIGINT) < CAST(${row.messageId} AS BIGINT))`,
       ));
+
+    await advanceDmParticipantReadCursor(tx, agentId, row.channelId, row.messageId);
   }
 }
 
diff --git a/packages/engine/src/engine/inbox.ts b/packages/engine/src/engine/inbox.ts
index ca75f2396..f958f5fc5 100644
--- a/packages/engine/src/engine/inbox.ts
+++ b/packages/engine/src/engine/inbox.ts
@@ -269,13 +269,10 @@ async function getInboxUnlabeled(db: Db, workspaceId: string, agentId: string) {
       ON dp.conversation_id = dc.id
      AND dp.agent_id = ${agentId}
      AND dp.left_at IS NULL
-    LEFT JOIN channel_members cm
-      ON cm.channel_id = dc.channel_id
-     AND cm.agent_id = ${agentId}
     JOIN messages m
       ON m.channel_id = dc.channel_id
      AND m.agent_id != ${agentId}
-     AND (cm.last_read_id IS NULL OR m.id > cm.last_read_id)
+     AND (dp.last_read_id IS NULL OR m.id > dp.last_read_id)
     WHERE dc.workspace_id = ${workspaceId}
     GROUP BY dc.id, dc.channel_id
     HAVING count(m.id) > 0
diff --git a/packages/engine/src/engine/receipt.ts b/packages/engine/src/engine/receipt.ts
index 72a77621f..2dd495adf 100644
--- a/packages/engine/src/engine/receipt.ts
+++ b/packages/engine/src/engine/receipt.ts
@@ -14,6 +14,27 @@ import { runAtomicWrites } from '../ports/database.js';
 
 type Db = ReturnType<typeof getDb>;
 
+/** Move a DM participant's unread cursor forward. Channel messages match no row. */
+export function advanceDmParticipantReadCursor(
+  db: Pick<Db, 'update'>,
+  agentId: string,
+  channelId: string,
+  messageId: string,
+) {
+  return db
+    .update(dmParticipants)
+    .set({ lastReadId: messageId })
+    .where(and(
+      eq(dmParticipants.agentId, agentId),
+      isNull(dmParticipants.leftAt),
+      sql`${dmParticipants.conversationId} IN (
+        SELECT ${dmConversations.id} FROM ${dmConversations}
+        WHERE ${dmConversations.channelId} = ${channelId}
+      )`,
+      sql`(${dmParticipants.lastReadId} IS NULL OR CAST(${dmParticipants.lastReadId} AS BIGINT) < CAST(${messageId} AS BIGINT))`,
+    ));
+}
+
 export async function markRead(
   db: Db,
   workspaceId: string,
@@ -84,6 +105,9 @@ export async function markRead(
           sql`(${channelMembers.lastReadId} IS NULL OR CAST(${channelMembers.lastReadId} AS BIGINT) < CAST(${messageId} AS BIGINT))`,
         ),
       ),
+
+    // DM participants have no channel_members row. The inbox reads this cursor.
+    advanceDmParticipantReadCursor(writeDb, agentId, msg.channelId, messageId),
   ]);
 
   const [receipt] = await db

From dfb5e392eda1061f76fbc5669db8c31f614c8aff Mon Sep 17 00:00:00 2001
From: Will Washburn <will.washburn@gmail.com>
Date: Wed, 28 Oct 2026 17:45:36 -0700
Subject: [PATCH 2/8] fix(engine): order DM unread cursors and ack them
 atomically

Inbox unread compares snowflake ids by length, then text.
A delivery ack commits the participant cursor in the same batch as the status change.
---
 packages/engine/CHANGELOG.md                  |  2 +-
 .../engine/src/__tests__/atomicity.test.ts    | 40 +++++++++++
 .../conformance/dmUnreadCursor.test.ts        | 29 ++++++++
 packages/engine/src/engine/delivery.ts        | 67 ++++++++++---------
 packages/engine/src/engine/inbox.ts           |  8 ++-
 5 files changed, 111 insertions(+), 35 deletions(-)

diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md
index eb4077880..b194e536a 100644
--- a/packages/engine/CHANGELOG.md
+++ b/packages/engine/CHANGELOG.md
@@ -11,7 +11,7 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
 
 ### Fixed
 
-- DM inbox unread counts follow `dm_participants.last_read_id` (migration `0078_dm_participant_read_cursor.sql`). `markRead` and delivery acknowledgement advance that cursor. A cursor on `channel_members` does not, because a DM participant has no channel membership row.
+- DM inbox unread counts follow `dm_participants.last_read_id` (migration `0078_dm_participant_read_cursor.sql`). `markRead` and delivery acknowledgement advance that cursor in the same atomic batch as the read. The inbox orders the cursor by snowflake length, then text. A cursor on `channel_members` does not, because a DM participant has no channel membership row.
 
 ## [12.2.0] - 2026-10-29
 
diff --git a/packages/engine/src/__tests__/atomicity.test.ts b/packages/engine/src/__tests__/atomicity.test.ts
index 760af438d..dd6666d51 100644
--- a/packages/engine/src/__tests__/atomicity.test.ts
+++ b/packages/engine/src/__tests__/atomicity.test.ts
@@ -17,6 +17,7 @@ import {
   channelMembers,
   channels,
   deliveries,
+  dmParticipants,
   files,
   messageAttachments,
   messageLogs,
@@ -31,6 +32,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 } from '../engine/delivery.js';
 import { rotateAgentIdentity } from '../engine/agentIdentity.js';
 import {
   applyStatusEventEffect,
@@ -196,6 +198,28 @@ describe('atomic write paths', () => {
       expect(await db.select().from(readReceipts)).toHaveLength(1);
     });
 
+    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();
 
@@ -397,6 +421,22 @@ describe('atomic write paths', () => {
       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);
diff --git a/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts b/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
index 790ecc4f0..d2e8855ae 100644
--- a/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
+++ b/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
@@ -102,6 +102,35 @@ describe('DM unread cursor', () => {
     expect(await unreadCount(bob.token, sent.conversation_id)).toBe(0);
   });
 
+  it('orders a DM cursor by snowflake length, then text', async () => {
+    const ws = await createWorkspace(stack.app, 'dm-unread-width');
+    const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice');
+    const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob');
+
+    const sentRes = await authed(alice.token).post('/v1/dm', { to: 'bob', text: 'current' });
+    expect(sentRes.status).toBeLessThan(300);
+    const sent = (await sentRes.json()).data as { id: string; conversation_id: string };
+    const sqlite = stack.runtime.handle.sqlite;
+    const row = sqlite.prepare(
+      'SELECT workspace_id, channel_id, agent_id FROM messages WHERE id = ?',
+    ).get(sent.id) as { workspace_id: string; channel_id: string; agent_id: string };
+    // 17 nines sort after an 18-digit id as text, and before it numerically.
+    const older = '9'.repeat(17);
+    sqlite.prepare(
+      'INSERT INTO messages (id, workspace_id, channel_id, agent_id, body) VALUES (?, ?, ?, ?, ?)',
+    ).run(older, row.workspace_id, row.channel_id, row.agent_id, 'older');
+
+    sqlite.prepare(
+      'UPDATE dm_participants SET last_read_id = ? WHERE agent_id = ? AND conversation_id = ?',
+    ).run(older, bob.agentId, sent.conversation_id);
+    expect(await unreadCount(bob.token, sent.conversation_id)).toBe(1);
+
+    sqlite.prepare(
+      'UPDATE dm_participants SET last_read_id = ? WHERE agent_id = ? AND conversation_id = ?',
+    ).run(sent.id, bob.agentId, sent.conversation_id);
+    expect(await unreadCount(bob.token, sent.conversation_id)).toBe(0);
+  });
+
   it('still clears channel unread from the channel membership cursor', async () => {
     const ws = await createWorkspace(stack.app, 'channel-unread');
     const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice');
diff --git a/packages/engine/src/engine/delivery.ts b/packages/engine/src/engine/delivery.ts
index c77dc6fc5..45e084fb9 100644
--- a/packages/engine/src/engine/delivery.ts
+++ b/packages/engine/src/engine/delivery.ts
@@ -4,7 +4,7 @@ import { eq, and, asc, exists, isNull, isNotNull, inArray, lt, ne, not, notInArr
 import type { getDb } from '../db/index.js';
 import { deliveries, messages, agents, nodes, readReceipts, channelMembers, channels, dmConversations } from '../db/schema.js';
 import type { DeliveryStatus } from '@relaycast/types';
-import { runAtomic } from '../ports/database.js';
+import { runAtomicWrites, type AtomicWrite } from '../ports/database.js';
 import type { NodeConnectionRegistry } from '../ports/realtime.js';
 import { isProviderAgentDeliveryReady } from '../ports/realtime.js';
 import { buildDeliverFrame, buildDeliverPayload, buildMessageCreatedEventData, buildThreadReplyEventData, buildDmReceivedEventData, buildGroupDmReceivedEventData } from './deliveryWire.js';
@@ -356,24 +356,22 @@ export async function deferDelivery(
   return resolveTransition(db, workspaceId, agentId, deliveryId, updated, existing.channelId);
 }
 
-async function markRowsRead(tx: Db, agentId: string, rows: DeliveryWithChannel[]) {
-  for (const row of rows) {
-    await tx
+function readCursorWrites(tx: Db, agentId: string, row: DeliveryWithChannel): AtomicWrite[] {
+  return [
+    tx
       .insert(readReceipts)
       .values({ messageId: row.messageId, agentId })
-      .onConflictDoNothing();
-
-    await tx
+      .onConflictDoNothing(),
+    tx
       .update(channelMembers)
       .set({ lastReadId: row.messageId })
       .where(and(
         eq(channelMembers.channelId, row.channelId),
         eq(channelMembers.agentId, agentId),
         sql`(${channelMembers.lastReadId} IS NULL OR CAST(${channelMembers.lastReadId} AS BIGINT) < CAST(${row.messageId} AS BIGINT))`,
-      ));
-
-    await advanceDmParticipantReadCursor(tx, agentId, row.channelId, row.messageId);
-  }
+      )),
+    advanceDmParticipantReadCursor(tx, agentId, row.channelId, row.messageId),
+  ];
 }
 
 async function ackRows(
@@ -388,23 +386,26 @@ async function ackRows(
   advanceCursorTo?: number,
 ): Promise<DeliveryRow[]> {
   const ids = rows.map((row) => row.id);
-  return runAtomic(db, async (tx) => {
-    await tx
-      .update(agents)
-      .set({
-        deliveryAckSeq:
-          advanceCursorTo === undefined
-            ? sql<number>`${agents.deliveryAckSeq}`
-            : sql<number>`CASE
-              WHEN ${agents.deliveryAckSeq} < ${advanceCursorTo} THEN ${advanceCursorTo}
-              ELSE ${agents.deliveryAckSeq}
-            END`,
-        lastSeen: new Date(),
-      })
-      .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agentId)));
-
-    const updated = ids.length > 0
-      ? await tx
+  // One batch with the ack. On D1, `runAtomic` would commit the ack before the
+  // cursor write, and a retry skips an already-acked delivery.
+  const results = await runAtomicWrites(db, (tx) => {
+    const writes: AtomicWrite[] = [
+      tx
+        .update(agents)
+        .set({
+          deliveryAckSeq:
+            advanceCursorTo === undefined
+              ? sql<number>`${agents.deliveryAckSeq}`
+              : sql<number>`CASE
+                WHEN ${agents.deliveryAckSeq} < ${advanceCursorTo} THEN ${advanceCursorTo}
+                ELSE ${agents.deliveryAckSeq}
+              END`,
+          lastSeen: new Date(),
+        })
+        .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agentId))),
+    ];
+    if (ids.length > 0) {
+      writes.push(tx
         .update(deliveries)
         .set({
           status: TERMINAL_SUCCESS_STATUS,
@@ -418,12 +419,12 @@ async function ackRows(
           inArray(deliveries.id, ids),
           notInArray(deliveries.status, ['acked', 'dead_lettered']),
         ))
-        .returning()
-      : [];
-
-    if (rows.length > 0) await markRowsRead(tx, agentId, rows);
-    return updated;
+        .returning());
+    }
+    for (const row of rows) writes.push(...readCursorWrites(tx, agentId, row));
+    return writes;
   });
+  return ids.length > 0 ? results[1] as DeliveryRow[] : [];
 }
 
 export async function ackDeliveriesUpToSeq(
diff --git a/packages/engine/src/engine/inbox.ts b/packages/engine/src/engine/inbox.ts
index f958f5fc5..f66fbb6a8 100644
--- a/packages/engine/src/engine/inbox.ts
+++ b/packages/engine/src/engine/inbox.ts
@@ -272,7 +272,13 @@ async function getInboxUnlabeled(db: Db, workspaceId: string, agentId: string) {
     JOIN messages m
       ON m.channel_id = dc.channel_id
      AND m.agent_id != ${agentId}
-     AND (dp.last_read_id IS NULL OR m.id > dp.last_read_id)
+     -- Snowflake ids are unpadded decimal text. Length, then text, is
+     -- chronological; a bare text compare reverses order across a digit boundary.
+     AND (
+       dp.last_read_id IS NULL
+       OR length(m.id) > length(dp.last_read_id)
+       OR (length(m.id) = length(dp.last_read_id) AND m.id > dp.last_read_id)
+     )
     WHERE dc.workspace_id = ${workspaceId}
     GROUP BY dc.id, dc.channel_id
     HAVING count(m.id) > 0

From c01c38cf265ee6436093bb38e3b68d3b8fb1e049 Mon Sep 17 00:00:00 2001
From: Will Washburn <will.washburn@gmail.com>
Date: Wed, 28 Oct 2026 17:55:38 -0700
Subject: [PATCH 3/8] fix(engine): skip DM read cursors when the ack does not
 land

A dead-lettered delivery stays unread. The cursor writes observe
the delivery status before the acknowledgement updates it.
---
 packages/engine/CHANGELOG.md                  |  2 +-
 .../engine/src/__tests__/atomicity.test.ts    | 15 +++++++
 packages/engine/src/engine/delivery.ts        | 40 ++++++++++++++++---
 packages/engine/src/engine/receipt.ts         |  4 +-
 4 files changed, 54 insertions(+), 7 deletions(-)

diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md
index b194e536a..ad1a36f29 100644
--- a/packages/engine/CHANGELOG.md
+++ b/packages/engine/CHANGELOG.md
@@ -11,7 +11,7 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
 
 ### Fixed
 
-- DM inbox unread counts follow `dm_participants.last_read_id` (migration `0078_dm_participant_read_cursor.sql`). `markRead` and delivery acknowledgement advance that cursor in the same atomic batch as the read. The inbox orders the cursor by snowflake length, then text. A cursor on `channel_members` does not, because a DM participant has no channel membership row.
+- DM inbox unread counts follow `dm_participants.last_read_id` (migration `0078_dm_participant_read_cursor.sql`). `markRead` and a successful delivery acknowledgement advance that cursor in the same atomic batch as the read. A dead-lettered delivery does not. The inbox orders the cursor by snowflake length, then text. A cursor on `channel_members` does not, because a DM participant has no channel membership row.
 
 ## [12.2.0] - 2026-10-29
 
diff --git a/packages/engine/src/__tests__/atomicity.test.ts b/packages/engine/src/__tests__/atomicity.test.ts
index dd6666d51..79812a122 100644
--- a/packages/engine/src/__tests__/atomicity.test.ts
+++ b/packages/engine/src/__tests__/atomicity.test.ts
@@ -198,6 +198,21 @@ 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('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' });
diff --git a/packages/engine/src/engine/delivery.ts b/packages/engine/src/engine/delivery.ts
index 45e084fb9..ed5f500ae 100644
--- a/packages/engine/src/engine/delivery.ts
+++ b/packages/engine/src/engine/delivery.ts
@@ -356,11 +356,39 @@ export async function deferDelivery(
   return resolveTransition(db, workspaceId, agentId, deliveryId, updated, existing.channelId);
 }
 
-function readCursorWrites(tx: Db, agentId: string, row: DeliveryWithChannel): AtomicWrite[] {
+/** The delivery row this ack is allowed to settle. A dead-lettered row is not one of them. */
+function deliveryStillAckable(workspaceId: string, deliveryId: string) {
+  return sql`EXISTS (
+    SELECT 1 FROM ${deliveries}
+    WHERE ${deliveries.id} = ${deliveryId}
+      AND ${deliveries.workspaceId} = ${workspaceId}
+      AND ${deliveries.status} NOT IN ('acked', 'dead_lettered')
+  )`;
+}
+
+function readCursorWrites(
+  tx: Db,
+  workspaceId: string,
+  agentId: string,
+  row: DeliveryWithChannel,
+): AtomicWrite[] {
+  // Observe the pre-ack status. These statements run before the status update;
+  // after it, the row is already `acked` and this predicate would match nothing.
+  const ackable = deliveryStillAckable(workspaceId, row.id);
   return [
     tx
       .insert(readReceipts)
-      .values({ messageId: row.messageId, agentId })
+      .select(
+        tx.select({
+          messageId: sql<string>`${row.messageId}`.as('message_id'),
+          agentId: sql<string>`${agentId}`.as('agent_id'),
+          readAt: sql<Date>`unixepoch()`.as('read_at'),
+        }).from(deliveries).where(and(
+          eq(deliveries.id, row.id),
+          eq(deliveries.workspaceId, workspaceId),
+          notInArray(deliveries.status, ['acked', 'dead_lettered']),
+        )),
+      )
       .onConflictDoNothing(),
     tx
       .update(channelMembers)
@@ -369,8 +397,9 @@ function readCursorWrites(tx: Db, agentId: string, row: DeliveryWithChannel): At
         eq(channelMembers.channelId, row.channelId),
         eq(channelMembers.agentId, agentId),
         sql`(${channelMembers.lastReadId} IS NULL OR CAST(${channelMembers.lastReadId} AS BIGINT) < CAST(${row.messageId} AS BIGINT))`,
+        ackable,
       )),
-    advanceDmParticipantReadCursor(tx, agentId, row.channelId, row.messageId),
+    advanceDmParticipantReadCursor(tx, agentId, row.channelId, row.messageId, ackable),
   ];
 }
 
@@ -404,6 +433,7 @@ async function ackRows(
         })
         .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agentId))),
     ];
+    for (const row of rows) writes.push(...readCursorWrites(tx, workspaceId, agentId, row));
     if (ids.length > 0) {
       writes.push(tx
         .update(deliveries)
@@ -421,10 +451,10 @@ async function ackRows(
         ))
         .returning());
     }
-    for (const row of rows) writes.push(...readCursorWrites(tx, agentId, row));
     return writes;
   });
-  return ids.length > 0 ? results[1] as DeliveryRow[] : [];
+  // The status update is the last statement, so its RETURNING rows are last.
+  return ids.length > 0 ? results[results.length - 1] as DeliveryRow[] : [];
 }
 
 export async function ackDeliveriesUpToSeq(
diff --git a/packages/engine/src/engine/receipt.ts b/packages/engine/src/engine/receipt.ts
index 2dd495adf..9893c76e2 100644
--- a/packages/engine/src/engine/receipt.ts
+++ b/packages/engine/src/engine/receipt.ts
@@ -1,4 +1,4 @@
-import { eq, and, sql, isNull, inArray } from 'drizzle-orm';
+import { eq, and, sql, isNull, inArray, type SQL } from 'drizzle-orm';
 import type { getDb } from '../db/index.js';
 import {
   readReceipts,
@@ -20,6 +20,7 @@ export function advanceDmParticipantReadCursor(
   agentId: string,
   channelId: string,
   messageId: string,
+  when?: SQL,
 ) {
   return db
     .update(dmParticipants)
@@ -32,6 +33,7 @@ export function advanceDmParticipantReadCursor(
         WHERE ${dmConversations.channelId} = ${channelId}
       )`,
       sql`(${dmParticipants.lastReadId} IS NULL OR CAST(${dmParticipants.lastReadId} AS BIGINT) < CAST(${messageId} AS BIGINT))`,
+      when,
     ));
 }
 

From e693d4913171c9bc1fc017fb2904e29915c5030c Mon Sep 17 00:00:00 2001
From: Will Washburn <will.washburn@gmail.com>
Date: Wed, 28 Oct 2026 17:59:51 -0700
Subject: [PATCH 4/8] docs(engine): shorten the DM unread changelog entry

---
 packages/engine/CHANGELOG.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md
index ad1a36f29..3cca076c6 100644
--- a/packages/engine/CHANGELOG.md
+++ b/packages/engine/CHANGELOG.md
@@ -11,7 +11,7 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
 
 ### Fixed
 
-- DM inbox unread counts follow `dm_participants.last_read_id` (migration `0078_dm_participant_read_cursor.sql`). `markRead` and a successful delivery acknowledgement advance that cursor in the same atomic batch as the read. A dead-lettered delivery does not. The inbox orders the cursor by snowflake length, then text. A cursor on `channel_members` does not, because a DM participant has no channel membership row.
+- 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`).
 
 ## [12.2.0] - 2026-10-29
 

From 423d98ccf8c0664004a60900cf69fad09190c40c Mon Sep 17 00:00:00 2001
From: Will Washburn <will.washburn@gmail.com>
Date: Wed, 28 Oct 2026 18:13:40 -0700
Subject: [PATCH 5/8] fix(engine): chunk cumulative delivery acks

A node up_to_seq ack writes each chunk as its own batch.
The cumulative cursor advances only with the last chunk.
---
 .../engine/src/__tests__/atomicity.test.ts    | 50 ++++++++++++++++++-
 packages/engine/src/engine/delivery.ts        | 25 +++++++---
 2 files changed, 66 insertions(+), 9 deletions(-)

diff --git a/packages/engine/src/__tests__/atomicity.test.ts b/packages/engine/src/__tests__/atomicity.test.ts
index 79812a122..b0e2091d7 100644
--- a/packages/engine/src/__tests__/atomicity.test.ts
+++ b/packages/engine/src/__tests__/atomicity.test.ts
@@ -22,6 +22,7 @@ import {
   messageAttachments,
   messageLogs,
   messages,
+  nodes,
   readReceipts,
   pendingEvents,
   sessionEvents,
@@ -32,7 +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 } from '../engine/delivery.js';
+import { ackDelivery, ackDeliveriesUpToSeq, DELIVERY_EXPIRY_BATCH_SIZE } from '../engine/delivery.js';
 import { rotateAgentIdentity } from '../engine/agentIdentity.js';
 import {
   applyStatusEventEffect,
@@ -213,6 +214,53 @@ describe('atomic write paths', () => {
       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('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' });
diff --git a/packages/engine/src/engine/delivery.ts b/packages/engine/src/engine/delivery.ts
index ed5f500ae..f156d966d 100644
--- a/packages/engine/src/engine/delivery.ts
+++ b/packages/engine/src/engine/delivery.ts
@@ -493,14 +493,23 @@ export async function ackDeliveriesUpToSeq(
     ))
     .orderBy(asc(deliveries.seq));
 
-  const updated = await ackRows(
-    db,
-    workspaceId,
-    agent.id,
-    rows.map((row) => ({ ...row.delivery, channelId: row.channelId })),
-    upToSeq,
-  );
-  return { agent_id: agent.id, agent_name: agent.name, up_to_seq: upToSeq, acked: updated.length };
+  const pending = rows.map((row) => ({ ...row.delivery, channelId: row.channelId }));
+  // Each row adds three cursor statements, and `up_to_seq` is caller-supplied.
+  // One batch for the whole backlog exceeds D1's bound-parameter cap and holds
+  // the write lock for every row. The cumulative cursor moves only with the
+  // last chunk, so a later chunk failing cannot skip rows still queued.
+  if (pending.length === 0) {
+    await ackRows(db, workspaceId, agent.id, [], upToSeq);
+    return { agent_id: agent.id, agent_name: agent.name, up_to_seq: upToSeq, acked: 0 };
+  }
+  let acked = 0;
+  for (let offset = 0; offset < pending.length; offset += DELIVERY_EXPIRY_BATCH_SIZE) {
+    const chunk = pending.slice(offset, offset + DELIVERY_EXPIRY_BATCH_SIZE);
+    const last = offset + chunk.length === pending.length;
+    const updated = await ackRows(db, workspaceId, agent.id, chunk, last ? upToSeq : undefined);
+    acked += updated.length;
+  }
+  return { agent_id: agent.id, agent_name: agent.name, up_to_seq: upToSeq, acked };
 }
 
 /** Mark still-queued rows delivered using bounded, explicit ID-index writes. */

From c0464282c3ac3a38e7fd07e005c29ebb23aad27f Mon Sep 17 00:00:00 2001
From: Barry Cape <barryonthecape@icloud.com>
Date: Wed, 28 Oct 2026 18:29:26 -0700
Subject: [PATCH 6/8] fix(engine): seed DM read cursors and order channel
 unread numerically

Migration 0078 seeds each DM participant cursor from its newest read
receipt, so already-read conversations do not come back unread.
Channel unread shares the length-then-text cursor predicate.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  1 +
 packages/engine/CHANGELOG.md                  |  3 +-
 .../conformance/dmUnreadCursor.test.ts        | 27 ++++++++++++
 .../__tests__/dmReadCursorMigration.test.ts   | 41 +++++++++++++++++++
 .../0078_dm_participant_read_cursor.sql       | 12 ++++++
 packages/engine/src/engine/inbox.ts           | 19 +++++----
 6 files changed, 94 insertions(+), 9 deletions(-)
 create mode 100644 packages/engine/src/db/__tests__/dmReadCursorMigration.test.ts

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 57ee42ba0..4a7b4a5cf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,7 @@ Packages without a separate changelog are covered by the cross-package notes bel
 ## [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
 
diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md
index 3cca076c6..8a79badca 100644
--- a/packages/engine/CHANGELOG.md
+++ b/packages/engine/CHANGELOG.md
@@ -11,7 +11,8 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
 
 ### 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`).
+- 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
 
diff --git a/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts b/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
index d2e8855ae..897c90a9b 100644
--- a/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
+++ b/packages/engine/src/__tests__/conformance/dmUnreadCursor.test.ts
@@ -155,4 +155,31 @@ describe('DM unread cursor', () => {
     };
     expect(afterInbox.unread_channels.find((row) => row.channel_name === 'general')).toBeUndefined();
   });
+
+  it('orders a channel cursor by snowflake length, then text', async () => {
+    const ws = await createWorkspace(stack.app, 'channel-unread-width');
+    const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice');
+    const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob');
+
+    const post = await authed(alice.token).post('/v1/channels/general/messages', { text: 'current' });
+    expect(post.status).toBeLessThan(300);
+    const messageId = ((await post.json()) as { data: { id: string } }).data.id;
+    const channelUnread = async () => {
+      const res = await authed(bob.token).get('/v1/inbox');
+      const inbox = (await res.json()).data as {
+        unread_channels: Array<{ channel_name: string; unread_count: number }>;
+      };
+      return inbox.unread_channels.find((row) => row.channel_name === 'general')?.unread_count ?? 0;
+    };
+    const sqlite = stack.runtime.handle.sqlite;
+    const { channel_id: channelId } = sqlite.prepare(
+      'SELECT channel_id FROM messages WHERE id = ?',
+    ).get(messageId) as { channel_id: string };
+
+    // 17 nines sort after an 18-digit id as text, and before it numerically.
+    sqlite.prepare(
+      'UPDATE channel_members SET last_read_id = ? WHERE agent_id = ? AND channel_id = ?',
+    ).run('9'.repeat(17), bob.agentId, channelId);
+    expect(await channelUnread()).toBe(1);
+  });
 });
diff --git a/packages/engine/src/db/__tests__/dmReadCursorMigration.test.ts b/packages/engine/src/db/__tests__/dmReadCursorMigration.test.ts
new file mode 100644
index 000000000..f873c52e4
--- /dev/null
+++ b/packages/engine/src/db/__tests__/dmReadCursorMigration.test.ts
@@ -0,0 +1,41 @@
+import { readdirSync, readFileSync } from 'node:fs';
+import { fileURLToPath } from 'node:url';
+import { expect, it } from 'vitest';
+import { getSqliteDb, runMigrations } from '../../adapters/node/database.js';
+
+it('seeds DM read cursors from the newest read receipt, ordered numerically', () => {
+  const directory = fileURLToPath(new URL('../migrations/', import.meta.url));
+  const handle = getSqliteDb(':memory:');
+  const db = handle.sqlite;
+  try {
+    db.exec('CREATE TABLE _engine_migrations(name TEXT PRIMARY KEY, applied_at INTEGER NOT NULL)');
+    for (const name of readdirSync(directory).filter(name => name.endsWith('.sql') && name < '0078').sort()) {
+      db.exec(readFileSync(directory + name, 'utf8'));
+      db.prepare('INSERT INTO _engine_migrations VALUES (?,0)').run(name);
+    }
+    // 17 nines sort after the 18-digit id as text, and before it numerically.
+    const older = '9'.repeat(17);
+    const newer = '1' + '0'.repeat(17);
+    db.exec(`INSERT INTO workspaces(id,name,api_key_hash) VALUES ('ws','fixture','key');
+      INSERT INTO agents(id,workspace_id,name,token_hash) VALUES ('alice','ws','alice','a'),('bob','ws','bob','b'),('cara','ws','cara','c');
+      INSERT INTO channels(id,workspace_id,name,channel_type) VALUES ('dmch','ws','dm-alice-bob',1),('gen','ws','general',0);
+      INSERT INTO dm_conversations(id,workspace_id,channel_id) VALUES ('dm','ws','dmch');
+      INSERT INTO dm_participants(conversation_id,agent_id) VALUES ('dm','alice'),('dm','bob');
+      INSERT INTO messages(id,workspace_id,channel_id,agent_id,body) VALUES
+        ('${older}','ws','dmch','alice','older'),('${newer}','ws','dmch','alice','newer'),
+        ('2${'0'.repeat(17)}','ws','gen','cara','channel');
+      INSERT INTO read_receipts(message_id,agent_id) VALUES
+        ('${older}','bob'),('${newer}','bob'),('2${'0'.repeat(17)}','alice');`);
+
+    expect(runMigrations(handle).applied).toEqual(['0078_dm_participant_read_cursor.sql']);
+
+    const cursors = db.prepare('SELECT agent_id, last_read_id FROM dm_participants ORDER BY agent_id').all();
+    // A receipt in another channel does not seed a DM cursor.
+    expect(cursors).toEqual([
+      { agent_id: 'alice', last_read_id: null },
+      { agent_id: 'bob', last_read_id: newer },
+    ]);
+  } finally {
+    db.close();
+  }
+});
diff --git a/packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql b/packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql
index 9187c4685..5c9eec700 100644
--- a/packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql
+++ b/packages/engine/src/db/migrations/0078_dm_participant_read_cursor.sql
@@ -1,3 +1,15 @@
 -- DM unread counts read a per-participant cursor. DM membership lives in
 -- dm_participants, so a cursor stored only on channel_members never moves.
 ALTER TABLE dm_participants ADD COLUMN last_read_id TEXT;
+
+-- Seed each cursor from the participant's newest read receipt in that DM, so
+-- conversations read before this column existed do not come back unread.
+-- Walks each DM's messages once per participant and probes the receipt
+-- primary key. Snowflake ids order numerically, not as text.
+UPDATE dm_participants SET last_read_id = (
+  SELECT CAST(MAX(CAST(m.id AS INTEGER)) AS TEXT)
+  FROM dm_conversations dc
+  JOIN messages m ON m.channel_id = dc.channel_id
+  JOIN read_receipts rr ON rr.message_id = m.id AND rr.agent_id = dm_participants.agent_id
+  WHERE dc.id = dm_participants.conversation_id
+);
diff --git a/packages/engine/src/engine/inbox.ts b/packages/engine/src/engine/inbox.ts
index f66fbb6a8..97165915c 100644
--- a/packages/engine/src/engine/inbox.ts
+++ b/packages/engine/src/engine/inbox.ts
@@ -116,6 +116,15 @@ async function backfillMessageMentionFtsUnlabeled(db: Db, limit: number): Promis
   return copied;
 }
 
+/**
+ * Unread predicate for a read cursor. Snowflake ids are unpadded decimal text:
+ * length, then text, is chronological; a bare text compare reverses order
+ * across a digit boundary. Both arguments are fixed column references.
+ */
+function messageAfterCursor(id: string, cursor: string) {
+  return sql.raw(`(${cursor} IS NULL OR length(${id}) > length(${cursor}) OR (length(${id}) = length(${cursor}) AND ${id} > ${cursor}))`);
+}
+
 export async function getInbox(db: Db, workspaceId: string, agentId: string) {
   return withEngineOperation('inbox.get', workspaceId, () => getInboxUnlabeled(db, workspaceId, agentId));
 }
@@ -132,7 +141,7 @@ async function getInboxUnlabeled(db: Db, workspaceId: string, agentId: string) {
       AND ch.is_archived = false
       AND m.thread_id IS NULL
       AND m.agent_id != ${agentId}
-      AND (cm.last_read_id IS NULL OR m.id > cm.last_read_id)
+      AND ${messageAfterCursor('m.id', 'cm.last_read_id')}
     GROUP BY cm.channel_id, ch.name
     HAVING count(*) > 0
   `);
@@ -272,13 +281,7 @@ async function getInboxUnlabeled(db: Db, workspaceId: string, agentId: string) {
     JOIN messages m
       ON m.channel_id = dc.channel_id
      AND m.agent_id != ${agentId}
-     -- Snowflake ids are unpadded decimal text. Length, then text, is
-     -- chronological; a bare text compare reverses order across a digit boundary.
-     AND (
-       dp.last_read_id IS NULL
-       OR length(m.id) > length(dp.last_read_id)
-       OR (length(m.id) = length(dp.last_read_id) AND m.id > dp.last_read_id)
-     )
+     AND ${messageAfterCursor('m.id', 'dp.last_read_id')}
     WHERE dc.workspace_id = ${workspaceId}
     GROUP BY dc.id, dc.channel_id
     HAVING count(m.id) > 0

From fb9e6ddc8c1c733a223ea45751be3eb653a635fc Mon Sep 17 00:00:00 2001
From: Will Washburn <will.washburn@gmail.com>
Date: Wed, 28 Oct 2026 18:41:03 -0700
Subject: [PATCH 7/8] test(engine): cover DM cursor progress across ack chunks

A failed later chunk leaves the participant cursor on the last
message the first chunk acked.
---
 .../engine/src/__tests__/atomicity.test.ts    | 45 +++++++++++++++++++
 1 file changed, 45 insertions(+)

diff --git a/packages/engine/src/__tests__/atomicity.test.ts b/packages/engine/src/__tests__/atomicity.test.ts
index b0e2091d7..00288e89a 100644
--- a/packages/engine/src/__tests__/atomicity.test.ts
+++ b/packages/engine/src/__tests__/atomicity.test.ts
@@ -261,6 +261,51 @@ describe('atomic write paths', () => {
       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' });

From e031b04dce04ddb42fa826c6d8fd0c186c160f06 Mon Sep 17 00:00:00 2001
From: Will Washburn <will.washburn@gmail.com>
Date: Wed, 28 Oct 2026 18:44:59 -0700
Subject: [PATCH 8/8] docs: describe when a DM inbox unread count drops

---
 README.md    |  4 +++-
 openapi.yaml | 13 ++++++++-----
 2 files changed, 11 insertions(+), 6 deletions(-)

diff --git a/README.md b/README.md
index ae6aa1f08..b6736da2d 100644
--- a/README.md
+++ b/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`.
diff --git a/openapi.yaml b/openapi.yaml
index fcb09a746..da23972bf 100644
--- a/openapi.yaml
+++ b/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: