fix(digest): a consumed wake reads its pod's whole digest; silent_or_empty is a read

backend/__tests__/unit/scripts/sweep-stale-digest.test.js

@@ -0,0 +1,84 @@
+const mongoose = require('mongoose');
+const AgentEvent = require('../../../models/AgentEvent');
+const { sweepStaleDigest, parseArgs, SWEEP_REASON } = require('../../../scripts/sweep-stale-digest');
+const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../../utils/testUtils');
+
+const HOUR_MS = 60 * 60 * 1000;
+const NOW = new Date('2026-10-26T12:00:00.000Z');
+const POD_ID = new mongoose.Types.ObjectId();
+
+describe('sweep-stale-digest', () => {
+  beforeAll(async () => { await setupMongoDb(); });
+  afterEach(async () => { await clearMongoDb(); });
+  afterAll(async () => { await closeMongoDb(); });
+
+  // Raw insert, so each row keeps the createdAt it is given.
+  const row = (overrides) => ({
+    agentName: 'rhea',
+    instanceId: 'default',
+    podId: POD_ID,
+    type: 'message.posted',
+    deliveryMode: 'digest',
+    status: 'pending',
+    attempts: 0,
+    payload: { content: 'ambient' },
+    ...overrides,
+  });
+  const seed = async () => {
+    const ago = (hours) => new Date(NOW.getTime() - hours * HOUR_MS);
+    const docs = [
+      row({ createdAt: ago(30) }),
+      row({ createdAt: ago(25) }),
+      row({ createdAt: ago(48), status: 'delivered', deliveryNonce: 'n-1' }),
+      row({ createdAt: ago(26), agentName: 'quire' }),
+      row({ createdAt: ago(2) }), // inside the day: stays unread
+      row({ createdAt: ago(30), deliveryMode: 'wake', type: 'chat.mention' }), // a task, never swept
+      row({ createdAt: ago(30), status: 'acked' }), // already read
+    ];
+    const { insertedIds } = await AgentEvent.collection.insertMany(docs);
+    return Object.values(insertedIds);
+  };
+  const statuses = async (ids) => Promise.all(ids.map(async (id) => (await AgentEvent.findById(id).lean()).status));
+
+  test('a dry run reports per seat and changes nothing', async () => {
+    const ids = await seed();
+    const result = await sweepStaleDigest({ now: NOW });
+    expect(result).toEqual({
+      apply: false,
+      cutoff: '2026-10-25T12:00:00.000Z',
+      matched: 4,
+      settled: 0,
+      seats: [
+        { agentName: 'rhea', instanceId: 'default', count: 3 },
+        { agentName: 'quire', instanceId: 'default', count: 1 },
+      ],
+    });
+    expect(await statuses(ids)).toEqual(['pending', 'pending', 'delivered', 'pending', 'pending', 'pending', 'acked']);
+  });
+
+  test('apply settles only digest rows older than a day, with the sweep as their reason', async () => {
+    const ids = await seed();
+    const result = await sweepStaleDigest({ apply: true, now: NOW });
+    expect(result.matched).toBe(4);
+    expect(result.settled).toBe(4);
+    expect(await statuses(ids)).toEqual(['acked', 'acked', 'acked', 'acked', 'pending', 'pending', 'acked']);
+    const swept = await AgentEvent.findById(ids[2]).lean();
+    expect(swept.deliveryNonce).toBeNull();
+    expect(swept.delivery).toEqual(expect.objectContaining({ outcome: 'no_action', reason: SWEEP_REASON }));
+    expect((await AgentEvent.findById(ids[6]).lean()).delivery).toBeUndefined();
+  });
+
+  test('a second apply finds nothing left', async () => {
+    await seed();
+    await sweepStaleDigest({ apply: true, now: NOW });
+    expect(await sweepStaleDigest({ apply: true, now: NOW }))
+      .toEqual(expect.objectContaining({ matched: 0, settled: 0, seats: [] }));
+  });
+
+  test('--dry-run wins over --apply, and no flag is a dry run', () => {
+    expect(parseArgs([])).toEqual({ apply: false });
+    expect(parseArgs(['--dry-run'])).toEqual({ apply: false });
+    expect(parseArgs(['--apply'])).toEqual({ apply: true });
+    expect(parseArgs(['--apply', '--dry-run'])).toEqual({ apply: false });
+  });
+});

backend/__tests__/unit/services/agentEventService.digestBacklog.test.js

@@ -0,0 +1,233 @@
+// A consumed wake reads its pod's whole unread digest, not only the page it
+// quoted. Pinned against mongodb-memory-server.
+//
+// The loop this exists for (rhea, 2026-10-25): a wake claimed up to 9 digest
+// companions, the seat answered NO_REPLY, the server judged the reply
+// `silent_or_empty`, and both the CLI and the ack treated that as "not read".
+// The requeue put the companions back, the unread-digest heartbeat fired an
+// hour later, and it repeated. Even a consumed wake settled only the rows it
+// had quoted, so a backlog drained 9 rows per wake, oldest first, while about
+// 1,000 rows arrived per seat per day. 22,568 were pending fleet-wide.
+const { MongoMemoryServer } = require('mongodb-memory-server');
+const mongoose = require('mongoose');
+
+jest.mock('../../../services/agentMemoryService', () => ({
+  buildMemoryDigestBundle: jest.fn(() => ({})),
+}));
+jest.mock('../../../services/agentProvisionerService', () => ({
+  getAgentSessionSizes: jest.fn(),
+  clearAgentRuntimeSessions: jest.fn(),
+  restartAgentRuntime: jest.fn(),
+  resolveOpenClawAccountId: jest.fn(() => 'acct'),
+}));
+jest.mock('../../../services/agentIdentityService', () => ({
+  getAgentTypeConfig: jest.fn(() => ({ runtime: 'moltbot' })),
+}));
+jest.mock('../../../services/nativeRuntimeService', () => ({ runAgent: jest.fn() }));
+
+let mongod;
+let AgentEvent;
+let AgentEventService;
+
+const AGENT = 'pixel';
+const INSTANCE = 'default';
+const POD_ID = new mongoose.Types.ObjectId();
+const BACKLOG = 30;
+
+beforeAll(async () => {
+  mongod = await MongoMemoryServer.create();
+  await mongoose.connect(mongod.getUri());
+  AgentEvent = require('../../../models/AgentEvent');
+  AgentEventService = require('../../../services/agentEventService');
+});
+
+afterAll(async () => {
+  await mongoose.disconnect();
+  await mongod.stop();
+});
+
+beforeEach(async () => {
+  await AgentEvent.deleteMany({});
+});
+
+// Raw insert so each row keeps the createdAt it is given: oldest first, one
+// minute apart, each in its own thread so the frame can name which it quoted.
+const seedBacklog = async () => {
+  const now = Date.now();
+  await AgentEvent.collection.insertMany(Array.from({ length: BACKLOG }, (_, index) => ({
+    agentName: AGENT,
+    instanceId: INSTANCE,
+    podId: POD_ID,
+    type: 'message.posted',
+    deliveryMode: 'digest',
+    status: 'pending',
+    attempts: 0,
+    payload: { messageId: `m${index + 1}`, threadRootId: index + 1, content: 'ambient body' },
+    createdAt: new Date(now - (BACKLOG - index) * 60000),
+    updatedAt: new Date(now - (BACKLOG - index) * 60000),
+  })));
+};
+
+const seedWake = () => AgentEvent.create({
+  agentName: AGENT, instanceId: INSTANCE, podId: POD_ID,
+  type: 'chat.mention', payload: { content: 'question' },
+});
+
+const unread = () => AgentEvent.countDocuments({
+  deliveryMode: 'digest', status: { $in: ['pending', 'delivered'] },
+});
+
+// The wrapper CLI's path: one pod-scoped page, wake plus digest companions.
+const claimPage = async () => {
+  const page = await AgentEventService.listInboxPage({
+    agentName: AGENT, instanceId: INSTANCE, podIds: [POD_ID], limit: 10,
+  });
+  const [wake] = page.events.filter((event) => event.deliveryMode !== 'digest');
+  return { page, wake };
+};
+
+describe('a consumed wake settles the digest backlog up to its snapshot', () => {
+  test.each([
+    [{ outcome: 'posted' }],
+    [{ outcome: 'acknowledged' }],
+    [{ outcome: 'no_action' }],
+    [{ outcome: 'no_action', reason: 'silent_or_empty' }],
+  ])('a 30-row backlog drains in one consumed ack: %j', async (result) => {
+    await seedBacklog();
+    await seedWake();
+    const { page, wake } = await claimPage();
+    expect(page.events.filter((event) => event.deliveryMode === 'digest')).toHaveLength(9);
+    expect((await AgentEvent.findById(wake._id).lean()).digestSnapshotAt).toEqual(expect.any(Date));
+
+    await AgentEventService.acknowledge(wake._id, AGENT, INSTANCE, result, wake.deliveryNonce);
+    expect(await unread()).toBe(0);
+  });
+
+  test.each([
+    [{ outcome: 'no_action', reason: 'claim-held' }],
+    [{ outcome: 'no_action', reason: 'duplicate_recent' }],
+    // Empty output is not the NO_REPLY sentinel: a seat failing silently every
+    // turn must leave its backlog pending, not drain it (sable, #2753).
+    [{ outcome: 'no_action', reason: 'empty-output' }],
+    [{ outcome: 'error' }],
+  ])('a non-consuming outcome settles nothing: %j', async (result) => {
+    await seedBacklog();
+    await seedWake();
+    const { wake } = await claimPage();
+
+    await AgentEventService.acknowledge(wake._id, AGENT, INSTANCE, result, wake.deliveryNonce);
+    expect(await unread()).toBe(BACKLOG);
+    expect(await AgentEvent.countDocuments({ deliveryMode: 'digest', status: 'acked' })).toBe(0);
+  });
+
+  test('a row created after the snapshot stays pending', async () => {
+    await seedBacklog();
+    await seedWake();
+    const { wake } = await claimPage();
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    const late = await AgentEvent.create({
+      agentName: AGENT, instanceId: INSTANCE, podId: POD_ID,
+      type: 'message.posted', deliveryMode: 'digest', payload: { content: 'arrived mid-turn' },
+    });
+
+    await AgentEventService.acknowledge(wake._id, AGENT, INSTANCE,
+      { outcome: 'posted' }, wake.deliveryNonce);
+    expect((await AgentEvent.findById(late._id).lean()).status).toBe('pending');
+    expect(await unread()).toBe(1);
+  });
+
+  test('another pod\'s backlog is untouched', async () => {
+    await seedBacklog();
+    const otherPod = new mongoose.Types.ObjectId();
+    const other = await AgentEvent.create({
+      agentName: AGENT, instanceId: INSTANCE, podId: otherPod,
+      type: 'message.posted', deliveryMode: 'digest', payload: { content: 'elsewhere' },
+    });
+    await AgentEvent.updateOne({ _id: other._id }, { $set: { createdAt: new Date(Date.now() - 3600000) } });
+    await seedWake();
+    const { wake } = await claimPage();
+
+    await AgentEventService.acknowledge(wake._id, AGENT, INSTANCE,
+      { outcome: 'posted' }, wake.deliveryNonce);
+    expect((await AgentEvent.findById(other._id).lean()).status).toBe('pending');
+  });
+
+  test('a late companion ack after the settle is still a success', async () => {
+    await seedBacklog();
+    await seedWake();
+    const { page, wake } = await claimPage();
+    const companion = page.events.find((event) => event.deliveryMode === 'digest');
+
+    await AgentEventService.acknowledge(wake._id, AGENT, INSTANCE,
+      { outcome: 'posted' }, wake.deliveryNonce);
+    expect(await AgentEventService.acknowledge(companion._id, AGENT, INSTANCE,
+      { outcome: 'no_action', reason: 'digest-read' }, companion.deliveryNonce)).toBeNull();
+    expect(await AgentEventService.isSupersededDelivery(
+      companion._id, AGENT, INSTANCE, companion.deliveryNonce,
+    )).toBe(false);
+  });
+});
+
+describe('the legacy frame shows the newest rows and the true total', () => {
+  test('quotes the newest 8 rows and counts every unread row', async () => {
+    await seedBacklog();
+    await seedWake();
+    const [delivered] = await AgentEventService.list({ agentName: AGENT, instanceId: INSTANCE });
+    const summary = delivered.payload.digestSummary;
+
+    expect(summary).toContain(`${BACKLOG} ambient updates`);
+    expect(summary).toContain(`older pod activity: ${BACKLOG - 8} unread`);
+    expect(summary).toContain('thread 23: 1 unread');
+    expect(summary).not.toContain('thread 22:');
+    expect(summary).not.toContain('thread 1:');
+    expect(summary).not.toContain('ambient body');
+
+    await AgentEventService.acknowledge(delivered._id, AGENT, INSTANCE,
+      { outcome: 'no_action', reason: 'silent_or_empty' }, delivered.payload.deliveryId);
+    expect(await unread()).toBe(0);
+  });
+});
+
+describe('a native turn settles the backlog it was shown', () => {
+  const { runAgent } = require('../../../services/nativeRuntimeService');
+  const { AgentInstallation } = require('../../../models/AgentRegistry');
+  const nativeInstall = {
+    _id: new mongoose.Types.ObjectId(),
+    agentName: AGENT,
+    instanceId: INSTANCE,
+    podId: POD_ID,
+    status: 'active',
+    config: { runtime: { runtimeType: 'native' } },
+  };
+
+  afterEach(() => { jest.restoreAllMocks(); });
+
+  const runNative = async (runResult) => {
+    runAgent.mockReset();
+    runAgent.mockResolvedValue(runResult);
+    jest.spyOn(AgentInstallation, 'findOne').mockReturnValue({ lean: () => Promise.resolve(nativeInstall) });
+    const event = await AgentEventService.enqueue({
+      agentName: AGENT, instanceId: INSTANCE, podId: POD_ID, type: 'chat.mention', payload: { text: 'hi' },
+    });
+    const deadline = Date.now() + 5000;
+    while (Date.now() < deadline
+      && !['acked', 'failed'].includes((await AgentEvent.findById(event._id).lean()).status)) {
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+    expect(runAgent).toHaveBeenCalledTimes(1);
+    return runAgent.mock.calls[0][1].payload.content;
+  };
+
+  test('a succeeded turn reads every row up to its snapshot', async () => {
+    await seedBacklog();
+    const content = await runNative({ status: 'succeeded', runId: 'run-1', totalTurns: 1 });
+    expect(content).toContain(`${BACKLOG} ambient updates`);
+    expect(await unread()).toBe(0);
+  });
+
+  test('a turn that never ran the model leaves the backlog unread', async () => {
+    await seedBacklog();
+    await runNative({ status: 'succeeded', runId: null, totalTurns: 0 });
+    expect(await unread()).toBe(BACKLOG);
+  });
+});

backend/__tests__/unit/services/agentEventService.lifecycle.test.js

@@ -480,6 +480,7 @@ describe('driver paths reach a terminal state', () => {
     };
     const duplicate = { ...ambient, _id: 'ambient-2' };
     AgentEvent.find.mockReturnValue({ sort: () => ({ limit: () => ({ lean: async () => [ambient, duplicate] }) }) });
+    AgentEvent.countDocuments.mockResolvedValue(2);
     AgentEvent.updateOne.mockResolvedValue({ matchedCount: 1 });
     AgentEvent.updateMany.mockResolvedValue({ modifiedCount: 1 });
     AgentEvent.create.mockResolvedValue({
@@ -501,8 +502,11 @@ describe('driver paths reach a terminal state', () => {
     expect(body.payload.digestSummary).toContain('thread 40: 1 unread');
     expect(body.payload.digestSummary).toContain('1 ambient update in 1 conversation');
     expect(body.payload.digestSummary).not.toContain('ambient note');
+    // Accepted delivery reads the pod's whole backlog up to the wake's snapshot.
     expect(AgentEvent.updateMany).toHaveBeenCalledWith(
-      expect.objectContaining({ _id: { $in: ['ambient-1', 'ambient-2'] }, deliveryMode: 'digest' }),
+      expect.objectContaining({
+        agentName: 'hookbot', podId: POD_P1, deliveryMode: 'digest', createdAt: { $lte: expect.any(Date) },
+      }),
       expect.objectContaining({ $set: expect.objectContaining({ status: 'acked' }) }),
     );
   });
@@ -575,9 +579,10 @@ describe('driver paths reach a terminal state', () => {
       expect(runAgent).not.toHaveBeenCalled();
     });
 
-    test('a later native wake carries digest context and settles only included rows on success', async () => {
+    test('a later native wake carries digest context and settles its backlog only on success', async () => {
       const digest = { _id: 'evt-d', payload: { username: 'alice', messageId: '42', content: 'daily update' } };
       AgentEvent.find.mockReturnValue({ sort: () => ({ limit: () => ({ lean: () => Promise.resolve([digest]) }) }) });
+      AgentEvent.countDocuments.mockResolvedValue(1);
       AgentEvent.updateMany.mockResolvedValue({ modifiedCount: 1 });
       runAgent.mockResolvedValue({ status: 'succeeded', runId: 'run-1', totalTurns: 1 });
 
@@ -586,8 +591,8 @@ describe('driver paths reach a terminal state', () => {
       expect(runAgent.mock.calls[0][1].payload.content).toContain('pod conversation: 1 unread');
       expect(runAgent.mock.calls[0][1].payload.content).not.toContain('daily update');
       expect(AgentEvent.updateMany).toHaveBeenCalledWith(
-        { _id: { $in: ['evt-d'] }, status: 'pending', deliveryMode: 'digest' },
-        { $set: { status: 'acked' } },
+        expect.objectContaining({ agentName: 'clerk', deliveryMode: 'digest', createdAt: { $lte: expect.any(Date) } }),
+        { $set: expect.objectContaining({ status: 'acked' }) },
       );
 
       jest.clearAllMocks();

backend/__tests__/unit/services/agentEventService.test.js

@@ -423,10 +423,18 @@ describe('AgentEventService', () => {
         status: 'delivered', payload: { messageId: '42' },
       }) });
     readFoldedDigest.mockResolvedValueOnce({ unreadCount: 3, lastEventIds: ['older-1'], revision: 4 });
-    AgentEvent.updateOne.mockResolvedValueOnce({ modifiedCount: 1 });
+    AgentEvent.updateOne
+      .mockResolvedValueOnce({ modifiedCount: 1 }) // the wake's digest snapshot
+      .mockResolvedValueOnce({ modifiedCount: 1 }); // the folded rollup it carries
     const page = await AgentEventService.listInboxPage({ agentName: 'inbox', podId: 'pod-a' });
     expect(page.events.map((event) => event._id)).toEqual(['wake', 'digest']);
     expect(page.inboxCount).toBe(2);
+    // A wake that carried digest records when it read it, so its ack settles
+    // the pod's backlog up to that moment.
+    expect(AgentEvent.updateOne).toHaveBeenNthCalledWith(1,
+      expect.objectContaining({ _id: 'wake', status: 'delivered' }),
+      { $set: { digestSnapshotAt: expect.any(Date) } });
+    expect(page.events[0].digestSnapshotAt).toEqual(expect.any(Date));
     expect(page.events[0].payload.digestRollup).toEqual({ unreadCount: 3, lastEventIds: ['older-1'] });
     expect(page.events[1].payload.digestRollup).toBeUndefined();
     expect(AgentEvent.find).toHaveBeenNthCalledWith(3, expect.objectContaining({