Share digest frame formatter across CLI wake consumers

cli/__tests__/lib.test.mjs

@@ -542,7 +542,7 @@ describe('poller.js', () => {
     const digest = {
       ...makeEvent('digest-1'), deliveryMode: 'digest',
       type: 'message.posted', payload: {
-        content: 'private ambient body', username: 'Ava', messageId: '42', threadRootId: 21,
+        content: 'private ambient body', username: 'Ava', messageId: '42', threadRootId: '21',
         deliveryId: 'digest-nonce',
       },
     };

cli/__tests__/run-loop.test.mjs

@@ -309,7 +309,7 @@ describe('performRun', () => {
           conversations: [{ label: 'thread 40', count: 3 }] },
       } }),
       makeEvent({ _id: 'evt-digest', type: 'message.posted', deliveryMode: 'digest',
-        payload: { content: 'daily update', username: 'alice', messageId: '42', threadRootId: 40 } }),
+        payload: { content: 'daily update', username: 'alice', messageId: '42', threadRootId: '40' } }),
     ];
     const get = jest.fn(async (route) => {
       if (route.endsWith('/messages')) return { messages: [] };

cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@commonlyai/cli",
-  "version": "0.1.154",
+  "version": "0.1.155",
   "license": "Apache-2.0",
   "description": "The Commonly CLI — connect agents, manage pods, iterate fast",
   "type": "module",

cli/src/commands/agent.js

@@ -27,6 +27,7 @@ import {
 } from '../lib/spawn-credential.js';
 import { getToken, resolveInstanceUrl } from '../lib/config.js';
 import { startPoller, terminalDeliveryAckError } from '../lib/poller.js';
+import { formatDigestFrame } from '../lib/digest-frame.js';
 import { startWebhookServer, forwardToLocalWebhook } from '../lib/webhook-server.js';
 import { getAdapter, listAdapterNames } from '../lib/adapters/index.js';
 import {
@@ -2027,42 +2028,13 @@ export const performRun = ({
       return `${prefix}\n${claimContext}${body}`;
     });
     const rollup = wakeEntries.find((entry) => entry.event.payload?.digestRollup)?.event.payload?.digestRollup;
-    const conversationCounts = new Map();
-    let knownOlder = 0;
-    for (const conversation of Array.isArray(rollup?.conversations) ? rollup.conversations : []) {
-      if (typeof conversation.label !== 'string' || !Number.isInteger(conversation.count)
-        || conversation.count < 1) continue;
-      conversationCounts.set(conversation.label,
-        (conversationCounts.get(conversation.label) || 0) + conversation.count);
-      knownOlder += conversation.count;
-    }
-    if (rollup?.unreadCount > knownOlder) {
-      conversationCounts.set('older pod activity', rollup.unreadCount - knownOlder);
-    }
-    const seenDigestSources = new Set();
-    for (const entry of digestEntries) {
-      const { threadRootId, taskId, taskKind, messageId, content } = entry.event.payload || {};
-      const sourceId = messageId != null ? `message:${messageId}`
-        : (typeof taskId === 'string' ? `task:${taskId}:${taskKind || ''}:${content || ''}` : null);
-      if (sourceId && seenDigestSources.has(sourceId)) continue;
-      if (sourceId) seenDigestSources.add(sourceId);
-      const label = Number.isInteger(Number(threadRootId)) && Number(threadRootId) > 0
-        ? `thread ${threadRootId}`
-        : (/^TASK-\d+$/.test(String(taskId || '')) ? taskId : 'pod conversation');
-      conversationCounts.set(label, (conversationCounts.get(label) || 0) + 1);
-    }
-    const digestLines = [...conversationCounts].slice(0, 8)
-      .map(([label, count]) => `- ${label}: ${count} unread`);
-    const digestCount = [...conversationCounts.values()].reduce((sum, count) => sum + count, 0);
+    const digestFrame = formatDigestFrame(digestEntries.map((entry) => entry.event), rollup);
     const omitted = Math.max(0, inboxCount - entries.length);
     return [
       `[Inbox batch: ${inboxCount} new event${inboxCount === 1 ? '' : 's'} in this pod. `
         + 'This is one turn: read the included items together, then decide what needs YOU.]',
       ...bodies,
-      ...(digestEntries.length > 0 || rollup?.unreadCount > 0 ? [
-        `[Since you last looked: ${digestCount} ambient update${digestCount === 1 ? '' : 's'} in ${conversationCounts.size} conversation${conversationCounts.size === 1 ? '' : 's'}; context only, no reply needed.]
-${digestLines.join('\n')}${conversationCounts.size > 8 ? `\n- ${conversationCounts.size - 8} more conversations omitted` : ''}\nPull relevant messages with commonly_get_messages. On resume, read oldest unread first; at most 8 conversations, 5 messages each, 20 messages total.`,
-      ] : []),
+      ...(digestFrame ? [digestFrame] : []),
       ...(omitted > 0
         ? [`[${omitted} more event${omitted === 1 ? '' : 's'} arrived. Their count is real; pull pod context if needed.]`]
         : []),

cli/src/lib/digest-frame.js

@@ -0,0 +1,42 @@
+// One counts-only frame for both the wrapper run loop and agent connect.
+// Ambient bodies stay in the inbox; waking turns receive conversation counts.
+export const formatDigestFrame = (events, rollup) => {
+  if (events.length === 0 && !(rollup?.unreadCount > 0)) return null;
+
+  const counts = new Map();
+  let knownOlder = 0;
+  for (const conversation of Array.isArray(rollup?.conversations) ? rollup.conversations : []) {
+    if (typeof conversation.label !== 'string' || !Number.isInteger(conversation.count)
+      || conversation.count < 1) continue;
+    counts.set(conversation.label, (counts.get(conversation.label) || 0) + conversation.count);
+    knownOlder += conversation.count;
+  }
+  if (rollup?.unreadCount > knownOlder) {
+    counts.set('older pod activity',
+      (counts.get('older pod activity') || 0) + rollup.unreadCount - knownOlder);
+  }
+
+  const seen = new Set();
+  for (const event of events) {
+    const payload = event.payload || {};
+    const sourceId = payload.messageId != null ? `message:${payload.messageId}`
+      : (typeof payload.taskId === 'string'
+        ? `task:${payload.taskId}:${String(payload.taskKind || '')}:${String(payload.content || '')}` : null);
+    if (sourceId && seen.has(sourceId)) continue;
+    if (sourceId) seen.add(sourceId);
+
+    const thread = payload.threadRootId;
+    const task = payload.taskId;
+    const validThread = (typeof thread === 'number' && Number.isInteger(thread) && thread > 0)
+      || (typeof thread === 'string' && /^[1-9]\d*$/.test(thread));
+    const label = validThread ? `thread ${thread}`
+      : (typeof task === 'string' && /^TASK-\d+$/.test(task) ? task : 'pod conversation');
+    counts.set(label, (counts.get(label) || 0) + 1);
+  }
+
+  const lines = [...counts].slice(0, 8).map(([label, count]) => `- ${label}: ${count} unread`);
+  if (counts.size > 8) lines.push(`- ${counts.size - 8} more conversations omitted`);
+  const total = [...counts.values()].reduce((sum, count) => sum + count, 0);
+  return `[Since you last looked: ${total} ambient update${total === 1 ? '' : 's'} in ${counts.size} conversation${counts.size === 1 ? '' : 's'}; context only, no reply needed.]\n`
+    + `${lines.join('\n')}\nPull relevant messages with commonly_get_messages. On resume, read oldest unread first; at most 8 conversations, 5 messages each, 20 messages total.`;
+};

cli/src/lib/poller.js

@@ -7,6 +7,7 @@
  */
 
 import { createClient } from './api.js';
+import { formatDigestFrame } from './digest-frame.js';
 
 // Delivery failures that must stop a consumer rather than fall through to the
 // normal at-least-once retry path. A stale delivery belongs to a replacement
@@ -28,40 +29,6 @@ export const terminalDeliveryAckError = (ackErr, eventId, runner = 'poller') =>
   return null;
 };
 
-const digestFrame = (events, rollup) => {
-  const counts = new Map();
-  let knownOlder = 0;
-  for (const conversation of Array.isArray(rollup?.conversations) ? rollup.conversations : []) {
-    if (typeof conversation.label !== 'string' || !Number.isInteger(conversation.count)
-      || conversation.count < 1) continue;
-    counts.set(conversation.label, (counts.get(conversation.label) || 0) + conversation.count);
-    knownOlder += conversation.count;
-  }
-  if (rollup?.unreadCount > knownOlder) {
-    counts.set('older pod activity',
-      (counts.get('older pod activity') || 0) + rollup.unreadCount - knownOlder);
-  }
-  const seen = new Set();
-  for (const event of events) {
-    const payload = event.payload || {};
-    const sourceId = payload.messageId != null ? `message:${payload.messageId}`
-      : (typeof payload.taskId === 'string'
-        ? `task:${payload.taskId}:${String(payload.taskKind || '')}:${String(payload.content || '')}` : null);
-    if (sourceId && seen.has(sourceId)) continue;
-    if (sourceId) seen.add(sourceId);
-    const thread = payload.threadRootId;
-    const task = payload.taskId;
-    const label = Number.isInteger(thread) && thread > 0 ? `thread ${thread}`
-      : (typeof task === 'string' && /^TASK-\d+$/.test(task) ? task : 'pod conversation');
-    counts.set(label, (counts.get(label) || 0) + 1);
-  }
-  const lines = [...counts].slice(0, 8).map(([label, count]) => `- ${label}: ${count} unread`);
-  if (counts.size > 8) lines.push(`- ${counts.size - 8} more conversations omitted`);
-  const total = [...counts.values()].reduce((sum, count) => sum + count, 0);
-  return `[Since you last looked: ${total} ambient update${total === 1 ? '' : 's'} in ${counts.size} conversation${counts.size === 1 ? '' : 's'}; context only, no reply needed.]\n`
-    + `${lines.join('\n')}\nPull relevant messages with commonly_get_messages. On resume, read oldest unread first; at most 8 conversations, 5 messages each, 20 messages total.`;
-};
-
 export const startPoller = ({
   instanceUrl,
   token,
@@ -97,8 +64,7 @@ export const startPoller = ({
       const digestEvents = events.filter((event) => event.deliveryMode === 'digest'
         && wakeEvents[0] && String(event.podId) === String(wakeEvents[0].podId));
       const rollup = wakeEvents[0]?.payload?.digestRollup;
-      const frame = digestEvents.length > 0 || rollup?.unreadCount > 0
-        ? digestFrame(digestEvents, rollup) : null;
+      const frame = formatDigestFrame(digestEvents, rollup);
       let digestConsumed = false;
       for (const [index, event] of wakeEvents.entries()) {
         const attachedDigest = index === 0 && frame !== null;