diff --git a/cli/__tests__/lib.test.mjs b/cli/__tests__/lib.test.mjs index 740cde2fa3..997016316d 100644 --- a/cli/__tests__/lib.test.mjs +++ b/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', }, }; diff --git a/cli/__tests__/run-loop.test.mjs b/cli/__tests__/run-loop.test.mjs index 27e06b7bf2..9753b978ba 100644 --- a/cli/__tests__/run-loop.test.mjs +++ b/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: [] }; diff --git a/cli/package-lock.json b/cli/package-lock.json index 65aca4a23d..7ed20a78f4 100644 --- a/cli/package-lock.json +++ b/cli/package-lock.json @@ -1,12 +1,12 @@ { "name": "@commonlyai/cli", - "version": "0.1.154", + "version": "0.1.155", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@commonlyai/cli", - "version": "0.1.154", + "version": "0.1.155", "license": "Apache-2.0", "dependencies": { "commander": "^12.0.0" diff --git a/cli/package.json b/cli/package.json index ae0b8ab562..2145947183 100644 --- a/cli/package.json +++ b/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", diff --git a/cli/src/commands/agent.js b/cli/src/commands/agent.js index cc9fe42781..8e9110a533 100644 --- a/cli/src/commands/agent.js +++ b/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.]`] : []), diff --git a/cli/src/lib/digest-frame.js b/cli/src/lib/digest-frame.js new file mode 100644 index 0000000000..135ad05bac --- /dev/null +++ b/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.`; +}; diff --git a/cli/src/lib/poller.js b/cli/src/lib/poller.js index 90e19d2f56..b7efe9a464 100644 --- a/cli/src/lib/poller.js +++ b/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;