diff --git a/changelog.d/coalescing-declined-freeze.fixed.md b/changelog.d/coalescing-declined-freeze.fixed.md new file mode 100644 index 000000000..13422a34d --- /dev/null +++ b/changelog.d/coalescing-declined-freeze.fixed.md @@ -0,0 +1 @@ +- A turn woken only by deferred coalescing batches is no longer started when none of them has anything for it to read. It used to run, calling the model although no new input had arrived for it, when the assembly found a batch's authority revoked (before or after its render-start commit), when the render-start commit failed, or when it found that the agent had left the batch's audience since the batch woke it. Such a turn is now withdrawn before inference, as a turn whose batches all rendered nothing already was, and traces `turn-withdrawn`. A batch whose render-start commit failed is still kept, and its recovery wake delivers its fallback once storage recovers. The stderr line for a withdrawn turn now reads `[coalescing] : its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`. It said `deferred render produced nothing`, which is untrue where no render ran. diff --git a/src/framework.ts b/src/framework.ts index 6d854a393..f957ca28c 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -12990,19 +12990,21 @@ export class AgentFramework { } } - // A turn whose only causes were deferred batches, all of which rendered - // nothing (empty or blank render, §5.2; cancelled; revoked), has nothing - // new to show: running it would wake the model to `[Continue]` or an - // older message, an uncaused wake. Stop before the checkpoint, locus, - // typing and compile. Any batch this assembly materialized keeps the - // turn, including one that was not among its causes (a batch whose wake - // the freeze withdrew, or one that never qualified for a wake). A cause - // this assembly did not settle (rendered elsewhere, not yet frozen) keeps - // the turn, as before. + // A turn whose only causes were deferred batches, none of which gave it + // anything to read (an empty or blank render, §5.2; cancelled; revoked; a + // render-start that could not be made durable, whose batch then waits for + // its recovery wake; a batch whose audience doesn't include the agent), + // has nothing new to show: running it would wake the model to + // `[Continue]` or an older message, an uncaused wake. Stop before the + // checkpoint, locus, typing and compile. Any batch this assembly + // materialized keeps the turn, including one that was not among its + // causes (a batch whose wake the freeze withdrew, or one that never + // qualified for a wake). A cause this assembly did not settle (rendered + // elsewhere, not yet frozen) keeps the turn, as before. const batchSubjects = trigger?.coalescingBatchSubjects; if (assembly && batchSubjects?.length && assembly.materialized.size === 0 && batchSubjects.every((s) => assembly.settled.has(s))) { - console.error(`[coalescing] ${agent.name}: deferred render produced nothing; turn not started (no wake cause left)`); + console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`); this.emitTrace({ type: 'mcpl:coalescing', kind: 'turn-withdrawn', agentName: agent.name, subjects: batchSubjects }); // Release the turn and land anything deferred while assembly awaited // the render, as a torn-down turn would. diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 9d76a930b..1bccc21df 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -129,11 +129,15 @@ interface Rendering { /** What an assembly did, per subject, for the turn being assembled. */ export interface AssemblyResult { - /** Subjects whose render this assembly waited on to completion. */ + /** Subjects this assembly finished with: it waited on their render to + * completion, or freeze() declined them for this agent. */ settled: Set; /** Of those, the subjects whose render or fallback was delivered as content. * A settled subject that is not here gave the turn nothing to read: empty - * or blank render (§5.2), cancelled, or revoked at response. */ + * or blank render (§5.2), cancelled, revoked before the render or at its + * response, a render-start that could not be made durable (the batch then + * waits for its recovery wake), or a batch whose audience doesn't include + * the agent. */ materialized: Set; } @@ -674,7 +678,8 @@ export class PushCoalescer { // be flattened by `then`, holding the lock until the render settled — // and the settlement itself needs the lock. const started = await this.locked(() => this.freeze(subject, agentName)); - if (started) { + if (started === 'declined') result.settled.add(subject); + else if (started) { work.push(started.done.then(() => { result.settled.add(subject); if (started.rendering.materialized) result.materialized.add(subject); @@ -690,27 +695,42 @@ export class PushCoalescer { * Under the lock: decide whether `subject`'s batch renders for `agentName`, * freeze it, make the render-start durable, re-check authority, and start * the RPC. Returns the render's completion (shared by every assembly that - * waits on this batch, vector 25) or undefined when nothing was started. + * waits on this batch, vector 25). Returns 'declined' when the turn being + * assembled gets nothing of the batch: the agent isn't in its audience, + * whether or not another assembly is rendering it (the batch is left as it + * is); authority was revoked before the render could start (the batch is + * dropped); or the render-start could not be made durable (the batch is + * kept for its recovery wake). Returns undefined when the batch was not + * this assembly's to decide on: there is none, the subject moved on during + * the audience lookup, or the host is stopping. */ - private async freeze(subject: string, agentName: string): Promise<{ done: Promise; rendering: Rendering } | undefined> { + private async freeze(subject: string, agentName: string): Promise<{ done: Promise; rendering: Rendering } | 'declined' | undefined> { const state = this.subjects.get(subject); if (!state || this.suspended) return undefined; if (state.rendering) { - // Another assembly froze this batch; share its outcome (vector 25). + // Another assembly froze this batch; share its outcome (vector 25), + // unless this agent isn't in its audience, as below. const rendering = state.rendering; - return (await this.host.audience(rendering.batch.latest)).includes(agentName) && state.rendering === rendering ? { done: rendering.done, rendering } : undefined; + const audience = await this.host.audience(rendering.batch.latest); + if (state.rendering !== rendering) return undefined; + if (!audience.includes(agentName)) return 'declined'; + return { done: rendering.done, rendering }; } const batch = state.batch; if (!batch) return undefined; // The audience lookup may spawn a fork. Under the lock no admission can // move the subject on meanwhile; the identity check guards the host. const audience = await this.host.audience(batch.latest); - if (!audience.includes(agentName) || state.batch !== batch || state.rendering) return undefined; + if (state.batch !== batch || state.rendering) return undefined; + // Not this agent's batch, or no longer: an agent woken for it can have left + // its audience since. This turn has nothing of it to show. The batch stays + // pending; nothing here wakes its current audience. + if (!audience.includes(agentName)) return 'declined'; state.batch = undefined; this.host.cancelWake(subject); if (!this.host.authorized(batch.latest)) { this.host.audit({ kind: 'revoked', subject, eventId: batch.latest.eventId }); - return undefined; + return 'declined'; } const rendering: Rendering = { batch, cancelled: false, done: Promise.resolve() }; state.rendering = rendering; @@ -728,7 +748,7 @@ export class PushCoalescer { // unrelated traffic happens by. Re-wake with backoff, so a persistent // storage outage does not become a tight loop of model turns. this.scheduleRecovery(subject, state); - return undefined; + return 'declined'; } this.clearRecovery(subject); if (this.suspended) { state.rendering = undefined; state.batch = batch; return undefined; } @@ -738,7 +758,7 @@ export class PushCoalescer { if (!this.host.authorized(batch.latest)) { this.host.audit({ kind: 'revoked', subject, eventId: batch.latest.eventId }); state.rendering = undefined; - return undefined; + return 'declined'; } rendering.done = this.render(subject, state, rendering, agentName); return { done: rendering.done, rendering }; diff --git a/test/helpers/coalescing-fixture.ts b/test/helpers/coalescing-fixture.ts index 39dee6a70..93148af4e 100644 --- a/test/helpers/coalescing-fixture.ts +++ b/test/helpers/coalescing-fixture.ts @@ -18,7 +18,8 @@ export const TS = '2026-09-30T00:00:00Z'; /** * How long the fixture's waits run before they fail: send() for a request's - * reply, answered() for the host's answers, eventually() for its condition. + * reply, answered() for the host's answers, eventually() for its condition, + * settles() for the work it's given. * Each is a hang guard: it turns a wait that never ends into a failure * instead of a hung run, and it isn't a check on how fast the host works. So * it's set wide, and the same for every wait: a loaded host's stall needn't @@ -145,6 +146,21 @@ export async function eventually(predicate: () => boolean, what = 'condition'): while (!predicate()) { assert(performance.now() < deadline, `${what} did not settle within ${REPLY_DEADLINE_MS / 1000} s`); await new Promise((r) => setTimeout(r, 10)); } } +/** + * Awaits `work`, failing with `what` if it hasn't settled within + * REPLY_DEADLINE_MS. CI runs without a test timeout, so a wait that never + * ends would hold the whole run, and this turns it into a failure that names + * what was awaited. When it fires, the work it waited on keeps running, and + * the test's close then runs under it. + */ +export async function settles(work: Promise, what: string): Promise { + let timer: ReturnType | undefined; + const late = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${what} did not settle within ${REPLY_DEADLINE_MS / 1000} s`)), REPLY_DEADLINE_MS); + }); + try { return await Promise.race([work, late]); } finally { clearTimeout(timer); } +} + /** Simulate a crash: stop without the snapshot flush a clean stop performs. */ export async function crash(f: Awaited>): Promise { diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index d8287e8a6..0f2512cec 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -13,7 +13,7 @@ import { tmpdir } from 'node:os'; import { join, dirname } from 'node:path'; import { fileURLToPath } from 'node:url'; import { createMockResponse } from './helpers/mock-membrane.js'; -import { fixture, eventually, crash, ok, TS, REPLY_DEADLINE_MS } from './helpers/coalescing-fixture.js'; +import { fixture, eventually, settles, crash, ok, TS, REPLY_DEADLINE_MS } from './helpers/coalescing-fixture.js'; // --------------------------------------------------------------------------- // Plain coalescing, feature-set scope (vectors 1–6, 35) @@ -340,9 +340,14 @@ test('vector 24: revoked authority before assembly drops the batch without a ren const f = await fixture(); t.after(f.close); await f.send('push/event', f.params('1', 'previously_permitted', { deferred: true })); await f.framework.stop(); await f.create(false); - await f.framework.runUntilIdle(); + let withdrawn = 0; + f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); + // A second model call would wait for a response the fixture never queues. + await settles(f.framework.runUntilIdle(), 'the drain after the restart'); assert.equal(f.renders.length, 0); assert(!f.context().includes('previously_permitted')); + assert.equal(f.membrane.calls.length, 0, 'and the revoked batch, the only cause, started no inference'); + assert.equal(withdrawn, 1, 'its turn started and was withdrawn'); }); test('vector 27i: a deferred notice displaces an unread plain occurrence', async (t) => { @@ -910,6 +915,8 @@ test('R11: a retraction during the render-start commit wait is final; the batch test('R12: endpoint reassignment during the render-start commit wait never sends the batch to the new peer', async (t) => { const f = await fixture(); t.after(f.close); await f.send('push/event', f.params('1', 'old_fallback', { deferred: true, data: { secret: 'private_notice_data' } })); + let withdrawn = 0; + f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); const held = holdNextCommit(f); const run = f.framework.runUntilIdle(); const release = await held; @@ -917,10 +924,13 @@ test('R12: endpoint reassignment during the render-start commit wait never sends await f.framework.restartMcplServer('editor', { ...config, url: `${config.url}/reassigned` } as never); await eventually(() => f.hostCaps.length === 2, 'reconnect to the new endpoint'); release(); - await run; + // A second model call would wait for a response the fixture never queues. + await settles(run, 'the drain after the reassignment'); assert.equal(f.renders.length, 0, 'no push/render to the new peer'); assert(!JSON.stringify(f.renders).includes('private_notice_data')); assert(!f.context().includes('old_fallback')); + assert.equal(f.membrane.calls.length, 0, 'and the revoked batch, the only cause, started no inference'); + assert.equal(withdrawn, 1, 'its turn started and was withdrawn'); }); // --------------------------------------------------------------------------- @@ -948,6 +958,51 @@ test('R13: a retraction during the audience lookup is final; the stale batch is assert(!f.context().includes('document_diff')); }); +test('a turn whose batch\'s audience no longer names its agent at the assembly starts no inference', async (t) => { + const f = await fixture(); t.after(f.close); + await f.send('push/event', f.params('1', 'moved_fallback', { deferred: true, initial: true })); + // The admission's wake has looked the audience up; the next lookup is the + // assembly's, and it names another agent instead, as when a conversation + // fork's channel is bound to a new fork in between. + const internals = f.framework as unknown as { coalescedAudience: (occ: unknown) => Promise; pushCoalescer: { pendingBatches(): number } }; + const orig = internals.coalescedAudience.bind(internals); + internals.coalescedAudience = async () => { internals.coalescedAudience = orig; return ['conversation-chat-g2']; }; + let withdrawn = 0; + f.framework.onTrace((e) => { if (e.type === 'mcpl:coalescing' && (e as { kind?: string }).kind === 'turn-withdrawn') withdrawn++; }); + // A second model call would wait for a response the fixture never queues. + await settles(f.framework.runUntilIdle(), 'the drain after the wake'); + assert.equal(f.renders.length, 0); + assert(!f.context().includes('moved_fallback')); + assert.equal(f.membrane.calls.length, 0, 'the batch, the only cause, started no inference'); + assert.equal(withdrawn, 1, 'its turn started and was withdrawn'); + assert.equal(internals.pushCoalescer.pendingBatches(), 1, 'the batch stays pending'); +}); + +test('an assembly for an agent outside the audience of a batch another assembly is rendering declines it', async (t) => { + const f = await fixture({ agents: [{ name: 'agent', model: 'test', systemPrompt: 'test' }, { name: 'other', model: 'test', systemPrompt: 'test' }] }); t.after(f.close); + let release!: (value: unknown) => void; + let began!: () => void; + const started = new Promise((resolve) => { began = resolve; }); + f.renderer(() => { began(); return new Promise((resolve) => { release = resolve; }); }); + await f.send('push/event', f.params('1', 'shared_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as { coalescedAudience: (occ: unknown) => Promise; pushCoalescer: { assemble(agent: string): Promise<{ settled: Set; materialized: Set }> } }; + // From here on the audience is 'other' alone, as when a conversation fork's + // channel has moved to a new fork. 'other' freezes the batch and its render + // is held; 'agent' then assembles while it's in flight. + internals.coalescedAudience = async () => ['other']; + const rendering = internals.pushCoalescer.assemble('other'); + await settles(started, "the render for 'other'"); + // Joining the held render instead would wait for a release that comes + // only after this call. + const declined = await settles(internals.pushCoalescer.assemble('agent'), "the assembly for 'agent'"); + assert.equal(declined.settled.size, 1, 'the subject is settled for the agent outside the audience'); + assert.equal(declined.materialized.size, 0, 'and gave it nothing'); + release({ content: [{ type: 'text', text: 'shared_render' }] }); + const rendered = await settles(rendering, "the assembly for 'other'"); + assert.equal(rendered.settled.size, 1); + assert.equal(rendered.materialized.size, 1, 'the assembly in the audience delivered the render'); +}); + 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' })); @@ -987,7 +1042,7 @@ test('R15: a retraction arriving between the render result and its publication w // Review round 8 (#197): liveness after a failed render-start commit // --------------------------------------------------------------------------- -test('R16: a batch whose render-start commit failed is re-woken once storage recovers', async (t) => { +test('R16: a batch whose render-start commit failed is re-woken once storage recovers', { timeout: 120_000 }, async (t) => { const f = await fixture(); t.after(f.close); const coalescer = (f.framework as unknown as { pushCoalescer: { options: { recoveryBackoffMs?: number }; pendingBatches(): number } }).pushCoalescer; coalescer.options.recoveryBackoffMs = 50; @@ -996,15 +1051,36 @@ test('R16: a batch whose render-start commit failed is re-woken once storage rec const orig = store.sync.bind(store); let failing = 0; store.sync = () => { if (failing > 0) { failing--; throw new Error('EIO simulated'); } orig(); }; + const traced: Array<{ kind: string; source?: string }> = []; + f.framework.onTrace((e) => { + if (e.type === 'mcpl:coalescing') traced.push({ kind: (e as { kind: string }).kind, source: (e as { source?: string }).source }); + }); + // Every turn that reaches the model is answered, so a model call nothing + // should have caused is counted rather than left waiting for a response. + f.alwaysRespond(); await f.send('push/event', f.params('1', 'asleep_fallback', { deferred: true, initial: true })); failing = 1; // the render-start commit fails; storage then recovers - await f.framework.runUntilIdle(); + // The recovery wake can fire inside the first drain or after it, so drive + // until the batch renders rather than for a fixed number of drains. + await settles(f.framework.runUntilIdle(), 'the first drain'); + for (let pass = 0; !traced.some((e) => e.kind === 'rendered'); pass++) { + assert(pass < 5, 'the batch rendered within five recovery wakes'); + await eventually(() => internals.pendingRequests.length > 0, 'recovery wake'); + await settles(f.framework.runUntilIdle(), `the drain after recovery wake ${pass + 1}`); + // A macrotask turn, so timers (this test's timeout among them) can fire + // even when both waits above return at once. + await new Promise((resolve) => setImmediate(resolve)); + } + assert.equal(f.membrane.calls.length, 1, 'one model call, the recovery turn\'s: the wake whose render-start commit failed, the only cause, started none'); + assert(f.lastRequest().includes('asleep_fallback'), 'the recovery turn delivered the batch'); + const kinds = traced.map((e) => e.kind); + const order = (...steps: string[]) => kinds.filter((k) => steps.includes(k)); + assert.deepEqual(order('persist-failed', 'turn-withdrawn', 'rendered'), ['persist-failed', 'turn-withdrawn', 'rendered'], 'the failed freeze withdrew its turn'); + assert.deepEqual(order('persist-failed', 'recovery-wake', 'rendered'), ['persist-failed', 'recovery-wake', 'rendered'], 're-woken by the recovery timer, without unrelated traffic'); + // The failed commit can't prove that no render started, so the batch takes + // its fallback and the server is never asked to render it. + assert.equal(traced.find((e) => e.kind === 'rendered')?.source, 'fallback'); assert.equal(f.renders.length, 0); - assert.equal(coalescer.pendingBatches(), 1, 'the batch is kept'); - assert(!f.lastRequest().includes('asleep_fallback')); - await eventually(() => internals.pendingRequests.length > 0, 'recovery wake'); - await f.turn(); - assert(f.lastRequest().includes('asleep_fallback'), 'delivered without unrelated traffic'); assert.equal(coalescer.pendingBatches(), 0); });