From 1c49cbde2024af35c2338f006456b19f46f87b80 Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 05:56:34 -0400 Subject: [PATCH 1/8] fix(mcpl): a batch-only turn whose batch is gone, rendering elsewhere or queued elsewhere starts no inference MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A turn woken only by deferred coalescing batches is withdrawn before inference when its batches give it nothing to read (RFC-006 §5.2). Three paths still reached the model with nothing new: - The gate assembled only while some batch was pending. A batch retracted or replaced after its wake was taken, or one rendering for another agent's assembly, left nothing pending, so the turn skipped the assembly and called the model. A batch-only turn now always assembles, and one whose batch is rendering for another assembly shares that render. - assemble() left a cause whose batch was gone unsettled, which kept the turn. It now takes the turn's batch subjects, and a cause whose batch is gone settles: a request already consumed it (the batch's one consumer, as freeze's withdrawal of queued wakes already makes it), or a retraction or replacement withdrew it. The turn reads it only when a delivery of the subject is stored and still unread (the new CoalescerHost.isStoredUnread), such as a render that waited in a deferred queue and has since been flushed. - A render placed in another agent's deferred queue, because that agent's turn was alive, counted as materialized for the assembling turn, which then called the model without it. It no longer counts, and since the freeze withdrew the queue's agent's wake, the coalescer wakes that agent for the render, unless a newer batch of the subject has wakes of its own. Co-Authored-By: Claude Opus 5.5 --- .../coalescing-batch-turn-content.fixed.md | 1 + src/framework.ts | 39 ++-- src/mcpl/push-coalescer.ts | 70 ++++-- test/mcpl-coalescing.test.ts | 201 ++++++++++++++++++ 4 files changed, 283 insertions(+), 28 deletions(-) create mode 100644 changelog.d/coalescing-batch-turn-content.fixed.md diff --git a/changelog.d/coalescing-batch-turn-content.fixed.md b/changelog.d/coalescing-batch-turn-content.fixed.md new file mode 100644 index 000000000..91391e704 --- /dev/null +++ b/changelog.d/coalescing-batch-turn-content.fixed.md @@ -0,0 +1 @@ +- A turn woken only by deferred coalescing batches is withdrawn before inference in three more cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken, its batch was rendering for another agent's assembly (the turn now waits for that render and reads it), or the render was placed in another agent's deferred queue because that agent's turn was running. In the last case that agent is now woken for the render; before, nothing woke it for it once its turn ended. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. diff --git a/src/framework.ts b/src/framework.ts index ae7f1231e..fd0293db1 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11355,6 +11355,7 @@ export class AgentFramework { let unreadable: { agent: string; error: unknown } | undefined; const coalescer = new PushCoalescer({ isUnread: (p) => this.isUnreadPlacement(p), + isStoredUnread: (p) => this.isStoredUnreadPlacement(p), remove: (p) => this.removePlacement(p), deliver: (occ, materialized, assemblingFor) => this.deliverCoalesced(occ, materialized, assemblingFor), wakeForBatch: (occ) => this.wakeForCoalescedBatch(occ), @@ -11749,7 +11750,13 @@ export class AgentFramework { private isUnreadPlacement(p: CoalescingPlacement): boolean { this.resolvePlacement(p); if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId); - if (!p.messageId) return false; + return this.isStoredUnreadPlacement(p); + } + + /** Stored, a deferred entry once its flush has stored it, and unread. */ + private isStoredUnreadPlacement(p: CoalescingPlacement): boolean { + this.resolvePlacement(p); + if (p.deferredId || !p.messageId) return false; const agent = this.agents.get(p.agent); return !!agent && this.isUnreadStoredMessage(agent, p.messageId); } @@ -13186,10 +13193,16 @@ export class AgentFramework { // boundary — after the deferred flush (same window: turn alive, compile // not yet run) and before the checkpoint (they are the turn's inputs). // Not on a context-budget restart: that continues the same logical turn. + // A turn whose only causes are deferred batches assembles even when no + // batch is pending: its batch may have been withdrawn after its wake was + // dispatched, or be rendering for another agent's assembly, and only the + // assembly can say whether the turn has anything to read. + const batchSubjects = trigger?.coalescingBatchSubjects; let assembly: AssemblyResult | undefined; - if (attempt === 0 && !resumesTurn && this.pushCoalescer?.pendingBatches()) { + if (attempt === 0 && !resumesTurn && this.pushCoalescer + && (batchSubjects?.length || this.pushCoalescer.pendingBatches())) { try { - assembly = await this.pushCoalescer.assemble(agent.name); + assembly = await this.pushCoalescer.assemble(agent.name, batchSubjects); } catch (err) { console.error(`[coalescing] assembly for ${agent.name} failed:`, err); } @@ -13198,15 +13211,17 @@ export class AgentFramework { // 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; + // its recovery wake; a batch whose audience doesn't include the agent; a + // render placed in another agent's deferred queue; a batch withdrawn, + // displaced or consumed by another request before this assembly), 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), and so + // does a cause whose delivery is stored and still unread. A cause this + // assembly did not settle (the host stopping, or its subject moving on + // during the audience lookup) keeps the turn, as before. if (assembly && batchSubjects?.length && assembly.materialized.size === 0 && batchSubjects.every((s) => assembly.settled.has(s))) { console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`); diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 1bccc21df..3905c12a4 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -130,14 +130,18 @@ 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, or freeze() declined them for this agent. */ + * completion, freeze() declined them for this agent, or they were among + * the turn's causes and their batch was gone. */ settled: Set; - /** Of those, the subjects whose render or fallback was delivered as content. + /** Of those, the subjects whose content the turn being assembled reads: a + * render or fallback delivered where its compile reads it, or, for a cause + * whose batch was gone, a delivery of the subject stored and still unread. * A settled subject that is not here gave the turn nothing to read: empty * 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. */ + * waits for its recovery wake), a batch whose audience doesn't include + * the agent, a render placed in another agent's deferred queue, or a cause + * whose batch was withdrawn, displaced or consumed by another request. */ materialized: Set; } @@ -185,6 +189,10 @@ export interface CoalescingReceiptRecord { export interface CoalescerHost { /** True iff the placement still exists and no model request has included it. */ isUnread(placement: CoalescingPlacement): boolean; + /** True iff the placement is stored in a context, no longer in the deferred + * queue, and no model request has included it: the next compile of a turn + * that reads that context shows it as new. */ + isStoredUnread(placement: CoalescingPlacement): boolean; /** False when the unread occurrence could not be removed (it may still be read). */ remove(placement: CoalescingPlacement): boolean; /** Ordinary delivery of the event; undefined when routing delivered nowhere. @@ -666,19 +674,29 @@ export class PushCoalescer { * `agentName`. Called by the host at a turn's assembly boundary, before the * compile. Bounded by the host's render timeout; a failed or late render * materializes the admitted fallback (§5.3). Reports which subjects - * settled and which of them produced content, so the host can tell a turn - * whose only cause rendered nothing. + * settled and which of them gave the turn content, so the host can tell a + * turn whose only causes gave it nothing to read. + * + * `causes` are the subjects whose batch wakes started the turn. A cause + * whose batch is gone by now settles: it was withdrawn or displaced after + * its wake was dispatched, or another request consumed it (§3.2), and it + * gives the turn content only through a delivery of the subject that is + * stored and still unread. One whose batch is rendering for another + * assembly shares that render's outcome, as any assembly in its audience + * does. */ - async assemble(agentName: string): Promise { + async assemble(agentName: string, causes: readonly string[] = []): Promise { const result: AssemblyResult = { settled: new Set(), materialized: new Set() }; if (this.suspended) return result; const work: Promise[] = []; - for (const subject of [...this.subjects.keys()]) { + const wanted = new Set(causes); + for (const subject of new Set([...this.subjects.keys(), ...causes])) { // The handle is wrapped: a bare promise returned through `locked` would // 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)); + const started = await this.locked(() => this.freeze(subject, agentName, wanted.has(subject))); if (started === 'declined') result.settled.add(subject); + else if (started === 'unread') { result.settled.add(subject); result.materialized.add(subject); } else if (started) { work.push(started.done.then(() => { result.settled.add(subject); @@ -700,13 +718,27 @@ export class PushCoalescer { * 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. + * kept for its recovery wake). For one of the turn's causes (`cause`) + * whose batch is gone, neither pending nor rendering, it returns 'unread' + * when a delivery of the subject is stored and still unread, so the turn + * reads it, and 'declined' otherwise. Returns undefined when the batch was + * not this assembly's to decide on: there is none and it wasn't a cause, + * 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> { + private async freeze(subject: string, agentName: string, cause = false): Promise<{ done: Promise; rendering: Rendering } | 'declined' | 'unread' | undefined> { + if (this.suspended) return undefined; const state = this.subjects.get(subject); - if (!state || this.suspended) return undefined; + if (!state || (!state.batch && !state.rendering)) { + if (!cause) return undefined; + // The batch whose wake started this turn is gone: a retraction or a + // displacement withdrew it after the wake was dispatched, or another + // request consumed it, and that consumption is the batch's one (§3.2), + // as freeze's withdrawal of the other queued wakes already makes it. + // What the turn can still read of it is a delivery stored and unread: a + // render that waited in another agent's deferred queue and has been + // flushed since, a replacement, or a notice. + return state?.occupant && this.host.isStoredUnread(state.occupant.placement) ? 'unread' : 'declined'; + } if (state.rendering) { // Another assembly froze this batch; share its outcome (vector 25), // unless this agent isn't in its audience, as below. @@ -815,15 +847,21 @@ export class PushCoalescer { this.host.audit({ kind: 'rendered', subject, eventId: occurrence.eventId, source, empty }); if (empty) return; // §5.2: nothing happened (blank text included) const placement = await this.host.deliver({ ...occurrence, timestamp, content }, content, assemblingFor); - rendering.materialized = true; if (placement?.deferredId) { // Landed in another agent's deferred queue (its turn is alive): still - // unread there, so it stays replaceable and withdrawable. + // unread there, so it stays replaceable and withdrawable. The turn + // being assembled doesn't read it, so it isn't materialized for that + // turn. The freeze withdrew the queue's agent's wake for this batch; + // wake it again, so its next turn, once the flush has stored the + // content, reads it. A newer batch that opened during the render + // has a wake of its own. state.occupant = { eventId: occurrence.eventId, placement }; state.audienceAgent = placement.agent; + if (!state.batch) await this.host.wakeForBatch({ ...occurrence, deliverTo: placement.agent }); return; } // The materialized occurrence is consumed by the request being assembled. + rendering.materialized = true; state.history = 'some'; state.consumedEventId = occurrence.eventId; state.occupant = undefined; diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index 4b8fc4dbc..ac2a39a41 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1003,6 +1003,207 @@ test('an assembly for an agent outside the audience of a batch another assembly assert.equal(rendered.materialized.size, 1, 'the assembly in the audience delivered the render'); }); +// --------------------------------------------------------------------------- +// A batch-only turn whose batch is gone, rendering elsewhere, or placed in +// another agent's queue (a turn with nothing to read never reaches the model) +// --------------------------------------------------------------------------- + +type TurnInternals = { + startAgentStream: (...a: unknown[]) => Promise; + processInferenceRequests: () => Promise; + activeTurnTokens: Map; + pendingRequests: Array<{ agentName: string; coalescingSubject?: string; coalescingBatch?: boolean }>; + pushCoalescer: { + pendingBatches(): number; + assemble(agent: string, causes?: readonly string[]): Promise<{ settled: Set; materialized: Set }>; + }; +}; + +const TWO_RESIDENTS = [ + { name: 'agent', model: 'test', systemPrompt: 'test' }, + { name: 'other', model: 'test', systemPrompt: 'test' }, +]; + +function withdrawals(f: Awaited>): string[] { + const agents: string[] = []; + f.framework.onTrace((e) => { + const ev = e as { type: string; kind?: string; agentName?: string }; + if (ev.type === 'mcpl:coalescing' && ev.kind === 'turn-withdrawn') agents.push(ev.agentName ?? ''); + }); + return agents; +} + +test('a batch-only turn whose batch is retracted after its wake is dispatched starts no inference', async (t) => { + const f = await fixture(); t.after(f.close); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'withdrawn_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + const start = internals.startAgentStream.bind(internals); + // The retraction lands after the scheduler took the wake and before the + // turn reaches its assembly, as it can while the wake waits behind an + // auxiliary call. No batch is pending by the time the turn gets there. + internals.startAgentStream = async (...a: unknown[]) => { + internals.startAgentStream = start; + const r = await f.send('push/event', f.params('2', '', { retract: true })); + assert.equal(r.result.coalesce.outcome, 'retracted'); + return start(...a); + }; + await settles(f.framework.runUntilIdle(), 'the drain after the wake'); + assert.equal(f.membrane.calls.length, 0, 'the withdrawn batch, the only cause, started no inference'); + assert.deepEqual(withdrawn, ['agent'], 'its turn started and was withdrawn'); +}); + +test('a batch-only turn whose batch is retracted while its assembly waits for the coalescer starts no inference', async (t) => { + const f = await fixture(); t.after(f.close); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'withdrawn_fallback', { deferred: true, initial: true })); + const coalescer = (f.framework as unknown as TurnInternals).pushCoalescer; + const assemble = coalescer.assemble.bind(coalescer); + // The retraction lands after the gate saw the batch pending and before the + // batch's freeze, as it can while another admission holds the lock. + coalescer.assemble = async (...a) => { + coalescer.assemble = assemble; + const r = await f.send('push/event', f.params('2', '', { retract: true })); + assert.equal(r.result.coalesce.outcome, 'retracted'); + return assemble(...a); + }; + await settles(f.framework.runUntilIdle(), 'the drain after the wake'); + assert.equal(f.membrane.calls.length, 0, 'the withdrawn batch, the only cause, started no inference'); + assert.deepEqual(withdrawn, ['agent'], 'its turn started and was withdrawn'); +}); + +test('a batch-only turn whose batch a plain occurrence replaces while its assembly waits gives way to the replacement\'s own turn', async (t) => { + const f = await fixture(); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'replaced_fallback', { deferred: true, initial: true })); + const coalescer = (f.framework as unknown as TurnInternals).pushCoalescer; + const assemble = coalescer.assemble.bind(coalescer); + // The replacement lands while the turn is alive, so it waits in the + // deferred queue: this turn's compile wouldn't show it. + coalescer.assemble = async (...a) => { + coalescer.assemble = assemble; + const r = await f.send('push/event', f.params('2', 'plain_replacement')); + assert.equal(r.result.coalesce.outcome, 'replaced'); + return assemble(...a); + }; + await settles(f.framework.runUntilIdle(), 'the drain'); + assert.deepEqual(withdrawn, ['agent'], 'the batch-only turn was withdrawn'); + assert.equal(f.membrane.calls.length, 1, "the replacement's own wake started one turn"); + assert(f.lastRequest().includes('plain_replacement'), 'which read the replacement'); +}); + +test('a batch-only turn that reaches its assembly while the batch renders for another resident shares that render: one call reads it', async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + 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 TurnInternals; + const coalescer = internals.pushCoalescer; + const assemble = coalescer.assemble.bind(coalescer); + let joined!: () => void; + const agentAssembles = new Promise((resolve) => { joined = resolve; }); + coalescer.assemble = async (agent, causes) => { + if (agent === 'agent' && causes?.length) joined(); + return assemble(agent, causes); + }; + // Both residents' wakes have been taken. Before 'agent''s turn reaches its + // assembly, 'other''s assembly freezes the batch, and its render is held. + let otherAssembly!: Promise; + const start = internals.startAgentStream.bind(internals); + internals.startAgentStream = async (...a: unknown[]) => { + internals.startAgentStream = start; + otherAssembly = assemble('other'); + await settles(started, "the render for 'other'"); + assert.equal(coalescer.pendingBatches(), 0, 'no batch is pending: it is rendering'); + return start(...a); + }; + const run = f.framework.runUntilIdle(); + // On a host that skips the assembly with nothing pending, the drain ends + // first: both turns called the model with nothing to read. + await settles(Promise.race([agentAssembles, run]), "'agent''s assembly or the drain"); + release({ content: [{ type: 'text', text: 'shared_render' }] }); + await settles(run, 'the drain'); + await settles(otherAssembly, "the assembly for 'other'"); + await settles(f.framework.runUntilIdle(), 'the drain after the render'); + assert.equal(f.membrane.calls.length, 1, 'the batch got one model call'); + assert(f.lastRequest().includes('shared_render'), 'and that call read the render'); + assert.equal(internals.pendingRequests.length, 0, 'and nothing is left queued'); +}); + +test("a render placed in a live turn's deferred queue isn't read by the resident that rendered it; the queue's agent is woken for it", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // The primary's turn is alive, as while it waits on a tool call; this + // token stands in for one. 'other''s assembly renders the batch, and the + // delivery, bound for the primary's context, waits in its queue. + internals.activeTurnTokens.set('agent', -1); + await settles(internals.processInferenceRequests(), 'one scheduler pass'); + assert.equal(f.membrane.calls.length, 0, "'other' read nothing, so its turn started no inference"); + assert.deepEqual(withdrawn, ['other']); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], 'the primary is woken for the render'); + assert(!f.context().includes('document_diff'), 'which is still in its queue'); + internals.activeTurnTokens.delete('agent'); + await settles(f.framework.runUntilIdle(), "the primary's turn"); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('document_diff'), "the primary's turn read the render once its queue was flushed"); +}); + +test("a render placed in a live turn's deferred queue leaves the wakes of a newer batch of its subject as they are", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + let release!: (value: unknown) => void; + let began!: () => void; + const started = new Promise((resolve) => { began = resolve; }); + let renders = 0; + f.renderer(() => { + if (++renders > 1) return Promise.resolve({ content: [{ type: 'text', text: 'second_render' }] }); + began(); + return new Promise((resolve) => { release = resolve; }); + }); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // As above, the primary's turn is alive and 'other''s assembly renders the + // batch, here held. A notice during the render opens a newer batch of the + // subject, which wakes its audience. + internals.activeTurnTokens.set('agent', -1); + const pass = internals.processInferenceRequests(); + await settles(started, "the render for 'other'"); + await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); + const woken = internals.pendingRequests.map((r) => r.agentName).sort(); + assert.deepEqual(woken, ['agent', 'other']); + release({ content: [{ type: 'text', text: 'first_render' }] }); + await settles(pass, 'the scheduler pass'); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName).sort(), woken, "the newer batch's wakes are as they were"); + internals.activeTurnTokens.delete('agent'); + await settles(f.framework.runUntilIdle(), 'the drain'); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('first_render') && f.lastRequest().includes('second_render'), 'one call read both renders'); +}); + +test("a batch consumed by one resident's request gives another resident's batch-only turn nothing", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'consumed_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent', 'other'], 'both residents are woken'); + // One scheduler pass takes both wakes. 'agent''s assembly renders the + // batch and its request reads it; 'other''s turn then finds the batch + // consumed (RFC-006 §3.2), as if its wake had still been queued when the + // freeze withdrew it. + await settles(f.framework.runUntilIdle(), 'the drain'); + assert.equal(f.membrane.calls.length, 1, 'one call for the batch'); + assert(f.lastRequest().includes('document_diff')); + assert.deepEqual(withdrawn, ['other']); +}); + 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' })); From 16b5c87879eab4441af7a1a6a760c3ee12672fdc Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 06:07:27 -0400 Subject: [PATCH 2/8] fix(mcpl): keep isUnreadPlacement as it was; isStoredUnreadPlacement reads through it The first commit folded isUnreadPlacement's stored-message branch into isStoredUnreadPlacement. Merged with #168, which adds a `by` parameter to isUnreadPlacement and passes it to isUnreadStoredMessage, git moved #168's `by` into isStoredUnreadPlacement, where it isn't in scope, and dropped it from isUnreadPlacement's own reading. isUnreadPlacement now stays as main has it, and isStoredUnreadPlacement resolves the placement and answers `!p.deferredId && this.isUnreadPlacement(p)`, which reads by the consumed watermark, the default #168 keeps. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index fd0293db1..67cee56ef 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11750,15 +11750,15 @@ export class AgentFramework { private isUnreadPlacement(p: CoalescingPlacement): boolean { this.resolvePlacement(p); if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId); - return this.isStoredUnreadPlacement(p); + if (!p.messageId) return false; + const agent = this.agents.get(p.agent); + return !!agent && this.isUnreadStoredMessage(agent, p.messageId); } /** Stored, a deferred entry once its flush has stored it, and unread. */ private isStoredUnreadPlacement(p: CoalescingPlacement): boolean { this.resolvePlacement(p); - if (p.deferredId || !p.messageId) return false; - const agent = this.agents.get(p.agent); - return !!agent && this.isUnreadStoredMessage(agent, p.messageId); + return !p.deferredId && this.isUnreadPlacement(p); } /** From 7d00a52627e37b4c5fa27fa1e4e24e0c69ba2698 Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 07:09:32 -0400 Subject: [PATCH 3/8] fix(mcpl): a batch has one consumer, even when two residents' assemblies overlap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The internal review of 16b5c878 (af-f103-f104) found that the first round's mechanism let a second request take a batch's content: - An assembly that joined another agent's in-flight render took that render's materialized flag. Two residents' assemblies overlap through the auxiliary park (a wake parked behind an auxiliary call starts its turn outside the scheduler's pass), so with the primary rendering, a resident whose parked turn joined called the model too. A joined render now settles the subject without counting as the joining assembly's own; the joining assembly still waits for it, as before. - A gone cause took any stored, unread delivery of its subject as content, in any context, so a sibling could read a render queued for the primary once its flush had stored it. The only content a gone cause gets now is a render no request consumed that was placed in its own agent's deferred queue (the occupant's new `reader`), read before the freeze, which may render a newer batch. A replacement or a retraction's notice is read through its own delivery and wake, under its own triggerInference. - The wake for the queue's agent was skipped whenever a newer batch of the subject existed, though that batch might not wake the agent (no triggerInference, or an audience without it). It is now given whenever the frozen batch's own wake would wake the agent, through the new CoalescerHost.wakeToRead, which adds that agent's wake and leaves the subject's other wakes as they are (wakeForCoalescedBatch is split into its cancel and queueCoalescedWakes). The snapshot no longer carries the occupant's reader: a restart counts every occupant as read. Comments that cited RFC-006 §3.2 for consumption now cite §5.2, where the rule is. Tests: the rendering-elsewhere test now runs two real turns that overlap through a real auxiliary park; new tests cover a stored replacement giving a gone cause nothing, a newer batch that wakes no one, a queued render its agent's running turn already read, and a third resident that must not read the primary's queued render. Turns kept alive by stand-in tokens now end with the flush a torn-down turn does, in finally blocks. Co-Authored-By: Claude Opus 5.5 --- .../coalescing-batch-turn-content.fixed.md | 2 +- src/framework.ts | 31 ++- src/mcpl/push-coalescer.ts | 135 +++++----- test/mcpl-coalescing.test.ts | 235 +++++++++++++++--- 4 files changed, 293 insertions(+), 110 deletions(-) diff --git a/changelog.d/coalescing-batch-turn-content.fixed.md b/changelog.d/coalescing-batch-turn-content.fixed.md index 91391e704..2eb536759 100644 --- a/changelog.d/coalescing-batch-turn-content.fixed.md +++ b/changelog.d/coalescing-batch-turn-content.fixed.md @@ -1 +1 @@ -- A turn woken only by deferred coalescing batches is withdrawn before inference in three more cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken, its batch was rendering for another agent's assembly (the turn now waits for that render and reads it), or the render was placed in another agent's deferred queue because that agent's turn was running. In the last case that agent is now woken for the render; before, nothing woke it for it once its turn ended. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. +- A turn woken only by deferred coalescing batches is withdrawn before inference in more of the cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken (the replacement or the retraction's notice wakes agents as any push does, by its own `triggerInference`), its batch was being rendered for another agent's assembly (the turn waits for that render, which the other agent's request reads), or the render was placed in another agent's deferred queue because that agent's turn was running. In that last case that agent is now woken for the render, as the batch's own wake would have woken it; before, nothing woke it for the render. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. diff --git a/src/framework.ts b/src/framework.ts index 2b961b238..f94f6f95e 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11712,6 +11712,7 @@ export class AgentFramework { remove: (p) => this.removePlacement(p), deliver: (occ, materialized, assemblingFor) => this.deliverCoalesced(occ, materialized, assemblingFor), wakeForBatch: (occ) => this.wakeForCoalescedBatch(occ), + wakeToRead: (occ, agentName) => this.queueCoalescedWakes({ ...occ, deliverTo: agentName }), cancelWake: (subject) => this.cancelCoalescedWakes(subject), authorized: (occ) => this.coalescedAuthorized(occ), audience: (occ) => this.coalescedAudience(occ), @@ -12099,8 +12100,13 @@ export class AgentFramework { } private async wakeForCoalescedBatch(occ: CoalescedOccurrence): Promise { + this.cancelCoalescedWakes(coalescingSubjectKey(occ.serverId, occ.binding, occ.scope, occ.key)); + await this.queueCoalescedWakes(occ); + } + + /** Queue a wake for each agent of the batch's audience, when its latest occurrence triggers inference. */ + private async queueCoalescedWakes(occ: CoalescedOccurrence): Promise { const subject = coalescingSubjectKey(occ.serverId, occ.binding, occ.scope, occ.key); - this.cancelCoalescedWakes(subject); if (occ.event.lane !== 'push' || !occ.event.event.triggerInference) return; const origin = occ.event.event.origin; // The wake carries the batch's channel so the turn freezes its reply @@ -14756,7 +14762,8 @@ export class AgentFramework { // Not on a context-budget restart: that continues the same logical turn. // A turn whose only causes are deferred batches assembles even when no // batch is pending: its batch may have been withdrawn after its wake was - // dispatched, or be rendering for another agent's assembly, and only the + // dispatched or be rendering for another agent's assembly, or the turn may + // be the one woken to read a render queued for this agent, and only the // assembly can say whether the turn has anything to read. const batchSubjects = trigger?.coalescingBatchSubjects; let assembly: AssemblyResult | undefined; @@ -14773,16 +14780,16 @@ export class AgentFramework { // 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; a - // render placed in another agent's deferred queue; a batch withdrawn, - // displaced or consumed by another request before this assembly), 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), and so - // does a cause whose delivery is stored and still unread. A cause this - // assembly did not settle (the host stopping, or its subject moving on - // during the audience lookup) keeps the turn, as before. + // render placed in another agent's deferred queue; a batch another + // assembly froze, or one withdrawn, displaced or consumed by another + // request before this assembly), 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), and so does a render queued for this agent that + // the flush has stored and nothing has read. A cause this assembly did not + // settle (the host stopping) keeps the turn, as before. if (assembly && batchSubjects?.length && assembly.materialized.size === 0 && batchSubjects.every((s) => assembly.settled.has(s))) { console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`); diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 60fb5e681..0cd569aa3 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -131,23 +131,28 @@ interface Rendering { export interface AssemblyResult { /** Subjects this assembly finished with: it waited on their render to * completion, freeze() declined them for this agent, or they were among - * the turn's causes and their batch was gone. */ + * the turn's causes and their batch was gone or another assembly's. */ settled: Set; - /** Of those, the subjects whose content the turn being assembled reads: a - * render or fallback delivered where its compile reads it, or, for a cause - * whose batch was gone, a delivery of the subject stored and still unread. - * A settled subject that is not here gave the turn nothing to read: empty - * 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), a batch whose audience doesn't include - * the agent, a render placed in another agent's deferred queue, or a cause - * whose batch was withdrawn, displaced or consumed by another request. */ + /** Of those, the subjects whose content the turn being assembled reads as + * its own: a render or fallback this assembly delivered where its compile + * reads it, or, for a cause, a render queued for this agent that no request + * consumed, which the flush has stored and nothing has read yet. A settled + * subject that is not here gave the turn nothing to read: empty 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), a batch whose audience doesn't include the agent, a + * render placed in another agent's deferred queue, a render another + * assembly froze, or a cause whose batch was withdrawn, displaced or + * consumed by another request. */ materialized: Set; } interface SubjectState { history: 'none' | 'some' | 'unknown'; - occupant?: { eventId: string; placement: CoalescingPlacement }; + /** The subject's delivery that may still be unread. `reader` marks a render + * no request consumed, placed in that agent's deferred queue: the agent + * woken to read it (§5.2). */ + occupant?: { eventId: string; placement: CoalescingPlacement; reader?: string }; batch?: DeferredBatch; rendering?: Rendering; consumedEventId?: string; @@ -201,6 +206,10 @@ export interface CoalescerHost { deliver(occurrence: CoalescedOccurrence, materialized?: McplContentBlock[], assemblingFor?: string): Promise; /** Queue a wake for a batch that has no model-visible content yet. */ wakeForBatch(occurrence: CoalescedOccurrence): Promise; + /** Queue a wake for `agent` to read a render of the batch that no request + * consumed, as the batch's own wake would wake it; the subject's other + * wakes stay. */ + wakeToRead(occurrence: CoalescedOccurrence, agent: string): Promise; /** Withdraw unstarted wakes whose sole cause is this subject. */ cancelWake(subject: string): void; /** The batch's audience still holds the authority it was admitted under. */ @@ -686,13 +695,13 @@ export class PushCoalescer { * settled and which of them gave the turn content, so the host can tell a * turn whose only causes gave it nothing to read. * - * `causes` are the subjects whose batch wakes started the turn. A cause - * whose batch is gone by now settles: it was withdrawn or displaced after - * its wake was dispatched, or another request consumed it (§3.2), and it - * gives the turn content only through a delivery of the subject that is - * stored and still unread. One whose batch is rendering for another - * assembly shares that render's outcome, as any assembly in its audience - * does. + * `causes` are the subjects whose batch wakes started the turn. A batch has + * one consumer, the request it was frozen for (§5.2: the materialized + * occurrence is consumed by that request), so a cause whose batch is gone + * by now, or another assembly's render, settles and gives the turn nothing + * of it. The exception is a render no request consumed, placed in this + * agent's deferred queue: this agent was woken to read it, and once the + * flush has stored it, the turn reads it as its own. */ async assemble(agentName: string, causes: readonly string[] = []): Promise { const result: AssemblyResult = { settled: new Set(), materialized: new Set() }; @@ -700,16 +709,22 @@ export class PushCoalescer { const work: Promise[] = []; const wanted = new Set(causes); for (const subject of new Set([...this.subjects.keys(), ...causes])) { + const cause = wanted.has(subject); // The handle is wrapped: a bare promise returned through `locked` would // 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, wanted.has(subject))); + // and the settlement itself needs the lock. The queued render is read + // before the freeze, whose render of a newer batch would clear it. + const { queued, started } = await this.locked(async () => ({ + queued: cause && this.queuedFor(subject, agentName), + started: await this.freeze(subject, agentName, cause), + })); + if (queued) { result.settled.add(subject); result.materialized.add(subject); } if (started === 'declined') result.settled.add(subject); - else if (started === 'unread') { result.settled.add(subject); result.materialized.add(subject); } else if (started) { work.push(started.done.then(() => { result.settled.add(subject); - if (started.rendering.materialized) result.materialized.add(subject); + // A render another assembly froze is that request's to read. + if (started.rendering.materialized && !started.joined) result.materialized.add(subject); })); } } @@ -718,44 +733,51 @@ export class PushCoalescer { return result; } + /** + * A render of `subject` that no request consumed, placed in `agentName`'s + * deferred queue, which the flush has stored and nothing has read yet: + * `agentName` was woken to read it. + */ + private queuedFor(subject: string, agentName: string): boolean { + const occupant = this.subjects.get(subject)?.occupant; + return occupant?.reader === agentName && this.host.isStoredUnread(occupant.placement); + } + /** * 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). 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). For one of the turn's causes (`cause`) - * whose batch is gone, neither pending nor rendering, it returns 'unread' - * when a delivery of the subject is stored and still unread, so the turn - * reads it, and 'declined' otherwise. Returns undefined when the batch was - * not this assembly's to decide on: there is none and it wasn't a cause, - * the subject moved on during the audience lookup, or the host is stopping. + * the RPC. Returns the render's completion, which an assembly that joins + * the render waits for too (`joined`: vector 25, one render). 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); the render-start could not + * be made durable (the batch is kept for its recovery wake); or, for one of + * the turn's causes (`cause`), the batch is gone. Returns undefined when + * the batch was not this assembly's to decide on: there is none and it + * wasn't a cause, the subject moved on during the audience lookup, or the + * host is stopping. */ - private async freeze(subject: string, agentName: string, cause = false): Promise<{ done: Promise; rendering: Rendering } | 'declined' | 'unread' | undefined> { + private async freeze(subject: string, agentName: string, cause = false): Promise<{ done: Promise; rendering: Rendering; joined?: true } | 'declined' | undefined> { if (this.suspended) return undefined; const state = this.subjects.get(subject); - if (!state || (!state.batch && !state.rendering)) { - if (!cause) return undefined; - // The batch whose wake started this turn is gone: a retraction or a - // displacement withdrew it after the wake was dispatched, or another - // request consumed it, and that consumption is the batch's one (§3.2), - // as freeze's withdrawal of the other queued wakes already makes it. - // What the turn can still read of it is a delivery stored and unread: a - // render that waited in another agent's deferred queue and has been - // flushed since, a replacement, or a notice. - return state?.occupant && this.host.isStoredUnread(state.occupant.placement) ? 'unread' : 'declined'; - } + // The batch whose wake started this turn is gone: a retraction or a + // displacement withdrew it after the wake was dispatched, or another + // request consumed it, the batch's one consumption, as the freeze's + // withdrawal of the other queued wakes already makes it. Whatever + // replaced it is read through its own delivery and wake. + if (!state || (!state.batch && !state.rendering)) return cause ? 'declined' : undefined; if (state.rendering) { - // Another assembly froze this batch; share its outcome (vector 25), - // unless this agent isn't in its audience, as below. + // Another assembly froze this batch, and the request it assembles + // consumes it. An assembly in its audience waits for that render, so + // its compile shows the content (vector 25), but the content isn't its + // own to read, cause or not; unless this agent isn't in its audience, + // as below. const rendering = state.rendering; 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 }; + return { done: rendering.done, rendering, joined: true }; } const batch = state.batch; if (!batch) return undefined; @@ -860,13 +882,13 @@ export class PushCoalescer { // Landed in another agent's deferred queue (its turn is alive): still // unread there, so it stays replaceable and withdrawable. The turn // being assembled doesn't read it, so it isn't materialized for that - // turn. The freeze withdrew the queue's agent's wake for this batch; - // wake it again, so its next turn, once the flush has stored the - // content, reads it. A newer batch that opened during the render - // has a wake of its own. - state.occupant = { eventId: occurrence.eventId, placement }; + // turn, and no request has consumed it. The freeze withdrew the + // queue's agent's wake for this batch; wake it again, beside any + // wakes a newer batch has, so that its next turn, once the flush has + // stored the content, reads it. + state.occupant = { eventId: occurrence.eventId, placement, reader: placement.agent }; state.audienceAgent = placement.agent; - if (!state.batch) await this.host.wakeForBatch({ ...occurrence, deliverTo: placement.agent }); + await this.host.wakeToRead(occurrence, placement.agent); return; } // The materialized occurrence is consumed by the request being assembled. @@ -971,7 +993,8 @@ export class PushCoalescer { subjects: [...this.subjects].map(([subject, s]) => ({ subject, history: s.history, consumedEventId: s.consumedEventId, identity: s.identity, ...(s.audienceAgent ? { audienceAgent: s.audienceAgent } : {}), - ...(s.occupant ? { occupant: s.occupant } : {}), + // A restart counts the occupant as read (restore), so its reader isn't kept. + ...(s.occupant ? { occupant: { eventId: s.occupant.eventId, placement: s.occupant.placement } } : {}), ...(s.batch ? { batch: { latest: s.batch.latest as CoalescedOccurrence, notices: s.batch.notices, dropped: s.batch.dropped, ...(s.batch.noRender ? { rendering: true } : {}) } } : s.rendering ? { batch: { latest: s.rendering.batch.latest as CoalescedOccurrence, notices: s.rendering.batch.notices, dropped: s.rendering.batch.dropped, rendering: true } } : {}), diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index ac2a39a41..7423e6ef9 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1013,6 +1013,10 @@ type TurnInternals = { processInferenceRequests: () => Promise; activeTurnTokens: Map; pendingRequests: Array<{ agentName: string; coalescingSubject?: string; coalescingBatch?: boolean }>; + providerGate(agent: string): { auxiliaryInFlight: number; idleWaiters: Array<() => void> }; + drainDeferredFor(agent: string): unknown[]; + readdDrained(entries: unknown[], where: string, opts?: { bypassDeferral?: boolean }): unknown[]; + ackDeferredWrites(): void; pushCoalescer: { pendingBatches(): number; assemble(agent: string, causes?: readonly string[]): Promise<{ settled: Set; materialized: Set }>; @@ -1024,6 +1028,17 @@ const TWO_RESIDENTS = [ { name: 'other', model: 'test', systemPrompt: 'test' }, ]; +/** A trigger filter under which a push whose text says 'quiet' triggers no inference. */ +const QUIET_PUSHES_DONT_TRIGGER = { shouldTriggerInference: (text: string) => !text.includes('quiet') }; + +/** Ends the turn a stand-in token kept alive, storing what waited in its + * deferred queue, as a torn-down turn's flush does. */ +function endStandInTurn(internals: TurnInternals, agent = 'agent'): void { + if (!internals.activeTurnTokens.delete(agent)) return; + internals.readdDrained(internals.drainDeferredFor(agent), 'the stand-in turn ends'); + internals.ackDeferredWrites(); +} + function withdrawals(f: Awaited>): string[] { const agents: string[] = []; f.framework.onTrace((e) => { @@ -1060,7 +1075,9 @@ test('a batch-only turn whose batch is retracted while its assembly waits for th const coalescer = (f.framework as unknown as TurnInternals).pushCoalescer; const assemble = coalescer.assemble.bind(coalescer); // The retraction lands after the gate saw the batch pending and before the - // batch's freeze, as it can while another admission holds the lock. + // batch's freeze, as it can while another admission holds the coalescer's + // lock; this seam, before the assembly takes the lock, stands in for that + // wait. coalescer.assemble = async (...a) => { coalescer.assemble = assemble; const r = await f.send('push/event', f.params('2', '', { retract: true })); @@ -1093,9 +1110,32 @@ test('a batch-only turn whose batch a plain occurrence replaces while its assemb assert(f.lastRequest().includes('plain_replacement'), 'which read the replacement'); }); -test('a batch-only turn that reaches its assembly while the batch renders for another resident shares that render: one call reads it', async (t) => { +test("a batch-only turn whose batch a stored replacement displaced gets nothing of it; the replacement wakes whom its own trigger wakes", async (t) => { + const f = await fixture({ server: QUIET_PUSHES_DONT_TRIGGER }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'replaced_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + const start = internals.startAgentStream.bind(internals); + // The replacement, which triggers no inference, lands after the scheduler + // took the batch's wake and before the turn is alive, so it's stored and + // unread when the turn assembles. + internals.startAgentStream = async (...a: unknown[]) => { + internals.startAgentStream = start; + const r = await f.send('push/event', f.params('2', 'quiet_replacement')); + assert.equal(r.result.coalesce.outcome, 'replaced'); + return start(...a); + }; + await settles(f.framework.runUntilIdle(), 'the drain'); + assert(f.context().includes('quiet_replacement'), 'the replacement is in context'); + assert.equal(f.membrane.calls.length, 0, "the batch's turn didn't take the replacement as its own"); + assert.deepEqual(withdrawn, ['agent']); +}); + +test("a batch-only turn that reaches its assembly while another resident's assembly renders its batch waits for it and gets nothing of it: one call reads the render", async (t) => { const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); f.alwaysRespond(); + const withdrawn = withdrawals(f); let release!: (value: unknown) => void; let began!: () => void; const started = new Promise((resolve) => { began = resolve; }); @@ -1103,35 +1143,48 @@ test('a batch-only turn that reaches its assembly while the batch renders for an await f.send('push/event', f.params('1', 'shared_fallback', { deferred: true, initial: true })); const internals = f.framework as unknown as TurnInternals; const coalescer = internals.pushCoalescer; + // 'other''s wake parks behind an auxiliary call of its own, so its turn + // starts outside the scheduler's pass, once the call ends: the one way two + // residents' assemblies overlap. The primary's turn is alive (this token + // stands in for one), so the same pass leaves the primary's wake queued. + const otherGate = internals.providerGate('other'); + otherGate.auxiliaryInFlight = 1; + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'the pass that parks the wake'); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], "the primary's wake waits for its turn to end"); + } finally { + endStandInTurn(internals); + } + // The primary's turn starts, and its assembly renders the batch, held. + const pass = internals.processInferenceRequests(); + await settles(started, "the primary's render"); + // The auxiliary call ends, and 'other''s turn reaches its assembly while + // the render is in flight. A host that skips the assembly with no batch + // pending calls the model for 'other' instead. const assemble = coalescer.assemble.bind(coalescer); - let joined!: () => void; - const agentAssembles = new Promise((resolve) => { joined = resolve; }); - coalescer.assemble = async (agent, causes) => { - if (agent === 'agent' && causes?.length) joined(); - return assemble(agent, causes); - }; - // Both residents' wakes have been taken. Before 'agent''s turn reaches its - // assembly, 'other''s assembly freezes the batch, and its render is held. - let otherAssembly!: Promise; + let joins!: () => void; + const otherAssembles = new Promise((resolve) => { joins = resolve; }); + coalescer.assemble = (agent, causes) => { if (agent === 'other') joins(); return assemble(agent, causes); }; const start = internals.startAgentStream.bind(internals); - internals.startAgentStream = async (...a: unknown[]) => { - internals.startAgentStream = start; - otherAssembly = assemble('other'); - await settles(started, "the render for 'other'"); - assert.equal(coalescer.pendingBatches(), 0, 'no batch is pending: it is rendering'); - return start(...a); + let otherTurn: Promise = new Promise(() => {}); + let otherStarts!: () => void; + const otherStarted = new Promise((resolve) => { otherStarts = resolve; }); + internals.startAgentStream = (...a: unknown[]) => { + const turn = start(...a); + if ((a[0] as { name: string }).name === 'other') { otherTurn = turn; otherStarts(); } + return turn; }; - const run = f.framework.runUntilIdle(); - // On a host that skips the assembly with nothing pending, the drain ends - // first: both turns called the model with nothing to read. - await settles(Promise.race([agentAssembles, run]), "'agent''s assembly or the drain"); + otherGate.auxiliaryInFlight = 0; + for (const resolve of otherGate.idleWaiters.splice(0)) resolve(); + await settles(Promise.race([otherAssembles, otherStarted.then(() => otherTurn)]), "'other''s assembly or its turn"); release({ content: [{ type: 'text', text: 'shared_render' }] }); - await settles(run, 'the drain'); - await settles(otherAssembly, "the assembly for 'other'"); - await settles(f.framework.runUntilIdle(), 'the drain after the render'); + await settles(pass, "the primary's turn"); + await settles(otherTurn, "'other''s turn"); + await settles(f.framework.runUntilIdle(), 'the drain'); assert.equal(f.membrane.calls.length, 1, 'the batch got one model call'); assert(f.lastRequest().includes('shared_render'), 'and that call read the render'); - assert.equal(internals.pendingRequests.length, 0, 'and nothing is left queued'); + assert.deepEqual(withdrawn, ['other'], "'other''s turn got nothing of the primary's render"); }); test("a render placed in a live turn's deferred queue isn't read by the resident that rendered it; the queue's agent is woken for it", async (t) => { @@ -1144,12 +1197,15 @@ test("a render placed in a live turn's deferred queue isn't read by the resident // token stands in for one. 'other''s assembly renders the batch, and the // delivery, bound for the primary's context, waits in its queue. internals.activeTurnTokens.set('agent', -1); - await settles(internals.processInferenceRequests(), 'one scheduler pass'); - assert.equal(f.membrane.calls.length, 0, "'other' read nothing, so its turn started no inference"); - assert.deepEqual(withdrawn, ['other']); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], 'the primary is woken for the render'); - assert(!f.context().includes('document_diff'), 'which is still in its queue'); - internals.activeTurnTokens.delete('agent'); + try { + await settles(internals.processInferenceRequests(), 'one scheduler pass'); + assert.equal(f.membrane.calls.length, 0, "'other' read nothing, so its turn started no inference"); + assert.deepEqual(withdrawn, ['other']); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], 'the primary is woken for the render'); + assert(!f.context().includes('document_diff'), 'which is still in its queue'); + } finally { + endStandInTurn(internals); + } await settles(f.framework.runUntilIdle(), "the primary's turn"); assert.equal(f.membrane.calls.length, 1); assert(f.lastRequest().includes('document_diff'), "the primary's turn read the render once its queue was flushed"); @@ -1173,20 +1229,117 @@ test("a render placed in a live turn's deferred queue leaves the wakes of a newe // batch, here held. A notice during the render opens a newer batch of the // subject, which wakes its audience. internals.activeTurnTokens.set('agent', -1); - const pass = internals.processInferenceRequests(); - await settles(started, "the render for 'other'"); - await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); - const woken = internals.pendingRequests.map((r) => r.agentName).sort(); - assert.deepEqual(woken, ['agent', 'other']); - release({ content: [{ type: 'text', text: 'first_render' }] }); - await settles(pass, 'the scheduler pass'); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName).sort(), woken, "the newer batch's wakes are as they were"); - internals.activeTurnTokens.delete('agent'); + try { + const pass = internals.processInferenceRequests(); + await settles(started, "the render for 'other'"); + await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); + const woken = internals.pendingRequests.map((r) => r.agentName).sort(); + assert.deepEqual(woken, ['agent', 'other']); + release({ content: [{ type: 'text', text: 'first_render' }] }); + await settles(pass, 'the scheduler pass'); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName).sort(), [...woken, 'agent'].sort(), + "the newer batch's wakes are as they were, beside the primary's wake for the render"); + } finally { + endStandInTurn(internals); + } await settles(f.framework.runUntilIdle(), 'the drain'); assert.equal(f.membrane.calls.length, 1); assert(f.lastRequest().includes('first_render') && f.lastRequest().includes('second_render'), 'one call read both renders'); }); +test("a render placed in a live turn's deferred queue wakes its agent though a newer batch of its subject wakes no one", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS, server: QUIET_PUSHES_DONT_TRIGGER }); t.after(f.close); + f.alwaysRespond(); + let release!: (value: unknown) => void; + let began!: () => void; + const started = new Promise((resolve) => { began = resolve; }); + let renders = 0; + f.renderer(() => { + if (++renders > 1) return Promise.resolve({ content: [{ type: 'text', text: 'second_render' }] }); + began(); + return new Promise((resolve) => { release = resolve; }); + }); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // As above, but the notice during the render triggers no inference: the + // newer batch it opens replaces the subject's wakes with none. + internals.activeTurnTokens.set('agent', -1); + try { + const pass = internals.processInferenceRequests(); + await settles(started, "the render for 'other'"); + await f.send('push/event', f.params('2', 'quiet_fallback', { deferred: true })); + assert.equal(internals.pendingRequests.length, 0, 'the newer batch wakes no one'); + release({ content: [{ type: 'text', text: 'first_render' }] }); + await settles(pass, 'the scheduler pass'); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], 'the primary is woken for the render in its queue'); + } finally { + endStandInTurn(internals); + } + await settles(f.framework.runUntilIdle(), "the primary's turn"); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('first_render'), "the primary's turn read the render"); +}); + +test("a render queued for an agent that its running turn already read gives the agent's re-woken turn nothing", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // As above, 'other''s assembly renders the batch into the primary's queue, + // and the primary is woken for it. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'one scheduler pass'); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent']); + // The primary's running turn reads the render before it ends, at a tool + // boundary that stores what waited in its queue, into the live turn's + // context, and marks it read. + internals.readdDrained(internals.drainDeferredFor('agent'), 'a tool boundary', { bypassDeferral: true }); + internals.ackDeferredWrites(); + f.framework.getAgent('agent')!.markContextConsumed(); + } finally { + endStandInTurn(internals); + } + await settles(f.framework.runUntilIdle(), "the primary's re-woken turn"); + assert(f.context().includes('document_diff'), 'the render was stored'); + assert.equal(f.membrane.calls.length, 0, 'the re-woken turn had nothing new to read'); + assert.deepEqual(withdrawn, ['other', 'agent']); +}); + +test("a render queued for one resident isn't another resident's to read once its flush has stored it", async (t) => { + const f = await fixture({ agents: [...TWO_RESIDENTS, { name: 'third', model: 'test', systemPrompt: 'test' }] }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // One pass takes all three wakes. The primary's turn is alive (a stand-in + // token), so its wake waits; 'other''s assembly renders the batch into the + // primary's queue, and the primary is woken for it. Before 'third''s turn + // starts, the primary's turn ends, and its flush stores the render, as a + // torn-down turn's does. + const start = internals.startAgentStream.bind(internals); + internals.startAgentStream = async (...a: unknown[]) => { + if ((a[0] as { name: string }).name === 'third') { + internals.startAgentStream = start; + endStandInTurn(internals); + } + return start(...a); + }; + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'the pass'); + } finally { + endStandInTurn(internals); + } + assert(f.context().includes('document_diff'), 'the render is stored'); + assert.equal(f.membrane.calls.length, 0, "'third' didn't read the primary's render"); + await settles(f.framework.runUntilIdle(), "the primary's turn"); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('document_diff'), 'the primary, woken for it, read it'); + assert.deepEqual(withdrawn, ['other', 'third']); +}); + test("a batch consumed by one resident's request gives another resident's batch-only turn nothing", async (t) => { const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); f.alwaysRespond(); @@ -1196,7 +1349,7 @@ test("a batch consumed by one resident's request gives another resident's batch- assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent', 'other'], 'both residents are woken'); // One scheduler pass takes both wakes. 'agent''s assembly renders the // batch and its request reads it; 'other''s turn then finds the batch - // consumed (RFC-006 §3.2), as if its wake had still been queued when the + // consumed (RFC-006 §5.2), as if its wake had still been queued when the // freeze withdrew it. await settles(f.framework.runUntilIdle(), 'the drain'); assert.equal(f.membrane.calls.length, 1, 'one call for the batch'); From ec241c766896c5e66a4671013cacca57e8866eca Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 07:44:17 -0400 Subject: [PATCH 4/8] fix(mcpl): a wake to read a queued render outlives withdrawals of its subject's wakes The review of 7d00a526 (af-f103-f104-r2) found that the reader's wake, queued for the agent whose deferred queue holds a render no request consumed, carried the subject like every batch wake, so any withdrawal of the subject's wakes took it too: another agent's freeze of a newer batch, a producer's retry, a recovery timer. When that newer batch then gave nothing (an empty, cancelled or revoked render), nothing woke the reader for the render it held. The wake is now marked (InferenceRequest's new host-owned coalescingRead), and cancelCoalescedWakes leaves it: it is for that render, which a later batch doesn't replace, and a stale one costs a turn whose assembly finds nothing of it and is withdrawn. Tests: a newer batch frozen by the other agent's assembly, rendering nothing, leaves the primary's wake for its queued render; a render still in the deferred queue because no flush could store it gives the re-woken turn nothing (Meg-2646's suggestion, pinning isStoredUnread's deferred half); the path-2 test now checks that the joining assembly waits for the render in flight and settles without taking it. Comments: the queued render is read under the freeze's lock, before any render it starts can settle; the gate names both reasons a cause can stay unsettled. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 21 ++++++--- src/mcpl/push-coalescer.ts | 3 +- src/types/agent.ts | 5 +++ test/mcpl-coalescing.test.ts | 85 +++++++++++++++++++++++++++++++++++- 4 files changed, 105 insertions(+), 9 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index f94f6f95e..5b29ec2b2 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11712,7 +11712,7 @@ export class AgentFramework { remove: (p) => this.removePlacement(p), deliver: (occ, materialized, assemblingFor) => this.deliverCoalesced(occ, materialized, assemblingFor), wakeForBatch: (occ) => this.wakeForCoalescedBatch(occ), - wakeToRead: (occ, agentName) => this.queueCoalescedWakes({ ...occ, deliverTo: agentName }), + wakeToRead: (occ, agentName) => this.queueCoalescedWakes({ ...occ, deliverTo: agentName }, true), cancelWake: (subject) => this.cancelCoalescedWakes(subject), authorized: (occ) => this.coalescedAuthorized(occ), audience: (occ) => this.coalescedAudience(occ), @@ -12104,8 +12104,9 @@ export class AgentFramework { await this.queueCoalescedWakes(occ); } - /** Queue a wake for each agent of the batch's audience, when its latest occurrence triggers inference. */ - private async queueCoalescedWakes(occ: CoalescedOccurrence): Promise { + /** Queue a wake for each agent of the batch's audience, when its latest + * occurrence triggers inference; `read`: for a render queued for it. */ + private async queueCoalescedWakes(occ: CoalescedOccurrence, read = false): Promise { const subject = coalescingSubjectKey(occ.serverId, occ.binding, occ.scope, occ.key); if (occ.event.lane !== 'push' || !occ.event.event.triggerInference) return; const origin = occ.event.event.origin; @@ -12124,15 +12125,20 @@ export class AgentFramework { addressed: isAddressedMessage(occ.tags, origin), coalescingSubject: subject, coalescingBatch: true, + ...(read ? { coalescingRead: true } : {}), }); } } - /** Withdraw unstarted wakes whose sole cause is this subject (§4.2). */ + /** Withdraw unstarted wakes whose sole cause is this subject (§4.2). A wake + * to read a render queued for its agent stays: it is for that render, + * which a later batch of the subject doesn't replace, and if the render is + * gone or read by then, the turn finds nothing of it at its assembly. */ private cancelCoalescedWakes(subject: string): void { - this.pendingRequests = this.pendingRequests.filter((r) => r.coalescingSubject !== subject); + const keep = (r: InferenceRequest) => r.coalescingSubject !== subject || r.coalescingRead === true; + this.pendingRequests = this.pendingRequests.filter(keep); for (const cooldown of this.providerAccelerationCooldowns.values()) { - cooldown.heldRequests = cooldown.heldRequests.filter((r) => r.coalescingSubject !== subject); + cooldown.heldRequests = cooldown.heldRequests.filter(keep); } } @@ -14789,7 +14795,8 @@ export class AgentFramework { // its causes (a batch whose wake the freeze withdrew, or one that never // qualified for a wake), and so does a render queued for this agent that // the flush has stored and nothing has read. A cause this assembly did not - // settle (the host stopping) keeps the turn, as before. + // settle (the host stopping, or its subject moving on during the audience + // lookup) keeps the turn, as before. if (assembly && batchSubjects?.length && assembly.materialized.size === 0 && batchSubjects.every((s) => assembly.settled.has(s))) { console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`); diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 0cd569aa3..6d28a502a 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -713,7 +713,8 @@ export class PushCoalescer { // The handle is wrapped: a bare promise returned through `locked` would // be flattened by `then`, holding the lock until the render settled — // and the settlement itself needs the lock. The queued render is read - // before the freeze, whose render of a newer batch would clear it. + // under the same lock, before any render this freeze starts can settle + // and take the occupant's place. const { queued, started } = await this.locked(async () => ({ queued: cause && this.queuedFor(subject, agentName), started: await this.freeze(subject, agentName, cause), diff --git a/src/types/agent.ts b/src/types/agent.ts index 3e71079a2..876e0cb37 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -351,6 +351,11 @@ export interface InferenceRequest { /** Host-owned: this wake's cause is an RFC-006 deferred batch, which has * no model-visible content until it renders at the turn's assembly. */ coalescingBatch?: boolean; + /** Host-owned, with `coalescingBatch`: the wake is for a render of the + * subject that no request consumed, placed in this agent's deferred + * queue. Withdrawing the subject's wakes leaves it, since it is about that + * render, not the subject's batch. */ + coalescingRead?: boolean; /** Host-owned, set on a turn's trigger when EVERY request batched into it * was a deferred-batch wake: these are their subjects. If assembly settles * all of them without materializing content, the turn has no cause left diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index 7423e6ef9..138ab41e8 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1165,7 +1165,12 @@ test("a batch-only turn that reaches its assembly while another resident's assem const assemble = coalescer.assemble.bind(coalescer); let joins!: () => void; const otherAssembles = new Promise((resolve) => { joins = resolve; }); - coalescer.assemble = (agent, causes) => { if (agent === 'other') joins(); return assemble(agent, causes); }; + let otherAssembly: ReturnType | undefined; + coalescer.assemble = (agent, causes) => { + const assembly = assemble(agent, causes); + if (agent === 'other') { otherAssembly = assembly; joins(); } + return assembly; + }; const start = internals.startAgentStream.bind(internals); let otherTurn: Promise = new Promise(() => {}); let otherStarts!: () => void; @@ -1178,6 +1183,13 @@ test("a batch-only turn that reaches its assembly while another resident's assem otherGate.auxiliaryInFlight = 0; for (const resolve of otherGate.idleWaiters.splice(0)) resolve(); await settles(Promise.race([otherAssembles, otherStarted.then(() => otherTurn)]), "'other''s assembly or its turn"); + if (otherAssembly) { + // 'other' joined the render in flight, so its assembly waits for it. + let settledEarly = false; + void otherAssembly.then(() => { settledEarly = true; }); + await new Promise((resolve) => setImmediate(resolve)); + assert(!settledEarly, "'other''s assembly waits for the render in flight"); + } release({ content: [{ type: 'text', text: 'shared_render' }] }); await settles(pass, "the primary's turn"); await settles(otherTurn, "'other''s turn"); @@ -1185,6 +1197,9 @@ test("a batch-only turn that reaches its assembly while another resident's assem assert.equal(f.membrane.calls.length, 1, 'the batch got one model call'); assert(f.lastRequest().includes('shared_render'), 'and that call read the render'); assert.deepEqual(withdrawn, ['other'], "'other''s turn got nothing of the primary's render"); + const joined = await settles(otherAssembly!, "'other''s assembly"); + assert.equal(joined.settled.size, 1, "'other''s assembly settled the subject once the render did"); + assert.equal(joined.materialized.size, 0, 'and took nothing of it as its own'); }); test("a render placed in a live turn's deferred queue isn't read by the resident that rendered it; the queue's agent is woken for it", async (t) => { @@ -1280,6 +1295,44 @@ test("a render placed in a live turn's deferred queue wakes its agent though a n assert(f.lastRequest().includes('first_render'), "the primary's turn read the render"); }); +test("a render placed in a live turn's deferred queue keeps its agent's wake when another agent's assembly freezes a newer batch of its subject that gives nothing", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + let release!: (value: unknown) => void; + let began!: () => void; + const started = new Promise((resolve) => { began = resolve; }); + let renders = 0; + f.renderer(() => { + if (++renders > 1) return Promise.resolve({ content: [] }); + began(); + return new Promise((resolve) => { release = resolve; }); + }); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // As above: the primary's turn is alive, 'other''s assembly renders the + // batch into the primary's queue, and a notice during the render opens a + // newer batch, which wakes both. In the next pass the primary is still + // busy, and 'other''s assembly freezes the newer batch, whose render is + // empty: the freeze withdraws the subject's wakes, and nothing is placed. + internals.activeTurnTokens.set('agent', -1); + try { + const pass = internals.processInferenceRequests(); + await settles(started, "the render for 'other'"); + await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); + release({ content: [{ type: 'text', text: 'first_render' }] }); + await settles(pass, 'the first pass'); + await settles(internals.processInferenceRequests(), 'the second pass'); + assert.deepEqual(withdrawn, ['other', 'other']); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], "the primary's wake for the render outlived the newer batch's freeze"); + } finally { + endStandInTurn(internals); + } + await settles(f.framework.runUntilIdle(), "the primary's turn"); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('first_render'), "the primary's turn read the render"); +}); + test("a render queued for an agent that its running turn already read gives the agent's re-woken turn nothing", async (t) => { const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); f.alwaysRespond(); @@ -1307,6 +1360,36 @@ test("a render queued for an agent that its running turn already read gives the assert.deepEqual(withdrawn, ['other', 'agent']); }); +test("a render still in its agent's deferred queue, which the flush couldn't store, gives the agent's re-woken turn nothing", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals & { deferredMessages: unknown[] }; + // As above, 'other''s assembly renders the batch into the primary's queue, + // and the primary is woken for it. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'one scheduler pass'); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent']); + } finally { + internals.activeTurnTokens.delete('agent'); + } + // The store takes no writes, so every flush is cut short at its first + // entry and the render stays queued (writeDrainedInOrder): the primary's + // turn-start flush leaves it there, and its compile wouldn't show it. + const drain = internals.drainDeferredFor.bind(internals); + internals.drainDeferredFor = () => []; + try { + await settles(f.framework.runUntilIdle(), "the primary's re-woken turn"); + assert.equal(f.membrane.calls.length, 0, 'the re-woken turn had nothing it could read'); + assert.deepEqual(withdrawn, ['other', 'agent']); + assert.equal(internals.deferredMessages.length, 1, 'the render is still queued'); + } finally { + internals.drainDeferredFor = drain; + } +}); + test("a render queued for one resident isn't another resident's to read once its flush has stored it", async (t) => { const f = await fixture({ agents: [...TWO_RESIDENTS, { name: 'third', model: 'test', systemPrompt: 'test' }] }); t.after(f.close); f.alwaysRespond(); From 36940ce49ca45729cf83adc30eae07faa27d6aaa Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 07:47:13 -0400 Subject: [PATCH 5/8] fix(mcpl): the quiesce park keeps a reader's wake apart from its subject's other wakes Meg-2646's read of ec241c76 found the sibling of af-f103-f104-r2's I1 in the quiesce path: parked wakes are de-duplicated by reason, addressedness and subject, newest kept, so a wake to read a queued render and a newer batch wake of the same subject shared a key. When the batch wake was kept, a later withdrawal of the subject's wakes (a producer's retry, a recovery timer) removed it, and the reader had no wake for its render after resume. The key now carries coalescingRead, so the two stay distinct. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index 5b29ec2b2..02c1fcb14 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -12544,8 +12544,10 @@ export class AgentFramework { const newestByKey = new Map(); for (const request of parked) { // A coalesced wake is withdrawable by its subject (RFC-006 §4.2); it - // must not absorb an unrelated wake that would vanish with it. - const key = `${request.reason}|${request.addressed ? 'a' : 'n'}|${request.coalescingSubject ?? ''}`; + // must not absorb an unrelated wake that would vanish with it. A wake + // to read a queued render outlives the subject's other wakes, so it + // keeps its own key too. + const key = `${request.reason}|${request.addressed ? 'a' : 'n'}|${request.coalescingSubject ?? ''}|${request.coalescingRead ? 'r' : ''}`; const prev = newestByKey.get(key); if (!prev || request.timestamp >= prev.timestamp) { newestByKey.set(key, request); From 88edb1a519ff6db16cdbcaba4ae1c3639a089d51 Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 09:04:26 -0400 Subject: [PATCH 6/8] fix(mcpl): answer Greptile's review of #520: provider retries, shards, merged wakes, a taken render Greptile's review of 36940ce4 found four defects, each pinned now by a test that fails at 36940ce4 and passes here: - A provider hold's retry was withdrawn. The gate assembles every batch-only trigger at attempt 0, and a trigger an acceleration cooldown or a stated wait kept after its turn's call failed came back as one: its batches consumed, the gate withdrew it, and the agent never answered. A trigger that passed the gate is now kept by a hold as a retry of its turn (retainedForRetry, gatedTurnTriggers), without its coalesced-wake marks, so neither the gate nor a withdrawal of its subject's wakes drops it. The trigger itself isn't changed, since getActiveTurnTrigger reports it mid-turn. - A render stored as shards counted as read. isStoredUnreadPlacement read through isUnreadPlacement, whose shard rule is about replacing a message in place. The rule moves to isUnreadPlacement (isShard), and isUnreadStoredMessage answers whether anyone has read a message. - A read wake lost its protection where wakes are merged. The scheduler's merged trigger took coalescingRead from its first request only, so a read wake merged behind a batch wake and requeued from the auxiliary park was withdrawn with the subject; it now takes the mark from any request it merges. The provider cooldown's sameInferenceRequest now tells apart wakes of different subjects or read marks, as the quiesce park's key does. - A queued render could be taken away after its reader's assembly counted it. takeQueued (was queuedFor) now consumes the occupant under the lock, as the settlement does for a materialized render, so a retraction or replacement from then on appends. Co-Authored-By: Claude Opus 5.5 --- .../coalescing-batch-turn-content.fixed.md | 2 +- src/framework.ts | 52 +++++- src/mcpl/push-coalescer.ts | 27 ++- test/mcpl-coalescing.test.ts | 168 +++++++++++++++++- 4 files changed, 228 insertions(+), 21 deletions(-) diff --git a/changelog.d/coalescing-batch-turn-content.fixed.md b/changelog.d/coalescing-batch-turn-content.fixed.md index 2eb536759..231ae5eec 100644 --- a/changelog.d/coalescing-batch-turn-content.fixed.md +++ b/changelog.d/coalescing-batch-turn-content.fixed.md @@ -1 +1 @@ -- A turn woken only by deferred coalescing batches is withdrawn before inference in more of the cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken (the replacement or the retraction's notice wakes agents as any push does, by its own `triggerInference`), its batch was being rendered for another agent's assembly (the turn waits for that render, which the other agent's request reads), or the render was placed in another agent's deferred queue because that agent's turn was running. In that last case that agent is now woken for the render, as the batch's own wake would have woken it; before, nothing woke it for the render. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. +- A turn woken only by deferred coalescing batches is withdrawn before inference in more of the cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken (the replacement or the retraction's notice wakes agents as any push does, by its own `triggerInference`), its batch was being rendered for another agent's assembly (the turn waits for that render, which the other agent's request reads), or the render was placed in another agent's deferred queue because that agent's turn was running. In that last case that agent is now woken for the render, as the batch's own wake would have woken it; before, nothing woke it for the render. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. A turn that a provider hold kept after its call failed is retried when the hold ends, even if a newer batch of its subject arrived meanwhile; before, that batch's wake replaced it, and one that triggered no inference left the held turn unanswered. A provider cooldown also keeps two subjects' wakes of the same moment apart, where before it held only one of them. diff --git a/src/framework.ts b/src/framework.ts index 02c1fcb14..de069e25b 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -2166,6 +2166,10 @@ export class AgentFramework { * guessing from the newest message in the window. */ private activeTurnTriggers: Map = new Map(); + /** Triggers whose turn has passed the batch gate (beginAgentTurn): a + * provider hold keeps such a trigger as a retry of its turn + * (retainedForRetry). */ + private readonly gatedTurnTriggers = new WeakSet(); /** The trigger of `agentName`'s turn in progress, if a turn is running. */ getActiveTurnTrigger(agentName: string): InferenceRequest | undefined { @@ -3569,9 +3573,23 @@ export class AgentFramework { } return { delayMs: ownMs, ownMs, stated: false }; } + /** Two wakes that a withdrawal of a subject's wakes would treat differently + * aren't the same request, as at the quiesce park: a coalesced wake must + * not absorb another subject's wake, or a wake to read a queued render. */ private sameInferenceRequest(a: InferenceRequest, b: InferenceRequest): boolean { return a.agentName === b.agentName && a.reason === b.reason && a.source === b.source && - a.timestamp === b.timestamp && a.channelId === b.channelId; + a.timestamp === b.timestamp && a.channelId === b.channelId && + a.coalescingSubject === b.coalescingSubject && !!a.coalescingRead === !!b.coalescingRead; + } + /** The request a provider hold keeps for `request`. A trigger whose turn + * passed its batch gate is a retry of that turn, which has read its + * batches, rather than a coalesced wake: neither a withdrawal of its + * subject's wakes (§4.2 withdraws a wake whose sole cause is the subject) + * nor the gate, finding the batches consumed, may drop it. */ + private retainedForRetry(request: InferenceRequest): InferenceRequest { + if (!this.gatedTurnTriggers.has(request)) return request; + const { coalescingSubject: _subject, coalescingBatch: _batch, coalescingBatchSubjects: _subjects, coalescingRead: _read, ...retry } = request; + return retry; } private consultProviderHold(error: Error, agentName: string): { holdMs: number; reason: string } | undefined { if (!this.providerHoldHook) return undefined; @@ -3631,7 +3649,8 @@ export class AgentFramework { const delayMs = hostHold ? hostHold.holdMs : pacing!.delayMs; const existing = this.providerAccelerationCooldowns.get(agent.name); const held = existing?.heldRequests ?? []; - if (trigger && !held.some((r) => this.sameInferenceRequest(r, trigger))) held.push(trigger); + const retry = trigger && this.retainedForRetry(trigger); + if (retry && !held.some((r) => this.sameInferenceRequest(r, retry))) held.push(retry); if (existing) clearTimeout(existing.timer); const timer = setTimeout(() => this.releaseProviderAccelerationCooldown(agent.name), Math.min(delayMs, MAX_TIMER_DELAY_MS)); timer.unref?.(); this.providerAccelerationCooldowns.set(agent.name, { startedAt: existing?.startedAt ?? now, @@ -3828,7 +3847,7 @@ export class AgentFramework { if (until <= now) return false; const existing = this.providerAccelerationCooldowns.get(agent.name); const held = existing?.heldRequests ?? []; - for (const request of requests) { + for (const request of requests.map((r) => this.retainedForRetry(r))) { if (!held.some((r) => this.sameInferenceRequest(r, request))) held.push(request); } // An acceleration or host hold already parks the whole agent; its release @@ -12164,30 +12183,41 @@ export class AgentFramework { } } + /** Unread and replaceable in place: queued, or stored, unread and not a + * body-group shard, which the store never edits or removes alone. */ private isUnreadPlacement(p: CoalescingPlacement): boolean { this.resolvePlacement(p); if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId); if (!p.messageId) return false; const agent = this.agents.get(p.agent); - return !!agent && this.isUnreadStoredMessage(agent, p.messageId); + return !!agent && !this.isShard(agent, p.messageId) && this.isUnreadStoredMessage(agent, p.messageId); } - /** Stored, a deferred entry once its flush has stored it, and unread. */ + /** Stored, a deferred entry once its flush has stored it, and unread. A + * render stored as shards counts: whether a message can be replaced in + * place says nothing about whether anyone has read it. */ private isStoredUnreadPlacement(p: CoalescingPlacement): boolean { this.resolvePlacement(p); - return !p.deferredId && this.isUnreadPlacement(p); + if (p.deferredId || !p.messageId) return false; + const agent = this.agents.get(p.agent); + return !!agent && this.isUnreadStoredMessage(agent, p.messageId); + } + + /** A shard of a body group; conservative when the store can't say. */ + private isShard(agent: Agent, messageId: MessageId): boolean { + try { return agent.getContextManager().getMessage(messageId)?.bodyGroupId !== undefined; } catch { return true; } } /** * Unread = stored, above the agent's consumed watermark on the current - * branch, not a body-group shard (immutable), and not folded into any - * summary (compression is consumption, §3.3). Conservative on every doubt. + * branch, and not folded into any summary (compression is consumption, + * §3.3). Conservative on every doubt. */ private isUnreadStoredMessage(agent: Agent, messageId: MessageId): boolean { const cm = agent.getContextManager(); let message: ReturnType; try { message = cm.getMessage(messageId); } catch { return false; } - if (!message || message.bodyGroupId) return false; + if (!message) return false; // Every agent that can read this message must not have compiled past it. // Residents share one message slot, so a message one of them read is // history for all of them (the never-rewrite rule is about the message, @@ -12698,6 +12728,9 @@ export class AgentFramework { await this.startAgentStream(agent, { ...trigger, coalescingBatch: batchOnly || undefined, + // Likewise for the whole batch: a merged wake that holds a wake to + // read a queued render outlives its subject's withdrawal, requeued too. + coalescingRead: requests.some((r) => r.coalescingRead === true) || undefined, coalescingBatchSubjects: batchOnly ? [...new Set(requests.flatMap((r) => r.coalescingBatchSubjects ?? [r.coalescingSubject!]))] : undefined, @@ -14816,6 +14849,7 @@ export class AgentFramework { } return false; } + if (trigger) this.gatedTurnTriggers.add(trigger); // Record turn checkpoint before inference (only on first attempt, not retries) if (attempt === 0 && trigger?.reason !== 'tool_result_guard_retry') { diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 6d28a502a..ee1c045fb 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -701,7 +701,8 @@ export class PushCoalescer { * by now, or another assembly's render, settles and gives the turn nothing * of it. The exception is a render no request consumed, placed in this * agent's deferred queue: this agent was woken to read it, and once the - * flush has stored it, the turn reads it as its own. + * flush has stored it, this request consumes it, as it would a render it + * materialized. */ async assemble(agentName: string, causes: readonly string[] = []): Promise { const result: AssemblyResult = { settled: new Set(), materialized: new Set() }; @@ -712,11 +713,11 @@ export class PushCoalescer { const cause = wanted.has(subject); // The handle is wrapped: a bare promise returned through `locked` would // be flattened by `then`, holding the lock until the render settled — - // and the settlement itself needs the lock. The queued render is read + // and the settlement itself needs the lock. The queued render is taken // under the same lock, before any render this freeze starts can settle // and take the occupant's place. const { queued, started } = await this.locked(async () => ({ - queued: cause && this.queuedFor(subject, agentName), + queued: cause && this.takeQueued(subject, agentName), started: await this.freeze(subject, agentName, cause), })); if (queued) { result.settled.add(subject); result.materialized.add(subject); } @@ -735,13 +736,21 @@ export class PushCoalescer { } /** - * A render of `subject` that no request consumed, placed in `agentName`'s - * deferred queue, which the flush has stored and nothing has read yet: - * `agentName` was woken to read it. + * Under the lock: whether a render of `subject` that no request consumed + * waits in `agentName`'s context, placed in its deferred queue, stored by + * the flush and read by nothing yet. `agentName` was woken to read it, and + * the request being assembled consumes it, as the settlement records a + * materialized render, so a retraction or a replacement from here on + * appends rather than taking it away from that request. */ - private queuedFor(subject: string, agentName: string): boolean { - const occupant = this.subjects.get(subject)?.occupant; - return occupant?.reader === agentName && this.host.isStoredUnread(occupant.placement); + private takeQueued(subject: string, agentName: string): boolean { + const state = this.subjects.get(subject); + const occupant = state?.occupant; + if (!state || occupant?.reader !== agentName || !this.host.isStoredUnread(occupant.placement)) return false; + state.history = 'some'; + state.consumedEventId = occupant.eventId; + state.occupant = undefined; + return true; } /** diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index 138ab41e8..ec8630db5 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -12,7 +12,9 @@ import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join, dirname } from 'node:path'; import { fileURLToPath } from 'node:url'; -import { createMockResponse } from './helpers/mock-membrane.js'; +import { MembraneError } from '@animalabs/membrane'; +import type { StreamEvent, YieldingStream } from '@animalabs/membrane'; +import { createMockResponse, MockYieldingStream } from './helpers/mock-membrane.js'; import { fixture, eventually, settles, crash, ok, TS, REPLY_DEADLINE_MS } from './helpers/coalescing-fixture.js'; // --------------------------------------------------------------------------- @@ -1008,11 +1010,16 @@ test('an assembly for an agent outside the audience of a batch another assembly // another agent's queue (a turn with nothing to read never reaches the model) // --------------------------------------------------------------------------- +type Wake = { agentName: string; reason?: string; source?: string; timestamp?: number; coalescingSubject?: string; coalescingBatch?: boolean; coalescingRead?: boolean }; + type TurnInternals = { startAgentStream: (...a: unknown[]) => Promise; processInferenceRequests: () => Promise; activeTurnTokens: Map; - pendingRequests: Array<{ agentName: string; coalescingSubject?: string; coalescingBatch?: boolean }>; + pendingRequests: Wake[]; + cancelCoalescedWakes(subject: string): void; + providerAccelerationCooldowns: Map; heldRequests: Wake[]; reason: string; failures: number }>; + releaseProviderAccelerationCooldown(agent: string): void; providerGate(agent: string): { auxiliaryInFlight: number; idleWaiters: Array<() => void> }; drainDeferredFor(agent: string): unknown[]; readdDrained(entries: unknown[], where: string, opts?: { bypassDeferral?: boolean }): unknown[]; @@ -1440,6 +1447,163 @@ test("a batch consumed by one resident's request gives another resident's batch- assert.deepEqual(withdrawn, ['other']); }); +/** A stream that fails with `error` before any content. */ +class FailingStream implements YieldingStream { + isWaitingForTools = false; + pendingToolCallIds: string[] = []; + toolDepth = 0; + isCancelled = false; + constructor(private readonly error: Error) {} + provideToolResults(): void { throw new Error('not waiting'); } + cancel(): void { this.isCancelled = true; } + async *[Symbol.asyncIterator](): AsyncIterator { + yield { type: 'error', error: this.error } as StreamEvent; + } +} + +test("a batch-only turn whose call a provider's capacity limit held is retried when the hold ends, not withdrawn", async (t) => { + const f = await fixture(); t.after(f.close); + const withdrawn = withdrawals(f); + // The first call meets an organization acceleration 429, which holds the + // turn's trigger for one fresh compile once the cooldown ends. + let calls = 0; + (f.membrane as unknown as { streamYielding: unknown }).streamYielding = (request: unknown) => { + f.membrane.calls.push(request as never); + if (++calls > 1) return new MockYieldingStream([ok()]); + return new FailingStream(new MembraneError({ + type: 'rate_limit', retryable: true, httpStatus: 429, + message: "This request would exceed your organization's maximum usage increase rate for input tokens per minute", + rawError: { status: 429 }, rawRequest: request, + } as ConstructorParameters[0])); + }; + await f.send('push/event', f.params('1', 'held_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + await settles(f.framework.runUntilIdle(), 'the turn that meets the limit'); + assert.equal(f.membrane.calls.length, 1); + const hold = internals.providerAccelerationCooldowns.get('agent'); + assert.equal(hold?.heldRequests.length, 1, "the turn's trigger is held"); + // The cooldown's time is up, and its timer's work runs now (a release + // before its time only sets the timer again). + hold!.until = Date.now(); + internals.releaseProviderAccelerationCooldown('agent'); + await settles(f.framework.runUntilIdle(), 'the retry'); + assert.equal(f.membrane.calls.length, 2, 'the held turn was retried'); + assert(f.lastRequest().includes('document_diff'), 'and the retry read the render'); + assert.deepEqual(withdrawn, []); +}); + +test("a render placed in a live turn's deferred queue that its flush stores as shards is still the woken agent's to read", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + const withdrawn = withdrawals(f); + f.renderer(async () => ({ content: [{ type: 'text', text: 'long_render_part_one' }, { type: 'text', text: 'long_render_part_two' }] })); + // The store shards the render on ingest, one shard per block, as the + // autobiographical strategy's chunking shards a long message. + const cm = f.framework.getAgent('agent')!.getContextManager() as unknown as { strategy: Record }; + cm.strategy.chunkIngressMessage = (_participant: string, content: Array<{ text?: string }>) => + content[0]?.text === 'long_render_part_one' + ? { bodyGroupId: 'body-render', shards: content.map((block, shardIndex) => ({ content: [block], shardIndex })) } + : null; + await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // As above: 'other''s assembly renders the batch into the primary's queue + // while the primary's turn is alive, and the primary is woken for it. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'one scheduler pass'); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], 'the primary is woken for the render'); + } finally { + endStandInTurn(internals); + } + assert(f.context().includes('"bodyGroupId":"body-render"'), 'the flush stored the render as shards'); + await settles(f.framework.runUntilIdle(), "the primary's turn"); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('long_render_part_two'), "the primary's turn read the sharded render"); + assert.deepEqual(withdrawn, ['other']); +}); + +test('a wake to read a queued render keeps its protection when the scheduler merges it with a batch wake and requeues the merged wake', async (t) => { + const f = await fixture(); t.after(f.close); + const internals = f.framework as unknown as TurnInternals; + const subject = 'a-subject'; + // A newer batch's wake, then a wake to read a render queued for the agent, + // both for one subject: the scheduler merges them into one trigger. + const wake: Wake = { agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), coalescingSubject: subject, coalescingBatch: true }; + internals.pendingRequests.push(wake, { ...wake, coalescingRead: true }); + // The merged wake parks behind an auxiliary call, and another turn takes + // the agent meanwhile, so the parked wake is requeued rather than started. + const gate = internals.providerGate('agent'); + gate.auxiliaryInFlight = 1; + await settles(internals.processInferenceRequests(), 'the pass that parks the wake'); + internals.activeTurnTokens.set('agent', -1); + try { + gate.auxiliaryInFlight = 0; + for (const resolve of gate.idleWaiters.splice(0)) resolve(); + await eventually(() => internals.pendingRequests.length === 1, 'the requeued wake'); + // Another agent's freeze of the newer batch withdraws the subject's wakes. + internals.cancelCoalescedWakes(subject); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], "the merged wake kept the read wake's protection"); + } finally { + internals.activeTurnTokens.delete('agent'); + } +}); + +test("a provider cooldown holds a wake to read a queued render, and another subject's wake, apart from a batch wake of the same moment", async (t) => { + const f = await fixture(); t.after(f.close); + const internals = f.framework as unknown as TurnInternals; + const timer = setTimeout(() => {}, 60_000); + internals.providerAccelerationCooldowns.set('agent', { startedAt: Date.now(), until: Date.now() + 60_000, timer, heldRequests: [], reason: 'a test hold', failures: 1 }); + try { + const wake: Wake = { agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), coalescingSubject: 'a-subject', coalescingBatch: true }; + internals.pendingRequests.push(wake, { ...wake, coalescingRead: true }, { ...wake, coalescingSubject: 'another-subject' }); + await settles(internals.processInferenceRequests(), 'the pass that holds the wakes'); + // A freeze of the first subject's newer batch withdraws its wakes. + internals.cancelCoalescedWakes('a-subject'); + assert.deepEqual(internals.providerAccelerationCooldowns.get('agent')!.heldRequests.map((r) => `${r.coalescingSubject}${r.coalescingRead ? ' (read)' : ''}`), + ['a-subject (read)', 'another-subject'], 'the read wake and the other subject\'s wake are still held'); + } finally { + clearTimeout(timer); + internals.providerAccelerationCooldowns.delete('agent'); + } +}); + +test("a render queued for an agent is that agent's once its assembly takes it: a retraction while the assembly waits on another render doesn't take it away", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + let release!: (value: unknown) => void; + let began!: () => void; + const started = new Promise((resolve) => { began = resolve; }); + f.renderer((params) => { + if (params.key !== 'other_document') return Promise.resolve({ content: [{ type: 'text', text: 'queued_render' }] }); + began(); + return new Promise((resolve) => { release = resolve; }); + }); + await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // As above, a render of the first subject waits in the primary's queue, + // and the primary is woken for it. A batch of a second subject then wakes + // both residents, so the primary's next turn has both as causes. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'one scheduler pass'); + await f.send('push/event', { featureSet: 'doc', eventId: '2', timestamp: TS, coalesce: { key: 'other_document', deferred: true, initial: true }, + payload: { content: [{ type: 'text', text: 'other_fallback' }] } }); + assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent', 'agent', 'other']); + } finally { + endStandInTurn(internals); + } + // The primary's assembly takes the queued render and renders the second + // batch, held. The first subject is retracted meanwhile. + const pass = internals.processInferenceRequests(); + await settles(started, 'the render of the second batch'); + await f.send('push/event', f.params('3', '', { retract: true })); + release({ content: [] }); + await settles(pass, 'the scheduler pass'); + await settles(f.framework.runUntilIdle(), 'the drain'); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('queued_render'), "the primary's turn read the render its assembly took"); +}); + 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' })); From be014f401407875e1b8b8e491e4b3ae3ecce95c9 Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 10:08:36 -0400 Subject: [PATCH 7/8] fix(mcpl): an ordinary wake for a render's queue agent; a turn's trigger without batch marks past the gate Answers the internal review af-f103-f104-r3 of 7d00a526..88edb1a5 (26 issues) by replacing two mechanisms rather than patching them. The design calls are Ada-1017's (room-203 #144737, #144795). The gate copy is Meg-2646's (#144763), in place of a consumedBy record that her counterexample broke: a newer batch another agent consumed during a provider hold overwrote it, and the held retry was declined. 1. The reader goes. A render placed in another agent's deferred queue (that agent's turn is alive) now gets that agent an ordinary push wake (wakeToRead -> queueCoalescedWakes(occ, plain)), under the batch's own wake policy and without coalescingSubject or coalescingBatch, so no withdrawal of the subject's wakes reaches it. Removed with it, back to main: the occupant's reader, CoalescerHost.isStoredUnread and the host's isStoredUnreadPlacement, takeQueued, InferenceRequest.coalescingRead and its protection in cancelCoalescedWakes, the scheduler's merge, sameInferenceRequest and the quiesce park's key, the snapshot's occupant, and the shard rule's move. A cause whose batch is gone gets 'declined', with no exception. Given up: when the running turn reads the render at a tool boundary, or its flushes can't store it, the woken turn still runs, as for any push delivered during a live turn on main. r3 found four more ways the reader lost a render (I1, I7, I13, I14). 2. retainedForRetry and gatedTurnTriggers go. Once a turn passes the batch gate, beginAgentTurn goes on with a copy of its trigger without coalescingSubject, coalescingBatch and coalescingBatchSubjects. Every door a retry comes back by (a provider hold's release, the auxiliary park's requeue, the tool-result guard's restart, the scheduler's cooldown pass) then gives an ordinary wake, which the gate doesn't withdraw and no withdrawal by subject reaches. A stale wake never passes the gate, so it keeps its marks and is still withdrawn. This frame's activeTurnTriggers entry keeps the trigger as it came; a retry's frame records the copy it carries, so getActiveTurnTrigger shows a retry without these host-owned marks. Tests: the six tests of the reader's mechanisms go. A new test has a held turn retried though another resident's request consumed a newer batch of its subject during the hold (Meg's case): 'other' has an earlier wake, so the pass holds the primary's newer-batch wake after 'other''s freeze, since a release with nothing held queues a fresh compile. 13 tests in all for F103 + F104. Changelog: the fixed fragment drops the cooldown sentence, and its held-retry sentence says what main did (the newer batch's wake, or a fresh compile that names no channel or author). A changed fragment says what getActiveTurnTrigger shows during a retry (Ada-1017, room-203 #145068). --- .../coalescing-batch-turn-content.changed.md | 1 + .../coalescing-batch-turn-content.fixed.md | 2 +- src/framework.ts | 101 +++------ src/mcpl/push-coalescer.ts | 92 +++----- src/types/agent.ts | 5 - test/mcpl-coalescing.test.ts | 207 ++++-------------- 6 files changed, 104 insertions(+), 304 deletions(-) create mode 100644 changelog.d/coalescing-batch-turn-content.changed.md diff --git a/changelog.d/coalescing-batch-turn-content.changed.md b/changelog.d/coalescing-batch-turn-content.changed.md new file mode 100644 index 000000000..d6fec8049 --- /dev/null +++ b/changelog.d/coalescing-batch-turn-content.changed.md @@ -0,0 +1 @@ +- `getActiveTurnTrigger` returns the trigger of a turn that a coalesced batch woke without the host-owned `coalescingSubject`, `coalescingBatch` and `coalescingBatchSubjects` while that turn is retried: an in-turn retry, a provider hold's release, a requeue behind an auxiliary call, or the tool-result guard's restart. Those fields identify a wake that hasn't been consumed, for withdrawal, and such a turn has consumed its batches. Its other fields, such as `reason`, `source`, `channelId` and `counterparty`, are as before. diff --git a/changelog.d/coalescing-batch-turn-content.fixed.md b/changelog.d/coalescing-batch-turn-content.fixed.md index 231ae5eec..d0e6224c2 100644 --- a/changelog.d/coalescing-batch-turn-content.fixed.md +++ b/changelog.d/coalescing-batch-turn-content.fixed.md @@ -1 +1 @@ -- A turn woken only by deferred coalescing batches is withdrawn before inference in more of the cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken (the replacement or the retraction's notice wakes agents as any push does, by its own `triggerInference`), its batch was being rendered for another agent's assembly (the turn waits for that render, which the other agent's request reads), or the render was placed in another agent's deferred queue because that agent's turn was running. In that last case that agent is now woken for the render, as the batch's own wake would have woken it; before, nothing woke it for the render. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. A turn that a provider hold kept after its call failed is retried when the hold ends, even if a newer batch of its subject arrived meanwhile; before, that batch's wake replaced it, and one that triggered no inference left the held turn unanswered. A provider cooldown also keeps two subjects' wakes of the same moment apart, where before it held only one of them. +- A turn woken only by deferred coalescing batches is withdrawn before inference in more of the cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken (the replacement or the retraction's notice wakes agents as any push does, by its own `triggerInference`), its batch was being rendered for another agent's assembly (the turn waits for that render, which the other agent's request reads), or the render was placed in another agent's deferred queue because that agent's turn was running. In that last case that agent is now woken for the render, as the batch's own wake would have woken it; before, nothing woke it for the render. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. A turn woken by a batch that a provider hold kept after its call failed is now retried with its own trigger when the hold ends, even if a newer batch of its subject arrived meanwhile. Before, the newer batch withdrew it, and the agent's next turn came from the newer batch's wake or, with nothing left held, from a fresh compile that names no channel or author. diff --git a/src/framework.ts b/src/framework.ts index de069e25b..29865a05e 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -2166,10 +2166,6 @@ export class AgentFramework { * guessing from the newest message in the window. */ private activeTurnTriggers: Map = new Map(); - /** Triggers whose turn has passed the batch gate (beginAgentTurn): a - * provider hold keeps such a trigger as a retry of its turn - * (retainedForRetry). */ - private readonly gatedTurnTriggers = new WeakSet(); /** The trigger of `agentName`'s turn in progress, if a turn is running. */ getActiveTurnTrigger(agentName: string): InferenceRequest | undefined { @@ -3573,23 +3569,9 @@ export class AgentFramework { } return { delayMs: ownMs, ownMs, stated: false }; } - /** Two wakes that a withdrawal of a subject's wakes would treat differently - * aren't the same request, as at the quiesce park: a coalesced wake must - * not absorb another subject's wake, or a wake to read a queued render. */ private sameInferenceRequest(a: InferenceRequest, b: InferenceRequest): boolean { return a.agentName === b.agentName && a.reason === b.reason && a.source === b.source && - a.timestamp === b.timestamp && a.channelId === b.channelId && - a.coalescingSubject === b.coalescingSubject && !!a.coalescingRead === !!b.coalescingRead; - } - /** The request a provider hold keeps for `request`. A trigger whose turn - * passed its batch gate is a retry of that turn, which has read its - * batches, rather than a coalesced wake: neither a withdrawal of its - * subject's wakes (§4.2 withdraws a wake whose sole cause is the subject) - * nor the gate, finding the batches consumed, may drop it. */ - private retainedForRetry(request: InferenceRequest): InferenceRequest { - if (!this.gatedTurnTriggers.has(request)) return request; - const { coalescingSubject: _subject, coalescingBatch: _batch, coalescingBatchSubjects: _subjects, coalescingRead: _read, ...retry } = request; - return retry; + a.timestamp === b.timestamp && a.channelId === b.channelId; } private consultProviderHold(error: Error, agentName: string): { holdMs: number; reason: string } | undefined { if (!this.providerHoldHook) return undefined; @@ -3649,8 +3631,7 @@ export class AgentFramework { const delayMs = hostHold ? hostHold.holdMs : pacing!.delayMs; const existing = this.providerAccelerationCooldowns.get(agent.name); const held = existing?.heldRequests ?? []; - const retry = trigger && this.retainedForRetry(trigger); - if (retry && !held.some((r) => this.sameInferenceRequest(r, retry))) held.push(retry); + if (trigger && !held.some((r) => this.sameInferenceRequest(r, trigger))) held.push(trigger); if (existing) clearTimeout(existing.timer); const timer = setTimeout(() => this.releaseProviderAccelerationCooldown(agent.name), Math.min(delayMs, MAX_TIMER_DELAY_MS)); timer.unref?.(); this.providerAccelerationCooldowns.set(agent.name, { startedAt: existing?.startedAt ?? now, @@ -3847,7 +3828,7 @@ export class AgentFramework { if (until <= now) return false; const existing = this.providerAccelerationCooldowns.get(agent.name); const held = existing?.heldRequests ?? []; - for (const request of requests.map((r) => this.retainedForRetry(r))) { + for (const request of requests) { if (!held.some((r) => this.sameInferenceRequest(r, request))) held.push(request); } // An acceleration or host hold already parks the whole agent; its release @@ -11727,7 +11708,6 @@ export class AgentFramework { let unreadable: { agent: string; error: unknown } | undefined; const coalescer = new PushCoalescer({ isUnread: (p) => this.isUnreadPlacement(p), - isStoredUnread: (p) => this.isStoredUnreadPlacement(p), remove: (p) => this.removePlacement(p), deliver: (occ, materialized, assemblingFor) => this.deliverCoalesced(occ, materialized, assemblingFor), wakeForBatch: (occ) => this.wakeForCoalescedBatch(occ), @@ -12124,8 +12104,10 @@ export class AgentFramework { } /** Queue a wake for each agent of the batch's audience, when its latest - * occurrence triggers inference; `read`: for a render queued for it. */ - private async queueCoalescedWakes(occ: CoalescedOccurrence, read = false): Promise { + * occurrence triggers inference. `plain`: an ordinary push wake, without + * the subject a withdrawal of the batch's wakes goes by, for a render + * placed in its agent's deferred queue (wakeToRead). */ + private async queueCoalescedWakes(occ: CoalescedOccurrence, plain = false): Promise { const subject = coalescingSubjectKey(occ.serverId, occ.binding, occ.scope, occ.key); if (occ.event.lane !== 'push' || !occ.event.event.triggerInference) return; const origin = occ.event.event.origin; @@ -12142,22 +12124,16 @@ export class AgentFramework { channelId, counterparty: authorId ? `${occ.serverId}:user:${authorId}` : undefined, addressed: isAddressedMessage(occ.tags, origin), - coalescingSubject: subject, - coalescingBatch: true, - ...(read ? { coalescingRead: true } : {}), + ...(plain ? {} : { coalescingSubject: subject, coalescingBatch: true }), }); } } - /** Withdraw unstarted wakes whose sole cause is this subject (§4.2). A wake - * to read a render queued for its agent stays: it is for that render, - * which a later batch of the subject doesn't replace, and if the render is - * gone or read by then, the turn finds nothing of it at its assembly. */ + /** Withdraw unstarted wakes whose sole cause is this subject (§4.2). */ private cancelCoalescedWakes(subject: string): void { - const keep = (r: InferenceRequest) => r.coalescingSubject !== subject || r.coalescingRead === true; - this.pendingRequests = this.pendingRequests.filter(keep); + this.pendingRequests = this.pendingRequests.filter((r) => r.coalescingSubject !== subject); for (const cooldown of this.providerAccelerationCooldowns.values()) { - cooldown.heldRequests = cooldown.heldRequests.filter(keep); + cooldown.heldRequests = cooldown.heldRequests.filter((r) => r.coalescingSubject !== subject); } } @@ -12183,41 +12159,24 @@ export class AgentFramework { } } - /** Unread and replaceable in place: queued, or stored, unread and not a - * body-group shard, which the store never edits or removes alone. */ private isUnreadPlacement(p: CoalescingPlacement): boolean { this.resolvePlacement(p); if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId); if (!p.messageId) return false; const agent = this.agents.get(p.agent); - return !!agent && !this.isShard(agent, p.messageId) && this.isUnreadStoredMessage(agent, p.messageId); - } - - /** Stored, a deferred entry once its flush has stored it, and unread. A - * render stored as shards counts: whether a message can be replaced in - * place says nothing about whether anyone has read it. */ - private isStoredUnreadPlacement(p: CoalescingPlacement): boolean { - this.resolvePlacement(p); - if (p.deferredId || !p.messageId) return false; - const agent = this.agents.get(p.agent); return !!agent && this.isUnreadStoredMessage(agent, p.messageId); } - /** A shard of a body group; conservative when the store can't say. */ - private isShard(agent: Agent, messageId: MessageId): boolean { - try { return agent.getContextManager().getMessage(messageId)?.bodyGroupId !== undefined; } catch { return true; } - } - /** * Unread = stored, above the agent's consumed watermark on the current - * branch, and not folded into any summary (compression is consumption, - * §3.3). Conservative on every doubt. + * branch, not a body-group shard (immutable), and not folded into any + * summary (compression is consumption, §3.3). Conservative on every doubt. */ private isUnreadStoredMessage(agent: Agent, messageId: MessageId): boolean { const cm = agent.getContextManager(); let message: ReturnType; try { message = cm.getMessage(messageId); } catch { return false; } - if (!message) return false; + if (!message || message.bodyGroupId) return false; // Every agent that can read this message must not have compiled past it. // Residents share one message slot, so a message one of them read is // history for all of them (the never-rewrite rule is about the message, @@ -12574,10 +12533,8 @@ export class AgentFramework { const newestByKey = new Map(); for (const request of parked) { // A coalesced wake is withdrawable by its subject (RFC-006 §4.2); it - // must not absorb an unrelated wake that would vanish with it. A wake - // to read a queued render outlives the subject's other wakes, so it - // keeps its own key too. - const key = `${request.reason}|${request.addressed ? 'a' : 'n'}|${request.coalescingSubject ?? ''}|${request.coalescingRead ? 'r' : ''}`; + // must not absorb an unrelated wake that would vanish with it. + const key = `${request.reason}|${request.addressed ? 'a' : 'n'}|${request.coalescingSubject ?? ''}`; const prev = newestByKey.get(key); if (!prev || request.timestamp >= prev.timestamp) { newestByKey.set(key, request); @@ -12728,9 +12685,6 @@ export class AgentFramework { await this.startAgentStream(agent, { ...trigger, coalescingBatch: batchOnly || undefined, - // Likewise for the whole batch: a merged wake that holds a wake to - // read a queued render outlives its subject's withdrawal, requeued too. - coalescingRead: requests.some((r) => r.coalescingRead === true) || undefined, coalescingBatchSubjects: batchOnly ? [...new Set(requests.flatMap((r) => r.coalescingBatchSubjects ?? [r.coalescingSubject!]))] : undefined, @@ -14803,8 +14757,7 @@ export class AgentFramework { // Not on a context-budget restart: that continues the same logical turn. // A turn whose only causes are deferred batches assembles even when no // batch is pending: its batch may have been withdrawn after its wake was - // dispatched or be rendering for another agent's assembly, or the turn may - // be the one woken to read a render queued for this agent, and only the + // dispatched or be rendering for another agent's assembly, and only the // assembly can say whether the turn has anything to read. const batchSubjects = trigger?.coalescingBatchSubjects; let assembly: AssemblyResult | undefined; @@ -14828,10 +14781,8 @@ export class AgentFramework { // 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), and so does a render queued for this agent that - // the flush has stored and nothing has read. A cause this assembly did not - // settle (the host stopping, or its subject moving on during the audience - // lookup) keeps the turn, as before. + // qualified for a wake). A cause this assembly did not settle (the host + // stopping) keeps the turn, as before. if (assembly && batchSubjects?.length && assembly.materialized.size === 0 && batchSubjects.every((s) => assembly.settled.has(s))) { console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`); @@ -14849,7 +14800,19 @@ export class AgentFramework { } return false; } - if (trigger) this.gatedTurnTriggers.add(trigger); + // The turn has what its batches gave it, so from here it goes on with a + // copy of its trigger without a batch wake's marks. A retry of it then + // comes back as an ordinary wake by whatever door it takes (a provider + // hold's release, the auxiliary park's requeue, the tool-result guard's + // restart): this gate doesn't withdraw it for finding its batches + // consumed, and no withdrawal of the subject's wakes reaches it (§4.2 + // withdraws a wake whose sole cause is the subject). This frame's + // activeTurnTriggers entry keeps the trigger as it came; a retry's frame + // records the copy it carries. + if (trigger?.coalescingSubject !== undefined || trigger?.coalescingBatch) { + const { coalescingSubject: _subject, coalescingBatch: _batch, coalescingBatchSubjects: _subjects, ...retry } = trigger; + trigger = retry; + } // Record turn checkpoint before inference (only on first attempt, not retries) if (attempt === 0 && trigger?.reason !== 'tool_result_guard_retry') { diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index ee1c045fb..de00ad737 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -133,26 +133,21 @@ export interface AssemblyResult { * completion, freeze() declined them for this agent, or they were among * the turn's causes and their batch was gone or another assembly's. */ settled: Set; - /** Of those, the subjects whose content the turn being assembled reads as - * its own: a render or fallback this assembly delivered where its compile - * reads it, or, for a cause, a render queued for this agent that no request - * consumed, which the flush has stored and nothing has read yet. A settled - * subject that is not here gave the turn nothing to read: empty 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), a batch whose audience doesn't include the agent, a - * render placed in another agent's deferred queue, a render another - * assembly froze, or a cause whose batch was withdrawn, displaced or - * consumed by another request. */ + /** Of those, the subjects whose render or fallback this assembly delivered + * where its compile reads it, as content the turn reads as its own. A + * settled subject that is not here gave the turn nothing to read: empty 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), a batch whose audience doesn't include the + * agent, a render placed in another agent's deferred queue, a render + * another assembly froze, or a cause whose batch was withdrawn, displaced + * or consumed by another request. */ materialized: Set; } interface SubjectState { history: 'none' | 'some' | 'unknown'; - /** The subject's delivery that may still be unread. `reader` marks a render - * no request consumed, placed in that agent's deferred queue: the agent - * woken to read it (§5.2). */ - occupant?: { eventId: string; placement: CoalescingPlacement; reader?: string }; + occupant?: { eventId: string; placement: CoalescingPlacement }; batch?: DeferredBatch; rendering?: Rendering; consumedEventId?: string; @@ -194,10 +189,6 @@ export interface CoalescingReceiptRecord { export interface CoalescerHost { /** True iff the placement still exists and no model request has included it. */ isUnread(placement: CoalescingPlacement): boolean; - /** True iff the placement is stored in a context, no longer in the deferred - * queue, and no model request has included it: the next compile of a turn - * that reads that context shows it as new. */ - isStoredUnread(placement: CoalescingPlacement): boolean; /** False when the unread occurrence could not be removed (it may still be read). */ remove(placement: CoalescingPlacement): boolean; /** Ordinary delivery of the event; undefined when routing delivered nowhere. @@ -206,9 +197,10 @@ export interface CoalescerHost { deliver(occurrence: CoalescedOccurrence, materialized?: McplContentBlock[], assemblingFor?: string): Promise; /** Queue a wake for a batch that has no model-visible content yet. */ wakeForBatch(occurrence: CoalescedOccurrence): Promise; - /** Queue a wake for `agent` to read a render of the batch that no request - * consumed, as the batch's own wake would wake it; the subject's other - * wakes stay. */ + /** Queue an ordinary push wake for `agent`, under the batch's own wake + * policy, to read a render of the batch placed in its deferred queue that + * no request consumed. It carries no subject, so the subject's other wakes, + * and later withdrawals of them, leave it. */ wakeToRead(occurrence: CoalescedOccurrence, agent: string): Promise; /** Withdraw unstarted wakes whose sole cause is this subject. */ cancelWake(subject: string): void; @@ -697,12 +689,10 @@ export class PushCoalescer { * * `causes` are the subjects whose batch wakes started the turn. A batch has * one consumer, the request it was frozen for (§5.2: the materialized - * occurrence is consumed by that request), so a cause whose batch is gone - * by now, or another assembly's render, settles and gives the turn nothing - * of it. The exception is a render no request consumed, placed in this - * agent's deferred queue: this agent was woken to read it, and once the - * flush has stored it, this request consumes it, as it would a render it - * materialized. + * occurrence is consumed by that request). So a cause whose batch is gone + * by now settles and gives the turn nothing, and an assembly that joins + * another's render waits for it, so its compile shows the content, without + * taking it as its own. */ async assemble(agentName: string, causes: readonly string[] = []): Promise { const result: AssemblyResult = { settled: new Set(), materialized: new Set() }; @@ -710,17 +700,10 @@ export class PushCoalescer { const work: Promise[] = []; const wanted = new Set(causes); for (const subject of new Set([...this.subjects.keys(), ...causes])) { - const cause = wanted.has(subject); // The handle is wrapped: a bare promise returned through `locked` would // be flattened by `then`, holding the lock until the render settled — - // and the settlement itself needs the lock. The queued render is taken - // under the same lock, before any render this freeze starts can settle - // and take the occupant's place. - const { queued, started } = await this.locked(async () => ({ - queued: cause && this.takeQueued(subject, agentName), - started: await this.freeze(subject, agentName, cause), - })); - if (queued) { result.settled.add(subject); result.materialized.add(subject); } + // and the settlement itself needs the lock. + const started = await this.locked(() => this.freeze(subject, agentName, wanted.has(subject))); if (started === 'declined') result.settled.add(subject); else if (started) { work.push(started.done.then(() => { @@ -735,24 +718,6 @@ export class PushCoalescer { return result; } - /** - * Under the lock: whether a render of `subject` that no request consumed - * waits in `agentName`'s context, placed in its deferred queue, stored by - * the flush and read by nothing yet. `agentName` was woken to read it, and - * the request being assembled consumes it, as the settlement records a - * materialized render, so a retraction or a replacement from here on - * appends rather than taking it away from that request. - */ - private takeQueued(subject: string, agentName: string): boolean { - const state = this.subjects.get(subject); - const occupant = state?.occupant; - if (!state || occupant?.reader !== agentName || !this.host.isStoredUnread(occupant.placement)) return false; - state.history = 'some'; - state.consumedEventId = occupant.eventId; - state.occupant = undefined; - return true; - } - /** * Under the lock: decide whether `subject`'s batch renders for `agentName`, * freeze it, make the render-start durable, re-check authority, and start @@ -774,8 +739,10 @@ export class PushCoalescer { // The batch whose wake started this turn is gone: a retraction or a // displacement withdrew it after the wake was dispatched, or another // request consumed it, the batch's one consumption, as the freeze's - // withdrawal of the other queued wakes already makes it. Whatever - // replaced it is read through its own delivery and wake. + // withdrawal of the other queued wakes already makes it. Whatever replaced + // it is read through its own delivery and wake. (A retry of a turn that + // read its batch comes back without the marks of a batch wake, so it + // never reaches here as a cause: beginAgentTurn.) if (!state || (!state.batch && !state.rendering)) return cause ? 'declined' : undefined; if (state.rendering) { // Another assembly froze this batch, and the request it assembles @@ -893,10 +860,10 @@ export class PushCoalescer { // unread there, so it stays replaceable and withdrawable. The turn // being assembled doesn't read it, so it isn't materialized for that // turn, and no request has consumed it. The freeze withdrew the - // queue's agent's wake for this batch; wake it again, beside any - // wakes a newer batch has, so that its next turn, once the flush has - // stored the content, reads it. - state.occupant = { eventId: occurrence.eventId, placement, reader: placement.agent }; + // queue's agent's wake for this batch; give it an ordinary wake, as + // any delivery during its turn has, beside any wakes a newer batch + // has, so that its next turn reads the content. + state.occupant = { eventId: occurrence.eventId, placement }; state.audienceAgent = placement.agent; await this.host.wakeToRead(occurrence, placement.agent); return; @@ -1003,8 +970,7 @@ export class PushCoalescer { subjects: [...this.subjects].map(([subject, s]) => ({ subject, history: s.history, consumedEventId: s.consumedEventId, identity: s.identity, ...(s.audienceAgent ? { audienceAgent: s.audienceAgent } : {}), - // A restart counts the occupant as read (restore), so its reader isn't kept. - ...(s.occupant ? { occupant: { eventId: s.occupant.eventId, placement: s.occupant.placement } } : {}), + ...(s.occupant ? { occupant: s.occupant } : {}), ...(s.batch ? { batch: { latest: s.batch.latest as CoalescedOccurrence, notices: s.batch.notices, dropped: s.batch.dropped, ...(s.batch.noRender ? { rendering: true } : {}) } } : s.rendering ? { batch: { latest: s.rendering.batch.latest as CoalescedOccurrence, notices: s.rendering.batch.notices, dropped: s.rendering.batch.dropped, rendering: true } } : {}), diff --git a/src/types/agent.ts b/src/types/agent.ts index 876e0cb37..3e71079a2 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -351,11 +351,6 @@ export interface InferenceRequest { /** Host-owned: this wake's cause is an RFC-006 deferred batch, which has * no model-visible content until it renders at the turn's assembly. */ coalescingBatch?: boolean; - /** Host-owned, with `coalescingBatch`: the wake is for a render of the - * subject that no request consumed, placed in this agent's deferred - * queue. Withdrawing the subject's wakes leaves it, since it is about that - * render, not the subject's batch. */ - coalescingRead?: boolean; /** Host-owned, set on a turn's trigger when EVERY request batched into it * was a deferred-batch wake: these are their subjects. If assembly settles * all of them without materializing content, the turn has no cause left diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index ec8630db5..c2fc3c36a 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1010,14 +1010,13 @@ test('an assembly for an agent outside the audience of a batch another assembly // another agent's queue (a turn with nothing to read never reaches the model) // --------------------------------------------------------------------------- -type Wake = { agentName: string; reason?: string; source?: string; timestamp?: number; coalescingSubject?: string; coalescingBatch?: boolean; coalescingRead?: boolean }; +type Wake = { agentName: string; reason?: string; source?: string; timestamp?: number; coalescingSubject?: string; coalescingBatch?: boolean }; type TurnInternals = { startAgentStream: (...a: unknown[]) => Promise; processInferenceRequests: () => Promise; activeTurnTokens: Map; pendingRequests: Wake[]; - cancelCoalescedWakes(subject: string): void; providerAccelerationCooldowns: Map; heldRequests: Wake[]; reason: string; failures: number }>; releaseProviderAccelerationCooldown(agent: string): void; providerGate(agent: string): { auxiliaryInFlight: number; idleWaiters: Array<() => void> }; @@ -1340,63 +1339,6 @@ test("a render placed in a live turn's deferred queue keeps its agent's wake whe assert(f.lastRequest().includes('first_render'), "the primary's turn read the render"); }); -test("a render queued for an agent that its running turn already read gives the agent's re-woken turn nothing", async (t) => { - const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); - f.alwaysRespond(); - const withdrawn = withdrawals(f); - await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); - const internals = f.framework as unknown as TurnInternals; - // As above, 'other''s assembly renders the batch into the primary's queue, - // and the primary is woken for it. - internals.activeTurnTokens.set('agent', -1); - try { - await settles(internals.processInferenceRequests(), 'one scheduler pass'); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent']); - // The primary's running turn reads the render before it ends, at a tool - // boundary that stores what waited in its queue, into the live turn's - // context, and marks it read. - internals.readdDrained(internals.drainDeferredFor('agent'), 'a tool boundary', { bypassDeferral: true }); - internals.ackDeferredWrites(); - f.framework.getAgent('agent')!.markContextConsumed(); - } finally { - endStandInTurn(internals); - } - await settles(f.framework.runUntilIdle(), "the primary's re-woken turn"); - assert(f.context().includes('document_diff'), 'the render was stored'); - assert.equal(f.membrane.calls.length, 0, 'the re-woken turn had nothing new to read'); - assert.deepEqual(withdrawn, ['other', 'agent']); -}); - -test("a render still in its agent's deferred queue, which the flush couldn't store, gives the agent's re-woken turn nothing", async (t) => { - const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); - f.alwaysRespond(); - const withdrawn = withdrawals(f); - await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); - const internals = f.framework as unknown as TurnInternals & { deferredMessages: unknown[] }; - // As above, 'other''s assembly renders the batch into the primary's queue, - // and the primary is woken for it. - internals.activeTurnTokens.set('agent', -1); - try { - await settles(internals.processInferenceRequests(), 'one scheduler pass'); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent']); - } finally { - internals.activeTurnTokens.delete('agent'); - } - // The store takes no writes, so every flush is cut short at its first - // entry and the render stays queued (writeDrainedInOrder): the primary's - // turn-start flush leaves it there, and its compile wouldn't show it. - const drain = internals.drainDeferredFor.bind(internals); - internals.drainDeferredFor = () => []; - try { - await settles(f.framework.runUntilIdle(), "the primary's re-woken turn"); - assert.equal(f.membrane.calls.length, 0, 'the re-woken turn had nothing it could read'); - assert.deepEqual(withdrawn, ['other', 'agent']); - assert.equal(internals.deferredMessages.length, 1, 'the render is still queued'); - } finally { - internals.drainDeferredFor = drain; - } -}); - test("a render queued for one resident isn't another resident's to read once its flush has stored it", async (t) => { const f = await fixture({ agents: [...TWO_RESIDENTS, { name: 'third', model: 'test', systemPrompt: 'test' }] }); t.after(f.close); f.alwaysRespond(); @@ -1492,116 +1434,49 @@ test("a batch-only turn whose call a provider's capacity limit held is retried w assert.deepEqual(withdrawn, []); }); -test("a render placed in a live turn's deferred queue that its flush stores as shards is still the woken agent's to read", async (t) => { +test("a turn a provider hold kept is retried though another resident's request consumed a newer batch of its subject meanwhile", async (t) => { const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); - f.alwaysRespond(); const withdrawn = withdrawals(f); - f.renderer(async () => ({ content: [{ type: 'text', text: 'long_render_part_one' }, { type: 'text', text: 'long_render_part_two' }] })); - // The store shards the render on ingest, one shard per block, as the - // autobiographical strategy's chunking shards a long message. - const cm = f.framework.getAgent('agent')!.getContextManager() as unknown as { strategy: Record }; - cm.strategy.chunkIngressMessage = (_participant: string, content: Array<{ text?: string }>) => - content[0]?.text === 'long_render_part_one' - ? { bodyGroupId: 'body-render', shards: content.map((block, shardIndex) => ({ content: [block], shardIndex })) } - : null; - await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); - const internals = f.framework as unknown as TurnInternals; - // As above: 'other''s assembly renders the batch into the primary's queue - // while the primary's turn is alive, and the primary is woken for it. - internals.activeTurnTokens.set('agent', -1); - try { - await settles(internals.processInferenceRequests(), 'one scheduler pass'); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], 'the primary is woken for the render'); - } finally { - endStandInTurn(internals); - } - assert(f.context().includes('"bodyGroupId":"body-render"'), 'the flush stored the render as shards'); - await settles(f.framework.runUntilIdle(), "the primary's turn"); - assert.equal(f.membrane.calls.length, 1); - assert(f.lastRequest().includes('long_render_part_two'), "the primary's turn read the sharded render"); - assert.deepEqual(withdrawn, ['other']); -}); - -test('a wake to read a queued render keeps its protection when the scheduler merges it with a batch wake and requeues the merged wake', async (t) => { - const f = await fixture(); t.after(f.close); - const internals = f.framework as unknown as TurnInternals; - const subject = 'a-subject'; - // A newer batch's wake, then a wake to read a render queued for the agent, - // both for one subject: the scheduler merges them into one trigger. - const wake: Wake = { agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), coalescingSubject: subject, coalescingBatch: true }; - internals.pendingRequests.push(wake, { ...wake, coalescingRead: true }); - // The merged wake parks behind an auxiliary call, and another turn takes - // the agent meanwhile, so the parked wake is requeued rather than started. - const gate = internals.providerGate('agent'); - gate.auxiliaryInFlight = 1; - await settles(internals.processInferenceRequests(), 'the pass that parks the wake'); - internals.activeTurnTokens.set('agent', -1); - try { - gate.auxiliaryInFlight = 0; - for (const resolve of gate.idleWaiters.splice(0)) resolve(); - await eventually(() => internals.pendingRequests.length === 1, 'the requeued wake'); - // Another agent's freeze of the newer batch withdraws the subject's wakes. - internals.cancelCoalescedWakes(subject); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent'], "the merged wake kept the read wake's protection"); - } finally { - internals.activeTurnTokens.delete('agent'); - } -}); - -test("a provider cooldown holds a wake to read a queued render, and another subject's wake, apart from a batch wake of the same moment", async (t) => { - const f = await fixture(); t.after(f.close); - const internals = f.framework as unknown as TurnInternals; - const timer = setTimeout(() => {}, 60_000); - internals.providerAccelerationCooldowns.set('agent', { startedAt: Date.now(), until: Date.now() + 60_000, timer, heldRequests: [], reason: 'a test hold', failures: 1 }); - try { - const wake: Wake = { agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), coalescingSubject: 'a-subject', coalescingBatch: true }; - internals.pendingRequests.push(wake, { ...wake, coalescingRead: true }, { ...wake, coalescingSubject: 'another-subject' }); - await settles(internals.processInferenceRequests(), 'the pass that holds the wakes'); - // A freeze of the first subject's newer batch withdraws its wakes. - internals.cancelCoalescedWakes('a-subject'); - assert.deepEqual(internals.providerAccelerationCooldowns.get('agent')!.heldRequests.map((r) => `${r.coalescingSubject}${r.coalescingRead ? ' (read)' : ''}`), - ['a-subject (read)', 'another-subject'], 'the read wake and the other subject\'s wake are still held'); - } finally { - clearTimeout(timer); - internals.providerAccelerationCooldowns.delete('agent'); - } -}); - -test("a render queued for an agent is that agent's once its assembly takes it: a retraction while the assembly waits on another render doesn't take it away", async (t) => { - const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); - f.alwaysRespond(); - let release!: (value: unknown) => void; - let began!: () => void; - const started = new Promise((resolve) => { began = resolve; }); - f.renderer((params) => { - if (params.key !== 'other_document') return Promise.resolve({ content: [{ type: 'text', text: 'queued_render' }] }); - began(); - return new Promise((resolve) => { release = resolve; }); - }); - await f.send('push/event', f.params('1', 'queued_fallback', { deferred: true, initial: true })); + // The first call, the primary's (its wake comes first in the pass), meets + // an organization acceleration 429, which holds its turn's trigger; 'other' + // isn't held. A release with nothing held would queue a fresh compile, so + // the case needs a held wake of the subject: the newer batch's, below. + let calls = 0; + (f.membrane as unknown as { streamYielding: unknown }).streamYielding = (request: unknown) => { + f.membrane.calls.push(request as never); + if (++calls === 1) return new FailingStream(new MembraneError({ + type: 'rate_limit', retryable: true, httpStatus: 429, + message: "This request would exceed your organization's maximum usage increase rate for input tokens per minute", + rawError: { status: 429 }, rawRequest: request, + } as ConstructorParameters[0])); + return new MockYieldingStream([ok()]); + }; + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); const internals = f.framework as unknown as TurnInternals; - // As above, a render of the first subject waits in the primary's queue, - // and the primary is woken for it. A batch of a second subject then wakes - // both residents, so the primary's next turn has both as causes. - internals.activeTurnTokens.set('agent', -1); - try { - await settles(internals.processInferenceRequests(), 'one scheduler pass'); - await f.send('push/event', { featureSet: 'doc', eventId: '2', timestamp: TS, coalesce: { key: 'other_document', deferred: true, initial: true }, - payload: { content: [{ type: 'text', text: 'other_fallback' }] } }); - assert.deepEqual(internals.pendingRequests.map((r) => r.agentName), ['agent', 'agent', 'other']); - } finally { - endStandInTurn(internals); - } - // The primary's assembly takes the queued render and renders the second - // batch, held. The first subject is retracted meanwhile. - const pass = internals.processInferenceRequests(); - await settles(started, 'the render of the second batch'); - await f.send('push/event', f.params('3', '', { retract: true })); - release({ content: [] }); - await settles(pass, 'the scheduler pass'); - await settles(f.framework.runUntilIdle(), 'the drain'); - assert.equal(f.membrane.calls.length, 1); - assert(f.lastRequest().includes('queued_render'), "the primary's turn read the render its assembly took"); + // One pass: the primary's assembly consumes the batch and its call fails; + // 'other''s turn finds the batch consumed and is withdrawn. + await settles(internals.processInferenceRequests(), 'the pass whose primary call meets the limit'); + const hold = internals.providerAccelerationCooldowns.get('agent'); + assert.equal(hold?.heldRequests.length, 1, "the primary's trigger is held"); + // During the hold, 'other' already has a wake queued (a heartbeat), so its + // turn comes first in the next pass. A newer batch of the subject arrives; + // 'other''s assembly consumes it, and its freeze withdraws the subject's + // queued and held wakes. The primary's wake for the newer batch, which the + // same pass reaches after that freeze, is held. + internals.pendingRequests.push({ agentName: 'other', reason: 'heartbeat', source: 'framework', timestamp: Date.now() }); + f.renderer(async () => ({ content: [{ type: 'text', text: 'second_render' }] })); + await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); + await settles(internals.processInferenceRequests(), "the pass where 'other' reads the newer batch"); + assert(f.lastRequest().includes('second_render'), "'other''s request read the newer batch"); + assert(hold!.heldRequests.some((r) => r.coalescingSubject !== undefined), "the primary's wake for the newer batch is held"); + const callsBefore = f.membrane.calls.length; + // The hold's time is up, and its timer's work runs now. + hold!.until = Date.now(); + internals.releaseProviderAccelerationCooldown('agent'); + await settles(f.framework.runUntilIdle(), "the primary's retry"); + assert.equal(f.membrane.calls.length, callsBefore + 1, "the primary's held turn was retried"); + assert(f.lastRequest().includes('document_diff'), 'and the retry read the first render'); + assert(!withdrawn.includes('agent'), "the primary's retry wasn't withdrawn"); }); test('R14: two consecutive crash windows with an acceptance in between keep both recovered batches', async (t) => { From b553239983766af438d0cbda7f1ce50ba7885b3c Mon Sep 17 00:00:00 2001 From: Gideon-2613 Date: Sun, 18 Oct 2026 10:56:15 -0400 Subject: [PATCH 8/8] fix(mcpl): answer af-f103-f104-r4: a merged wake carries a subject only when it is every merged wake's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers the internal review af-f103-f104-r4 of 88edb1a5..be014f40 (13 issues). - I1, which is F206 reaching this change. Ada-1017 decided to fix it here (room-203 #145364), with Meg-2646's two runs (#145363). The scheduler's merged trigger was spread from requests[0]. So the queue agent's ordinary wake, or a retry's copy, merged behind a batch wake took that wake's subject, and a withdrawal of the subject that landed while the merge was held or requeued before its gate took them with it. The merged trigger now carries coalescingSubject only when it is every merged request's (RFC-006 §4.2: a wake is withdrawn only when the subject is its sole cause). This also fixes F206 on main: a merge of a batch wake and an ordinary wake no longer loses the ordinary cause, and a batch-only merge of two subjects no longer loses the second. Two tests: a mixed merge keeps its other cause, and a batch wake requeued before its gate is still withdrawn (removals R10 and R11). - I6: the first provider-hold test asserts that the held copy carries none of the three marks, and what getActiveTurnTrigger shows on the first attempt and on the retry. - I10: the second provider-hold test sets the held wake's timestamp before the newer batch's, so the hold's de-duplication by timestamp can't make the test depend on timing. - I9: AssemblyResult's doc now says a settled subject outside `materialized` gave the turn nothing new of its own to read; a joined render may show in its compile, as that request's. - I13: the gate comment now says the turn has passed its gate, not that it has what its batches gave it. The changed fragment says a retry belongs to a turn that has passed its gate. - Changelog: a fixed bullet for the merge; the changed fragment adds the merged turn's subject. A correction to be014f40's message: of r3's findings on the reader, the real losses were I1, I7, I9, I22 and I24. I13 was judged not real, and I14 is main's behaviour. --- .../coalescing-batch-turn-content.changed.md | 2 +- .../coalescing-batch-turn-content.fixed.md | 1 + src/framework.ts | 18 ++++-- src/mcpl/push-coalescer.ts | 15 ++--- test/mcpl-coalescing.test.ts | 59 ++++++++++++++++++- 5 files changed, 81 insertions(+), 14 deletions(-) diff --git a/changelog.d/coalescing-batch-turn-content.changed.md b/changelog.d/coalescing-batch-turn-content.changed.md index d6fec8049..429074e79 100644 --- a/changelog.d/coalescing-batch-turn-content.changed.md +++ b/changelog.d/coalescing-batch-turn-content.changed.md @@ -1 +1 @@ -- `getActiveTurnTrigger` returns the trigger of a turn that a coalesced batch woke without the host-owned `coalescingSubject`, `coalescingBatch` and `coalescingBatchSubjects` while that turn is retried: an in-turn retry, a provider hold's release, a requeue behind an auxiliary call, or the tool-result guard's restart. Those fields identify a wake that hasn't been consumed, for withdrawal, and such a turn has consumed its batches. Its other fields, such as `reason`, `source`, `channelId` and `counterparty`, are as before. +- `getActiveTurnTrigger` returns the trigger of a turn that a coalesced batch woke without the host-owned `coalescingSubject`, `coalescingBatch` and `coalescingBatchSubjects` while that turn is retried: an in-turn retry, a provider hold's release, a requeue behind an auxiliary call, or the tool-result guard's restart. Those fields identify a wake for withdrawal while it is unconsumed, and a retry belongs to a turn that has already passed its batch gate. A turn whose merged wakes don't all share one subject shows no `coalescingSubject` either. Its other fields, such as `reason`, `source`, `channelId` and `counterparty`, are as before. diff --git a/changelog.d/coalescing-batch-turn-content.fixed.md b/changelog.d/coalescing-batch-turn-content.fixed.md index d0e6224c2..e0836eda7 100644 --- a/changelog.d/coalescing-batch-turn-content.fixed.md +++ b/changelog.d/coalescing-batch-turn-content.fixed.md @@ -1 +1,2 @@ - A turn woken only by deferred coalescing batches is withdrawn before inference in more of the cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken (the replacement or the retraction's notice wakes agents as any push does, by its own `triggerInference`), its batch was being rendered for another agent's assembly (the turn waits for that render, which the other agent's request reads), or the render was placed in another agent's deferred queue because that agent's turn was running. In that last case that agent is now woken for the render, as the batch's own wake would have woken it; before, nothing woke it for the render. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. A turn woken by a batch that a provider hold kept after its call failed is now retried with its own trigger when the hold ends, even if a newer batch of its subject arrived meanwhile. Before, the newer batch withdrew it, and the agent's next turn came from the newer batch's wake or, with nothing left held, from a fresh compile that names no channel or author. +- When one agent's queued wakes merge into one turn, the turn's trigger now carries a coalescing subject only when it is every merged wake's, so a withdrawal of that subject's wakes, such as a retraction's or a newer batch's, takes the merged wake only when that subject was its sole cause. Before, the merged trigger carried its first wake's subject, and such a withdrawal, landing while the merged wake was held or requeued before its turn started, also dropped the wake's other causes, such as an ordinary message. diff --git a/src/framework.ts b/src/framework.ts index 29865a05e..02cefc3e5 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -12688,6 +12688,13 @@ export class AgentFramework { coalescingBatchSubjects: batchOnly ? [...new Set(requests.flatMap((r) => r.coalescingBatchSubjects ?? [r.coalescingSubject!]))] : undefined, + // A withdrawal of a subject's wakes takes a wake whose sole cause is + // that subject (§4.2), so the merged trigger carries a subject only + // when it is every merged request's: withdrawing it then never takes + // the merge's other causes, such as an ordinary wake, another + // subject's batch, or a retry past its gate. + coalescingSubject: requests.every((r) => r.coalescingSubject === trigger?.coalescingSubject) + ? trigger?.coalescingSubject : undefined, suppressProse: silentOnly ? trigger?.suppressProse : undefined, ephemeralSystemPrompt: silentOnly ? trigger?.ephemeralSystemPrompt : undefined, silentHeartbeat: silentOnly ? trigger?.silentHeartbeat : undefined, @@ -14800,13 +14807,14 @@ export class AgentFramework { } return false; } - // The turn has what its batches gave it, so from here it goes on with a - // copy of its trigger without a batch wake's marks. A retry of it then - // comes back as an ordinary wake by whatever door it takes (a provider - // hold's release, the auxiliary park's requeue, the tool-result guard's + // The turn has passed its gate, so from here it goes on with a copy of + // its trigger without a batch wake's marks. A retry of it then comes back + // as an ordinary wake by whatever door it takes (a provider hold's + // release, the auxiliary park's requeue, the tool-result guard's // restart): this gate doesn't withdraw it for finding its batches // consumed, and no withdrawal of the subject's wakes reaches it (§4.2 - // withdraws a wake whose sole cause is the subject). This frame's + // withdraws a wake whose sole cause is the subject, and a merge carries + // a subject only when it is every merged request's). This frame's // activeTurnTriggers entry keeps the trigger as it came; a retry's frame // records the copy it carries. if (trigger?.coalescingSubject !== undefined || trigger?.coalescingBatch) { diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index de00ad737..0080f48a2 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -135,13 +135,14 @@ export interface AssemblyResult { settled: Set; /** Of those, the subjects whose render or fallback this assembly delivered * where its compile reads it, as content the turn reads as its own. A - * settled subject that is not here gave the turn nothing to read: empty 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), a batch whose audience doesn't include the - * agent, a render placed in another agent's deferred queue, a render - * another assembly froze, or a cause whose batch was withdrawn, displaced - * or consumed by another request. */ + * settled subject that is not here gave the turn nothing new of its own to + * read: empty 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), a batch whose audience doesn't + * include the agent, a render placed in another agent's deferred queue, a + * render another assembly froze (its compile may show it, as that + * request's), or a cause whose batch was withdrawn, displaced or consumed + * by another request. */ materialized: Set; } diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index c2fc3c36a..6a5c77561 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1010,7 +1010,7 @@ test('an assembly for an agent outside the audience of a batch another assembly // another agent's queue (a turn with nothing to read never reaches the model) // --------------------------------------------------------------------------- -type Wake = { agentName: string; reason?: string; source?: string; timestamp?: number; coalescingSubject?: string; coalescingBatch?: boolean }; +type Wake = { agentName: string; reason?: string; source?: string; timestamp?: number; coalescingSubject?: string; coalescingBatch?: boolean; coalescingBatchSubjects?: string[] }; type TurnInternals = { startAgentStream: (...a: unknown[]) => Promise; @@ -1409,8 +1409,10 @@ test("a batch-only turn whose call a provider's capacity limit held is retried w // The first call meets an organization acceleration 429, which holds the // turn's trigger for one fresh compile once the cooldown ends. let calls = 0; + const shown: Array = []; (f.membrane as unknown as { streamYielding: unknown }).streamYielding = (request: unknown) => { f.membrane.calls.push(request as never); + shown.push(f.framework.getActiveTurnTrigger('agent') as Wake | undefined); if (++calls > 1) return new MockYieldingStream([ok()]); return new FailingStream(new MembraneError({ type: 'rate_limit', retryable: true, httpStatus: 429, @@ -1424,6 +1426,8 @@ test("a batch-only turn whose call a provider's capacity limit held is retried w assert.equal(f.membrane.calls.length, 1); const hold = internals.providerAccelerationCooldowns.get('agent'); assert.equal(hold?.heldRequests.length, 1, "the turn's trigger is held"); + const marks = (w: Wake | undefined) => [w?.coalescingSubject, w?.coalescingBatch, w?.coalescingBatchSubjects]; + assert.deepEqual(marks(hold!.heldRequests[0]), [undefined, undefined, undefined], 'as an ordinary wake, without the batch wake\'s marks'); // The cooldown's time is up, and its timer's work runs now (a release // before its time only sets the timer again). hold!.until = Date.now(); @@ -1432,6 +1436,8 @@ test("a batch-only turn whose call a provider's capacity limit held is retried w assert.equal(f.membrane.calls.length, 2, 'the held turn was retried'); assert(f.lastRequest().includes('document_diff'), 'and the retry read the render'); assert.deepEqual(withdrawn, []); + assert.equal(shown[0]?.coalescingBatch, true, "getActiveTurnTrigger shows the first attempt's batch wake as it came"); + assert.deepEqual(marks(shown[1]), [undefined, undefined, undefined], "and the retry's trigger without the batch wake's marks"); }); test("a turn a provider hold kept is retried though another resident's request consumed a newer batch of its subject meanwhile", async (t) => { @@ -1458,6 +1464,10 @@ test("a turn a provider hold kept is retried though another resident's request c await settles(internals.processInferenceRequests(), 'the pass whose primary call meets the limit'); const hold = internals.providerAccelerationCooldowns.get('agent'); assert.equal(hold?.heldRequests.length, 1, "the primary's trigger is held"); + // The hold de-duplicates by timestamp, among other fields: the held wake + // dates from before the newer batch's, as it would in a running host, where + // a render and a call lie between them, not a shared millisecond. + hold!.heldRequests[0].timestamp = Date.now() - 1000; // During the hold, 'other' already has a wake queued (a heartbeat), so its // turn comes first in the next pass. A newer batch of the subject arrives; // 'other''s assembly consumes it, and its freeze withdraws the subject's @@ -1479,6 +1489,53 @@ test("a turn a provider hold kept is retried though another resident's request c assert(!withdrawn.includes('agent'), "the primary's retry wasn't withdrawn"); }); +/** One pass merges the agent's queued wakes into one trigger; the wrapped + * `startAgentStream` sends it back to the queue before its gate, as the + * auxiliary park's requeue does when another turn has taken the agent. */ +async function mergeAndRequeue(f: Awaited>): Promise { + const internals = f.framework as unknown as TurnInternals; + const start = internals.startAgentStream.bind(internals); + let merged: Wake | undefined; + internals.startAgentStream = async (_agent: unknown, trigger: unknown) => { + internals.startAgentStream = start; + merged = trigger as Wake; + internals.pendingRequests.push(merged); + }; + await settles(internals.processInferenceRequests(), 'the pass that merges the wakes'); + assert(merged, 'the pass merged the wakes into one trigger'); + return merged!; +} + +test("a wake merged from a batch wake and another wake keeps the other cause when the batch's subject is withdrawn before its gate", async (t) => { + const f = await fixture(); t.after(f.close); + f.alwaysRespond(); + await f.send('push/event', f.params('1', 'merged_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // A heartbeat for the same agent, queued behind the batch's wake: the merge + // is spread from the batch's wake. + internals.pendingRequests.push({ agentName: 'agent', reason: 'heartbeat', source: 'framework', timestamp: Date.now() }); + const merged = await mergeAndRequeue(f); + // A retraction withdraws the wakes whose sole cause is the subject (§4.2). + const r = await f.send('push/event', f.params('2', '', { retract: true })); + assert.equal(r.result.coalesce.outcome, 'retracted'); + assert(internals.pendingRequests.includes(merged), "the merged wake, which the heartbeat shares, isn't withdrawn with the batch's subject"); + await settles(f.framework.runUntilIdle(), 'the merged turn'); + assert.equal(f.membrane.calls.length, 1, "the heartbeat's turn ran"); +}); + +test("a batch wake requeued before its gate is still withdrawn with its subject", async (t) => { + const f = await fixture(); t.after(f.close); + f.alwaysRespond(); + await f.send('push/event', f.params('1', 'stale_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + const merged = await mergeAndRequeue(f); + const r = await f.send('push/event', f.params('2', '', { retract: true })); + assert.equal(r.result.coalesce.outcome, 'retracted'); + assert(!internals.pendingRequests.includes(merged), 'the wake, whose sole cause was the batch, is withdrawn with its subject'); + await settles(f.framework.runUntilIdle(), 'the drain'); + assert.equal(f.membrane.calls.length, 0); +}); + 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' }));