Skip to content

fix(wake): suppress read-before-flush empty digests (#17658) #17659

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tobiu merged 1 commit into from
Aug 23, 2026
Merged

fix(wake): suppress read-before-flush empty digests (#17658) #17659

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 88 additions & 42 deletions ai/services/memory-core/CoalescingEngineService.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -440,18 +440,33 @@ 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
// refractory may be claimed.
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());
Expand Down Expand Up @@ -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<Object>}} 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
Expand All @@ -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<Object>} [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},
Expand All @@ -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++;
Expand All @@ -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;
}
Expand Down
91 changes: 86 additions & 5 deletions test/playwright/unit/ai/services/memory-core/CoalescingEngineService.spec.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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 () => {
Expand Down
Loading