From 680e075bd39e0fbadd650dfadaf113a5541a8214 Mon Sep 17 00:00:00 2001 From: Meg-2646 Date: Fri, 16 Oct 2026 16:18:34 -0400 Subject: [PATCH 1/4] fix(mcpl): withdraw a turn whose deferred batches give it nothing to read The framework withdraws a turn woken only by deferred coalescing batches, before inference, when every batch that caused it settled in the assembly and the assembly materialized nothing (the `turn-withdrawn` trace). assemble() counted a subject as settled only when it waited on the subject's render. So at four early returns of freeze() the turn went on and called the model, although nothing new had arrived for the agent: - freeze() takes the batch and finds its authority revoked; - the render-start commit fails, so the batch is kept, marked to take its fallback, for the recovery wake; - the authority check after that commit fails; - the agent is no longer in the batch's audience, although the batch woke it. freeze() now answers 'declined' at each, and assemble() counts that subject as settled with nothing materialized, so the withdrawal applies as it does to a batch whose render came back empty. A subject that moved on during the audience lookup, or that another assembly is rendering, still answers undefined and keeps the turn. The withdrawal's stderr line said "deferred render produced nothing", which would be untrue in these cases, since no render runs in them. It now says "its deferred batches gave this turn nothing to read". The persist-failed and revoked traces still name the cause. The second case also hung R16 under CPU load. The fixture queues one model response when it starts, and R16's first drain ran the turn whose commit failed, which took that response. When the 50 ms recovery timer fired inside the same drain, the recovery turn's model call found no response queued and waited for one that the test would push only after the drain returned. R16 now answers every model call and drives the drains until the batch renders. It then checks that there was one model call and that it carried the fallback, that each of persist-failed, turn-withdrawn, recovery-wake and rendered was traced once, with persist-failed first and rendered last, and that the server was never asked to render the batch. vector 24 and R12 now also check that the revoked batch, their only cause, started no inference, as R11 does for a retraction, and that its turn started and was withdrawn. A new test does the same for a turn whose agent left the batch's audience. With these tests on main's push-coalescer.ts and framework.ts, R16 fails with 2 model calls where it expects 1, and vector 24, R12 and the new test with 1 where they expect 0. Each 'declined' answer put back to undefined on its own fails its own site's test, and dropping assemble()'s count fails all four. Co-Authored-By: Claude Opus 5.5 --- .../coalescing-declined-freeze.fixed.md | 1 + src/framework.ts | 22 ++++---- src/mcpl/push-coalescer.ts | 31 ++++++++--- test/mcpl-coalescing.test.ts | 55 +++++++++++++++++-- 4 files changed, 84 insertions(+), 25 deletions(-) create mode 100644 changelog.d/coalescing-declined-freeze.fixed.md diff --git a/changelog.d/coalescing-declined-freeze.fixed.md b/changelog.d/coalescing-declined-freeze.fixed.md new file mode 100644 index 000000000..fc6f4aa89 --- /dev/null +++ b/changelog.d/coalescing-declined-freeze.fixed.md @@ -0,0 +1 @@ +- A turn woken only by deferred coalescing batches is no longer started when none of them has anything for it to read. It used to run, calling the model although no new input had arrived for it, when the assembly found a batch's authority revoked (before or after its render-start commit), when the render-start commit failed, or when the agent had left the batch's audience since the batch woke it. Such a turn is now withdrawn before inference, as a turn whose batches all rendered nothing already was, and traces `turn-withdrawn`. A batch whose render-start commit failed is still kept, and its recovery wake delivers its fallback once storage recovers. The stderr line for a withdrawn turn now reads `[coalescing] : its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`. It said `deferred render produced nothing`, which is untrue where no render ran. diff --git a/src/framework.ts b/src/framework.ts index 94f1035ff..f7937caba 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11706,19 +11706,21 @@ export class AgentFramework { } } - // A turn whose only causes were deferred batches, all of which rendered - // nothing (empty or blank render, §5.2; cancelled; revoked), has nothing - // new to show: running it would wake the model to `[Continue]` or an - // older message, an uncaused wake. Stop before the checkpoint, locus, - // typing and compile. Any batch this assembly materialized keeps the - // turn, including one that was not among its causes (a batch whose wake - // the freeze withdrew, or one that never qualified for a wake). A cause - // this assembly did not settle (rendered elsewhere, not yet frozen) keeps - // the turn, as before. + // A turn whose only causes were deferred batches, none of which gave it + // anything to read (an empty or blank render, §5.2; cancelled; revoked; a + // render-start that could not be made durable, whose batch then waits for + // its recovery wake; a batch whose audience doesn't include the agent), + // has nothing new to show: running it would wake the model to + // `[Continue]` or an older message, an uncaused wake. Stop before the + // checkpoint, locus, typing and compile. Any batch this assembly + // materialized keeps the turn, including one that was not among its + // causes (a batch whose wake the freeze withdrew, or one that never + // qualified for a wake). A cause this assembly did not settle (rendered + // elsewhere, not yet frozen) keeps the turn, as before. const batchSubjects = trigger?.coalescingBatchSubjects; if (assembly && batchSubjects?.length && assembly.materialized.size === 0 && batchSubjects.every((s) => assembly.settled.has(s))) { - console.error(`[coalescing] ${agent.name}: deferred render produced nothing; turn not started (no wake cause left)`); + console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`); this.emitTrace({ type: 'mcpl:coalescing', kind: 'turn-withdrawn', agentName: agent.name, subjects: batchSubjects }); // Release the turn and land anything deferred while assembly awaited // the render, as a torn-down turn would. diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 7615256b1..e82696785 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -129,11 +129,16 @@ interface Rendering { /** What an assembly did, per subject, for the turn being assembled. */ export interface AssemblyResult { - /** Subjects whose render this assembly waited on to completion. */ + /** Subjects this assembly finished with: it waited on their render to + * completion, took their batch and started no render, or found the agent + * outside their batch's audience. */ settled: Set; /** Of those, the subjects whose render or fallback was delivered as content. * A settled subject that is not here gave the turn nothing to read: empty - * or blank render (§5.2), cancelled, or revoked at response. */ + * or blank render (§5.2), cancelled, revoked before the render or at its + * response, a render-start that could not be made durable (the batch then + * waits for its recovery wake), or a batch whose audience doesn't include + * the agent. */ materialized: Set; } @@ -667,7 +672,8 @@ export class PushCoalescer { // be flattened by `then`, holding the lock until the render settled — // and the settlement itself needs the lock. const started = await this.locked(() => this.freeze(subject, agentName)); - if (started) { + if (started === 'declined') result.settled.add(subject); + else if (started) { work.push(started.done.then(() => { result.settled.add(subject); if (started.rendering.materialized) result.materialized.add(subject); @@ -683,9 +689,12 @@ export class PushCoalescer { * Under the lock: decide whether `subject`'s batch renders for `agentName`, * freeze it, make the render-start durable, re-check authority, and start * the RPC. Returns the render's completion (shared by every assembly that - * waits on this batch, vector 25) or undefined when nothing was started. + * waits on this batch, vector 25); 'declined' when the turn being assembled + * gets nothing of the batch, because the agent isn't in its audience or + * because this assembly took the batch and started no render; or undefined + * when the batch was not this assembly's to decide on. */ - private async freeze(subject: string, agentName: string): Promise<{ done: Promise; rendering: Rendering } | undefined> { + private async freeze(subject: string, agentName: string): Promise<{ done: Promise; rendering: Rendering } | 'declined' | undefined> { const state = this.subjects.get(subject); if (!state || this.suspended) return undefined; if (state.rendering) { @@ -698,12 +707,16 @@ export class PushCoalescer { // The audience lookup may spawn a fork. Under the lock no admission can // move the subject on meanwhile; the identity check guards the host. const audience = await this.host.audience(batch.latest); - if (!audience.includes(agentName) || state.batch !== batch || state.rendering) return undefined; + if (state.batch !== batch || state.rendering) return undefined; + // Not this agent's batch, or no longer: an agent woken for it can have left + // its audience since. The batch stays for its audience, and this turn has + // nothing of it to show. + if (!audience.includes(agentName)) return 'declined'; state.batch = undefined; this.host.cancelWake(subject); if (!this.host.authorized(batch.latest)) { this.host.audit({ kind: 'revoked', subject, eventId: batch.latest.eventId }); - return undefined; + return 'declined'; } const rendering: Rendering = { batch, cancelled: false, done: Promise.resolve() }; state.rendering = rendering; @@ -721,7 +734,7 @@ export class PushCoalescer { // unrelated traffic happens by. Re-wake with backoff, so a persistent // storage outage does not become a tight loop of model turns. this.scheduleRecovery(subject, state); - return undefined; + return 'declined'; } this.clearRecovery(subject); if (this.suspended) { state.rendering = undefined; state.batch = batch; return undefined; } @@ -731,7 +744,7 @@ export class PushCoalescer { if (!this.host.authorized(batch.latest)) { this.host.audit({ kind: 'revoked', subject, eventId: batch.latest.eventId }); state.rendering = undefined; - return undefined; + return 'declined'; } rendering.done = this.render(subject, state, rendering, agentName); return { done: rendering.done, rendering }; diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index c5bac52d1..812d84330 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -340,9 +340,13 @@ test('vector 24: revoked authority before assembly drops the batch without a ren const f = await fixture(); t.after(f.close); await f.send('push/event', f.params('1', 'previously_permitted', { deferred: true })); await f.framework.stop(); await f.create(false); + let withdrawn = 0; + f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); await f.framework.runUntilIdle(); assert.equal(f.renders.length, 0); assert(!f.context().includes('previously_permitted')); + assert.equal(f.membrane.calls.length, 0, 'and the revoked batch, the only cause, started no inference'); + assert.equal(withdrawn, 1, 'its turn started and was withdrawn'); }); test('vector 27i: a deferred notice displaces an unread plain occurrence', async (t) => { @@ -855,6 +859,8 @@ test('R11: a retraction during the render-start commit wait is final; the batch test('R12: endpoint reassignment during the render-start commit wait never sends the batch to the new peer', async (t) => { const f = await fixture(); t.after(f.close); await f.send('push/event', f.params('1', 'old_fallback', { deferred: true, data: { secret: 'private_notice_data' } })); + let withdrawn = 0; + f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); const held = holdNextCommit(f); const run = f.framework.runUntilIdle(); const release = await held; @@ -866,6 +872,8 @@ test('R12: endpoint reassignment during the render-start commit wait never sends assert.equal(f.renders.length, 0, 'no push/render to the new peer'); assert(!JSON.stringify(f.renders).includes('private_notice_data')); assert(!f.context().includes('old_fallback')); + assert.equal(f.membrane.calls.length, 0, 'and the revoked batch, the only cause, started no inference'); + assert.equal(withdrawn, 1, 'its turn started and was withdrawn'); }); // --------------------------------------------------------------------------- @@ -893,6 +901,24 @@ test('R13: a retraction during the audience lookup is final; the stale batch is assert(!f.context().includes('document_diff')); }); +test('a turn whose agent left the batch\'s audience before the assembly starts no inference', async (t) => { + const f = await fixture(); t.after(f.close); + await f.send('push/event', f.params('1', 'moved_fallback', { deferred: true, initial: true })); + // The admission's wake has looked the audience up; the next lookup is the + // assembly's, and it no longer names the agent, as when a conversation + // fork's channel is bound to a new fork in between. + const internals = f.framework as unknown as { coalescedAudience: (occ: unknown) => Promise }; + const orig = internals.coalescedAudience.bind(internals); + internals.coalescedAudience = async () => { internals.coalescedAudience = orig; return []; }; + let withdrawn = 0; + f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); + await f.framework.runUntilIdle(); + assert.equal(f.renders.length, 0); + assert(!f.context().includes('moved_fallback')); + assert.equal(f.membrane.calls.length, 0, 'the batch, the only cause, started no inference'); + assert.equal(withdrawn, 1, 'its turn started and was withdrawn'); +}); + test('R14: two consecutive crash windows with an acceptance in between keep both recovered batches', async (t) => { const f = await fixture(); t.after(f.close); await f.send('push/event', f.params('a', 'first_window_fallback', { deferred: true, key: 'A' })); @@ -932,7 +958,7 @@ test('R15: a retraction arriving between the render result and its publication w // Review round 8 (#197): liveness after a failed render-start commit // --------------------------------------------------------------------------- -test('R16: a batch whose render-start commit failed is re-woken once storage recovers', async (t) => { +test('R16: a batch whose render-start commit failed is re-woken once storage recovers', { timeout: 30_000 }, async (t) => { const f = await fixture(); t.after(f.close); const coalescer = (f.framework as unknown as { pushCoalescer: { options: { recoveryBackoffMs?: number }; pendingBatches(): number } }).pushCoalescer; coalescer.options.recoveryBackoffMs = 50; @@ -941,15 +967,32 @@ test('R16: a batch whose render-start commit failed is re-woken once storage rec const orig = store.sync.bind(store); let failing = 0; store.sync = () => { if (failing > 0) { failing--; throw new Error('EIO simulated'); } orig(); }; + const traced: Array<{ kind: string; source?: string }> = []; + f.framework.onTrace((e) => { + if (e.type === 'mcpl:coalescing') traced.push({ kind: (e as { kind: string }).kind, source: (e as { source?: string }).source }); + }); + // Every turn that reaches the model is answered, so a model call nothing + // should have caused is counted rather than left waiting for a response. + f.alwaysRespond(); await f.send('push/event', f.params('1', 'asleep_fallback', { deferred: true, initial: true })); failing = 1; // the render-start commit fails; storage then recovers + // The recovery wake can fire inside the first drain or after it, so drive + // until the batch renders rather than for a fixed number of drains. await f.framework.runUntilIdle(); + while (!traced.some((e) => e.kind === 'rendered')) { + await eventually(() => internals.pendingRequests.length > 0, 'recovery wake'); + await f.framework.runUntilIdle(); + } + assert.equal(f.membrane.calls.length, 1, 'the wake whose render-start commit failed, the only cause, started no inference'); + assert(f.lastRequest().includes('asleep_fallback'), 'the recovery turn delivered the batch'); + const kinds = traced.map((e) => e.kind); + const order = (...steps: string[]) => kinds.filter((k) => steps.includes(k)); + assert.deepEqual(order('persist-failed', 'turn-withdrawn', 'rendered'), ['persist-failed', 'turn-withdrawn', 'rendered'], 'the failed freeze withdrew its turn'); + assert.deepEqual(order('persist-failed', 'recovery-wake', 'rendered'), ['persist-failed', 'recovery-wake', 'rendered'], 're-woken by the recovery timer, without unrelated traffic'); + // The failed commit can't prove that no render started, so the batch takes + // its fallback and the server is never asked to render it. + assert.equal(traced.find((e) => e.kind === 'rendered')?.source, 'fallback'); assert.equal(f.renders.length, 0); - assert.equal(coalescer.pendingBatches(), 1, 'the batch is kept'); - assert(!f.lastRequest().includes('asleep_fallback')); - await eventually(() => internals.pendingRequests.length > 0, 'recovery wake'); - await f.turn(); - assert(f.lastRequest().includes('asleep_fallback'), 'delivered without unrelated traffic'); assert.equal(coalescer.pendingBatches(), 0); }); From 0627bf5df0ddc8207a518a8bb14dd7fb3cce9665 Mon Sep 17 00:00:00 2001 From: Meg-2646 Date: Sat, 17 Oct 2026 00:42:54 -0400 Subject: [PATCH 2/4] fix(mcpl): an agent outside a batch's audience gets one answer at both sites From an internal review of the commit before this one: - freeze()'s shared-render path answered undefined for an agent outside the audience of a batch another assembly is rendering, so the subject stayed unsettled, while the path before a render declines it. It now declines it too; a rendering that changed during the audience lookup still answers undefined. A new test calls assemble() for such an agent while another assembly's render is held, and checks that the subject is settled with nothing materialized. A turn whose only cause is such a batch still runs when no other batch is pending: startAgentStream assembles only when pendingBatches() is non-zero, and that counts batches, not renders. This commit doesn't change that gate. - The audience test's lookup now names another agent rather than none, the test checks that the declined batch stays pending, and its name says what it sets up. - R16's wait loop is bounded and gives timers a turn on each pass, so a regression that keeps re-waking the batch fails the test instead of spinning past its timeout. Its count message names the one call it expects, the recovery turn's. - The doc comments of AssemblyResult and freeze() name the cases each answer covers, the comment at the audience check says that nothing there wakes the batch's current audience, and the changelog fragment says the assembly found the agent outside the audience. With this commit's tests on main's push-coalescer.ts and framework.ts, the new assemble() test fails with its subject unsettled, and the audience test, R16, vector 24 and R12 fail at their model-call counts. With the second audience check put back to undefined, the new test fails, and with the declined batch dropped at the first, the audience test fails. Co-Authored-By: Claude Opus 5.5 --- .../coalescing-declined-freeze.fixed.md | 2 +- src/mcpl/push-coalescer.ts | 27 ++++++++----- test/mcpl-coalescing.test.ts | 40 ++++++++++++++++--- 3 files changed, 52 insertions(+), 17 deletions(-) diff --git a/changelog.d/coalescing-declined-freeze.fixed.md b/changelog.d/coalescing-declined-freeze.fixed.md index fc6f4aa89..13422a34d 100644 --- a/changelog.d/coalescing-declined-freeze.fixed.md +++ b/changelog.d/coalescing-declined-freeze.fixed.md @@ -1 +1 @@ -- A turn woken only by deferred coalescing batches is no longer started when none of them has anything for it to read. It used to run, calling the model although no new input had arrived for it, when the assembly found a batch's authority revoked (before or after its render-start commit), when the render-start commit failed, or when the agent had left the batch's audience since the batch woke it. Such a turn is now withdrawn before inference, as a turn whose batches all rendered nothing already was, and traces `turn-withdrawn`. A batch whose render-start commit failed is still kept, and its recovery wake delivers its fallback once storage recovers. The stderr line for a withdrawn turn now reads `[coalescing] : its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`. It said `deferred render produced nothing`, which is untrue where no render ran. +- A turn woken only by deferred coalescing batches is no longer started when none of them has anything for it to read. It used to run, calling the model although no new input had arrived for it, when the assembly found a batch's authority revoked (before or after its render-start commit), when the render-start commit failed, or when it found that the agent had left the batch's audience since the batch woke it. Such a turn is now withdrawn before inference, as a turn whose batches all rendered nothing already was, and traces `turn-withdrawn`. A batch whose render-start commit failed is still kept, and its recovery wake delivers its fallback once storage recovers. The stderr line for a withdrawn turn now reads `[coalescing] : its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`. It said `deferred render produced nothing`, which is untrue where no render ran. diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index e82696785..ed57365d1 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -130,8 +130,7 @@ interface Rendering { /** What an assembly did, per subject, for the turn being assembled. */ export interface AssemblyResult { /** Subjects this assembly finished with: it waited on their render to - * completion, took their batch and started no render, or found the agent - * outside their batch's audience. */ + * completion, or freeze() declined them for this agent. */ settled: Set; /** Of those, the subjects whose render or fallback was delivered as content. * A settled subject that is not here gave the turn nothing to read: empty @@ -689,18 +688,26 @@ export class PushCoalescer { * Under the lock: decide whether `subject`'s batch renders for `agentName`, * freeze it, make the render-start durable, re-check authority, and start * the RPC. Returns the render's completion (shared by every assembly that - * waits on this batch, vector 25); 'declined' when the turn being assembled - * gets nothing of the batch, because the agent isn't in its audience or - * because this assembly took the batch and started no render; or undefined - * when the batch was not this assembly's to decide on. + * waits on this batch, vector 25). Returns 'declined' when the turn being + * assembled gets nothing of the batch: the agent isn't in its audience, + * whether or not another assembly is rendering it (the batch is left as it + * is); authority was revoked before the render could start (the batch is + * dropped); or the render-start could not be made durable (the batch is + * kept for its recovery wake). Returns undefined when the batch was not + * this assembly's to decide on: there is none, the subject moved on during + * the audience lookup, or the host is stopping. */ private async freeze(subject: string, agentName: string): Promise<{ done: Promise; rendering: Rendering } | 'declined' | undefined> { const state = this.subjects.get(subject); if (!state || this.suspended) return undefined; if (state.rendering) { - // Another assembly froze this batch; share its outcome (vector 25). + // Another assembly froze this batch; share its outcome (vector 25), + // unless this agent isn't in its audience, as below. const rendering = state.rendering; - return (await this.host.audience(rendering.batch.latest)).includes(agentName) && state.rendering === rendering ? { done: rendering.done, rendering } : undefined; + const audience = await this.host.audience(rendering.batch.latest); + if (state.rendering !== rendering) return undefined; + if (!audience.includes(agentName)) return 'declined'; + return { done: rendering.done, rendering }; } const batch = state.batch; if (!batch) return undefined; @@ -709,8 +716,8 @@ export class PushCoalescer { const audience = await this.host.audience(batch.latest); if (state.batch !== batch || state.rendering) return undefined; // Not this agent's batch, or no longer: an agent woken for it can have left - // its audience since. The batch stays for its audience, and this turn has - // nothing of it to show. + // its audience since. This turn has nothing of it to show. The batch stays + // pending; nothing here wakes its current audience. if (!audience.includes(agentName)) return 'declined'; state.batch = undefined; this.host.cancelWake(subject); diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index 812d84330..0295f2383 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -901,15 +901,15 @@ test('R13: a retraction during the audience lookup is final; the stale batch is assert(!f.context().includes('document_diff')); }); -test('a turn whose agent left the batch\'s audience before the assembly starts no inference', async (t) => { +test('a turn whose batch\'s audience no longer names its agent at the assembly starts no inference', async (t) => { const f = await fixture(); t.after(f.close); await f.send('push/event', f.params('1', 'moved_fallback', { deferred: true, initial: true })); // The admission's wake has looked the audience up; the next lookup is the - // assembly's, and it no longer names the agent, as when a conversation + // assembly's, and it names another agent instead, as when a conversation // fork's channel is bound to a new fork in between. - const internals = f.framework as unknown as { coalescedAudience: (occ: unknown) => Promise }; + const internals = f.framework as unknown as { coalescedAudience: (occ: unknown) => Promise; pushCoalescer: { pendingBatches(): number } }; const orig = internals.coalescedAudience.bind(internals); - internals.coalescedAudience = async () => { internals.coalescedAudience = orig; return []; }; + internals.coalescedAudience = async () => { internals.coalescedAudience = orig; return ['conversation-chat-g2']; }; let withdrawn = 0; f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); await f.framework.runUntilIdle(); @@ -917,6 +917,30 @@ test('a turn whose agent left the batch\'s audience before the assembly starts n assert(!f.context().includes('moved_fallback')); assert.equal(f.membrane.calls.length, 0, 'the batch, the only cause, started no inference'); assert.equal(withdrawn, 1, 'its turn started and was withdrawn'); + assert.equal(internals.pushCoalescer.pendingBatches(), 1, 'the batch stays pending'); +}); + +test('an assembly for an agent outside the audience of a batch another assembly is rendering declines it', async (t) => { + const f = await fixture({ agents: [{ name: 'agent', model: 'test', systemPrompt: 'test' }, { name: 'other', model: 'test', systemPrompt: 'test' }] }); t.after(f.close); + let release!: (value: unknown) => void; + let began!: () => void; + const started = new Promise((resolve) => { began = resolve; }); + f.renderer(() => { began(); return new Promise((resolve) => { release = resolve; }); }); + await f.send('push/event', f.params('1', 'shared_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as { coalescedAudience: (occ: unknown) => Promise; pushCoalescer: { assemble(agent: string): Promise<{ settled: Set; materialized: Set }> } }; + // From here on the audience is 'other' alone, as when a conversation fork's + // channel has moved to a new fork. 'other' freezes the batch and its render + // is held; 'agent' then assembles while it's in flight. + internals.coalescedAudience = async () => ['other']; + const rendering = internals.pushCoalescer.assemble('other'); + await started; + const declined = await internals.pushCoalescer.assemble('agent'); + assert.equal(declined.settled.size, 1, 'the subject is settled for the agent outside the audience'); + assert.equal(declined.materialized.size, 0, 'and gave it nothing'); + release({ content: [{ type: 'text', text: 'shared_render' }] }); + const rendered = await rendering; + assert.equal(rendered.settled.size, 1); + assert.equal(rendered.materialized.size, 1, 'the assembly in the audience delivered the render'); }); test('R14: two consecutive crash windows with an acceptance in between keep both recovered batches', async (t) => { @@ -979,11 +1003,15 @@ test('R16: a batch whose render-start commit failed is re-woken once storage rec // The recovery wake can fire inside the first drain or after it, so drive // until the batch renders rather than for a fixed number of drains. await f.framework.runUntilIdle(); - while (!traced.some((e) => e.kind === 'rendered')) { + for (let pass = 0; !traced.some((e) => e.kind === 'rendered'); pass++) { + assert(pass < 5, 'the batch rendered within five recovery wakes'); await eventually(() => internals.pendingRequests.length > 0, 'recovery wake'); await f.framework.runUntilIdle(); + // A macrotask turn, so timers (this test's timeout among them) can fire + // even when both waits above return at once. + await new Promise((resolve) => setImmediate(resolve)); } - assert.equal(f.membrane.calls.length, 1, 'the wake whose render-start commit failed, the only cause, started no inference'); + assert.equal(f.membrane.calls.length, 1, 'one model call, the recovery turn\'s: the wake whose render-start commit failed, the only cause, started none'); assert(f.lastRequest().includes('asleep_fallback'), 'the recovery turn delivered the batch'); const kinds = traced.map((e) => e.kind); const order = (...steps: string[]) => kinds.filter((k) => steps.includes(k)); From bf7adf82f76550780219602608b186501286a03b Mon Sep 17 00:00:00 2001 From: Meg-2646 Date: Sat, 17 Oct 2026 01:14:04 -0400 Subject: [PATCH 3/4] test(mcpl): a coalescing test that would hang fails, naming its wait R16 as main has it can hang on a loaded host (5 of 12 runs beside six busy loops, each stopped at 60 s), and CI runs node --test with no test timeout, so a wait that never ends holds the whole run. Each wait in the tests this change touches that could hang now goes through settles(), beside eventually() in the fixture, which fails with what it was waiting for: R16's drains, the drains of vector 24, R12 and the audience test, where a second model call would wait for a response the fixture never queues, and the new assemble() test's waits on its held render. Every guard is 60 s (HANG_GUARD_MS), far above what these waits take on a heavily loaded host, so load isn't expected to fire it, while a hang still fails within a minute. No assertion rests on its timing. R16's test timeout goes from 30 s to 120 s, above its guards, so a guard names the wait before the timeout ends the test. With R16's alwaysRespond() taken out and main's push-coalescer.ts and framework.ts in place, the shape in which R16 hung on main, R16 now fails in about 61 s, naming the drain it was in ("the first drain" or "the drain after recovery wake 1 did not settle within 60 s"), in each of two runs, and its file ends. With the second audience check made to join the render whatever the audience, the new assemble() test fails in about 6 s at its check that the agent was given nothing, once push/render's own 5 s timeout ends the held render. With this commit's tests on main's source, the same five tests as before fail at their own checks, not at a guard. Co-Authored-By: Claude Opus 5.5 --- test/helpers/coalescing-fixture.ts | 20 ++++++++++++++++++++ test/mcpl-coalescing.test.ts | 25 +++++++++++++++---------- 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/test/helpers/coalescing-fixture.ts b/test/helpers/coalescing-fixture.ts index 4ff0038e2..8c0cb3ad5 100644 --- a/test/helpers/coalescing-fixture.ts +++ b/test/helpers/coalescing-fixture.ts @@ -123,6 +123,26 @@ export async function eventually(predicate: () => boolean, what = 'condition'): while (!predicate()) { assert(performance.now() < deadline, `${what} did not settle`); await new Promise((r) => setTimeout(r, 10)); } } +/** How long settles() lets a wait run: far above what any of these waits + * takes on a heavily loaded host, so load can't fire it, while a hang still + * fails within a minute. */ +export const HANG_GUARD_MS = 60_000; + +/** + * Awaits `work`, failing with `what` if it hasn't settled within + * HANG_GUARD_MS. A hang guard, not a timing check: CI runs without a test + * timeout, so a wait that never ends would hold the whole run, and this turns + * it into a failure that names what was awaited. When it fires, the work it + * waited on keeps running, and the test's close then runs under it. + */ +export async function settles(work: Promise, what: string): Promise { + let timer: ReturnType | undefined; + const late = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${what} did not settle within ${HANG_GUARD_MS / 1000} s`)), HANG_GUARD_MS); + }); + try { return await Promise.race([work, late]); } finally { clearTimeout(timer); } +} + /** Simulate a crash: stop without the snapshot flush a clean stop performs. */ export async function crash(f: Awaited>): Promise { diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index 0295f2383..e112964df 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -13,7 +13,7 @@ import { tmpdir } from 'node:os'; import { join, dirname } from 'node:path'; import { fileURLToPath } from 'node:url'; import { createMockResponse } from './helpers/mock-membrane.js'; -import { fixture, eventually, crash, ok, TS } from './helpers/coalescing-fixture.js'; +import { fixture, eventually, settles, crash, ok, TS } from './helpers/coalescing-fixture.js'; // --------------------------------------------------------------------------- // Plain coalescing, feature-set scope (vectors 1–6, 35) @@ -342,7 +342,8 @@ test('vector 24: revoked authority before assembly drops the batch without a ren await f.framework.stop(); await f.create(false); let withdrawn = 0; f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); - await f.framework.runUntilIdle(); + // A second model call would wait for a response the fixture never queues. + await settles(f.framework.runUntilIdle(), 'the drain after the restart'); assert.equal(f.renders.length, 0); assert(!f.context().includes('previously_permitted')); assert.equal(f.membrane.calls.length, 0, 'and the revoked batch, the only cause, started no inference'); @@ -868,7 +869,8 @@ test('R12: endpoint reassignment during the render-start commit wait never sends await f.framework.restartMcplServer('editor', { ...config, url: `${config.url}/reassigned` } as never); await eventually(() => f.hostCaps.length === 2, 'reconnect to the new endpoint'); release(); - await run; + // A second model call would wait for a response the fixture never queues. + await settles(run, 'the drain after the reassignment'); assert.equal(f.renders.length, 0, 'no push/render to the new peer'); assert(!JSON.stringify(f.renders).includes('private_notice_data')); assert(!f.context().includes('old_fallback')); @@ -912,7 +914,8 @@ test('a turn whose batch\'s audience no longer names its agent at the assembly s internals.coalescedAudience = async () => { internals.coalescedAudience = orig; return ['conversation-chat-g2']; }; let withdrawn = 0; f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); - await f.framework.runUntilIdle(); + // A second model call would wait for a response the fixture never queues. + await settles(f.framework.runUntilIdle(), 'the drain after the wake'); assert.equal(f.renders.length, 0); assert(!f.context().includes('moved_fallback')); assert.equal(f.membrane.calls.length, 0, 'the batch, the only cause, started no inference'); @@ -933,12 +936,14 @@ test('an assembly for an agent outside the audience of a batch another assembly // is held; 'agent' then assembles while it's in flight. internals.coalescedAudience = async () => ['other']; const rendering = internals.pushCoalescer.assemble('other'); - await started; - const declined = await internals.pushCoalescer.assemble('agent'); + await settles(started, "the render for 'other'"); + // Joining the held render instead would wait for a release that comes + // only after this call. + const declined = await settles(internals.pushCoalescer.assemble('agent'), "the assembly for 'agent'"); assert.equal(declined.settled.size, 1, 'the subject is settled for the agent outside the audience'); assert.equal(declined.materialized.size, 0, 'and gave it nothing'); release({ content: [{ type: 'text', text: 'shared_render' }] }); - const rendered = await rendering; + const rendered = await settles(rendering, "the assembly for 'other'"); assert.equal(rendered.settled.size, 1); assert.equal(rendered.materialized.size, 1, 'the assembly in the audience delivered the render'); }); @@ -982,7 +987,7 @@ test('R15: a retraction arriving between the render result and its publication w // Review round 8 (#197): liveness after a failed render-start commit // --------------------------------------------------------------------------- -test('R16: a batch whose render-start commit failed is re-woken once storage recovers', { timeout: 30_000 }, async (t) => { +test('R16: a batch whose render-start commit failed is re-woken once storage recovers', { timeout: 120_000 }, async (t) => { const f = await fixture(); t.after(f.close); const coalescer = (f.framework as unknown as { pushCoalescer: { options: { recoveryBackoffMs?: number }; pendingBatches(): number } }).pushCoalescer; coalescer.options.recoveryBackoffMs = 50; @@ -1002,11 +1007,11 @@ test('R16: a batch whose render-start commit failed is re-woken once storage rec failing = 1; // the render-start commit fails; storage then recovers // The recovery wake can fire inside the first drain or after it, so drive // until the batch renders rather than for a fixed number of drains. - await f.framework.runUntilIdle(); + await settles(f.framework.runUntilIdle(), 'the first drain'); for (let pass = 0; !traced.some((e) => e.kind === 'rendered'); pass++) { assert(pass < 5, 'the batch rendered within five recovery wakes'); await eventually(() => internals.pendingRequests.length > 0, 'recovery wake'); - await f.framework.runUntilIdle(); + await settles(f.framework.runUntilIdle(), `the drain after recovery wake ${pass + 1}`); // A macrotask turn, so timers (this test's timeout among them) can fire // even when both waits above return at once. await new Promise((resolve) => setImmediate(resolve)); From 1f1a4e118c1c6c7048441abeb156dc5a537b4512 Mon Sep 17 00:00:00 2001 From: Meg-2646 Date: Sun, 18 Oct 2026 02:03:16 -0400 Subject: [PATCH 4/4] test(mcpl): settles() waits the fixture's REPLY_DEADLINE_MS #427 gave the coalescing fixture REPLY_DEADLINE_MS, one 60 s bound for every wait it runs, and this branch had added HANG_GUARD_MS, a second 60 s bound for settles(). This PR's description said whichever of the two PRs merged second would take the other's constant and drop its own, so the fixture keeps one. #427 merged first, so settles() now waits REPLY_DEADLINE_MS, HANG_GUARD_MS is gone, and the constant's doc lists settles() among the waits it bounds. The bound is the same 60 s. With R16's alwaysRespond() taken out and the source before this PR's fix in place, R16 still fails at the bound, in about 60 s, naming the drain it was in, rather than hanging. Co-Authored-By: Claude Opus 5.5 --- test/helpers/coalescing-fixture.ts | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/test/helpers/coalescing-fixture.ts b/test/helpers/coalescing-fixture.ts index 1a117918d..93148af4e 100644 --- a/test/helpers/coalescing-fixture.ts +++ b/test/helpers/coalescing-fixture.ts @@ -18,7 +18,8 @@ export const TS = '2026-09-30T00:00:00Z'; /** * How long the fixture's waits run before they fail: send() for a request's - * reply, answered() for the host's answers, eventually() for its condition. + * reply, answered() for the host's answers, eventually() for its condition, + * settles() for the work it's given. * Each is a hang guard: it turns a wait that never ends into a failure * instead of a hung run, and it isn't a check on how fast the host works. So * it's set wide, and the same for every wait: a loaded host's stall needn't @@ -145,22 +146,17 @@ export async function eventually(predicate: () => boolean, what = 'condition'): while (!predicate()) { assert(performance.now() < deadline, `${what} did not settle within ${REPLY_DEADLINE_MS / 1000} s`); await new Promise((r) => setTimeout(r, 10)); } } -/** How long settles() lets a wait run: far above what any of these waits - * takes on a heavily loaded host, so load can't fire it, while a hang still - * fails within a minute. */ -export const HANG_GUARD_MS = 60_000; - /** * Awaits `work`, failing with `what` if it hasn't settled within - * HANG_GUARD_MS. A hang guard, not a timing check: CI runs without a test - * timeout, so a wait that never ends would hold the whole run, and this turns - * it into a failure that names what was awaited. When it fires, the work it - * waited on keeps running, and the test's close then runs under it. + * REPLY_DEADLINE_MS. CI runs without a test timeout, so a wait that never + * ends would hold the whole run, and this turns it into a failure that names + * what was awaited. When it fires, the work it waited on keeps running, and + * the test's close then runs under it. */ export async function settles(work: Promise, what: string): Promise { let timer: ReturnType | undefined; const late = new Promise((_, reject) => { - timer = setTimeout(() => reject(new Error(`${what} did not settle within ${HANG_GUARD_MS / 1000} s`)), HANG_GUARD_MS); + timer = setTimeout(() => reject(new Error(`${what} did not settle within ${REPLY_DEADLINE_MS / 1000} s`)), REPLY_DEADLINE_MS); }); try { return await Promise.race([work, late]); } finally { clearTimeout(timer); } }