TASK-414: separate digest delivery from agent wake

backend/__tests__/service/taskEventService.offeredAt.test.js

@@ -119,6 +119,20 @@ afterAll(async () => {
 });
 
 describe('kernel sweep offeredAt filter', () => {
+  it('still nudges unassigned work after its board change reached only digest', async () => {
+    const row = await createTask({ taskId: 'TASK-UNASSIGNED', updatedAt: new Date(NOW.getTime() - 30000) });
+
+    await notifyPodAgents(POD_ID, row, 'created', { isAgent: false });
+
+    expect(AgentEventService.enqueue.mock.calls[0][0].deliveryMode).toBe('digest');
+    expect((await Task.findById(row._id).lean()).offeredAt).toBeUndefined();
+    const sweep = await KernelWorkSweepService.wakeForFoundWork(NOW);
+    expect(sweep.woken).toBe(1);
+    expect(AgentEventService.enqueue.mock.calls[1][0].payload.foundWorkAt).toEqual(NOW);
+    const after = await Task.findById(row._id).lean();
+    expect(after.offeredAt).toEqual(after.updatedAt);
+  });
+
   it('still offers a revision deferred past the ten-minute scan window by the hourly cap', async () => {
     const old = await createTask({
       taskId: 'TASK-DEFERRED', updatedAt: new Date(NOW.getTime() - 45 * 60 * 1000),
@@ -287,7 +301,8 @@ describe('board wake assigned work', () => {
     expect(scout.payload.content).toContain('[Your assigned rows (pending/claimed; newest first):]');
     expect(scout.payload.content).toContain('TASK-ASSIGN — TASK-ASSIGN [pending] — Lily updated: reassigned to scout');
     expect(scout.payload.content).not.toContain('TASK-FOREIGN');
-    expect(peer).toBeUndefined();
+    expect(peer.deliveryMode).toBe('digest');
+    expect(peer.payload.content).toBe('TASK-ASSIGN updated: TASK-ASSIGN');
   });
 
   it('keeps the assigned list to five rows, newest first, after a folded change', async () => {
@@ -338,12 +353,14 @@ describe('board wake assigned work', () => {
     expect(anvil.payload.content).toContain('TASK-INSTANCE — Instance Do it now [pending]');
     expect(anvil.payload.content).not.toContain('Instance\nDo it now');
     expect(anvil.payload.content).not.toContain('TASK-SCOUT');
-    expect(scout).toBeUndefined();
+    expect(scout.deliveryMode).toBe('digest');
+    expect(scout.payload.content).toBe('TASK-INSTANCE updated: Instance Do it now');
   });
 });
 
 describe('own-row board wake routing', () => {
-  const wakeNames = () => AgentEventService.enqueue.mock.calls.map(([event]) => event.agentName);
+  const wakeNames = () => AgentEventService.enqueue.mock.calls
+    .filter(([event]) => event.deliveryMode !== 'digest').map(([event]) => event.agentName);
 
   beforeEach(() => setInstalls([install('scout'), install('sprint-impl')]));
 
@@ -381,7 +398,8 @@ describe('own-row board wake routing', () => {
     await notifyPodAgents(POD_ID, commented, 'updated', { isAgent: false });
 
     expect(wakeNames()).toEqual(['scout']);
-    expect(AgentEventService.enqueue.mock.calls[0][0].payload.content).toContain('Peer commented');
+    expect(AgentEventService.enqueue.mock.calls
+      .find(([event]) => event.deliveryMode !== 'digest')[0].payload.content).toContain('Peer commented');
   });
 
   it.each(['native', 'durable-object'])('sends a hosted %s owner a type its engine runs', async (engine) => {
@@ -394,7 +412,8 @@ describe('own-row board wake routing', () => {
     await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false });
 
     expect(wakeNames()).toEqual(['maya-agent']);
-    expect(AgentEventService.enqueue.mock.calls[0][0].type).toBe('chat.mention');
+    expect(AgentEventService.enqueue.mock.calls
+      .find(([event]) => event.deliveryMode !== 'digest')[0].type).toBe('chat.mention');
   });
 
   it('inherits hosted runtime for a mention-created pod projection', async () => {

backend/__tests__/unit/models/AgentEvent.deliveryMode.test.js

@@ -0,0 +1,14 @@
+const mongoose = require('mongoose');
+const AgentEvent = require('../../../models/AgentEvent');
+
+describe('AgentEvent delivery mode persistence shape', () => {
+  const base = {
+    agentName: 'flint', instanceId: 'default', podId: new mongoose.Types.ObjectId(),
+    type: 'message.posted', status: 'pending', attempts: 0,
+  };
+
+  test('legacy callers default to wake; digest survives strict schema conversion', () => {
+    expect(new AgentEvent(base).toObject().deliveryMode).toBe('wake');
+    expect(new AgentEvent({ ...base, deliveryMode: 'digest' }).toObject().deliveryMode).toBe('digest');
+  });
+});

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

@@ -0,0 +1,217 @@
+const mongoose = require('mongoose');
+const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../../utils/testUtils');
+const AgentEvent = require('../../../models/AgentEvent');
+const AgentDigestState = require('../../../models/AgentDigestState');
+const AgentEventService = require('../../../services/agentEventService');
+const { foldExpiredDigests, readFoldedDigest, settleFoldedDigest } = require('../../../services/agentDigestFoldService');
+
+const NOW = new Date('2026-10-23T23:00:00.000Z');
+const THRESHOLD = new Date(NOW.getTime() - 60 * 60 * 1000);
+
+describe('expired digest fold', () => {
+  beforeAll(async () => {
+    await setupMongoDb();
+    await AgentDigestState.init();
+  });
+  afterAll(closeMongoDb);
+  afterEach(async () => {
+    jest.restoreAllMocks();
+    await clearMongoDb();
+  });
+
+  const seed = async (createdAt, podId, payload = {}) => AgentEvent.collection.insertOne({
+    agentName: 'flint', instanceId: 'default', podId,
+    type: 'message.posted', deliveryMode: 'digest', status: 'pending',
+    payload,
+    attempts: 0, createdAt, updatedAt: createdAt,
+  });
+
+  it('drains multiple batches within one GC pass', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    await Promise.all(Array.from({ length: 205 }, () => seed(new Date(THRESHOLD.getTime() - 1000), podId)));
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(205);
+
+    expect(await AgentEvent.countDocuments({ podId })).toBe(0);
+    expect((await AgentDigestState.findOne({ podId }).lean()).unreadCount).toBe(205);
+  });
+
+  it('stops at its time budget and leaves remaining rows due for the next pass', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    await Promise.all(Array.from({ length: 4 }, () => seed(new Date(THRESHOLD.getTime() - 1000), podId)));
+    let elapsed = 0;
+    const realDelete = AgentEvent.deleteOne.bind(AgentEvent);
+    const deletion = jest.spyOn(AgentEvent, 'deleteOne').mockImplementation(async (...args) => {
+      const result = await realDelete(...args);
+      elapsed += 5000;
+      return result;
+    });
+
+    expect(await foldExpiredDigests(THRESHOLD, () => elapsed)).toBe(2);
+    expect(await AgentEvent.countDocuments({ podId })).toBe(2);
+    deletion.mockRestore();
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(2);
+    expect((await AgentDigestState.findOne({ podId }).lean()).unreadCount).toBe(4);
+  });
+
+  it('defers a contended scope without starving another scope beyond the first batch', async () => {
+    const blockedPod = new mongoose.Types.ObjectId();
+    const healthyPod = new mongoose.Types.ObjectId();
+    const old = new Date(THRESHOLD.getTime() - 1000);
+    await Promise.all(Array.from({ length: 102 }, () => seed(old, blockedPod)));
+    await seed(old, healthyPod);
+    await AgentDigestState.create({ agentName: 'flint', instanceId: 'default', podId: blockedPod });
+    const realUpdate = AgentDigestState.updateOne.bind(AgentDigestState);
+    const contention = jest.spyOn(AgentDigestState, 'updateOne').mockImplementation((filter, ...args) => (
+      String(filter.podId) === String(blockedPod)
+        ? Promise.resolve({ modifiedCount: 0 }) : realUpdate(filter, ...args)
+    ));
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(1);
+    expect(await AgentEvent.countDocuments({ podId: blockedPod })).toBe(102);
+    expect((await AgentDigestState.findOne({ podId: healthyPod }).lean()).unreadCount).toBe(1);
+    contention.mockRestore();
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(102);
+    expect((await AgentDigestState.findOne({ podId: blockedPod }).lean()).unreadCount).toBe(102);
+  });
+
+  it('retains unread count and last IDs while deleting expired full events', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    const first = await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+    const second = await seed(new Date(NOW.getTime() - 90 * 60 * 1000), podId);
+    const recent = await seed(NOW, podId);
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(2);
+    const state = await AgentDigestState.findOne({ agentName: 'flint', podId }).lean();
+    expect(state.unreadCount).toBe(2);
+    expect(state.unreadSince).toEqual(new Date(NOW.getTime() - 2 * 60 * 60 * 1000));
+    expect(state.lastEventIds).toEqual([String(first.insertedId), String(second.insertedId)]);
+    expect(state.conversations).toEqual([{ label: 'pod conversation', count: 2 }]);
+    expect(await AgentEvent.exists({ _id: recent.insertedId })).toBeTruthy();
+    expect(await AgentEvent.countDocuments({ podId })).toBe(1);
+  });
+
+  it('retains conversation counts and order after full event bodies expire', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    const old = new Date(NOW.getTime() - 2 * 60 * 60 * 1000);
+    await seed(old, podId, { threadRootId: 40, content: 'first body' });
+    await seed(old, podId, { threadRootId: 40, content: 'second body' });
+    await seed(old, podId, { taskId: 'TASK-7', content: 'board text' });
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(3);
+    const state = await readFoldedDigest({ agentName: 'flint', instanceId: 'default', podId });
+    expect(state.conversations).toEqual([
+      { label: 'thread 40', count: 2 },
+      { label: 'TASK-7', count: 1 },
+    ]);
+    expect(state.otherCount).toBe(0);
+    expect(JSON.stringify(state)).not.toContain('body');
+    expect(JSON.stringify(state)).not.toContain('board text');
+  });
+
+  it('folds old unacked deliveries and failures before ordinary GC can delete them', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    const delivered = await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+    const failed = await seed(new Date(NOW.getTime() - 90 * 60 * 1000), podId);
+    await AgentEvent.collection.updateOne({ _id: delivered.insertedId }, { $set: { status: 'delivered' } });
+    await AgentEvent.collection.updateOne({ _id: failed.insertedId }, { $set: { status: 'failed' } });
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(2);
+    expect((await AgentDigestState.findOne({ agentName: 'flint', podId }).lean()).unreadCount).toBe(2);
+  });
+
+  it('does not count a row twice after a crash between state update and delete', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    const event = await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+    const deleteOne = jest.spyOn(AgentEvent, 'deleteOne').mockRejectedValueOnce(new Error('lost connection'));
+    await expect(foldExpiredDigests(THRESHOLD)).rejects.toThrow('lost connection');
+    deleteOne.mockRestore();
+    expect(await AgentEvent.findById(event.insertedId).lean()).toMatchObject({ status: 'folding' });
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(1);
+    const state = await AgentDigestState.findOne({ agentName: 'flint', podId }).lean();
+    expect(state.unreadCount).toBe(1);
+    expect(state.lastEventIds).toEqual([String(event.insertedId)]);
+  });
+
+  it('counts each expired row once when two GC workers overlap', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+    await seed(new Date(NOW.getTime() - 90 * 60 * 1000), podId);
+
+    const deleted = await Promise.all([foldExpiredDigests(THRESHOLD), foldExpiredDigests(THRESHOLD)]);
+    expect(deleted[0] + deleted[1]).toBe(2);
+    const state = await AgentDigestState.findOne({ agentName: 'flint', podId }).lean();
+    expect(state.unreadCount).toBe(2);
+    expect(await AgentEvent.countDocuments({ podId })).toBe(0);
+  });
+
+  it('does not advance the cursor past an older row another worker claimed', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+    await seed(new Date(NOW.getTime() - 90 * 60 * 1000), podId);
+    const claim = jest.spyOn(AgentEvent, 'findOneAndUpdate').mockResolvedValueOnce(null);
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(0);
+    expect(await AgentDigestState.findOne({ agentName: 'flint', podId })).toBeNull();
+    claim.mockRestore();
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(2);
+  });
+
+  it('waits when an older ID has a newer retention timestamp', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    const olderId = await seed(NOW, podId);
+    await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(0);
+    expect(await AgentDigestState.findOne({ agentName: 'flint', podId })).toBeNull();
+    await AgentEvent.collection.updateOne(
+      { _id: olderId.insertedId },
+      { $set: { createdAt: new Date(NOW.getTime() - 3 * 60 * 60 * 1000) } },
+    );
+    expect(await foldExpiredDigests(THRESHOLD)).toBe(2);
+  });
+
+  it('settles the exact seen revision while preserving a newer fold', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    const scope = { agentName: 'flint', instanceId: 'default', podId };
+    await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+    await foldExpiredDigests(THRESHOLD);
+    const first = await readFoldedDigest(scope);
+    expect(first).toMatchObject({ unreadCount: 1, revision: 1 });
+
+    await seed(new Date(NOW.getTime() - 90 * 60 * 1000), podId);
+    await foldExpiredDigests(THRESHOLD);
+    await settleFoldedDigest(scope, first.revision);
+    expect((await readFoldedDigest(scope)).unreadCount).toBe(2);
+
+    const latest = await readFoldedDigest(scope);
+    await settleFoldedDigest(scope, latest.revision);
+    expect(await readFoldedDigest(scope)).toBeNull();
+    expect((await AgentDigestState.findOne(scope).lean()).revision).toBe(3);
+    expect((await AgentDigestState.findOne(scope).lean()).unreadSince).toBeNull();
+
+    const nextUnreadAt = new Date(NOW.getTime() - 70 * 60 * 1000);
+    await seed(nextUnreadAt, podId);
+    await foldExpiredDigests(THRESHOLD);
+    const nextState = await AgentDigestState.findOne(scope).lean();
+    expect(nextState.unreadCount).toBe(1);
+    expect(nextState.unreadSince).toEqual(nextUnreadAt);
+  });
+
+  it('attaches the folded summary to a real wake and clears it on that wake ack', async () => {
+    const podId = new mongoose.Types.ObjectId();
+    const scope = { agentName: 'flint', instanceId: 'default', podId };
+    await seed(new Date(NOW.getTime() - 2 * 60 * 60 * 1000), podId);
+    await foldExpiredDigests(THRESHOLD);
+    const wake = await AgentEvent.create({ ...scope, type: 'heartbeat',
+      status: 'delivered', deliveryMode: 'wake', attempts: 1, deliveryNonce: 'nonce-1' });
+
+    expect(await AgentEventService.attachFoldedDigest(wake)).toMatchObject({
+      unreadCount: 1, conversations: [{ label: 'pod conversation', count: 1 }],
+    });
+    expect((await AgentEvent.findById(wake._id).lean()).digestRollupRevision).toBe(1);
+    await AgentEventService.acknowledge(wake._id, 'flint', 'default', { outcome: 'no_action' }, 'nonce-1', 'cli');
+    expect(await readFoldedDigest(scope)).toBeNull();
+  });
+});