diff --git a/ai/services/memory-core/CoalescingEngineService.mjs b/ai/services/memory-core/CoalescingEngineService.mjs index f10c7b397f..c10ead03cc 100644 --- a/ai/services/memory-core/CoalescingEngineService.mjs +++ b/ai/services/memory-core/CoalescingEngineService.mjs @@ -440,9 +440,9 @@ class CoalescingEngineService extends Base { // replayed lane-claim is the highest-cost shape, since it points one peer at another over // territory both already hold correctly. Mailbox state stays untouched — old unread mail // remains listable; it simply cannot manufacture live interruption urgency. - const surviving = this._partitionExpiredMessageWakes(subscription, queue); + const ageAdmitted = this._partitionExpiredMessageWakes(subscription, queue); - if (surviving.length === 0) { + if (ageAdmitted.length === 0) { // Every queued message event was past the admission horizon (or carried no verifiable // `sentAt` — fail-closed, daemon-symmetric): consume the queue WITHOUT dispatching a // zero-event wake. No `lastFlushAtBySub` arming — nothing was delivered, so no @@ -450,8 +450,23 @@ class CoalescingEngineService extends Base { return; } - const digest = this._buildDigestEnvelope(subscription, surviving, firstQueuedAt); - const dispatchPromise = this._dispatchDigest(subscription, digest) + // Reconcile read-state before the digest exists, for the same reason freshness lives above: + // every outer field and rendered bucket must describe ONE admitted event set. The prior + // implementation skipped committed-read messages only while incrementing buckets, then + // derived `totalEvents`, source ids, digest identity, and the tail `logId` from the original + // queue — producing a turn-priced `[WAKE] … 1 events` header with zero detail when a message + // was read inside the coalescing window. + const {events: surviving, unopenableEvents} = this._reconcileMessageWakeReadState(subscription, ageAdmitted); + + if (surviving.length === 0) { + // A committed read is user authority, not a delivery failure. Consume the queue without + // dispatch/retry/refractory; the durable mailbox state is already the receipt. + return; + } + + const + digest = this._buildDigestEnvelope(subscription, surviving, firstQueuedAt, {unopenableEvents}), + dispatchPromise = this._dispatchDigest(subscription, digest) .then(outcome => { if (outcome === 'delivered') { this.lastFlushAtBySub.set(subscriptionId, Date.now()); @@ -504,6 +519,70 @@ class CoalescingEngineService extends Base { return events.filter(event => !suppressedSet.has(event)) } + /** + * @summary Reconciles queued MESSAGE wakes against committed recipient read-state before digest construction. + * + * Four outcomes stay deliberately distinct: + * + * - `{readAt}` — committed user action; remove the event from every digest field. + * - `{missing: true}` — a positive no-row finding; keep the queued fact/count but disqualify it + * from `latest`, preserving the existing openability contract. + * - `{}` / no resolver — graph state is UNKNOWN; keep the event fail-safe. + * - resolver throw — keep the event and warn; a visible extra wake is safer than a missing one. + * + * The pass is synchronous and its result feeds `_buildDigestEnvelope` immediately, so read-state + * cannot be consulted once for aggregate identity and again for rendered buckets. Suppression + * telemetry names only count + recipient/subscription; mailbox content never enters the log. + * + * @protected + * @param {Object} subscription Owning cached subscription. + * @param {Object[]} events Age-admitted queued wake envelopes. + * @returns {{events: Object[], unopenableEvents: Set}} One surviving set plus the + * positively-missing events that may count but cannot become a `latest` pointer. + */ + _reconcileMessageWakeReadState(subscription, events) { + const unopenableEvents = new Set(); + + if (!this.resolveDeliveryReadState || !subscription.agentIdentity) { + return {events, unopenableEvents} + } + + const suppressed = new Set(); + + for (const event of events) { + if (event?.eventType !== 'wake/sent_to_me') continue; + + const messageId = event.payload?.messageId; + + if (!messageId) continue; + + let state = null; + + try { + state = this.resolveDeliveryReadState(messageId, subscription.agentIdentity) + } catch (error) { + logger.warn?.(`[CoalescingEngine] read-state lookup failed for ${messageId}; rendering as unread: ${error.message}`); + continue + } + + if (state?.readAt) { + suppressed.add(event) + } else if (state?.missing === true) { + unopenableEvents.add(event) + } + } + + if (suppressed.size === 0) return {events, unopenableEvents}; + + logger.info(`[CoalescingEngine] Suppressed ${suppressed.size} already-read message wake event(s) for ` + + `${subscription.agentIdentity || subscription.id} at flush.`); + + return { + events: events.filter(event => !suppressed.has(event)), + unopenableEvents + } + } + /** * Builds the digest envelope per ADR §6.4.2. The structured payload reports counts * per trigger type plus the latest-of-each for context. Wraps in the standard @@ -526,9 +605,12 @@ class CoalescingEngineService extends Base { * @param {Object} subscription * @param {Object[]} events Queued event envelopes * @param {Number} firstQueuedAt Epoch ms when the first event in this window was enqueued + * @param {Object} [options] + * @param {Set} [options.unopenableEvents] Positively-missing MESSAGE events which count + * but cannot become the `latest` pointer. * @returns {Object} Digest envelope */ - _buildDigestEnvelope(subscription, events, firstQueuedAt) { + _buildDigestEnvelope(subscription, events, firstQueuedAt, {unopenableEvents = new Set()} = {}) { const breakdown = { sent_to_me : {count: 0, latest: null, latestTs: null, highestPriority: 'normal'}, task_state_changed: {count: 0, latest: null, latestTs: null}, @@ -550,42 +632,6 @@ class CoalescingEngineService extends Base { continue } - // Read-state reconciliation, `sent_to_me` only — the other buckets carry no mailbox row - // and therefore have no read-state to reconcile against. - // - // FAIL-SAFE at every branch. No resolver, no messageId, a resolver that throws, or a - // resolver returning `{}` (graph unavailable) all mean UNKNOWN, and unknown renders the - // event exactly as before. Only a committed `readAt` suppresses one. Suppressing on - // uncertainty would turn a mislabelled count into a missing wake, and a missing wake is - // visible to nobody. - let unopenable = false; - - if (bucketKey === 'sent_to_me' && this.resolveDeliveryReadState) { - const messageId = evt.payload?.messageId; - - if (messageId && subscription.agentIdentity) { - let state = null; - - try { - state = this.resolveDeliveryReadState(messageId, subscription.agentIdentity) - } catch (error) { - logger.warn?.(`[CoalescingEngine] read-state lookup failed for ${messageId}; rendering as unread: ${error.message}`) - } - - if (state?.readAt) { - continue - } - - // `missing` is NOT `{}`. The resolver reports it only after establishing that no - // MESSAGE row exists — a positive finding, not an absence of information. The - // event still counts (something was queued, and hiding that is the suppression - // failure mode), but it must never become `latest`: a `latest` is a pointer the - // recipient is invited to open, and naming one that cannot be opened sends them - // hunting for a message that is not there. That is AC-6. - unopenable = state?.missing === true - } - } - const bucket = breakdown[bucketKey]; bucket.count++; @@ -594,7 +640,7 @@ class CoalescingEngineService extends Base { // Recency wins over position; a timestamp-less candidate keeps last-write-wins. An // unopenable event is disqualified from the pointer only — it has already been counted. - if (!unopenable && (bucket.latest === null || ts === null || ts >= bucket.latestTs)) { + if (!unopenableEvents.has(evt) && (bucket.latest === null || ts === null || ts >= bucket.latestTs)) { bucket.latest = evt.payload; bucket.latestTs = ts ?? bucket.latestTs; } diff --git a/test/playwright/unit/ai/services/memory-core/CoalescingEngineService.spec.mjs b/test/playwright/unit/ai/services/memory-core/CoalescingEngineService.spec.mjs index 9c0e8eea94..c06e31ef04 100644 --- a/test/playwright/unit/ai/services/memory-core/CoalescingEngineService.spec.mjs +++ b/test/playwright/unit/ai/services/memory-core/CoalescingEngineService.spec.mjs @@ -510,7 +510,16 @@ test.describe('CoalescingEngineService', () => { // replaces, because a wrong number is visible in the wake and a missing wake is visible to nobody. // ----------------------------------------------------------------------------- - test('an already-read message is excluded from the count and cannot be named latest', async () => { + /** + * @summary A committed read before flush removes the event from every digest field. + * + * The original reconciliation skipped read messages while populating the breakdown, then built + * `totalEvents`, source ids, digest identity, and the tail `logId` from the unfiltered queue. A + * read HIGH tail therefore produced a normal-priority header claiming one extra event, with no + * corresponding detail when it was the only item. This mixed arm pins the one-surviving-set + * contract across every aggregate, not only the bucket that already worked. + */ + test('an already-read HIGH tail is excluded from every digest field and cannot be named latest', async () => { CoalescingEngineService.configure({ coalesceWindowSeconds : 30, flushRefractorySeconds: 120, @@ -519,16 +528,88 @@ test.describe('CoalescingEngineService', () => { resolveDeliveryReadState: messageId => messageId === 'M-READ' ? {readAt: '2026-08-10T10:00:00.000Z'} : {} }); - const sub = buildSubscription({harnessTargetMetadata: {url: 'https://example.com/wake', coalesceWindow: 0.05}}); - CoalescingEngineService.enqueue(sub, buildEnvelope('wake/sent_to_me', {messageId: 'M-READ', subject: 'handled hours ago'}, 10)); - CoalescingEngineService.enqueue(sub, buildEnvelope('wake/sent_to_me', {messageId: 'M-UNREAD', subject: 'actually new'}, 11)); + const + sub = buildSubscription({harnessTargetMetadata: {url: 'https://example.com/wake', coalesceWindow: 0.05}}), + unread = buildEnvelope('wake/sent_to_me', { + messageId: 'M-UNREAD', + subject : 'actually new', + priority : 'normal' + }, 10), + read = buildEnvelope('wake/sent_to_me', { + messageId: 'M-READ', + subject : 'handled before flush', + priority : 'high' + }, 11); + + CoalescingEngineService.enqueue(sub, unread); + CoalescingEngineService.enqueue(sub, read); await new Promise(resolve => setTimeout(resolve, 100)); - const digest = deliverCalls[0].eventData; + const + digest = deliverCalls[0].eventData, + expected = CoalescingEngineService._buildDigestEnvelope(sub, [unread], Date.now()); + expect(digest.payload.totalEvents).toBe(1); + expect(digest.payload.sourceEventIds).toEqual(['M-UNREAD']); + expect(digest.eventId, 'digest identity is derived from the admitted source ids').toBe(expected.eventId); + expect(digest.logId, 'the read queue tail cannot remain the delivery cursor').toBe(10); expect(digest.payload.breakdown.sent_to_me.count).toBe(1); expect(digest.payload.breakdown.sent_to_me.latest.messageId).toBe('M-UNREAD'); + expect(digest.payload.breakdown.sent_to_me.highestPriority, + 'a read HIGH message cannot shape interruption priority').toBe('normal') + }); + + /** + * @summary An all-read queue is consumed without dispatch, retry, or refractory state. + * + * This is the exact live regression: the direct message was read inside the 150-second window, + * yet the outer queue length still produced `[WAKE] … 1 events` while every rendered bucket was + * zero. Suppression telemetry is deliberately bounded to count + recipient; logging the subject + * or body would turn a wake-safety repair into a content leak. + */ + test('an all-read queue is consumed without a header-only wake or refractory claim', async () => { + CoalescingEngineService.configure({ + coalesceWindowSeconds : 30, + flushRefractorySeconds: 120, + flushHardCapSeconds : 300 + }, { + resolveDeliveryReadState: () => ({readAt: '2026-08-23T22:53:02.704Z'}) + }); + + const + logger = (await import('../../../../../../ai/mcp/server/memory-core/logger.mjs')).default, + originalInfo = logger.info, + infoCalls = [], + sub = buildSubscription({harnessTargetMetadata: {url: 'https://example.com/wake', coalesceWindow: 0.05}}); + + logger.info = message => infoCalls.push(String(message)); + + try { + CoalescingEngineService.enqueue(sub, buildEnvelope('wake/sent_to_me', { + messageId: 'M-READ', + subject : 'must never enter suppression telemetry', + body : 'nor may the body' + }, 10)); + + await new Promise(resolve => setTimeout(resolve, 100)); + } finally { + logger.info = originalInfo; + } + + expect(deliverCalls, 'a consumed message cannot price a harness turn').toEqual([]); + expect(CoalescingEngineService.coalesceState.size, 'the queue is consumed, not re-armed').toBe(0); + expect(CoalescingEngineService.lastFlushAtBySub.has(sub.id), + 'nothing delivered, so no refractory may be claimed').toBe(false); + expect(CoalescingEngineService.dispatchInFlight.has(sub.id), + 'no delivery means no retry-bearing dispatch promise').toBe(false); + + const telemetry = infoCalls.join('\n'); + + expect(telemetry).toContain('Suppressed 1 already-read message wake event(s) for @alice at flush'); + expect(telemetry).not.toContain('M-READ'); + expect(telemetry).not.toContain('must never enter suppression telemetry'); + expect(telemetry).not.toContain('nor may the body') }); test('NON-VACUITY — the same two events both count when neither is read', async () => {