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..429074e79 --- /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 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 new file mode 100644 index 000000000..e0836eda7 --- /dev/null +++ b/changelog.d/coalescing-batch-turn-content.fixed.md @@ -0,0 +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 06b698273..02cefc3e5 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11711,6 +11711,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 }, true), cancelWake: (subject) => this.cancelCoalescedWakes(subject), authorized: (occ) => this.coalescedAuthorized(occ), audience: (occ) => this.coalescedAudience(occ), @@ -12098,8 +12099,16 @@ 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. `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); - 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 @@ -12115,8 +12124,7 @@ export class AgentFramework { channelId, counterparty: authorId ? `${occ.serverId}:user:${authorId}` : undefined, addressed: isAddressedMessage(occ.tags, origin), - coalescingSubject: subject, - coalescingBatch: true, + ...(plain ? {} : { coalescingSubject: subject, coalescingBatch: true }), }); } } @@ -12680,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, @@ -14747,10 +14762,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); } @@ -14759,15 +14780,16 @@ 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 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). 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)`); @@ -14785,6 +14807,20 @@ export class AgentFramework { } return false; } + // 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, 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) { + 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 9787152f4..0080f48a2 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -130,14 +130,19 @@ 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 or another assembly's. */ settled: Set; - /** Of those, the subjects whose render or fallback was delivered as content. - * A settled subject that is not here gave the turn nothing to read: empty - * or blank render (§5.2), cancelled, 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. */ + /** 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 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; } @@ -193,6 +198,11 @@ 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 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; /** The batch's audience still holds the authority it was admitted under. */ @@ -675,23 +685,32 @@ 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 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 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): 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) { 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); })); } } @@ -703,27 +722,40 @@ export class PushCoalescer { /** * Under the lock: decide whether `subject`'s batch renders for `agentName`, * freeze it, make the render-start durable, re-check authority, and start - * the RPC. Returns the render's completion (shared by every assembly that - * waits on this batch, vector 25). Returns 'declined' when the turn being - * assembled gets nothing of the batch: the agent isn't in its audience, - * whether or not another assembly is rendering it (the batch is left as it - * is); authority was revoked before the render could start (the batch is - * dropped); or the render-start could not be made durable (the batch is - * kept for its recovery wake). Returns undefined when the batch was not - * this assembly's to decide on: there is none, the subject moved on during - * the audience lookup, or the host is stopping. + * 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): Promise<{ done: Promise; rendering: Rendering } | 'declined' | 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 || this.suspended) 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, 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. (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; 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; @@ -824,15 +856,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, and no request has consumed it. The freeze withdrew the + // 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; } // 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..6a5c77561 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'; // --------------------------------------------------------------------------- @@ -1003,6 +1005,537 @@ 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 Wake = { agentName: string; reason?: string; source?: string; timestamp?: number; coalescingSubject?: string; coalescingBatch?: boolean; coalescingBatchSubjects?: string[] }; + +type TurnInternals = { + startAgentStream: (...a: unknown[]) => Promise; + processInferenceRequests: () => Promise; + activeTurnTokens: Map; + pendingRequests: Wake[]; + 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[]; + ackDeferredWrites(): void; + 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' }, +]; + +/** 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) => { + 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 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 })); + 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 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; }); + 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; + // '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 joins!: () => void; + const otherAssembles = new Promise((resolve) => { joins = resolve; }); + 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; + 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; + }; + 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"); + 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.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) => { + 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); + 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"); +}); + +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); + 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 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 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(); + 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 §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'); + assert(f.lastRequest().includes('document_diff')); + 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; + 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, + 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"); + 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(); + 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, []); + 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) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + const withdrawn = withdrawals(f); + // 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; + // 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"); + // 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 + // 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"); +}); + +/** 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' }));