From 3253dab4a92d3598c8ee092eeab17549974d3d29 Mon Sep 17 00:00:00 2001 From: Crispin-2713 Date: Mon, 19 Oct 2026 09:14:36 -0400 Subject: [PATCH 1/2] fix(deferral): a write the coalescing assembly makes lands after what was deferred for its agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The coalescing assembly writes past its agent's deferral (addMessage's bypassDeferralFor), in the window the turn-start flush writes in, before the turn's checkpoint. Such a write was stored at once, so it could land ahead of a write still queued for the agent, and what was deferred while the assembly awaited a render missed the request: - a render another agent's assembly placed in the agent's queue during the agent's own assembly was read after the agent's next render of the subject, a delta that builds on it (RFC-006 §5.2, vector 49); - when the turn-start flush couldn't store a queued render, the assembly's next render landed ahead of it the same way; - a render the agent's assembly joined (vector 25), placed in the agent's queue by the assembly that froze it, waited for the turn's first tool boundary or its end. The same option also reached past the assembly: a render an assembly started and left running when it threw delivered into the live turn after its compile, and an assembly that spans a quiesce wrote what the quiesce held back. Now every write deferred for the agent from its turn's start to the end of its assembly lands in that request, in arrival order, up to a write that can't be stored and unless the host is quiesced, through one flush (flushForStart) that the turn-start flush uses too: a write the assembly makes into the agent's own window first writes what is queued for the agent, and what was deferred during the assembly is written once it ends, before the checkpoint. - If a queued write can't be stored, the assembly's write waits behind it in the queue, with a new seq and a deferredId, so the coalescer tracks it as an unread render and a retraction can still remove it. - bypassDeferralFor counts only while that agent's assembly runs (an `assembling` set entered around assemble() and left in its finally), and for a write from the queue. A render that settles after its assembly threw, while the turn is alive, is deferred like any write during the turn. - While the host is quiesced, neither of the assembly's flushes runs, and an assembled write with anything queued for its agent waits behind it. - A write from the queue, as stop()'s drain makes them, doesn't flush: the flush's ack would acknowledge that drain's entries before they're stored. - A render placed in an agent's queue gets a wake to read it; that wake now carries the render's deferredId (readsDeferredWrite). Where the scheduler would start a turn, such a wake is dropped on evidence that the agent has nothing left to read of it (deferredWriteReadOrGone): the row the write landed as is at or below the agent's consumed watermark on its current branch, or the write is gone: in neither the queue nor the store's current branch, and not drained and still to be stored. Any doubt keeps the wake: a watermark an undo left on another branch, or a store that can't be read, which is logged. Each drop is logged. If nothing else is left, no turn starts, and a provider recovery that released it ends as a policy drop ends one (endPendingPrimary). A merged trigger keeps the mark only when every merged request shares it, quiesce compaction keys parked wakes on it, and a provider hold's de-duplication (sameInferenceRequest) compares it and the batch's subject, so a marked wake never stands for another cause. A turn that has started drops the mark from its trigger, with a batch wake's marks, so its retry is an ordinary wake; its [Continue] prompt is recorded under the frame's activeTurnTriggers entry, where its readers look it up, not under that copy. writeDrainedInOrder's doc states the rule for that window, and what still lands ahead of a held entry outside it: a fresh tool_result, and a write while its agent is idle. The assemblingFor fields, the coalescer's deferred-placement branch, the turn-token invariant (here and in deferred-wake-integrity's header), the deferred flushes' target rule and the render-run test's stand-in host say what they now do. Tests: deferred-window-only checks that an assembled write lands after a write deferred before it, unheld by another target's entry and acknowledging what it stored; that it waits behind one that can't be stored and lands after it; that while quiesced it waits behind what is queued, or is stored when nothing is queued for its agent; that a write naming an assembly that has ended is deferred; and that stop()'s drain keeps its entries un-acked until it has stored them. mcpl-coalescing checks that a render whose write failed at its agent's turn start is read before the next render of its subject, in one request; that a render deferred behind a write the store keeps refusing stays tracked and retractable; that a render an assembly joins reaches the joining agent's request, in one turn, kept by an undo of that turn; that a module write deferred while an assembly awaits a render lands in that request, or stays queued while quiesced; that an assembly that throws closes its window; that a wake to read a render retracted unread starts no turn; that one whose render landed starts its turn above the agent's watermark, with no watermark, and after an undo has left the watermark on another branch, and none with the row at the watermark or below it, also while the stored write still awaits a sync; that a turn such a wake started is retried after a provider hold; that a store that can't be read keeps the wake; that a gone render's wake is dropped from a pass whose other wakes start the turn; that a provider recovery whose released wakes start no turn ends and reopens auxiliary admission, and that one beside a wake that starts the turn is left to that turn; that the wake's mark survives no merge with another cause or another render's wake; and that a provider hold keeps a wake to read a render, a batch wake and an ordinary wake apart though they share a millisecond. silent-heartbeat-rows checks that a [Continue] turn a wake to read a render started stamps its reply. present-while-acting's assembled delivery runs inside the assembly's window, as a real one does. The changelog fragment says what agents see. Co-Authored-By: Claude Opus 5.5 --- .../assembly-writes-in-arrival-order.fixed.md | 26 + src/framework.ts | 266 ++++++--- src/mcpl/push-coalescer.ts | 35 +- src/mcpl/push-handler.ts | 4 +- src/types/agent.ts | 4 + test/deferred-wake-integrity.test.ts | 5 +- test/deferred-window-only.test.ts | 176 +++++- test/mcpl-coalescer-render-run.test.ts | 3 +- test/mcpl-coalescing.test.ts | 503 +++++++++++++++++- test/present-while-acting.test.ts | 3 + test/silent-heartbeat-rows.test.ts | 17 + 11 files changed, 958 insertions(+), 84 deletions(-) create mode 100644 changelog.d/assembly-writes-in-arrival-order.fixed.md diff --git a/changelog.d/assembly-writes-in-arrival-order.fixed.md b/changelog.d/assembly-writes-in-arrival-order.fixed.md new file mode 100644 index 000000000..6f91310a9 --- /dev/null +++ b/changelog.d/assembly-writes-in-arrival-order.fixed.md @@ -0,0 +1,26 @@ +- What was deferred for an agent before a coalesced render or message the + agent's turn assembles now lands ahead of it, in the order it arrived. The + assembly's write used to be stored at once, ahead of a write still queued + for the agent, so a render could be read before the earlier render it + builds on: after a render landed in the agent's queue during its own + assembly, or after the turn's start couldn't store a queued render. When a + queued write can't be stored, the assembled one now waits behind it. What + is deferred for the agent while its assembly awaits a render, such as a + render another agent's assembly placed in its queue, now lands in that + turn's request, rather than at the turn's first tool boundary or its end. +- A render's wake to read it, given when the render lands in an agent's + deferred queue, no longer starts a turn if, when the scheduler comes to + start it, the agent has already read the render or the render was + retracted. +- A provider hold no longer keeps two wakes from the same millisecond as + one when one of them is a coalesced batch's wake or a wake to read a + render, so withdrawing the batch, or dropping the read render's wake, + no longer takes the other wake's cause with it. +- A reply to a `[Continue]` turn that a coalesced batch's wake or a wake to + read a render started now keeps its `promptedBy` stamp, so later + requests render `[Continue]` before it as they do for other such replies. +- A render that settles after its turn's assembly has failed, while the turn + is still alive, is now deferred like any other write during the turn, + rather than stored into the turn after its compile. While the host is + quiesced, the assembly's writes wait behind what is queued for the agent, + and land when it does. diff --git a/src/framework.ts b/src/framework.ts index d72da2323..6d52b0b40 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -572,7 +572,9 @@ interface CoalescedChannelEvent { coalescingSubject?: string; /** Occurrence id (RFC-006 §3.1); channel messages carry it only when coalesced. */ eventId?: string; - /** RFC-006 assembly: materialized for this agent's turn (store directly). */ + /** RFC-006 assembly: materialized for this agent's turn, stored directly + * while its assembly runs, after what is queued for it (addMessage's + * bypassDeferralFor). */ assemblingFor?: string; /** RFC-006 §3.2: deliver into this agent only (replacement / notice audience). */ deliverTo?: string; @@ -2314,8 +2316,13 @@ export class AgentFramework { // (falsified history — it later confabulated an apology for the skip), and // every subsequent compile diverged from the live prefix at that message // (KV bust). Invariant: nothing enters the window between a turn's dequeue - // and its settle except that turn's own blocks. + // and its settle except that turn's own blocks, and, before its checkpoint, + // the inputs its start writes (flushForStart). private activeTurnTokens: Map = new Map(); + /** Agents whose fresh turn's coalescing assembly is running + * (beginAgentTurn): the window in which addMessage honours a write's + * bypassDeferralFor naming that agent. */ + private readonly assembling = new Set(); /** * The InferenceRequest that started each agent's turn in progress — set and * cleared exactly where the turn token is. Read-only outside: the host @@ -3811,6 +3818,17 @@ export class AgentFramework { if (gate.primaryDepth > 0 || gate.primaryPending || this.cooldownBlocksAuxiliary(agentName)) return; for (const resolve of gate.auxiliaryWaiters.splice(0)) resolve(); } + /** A provider hold's released requests start no turn (the scheduler drops + * them): the primary turn the release left pending won't run, so the + * recovery ends and auxiliary admission reopens. */ + private endPendingPrimary(agentName: string): void { + const gate = this.providerGates.get(agentName); + if (gate?.primaryPending && this.providerAccelerationRecoveries.has(agentName)) { + gate.primaryPending = false; + this.providerAccelerationRecoveries.delete(agentName); + this.flushAuxiliaryAdmission(agentName); + } + } private async waitForAuxiliaryIdle(agentName: string): Promise { const gate = this.providerGate(agentName); if (gate.auxiliaryInFlight === 0) return; @@ -3941,9 +3959,14 @@ export class AgentFramework { } return { delayMs: ownMs, ownMs, stated: false }; } + /** Whether two requests are one wake, for a provider hold's dedup: the + * same agent, reason, source, time and channel, and the same + * coalescingSubject and readsDeferredWrite, since a wake with either can be + * withdrawn or dropped on its own, and must not stand for one without. */ 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.readsDeferredWrite === b.readsDeferredWrite; } private consultProviderHold(error: Error, agentName: string): { holdMs: number; reason: string } | undefined { if (!this.providerHoldHook) return undefined; @@ -10841,9 +10864,9 @@ export class AgentFramework { agent.toolResultGuard.storeResults(toolResultContent, membraneResults, currentState.toolResults, turnRowMetadata, maxChars); // Flush any messages that were deferred while this turn was in - // flight. Route to the PRIMARY agent — deferred messages are - // framework-level (module messages, push events, subagent return - // notifications) meant for the main conversation. + // flight: this agent's own entries, those addressed to it and, for + // the primary, those addressed to no one (module messages, push + // events, subagent return notifications). // // Hear-while-acting: messages flushed into THIS agent's own window // are also collected for mid-turn injection — passed to the membrane @@ -12261,7 +12284,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), + wakeToRead: (occ, agentName, deferredId) => this.queueCoalescedWakes({ ...occ, deliverTo: agentName }, true, deferredId), cancelWake: (subject) => this.cancelCoalescedWakes(subject), authorized: (occ) => this.coalescedAuthorized(occ), audience: (occ) => this.coalescedAudience(occ), @@ -12656,8 +12679,8 @@ export class AgentFramework { /** 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 { + * placed in its agent's deferred queue (wakeToRead), at `readsDeferredWrite`. */ + private async queueCoalescedWakes(occ: CoalescedOccurrence, plain = false, readsDeferredWrite?: string): 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; @@ -12674,7 +12697,7 @@ export class AgentFramework { channelId, counterparty: authorId ? `${occ.serverId}:user:${authorId}` : undefined, addressed: isAddressedMessage(occ.tags, origin), - ...(plain ? {} : { coalescingSubject: subject, coalescingBatch: true }), + ...(plain ? (readsDeferredWrite ? { readsDeferredWrite } : {}) : { coalescingSubject: subject, coalescingBatch: true }), }); } } @@ -13084,9 +13107,10 @@ export class AgentFramework { // route the resumed turn into the wrong channel. 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 ?? ''}`; + // A coalesced wake is withdrawable by its subject (RFC-006 §4.2), + // and a wake to read a render is dropped once it's read: neither + // may absorb an unrelated wake that would vanish with it. + const key = `${request.reason}|${request.addressed ? 'a' : 'n'}|${request.coalescingSubject ?? ''}|${request.readsDeferredWrite ?? ''}`; const prev = newestByKey.get(key); if (!prev || request.timestamp >= prev.timestamp) { newestByKey.set(key, request); @@ -13153,6 +13177,23 @@ export class AgentFramework { continue; } + // A wake whose sole cause was reading a write placed in this agent's + // queue (wakeToRead's readsDeferredWrite) has nothing left to do once + // the agent has read that write, or it was removed: it starts no turn. + const read = requests.filter((r) => r.readsDeferredWrite !== undefined + && this.deferredWriteReadOrGone(agent, r.readsDeferredWrite)); + if (read.length > 0) { + const left = requests.filter((r) => !read.includes(r)); + requests.length = 0; + requests.push(...left); + console.error(`[coalescing] ${agentName}: ${read.length} wake(s) to read a render it has read, or that was removed, dropped` + + (requests.length === 0 ? '; no turn starts' : '')); + if (requests.length === 0) { + this.endPendingPrimary(agentName); + continue; + } + } + // Check policy if (!this.inferencePolicy.shouldInfer(agentName, requests, state)) { // Loud drop: a queued request that dies here is otherwise invisible — @@ -13165,12 +13206,7 @@ export class AgentFramework { // A context-budget restart inherits the predecessor's EventGate liveness. // If policy drops the queued successor, no driveStream remains to release it. if (budgetRestart) this.eventGate?.onInferenceEnded(agentName); - const gate = this.providerGates.get(agentName); - if (gate?.primaryPending && this.providerAccelerationRecoveries.has(agentName)) { - gate.primaryPending = false; - this.providerAccelerationRecoveries.delete(agentName); - this.flushAuxiliaryAdmission(agentName); - } + this.endPendingPrimary(agentName); continue; } @@ -13248,6 +13284,10 @@ export class AgentFramework { // subject's batch, or a retry past its gate. coalescingSubject: requests.every((r) => r.coalescingSubject === trigger?.coalescingSubject) ? trigger?.coalescingSubject : undefined, + // Likewise a write to read: dropping the merge once it's read must + // take no other cause with it. + readsDeferredWrite: requests.every((r) => r.readsDeferredWrite === trigger?.readsDeferredWrite) + ? trigger?.readsDeferredWrite : undefined, suppressProse: silentOnly ? trigger?.suppressProse : undefined, silentHeartbeat: silentOnly ? trigger?.silentHeartbeat : undefined, channelId: channelReq?.channelId, @@ -15243,7 +15283,8 @@ export class AgentFramework { // Mark the turn alive before ANYTHING awaits (hooks, compile, stream // setup): from here until this turn's teardown, cross-turn writers defer // (addMessage guard) instead of appending — nothing may enter the window - // between a turn's dequeue and its settle except the turn's own blocks. + // between a turn's dequeue and its settle except the turn's own blocks, + // and, before its checkpoint, the inputs its start writes (flushForStart). // A retry re-enters with a fresh token; teardown clears token-matched. // // The try/finally is the leak-proofing: a token whose owner never clears @@ -15441,9 +15482,9 @@ export class AgentFramework { // skipped, and the DM then landed before the skip in the window — false // "saw it and ignored it" history plus KV divergence. Flushed here, the // payload rides its own wake, and the notice/payload can never split.) - // Primary agent only — deferred messages are framework-level, meant for - // the main conversation (same target rule as the tool-boundary flush). - // Fresh turns only: a context-budget restart continues the same logical + // This agent's own entries: those addressed to it, and, for the primary, + // those addressed to no one (the same target rule as the tool-boundary + // flush). Fresh turns only: a context-budget restart continues the same logical // turn, and flushing there would insert messages between its rounds. // Before recordTurnCheckpoint: these are the turn's inputs, not its // products — undoing the turn must not destroy them. @@ -15452,21 +15493,11 @@ export class AgentFramework { !resumesTurn && this.deferredMessages.length > 0 ) { - { - // One poison message, or a transient store-write failure, must not - // abort the turn or drop the messages behind it in the queue: the - // drain cuts there, and it and what follows stay queued, in order - // (writeDrainedInOrder). What was stored is weighed for the turn now - // starting and for every other resident's open turn. - const stored = this.writeDrainedInOrder(this.drainDeferredFor(agent.name), `turn-start flush for ${agent.name}`, (msg) => - agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); - for (const [msg, { id }] of stored) { - this.emitTrace({ type: 'message:added', messageId: id, source: 'deferred-flush:turn-start' }); - this.noteArrival(agent.name, msg.metadata, id); - this.noteForOtherResidents(agent, msg.metadata, id); - } - this.ackDeferredWrites(); - } + // One poison message, or a transient store-write failure, must not + // abort the turn or drop the messages behind it in the queue: the + // drain cuts there, and it and what follows stay queued, in order + // (writeDrainedInOrder). + this.flushForStart(agent, 'turn-start'); } // RFC-006 §5: deferred batches for this agent render now, at the assembly @@ -15479,12 +15510,20 @@ export class AgentFramework { // assembly can say whether the turn has anything to read. const batchSubjects = trigger?.coalescingBatchSubjects; let assembly: AssemblyResult | undefined; + let assembled = false; if (attempt === 0 && !resumesTurn && this.pushCoalescer && (batchSubjects?.length || this.pushCoalescer.pendingBatches())) { + assembled = true; + // An assembly that throws can leave renders it started running: their + // deliveries come after this window closes, as ordinary writes during + // the turn (addMessage's bypassDeferralFor counts only within it). + this.assembling.add(agent.name); try { assembly = await this.pushCoalescer.assemble(agent.name, batchSubjects); } catch (err) { console.error(`[coalescing] assembly for ${agent.name} failed:`, err); + } finally { + this.assembling.delete(agent.name); } } @@ -15518,18 +15557,28 @@ export class AgentFramework { } return false; } + // What was deferred for this agent while its assembly awaited renders is + // among this turn's inputs, before its checkpoint, so it lands in this + // request, in order, as a turn-start flush a moment later would have + // written it: among it, a render this assembly joined that another + // agent's assembly placed in this agent's queue (freeze's `joined`, + // vector 25). Not once the host is quiesced, when it waits for resume(). + if (assembled && !this.quiesced && this.deferredMessages.length > 0) this.flushForStart(agent, 'assembly'); + // 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 + // its trigger without a batch wake's marks or the mark of a wake to read + // a render. 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, 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), and + // the scheduler doesn't drop it as having nothing to read once this + // attempt's compile has moved the watermark past the render. 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; + if (trigger?.coalescingSubject !== undefined || trigger?.coalescingBatch || trigger?.readsDeferredWrite !== undefined) { + const { coalescingSubject: _subject, coalescingBatch: _batch, coalescingBatchSubjects: _subjects, readsDeferredWrite: _read, ...retry } = trigger; trigger = retry; } @@ -15846,10 +15895,13 @@ export class AgentFramework { // The [Continue] prompt this stream's first round follows, if any, as // the Agent read it from the request: the round's rows carry it // (silentTurnRowMetadata), and so do later rounds' until a tool - // boundary hands the stream messages after the results. - if (trigger) { - if (continuePrompt) this.turnPrompts.set(trigger, continuePrompt); - else this.turnPrompts.delete(trigger); + // boundary hands the stream messages after the results. Both read it + // by the frame's activeTurnTriggers entry, the trigger as it came, not + // the copy without a wake's marks that this frame goes on with. + const frameTrigger = this.activeTurnTriggers.get(agent.name); + if (frameTrigger) { + if (continuePrompt) this.turnPrompts.set(frameTrigger, continuePrompt); + else this.turnPrompts.delete(frameTrigger); } // The turn has a live stream from here, which carries a stop: this @@ -20021,7 +20073,13 @@ export class AgentFramework { /** * RFC-006 assembly: the named agent's turn is alive but its compile has * not run yet (the same window the turn-start deferred flush writes - * in). A message for THAT agent is stored directly instead of deferred. + * in). While that agent's assembly runs (`assembling`), a message for + * THAT agent is stored directly instead of deferred, after what is + * queued for the agent, which is written first (flushForStart). When + * some of that stays queued, because a write that failed cut it or the + * host is quiesced, the message is deferred behind it. Outside the + * assembly it counts only for a write from the queue (stop()'s drain): + * any other write is weighed as if it weren't set. */ bypassDeferralFor?: string; } @@ -20087,13 +20145,28 @@ export class AgentFramework { // slot with overlapping turns put each other's writes inside the other's // live turn whatever this check does: the limit #286 states. const midToolCycle = this.pendingAssistantBlocks.has(agent.name); + // A write assembled for the agent's start lands after what is queued for + // the agent: the queue is written first, as the turn-start flush writes it + // in the same window, and if a write that failed cut it, this write waits + // behind what stays queued (writeDrainedInOrder). While the host is + // quiesced, what is queued waits for resume(), so this write waits behind + // it. A write from the queue belongs to a drain in progress (stop()'s), + // whose entries stay un-acked until it has written them all, so it + // doesn't flush: the flush's ack would acknowledge them before they're + // stored. + const bypasses = opts?.bypassDeferralFor === agent.name && + (opts.fromQueue !== undefined || this.assembling.has(agent.name)); + const assembledForStart = bypasses && !opts?.fromQueue; + const behindQueue = assembledForStart && this.deferredMessages.length > 0 && + (this.quiesced ? this.queuedFor(agent.name) : this.flushForStart(agent, 'assembly')); if ( !freshToolResult && - opts?.bypassDeferralFor !== agent.name && - (this.quiesced || - midToolCycle || - this.activeTurnTokens.has(agent.name) || - this.activeStreams.has(agent.name)) + (behindQueue || + (!bypasses && + (this.quiesced || + midToolCycle || + this.activeTurnTokens.has(agent.name) || + this.activeStreams.has(agent.name)))) ) { // A re-deferral of a drained entry keeps its id and seq, and leaves the // un-acked set: one logical message, one durable identity, one place @@ -20214,11 +20287,24 @@ export class AgentFramework { * id at each flush, and holds its target's later queued writes rather than * letting them land out of order. * - * The hold is the queue's alone. A write that isn't deferred is stored at - * once and can land ahead of a held entry: a fresh tool_result, whatever - * its agent's state, and any write while its agent's turn, stream and tool - * cycle are all idle and the host isn't quiesced. So an entry that keeps - * failing holds only what is deferred behind it. + * From a fresh turn's start to the end of its coalescing assembly, before + * its checkpoint, every write deferred for its agent lands in that + * request, in arrival order, up to one that can't be stored and unless the + * host is quiesced (flushForStart). The turn-start flush writes the queue; + * while the assembly runs, each write it makes into the agent's own window + * (addMessage's bypassDeferralFor) first writes what was deferred since; + * and what was deferred during the assembly is written once it ends. A + * write the assembly makes while an entry is held waits behind it in the + * queue, so the hold covers the assembly's writes too, and so does one + * made while the host is quiesced with anything queued for the agent: a + * quiesce that begins while an assembly awaits a render lets that turn + * drain, and what is queued waits for resume(). Outside that window the + * hold is the queue's alone: a write that isn't deferred is stored at once + * and can land ahead of a held entry, as a fresh tool_result does whatever + * its agent's state, and any write does while its agent's turn, stream and + * tool cycle are all idle and the host isn't quiesced. So an entry that + * keeps failing holds what is deferred behind it and the assembly's writes + * for its agent. * * A write can throw after its message is stored: a store listener, the * strategy's handling of the add, or a later shard of a sharded body can @@ -20345,6 +20431,35 @@ export class AgentFramework { ); } + /** + * Write what is queued for `agent` into its window past its deferral, in + * queue order and cut at a write that fails (writeDrainedInOrder), between + * a fresh turn's start and its checkpoint, where what is stored is the + * turn's input: at the start, before each write the coalescing assembly + * makes for the agent, and once the assembly ends, the last two not while + * the host is quiesced, when what is queued waits for resume(). What is + * stored is weighed for the start now beginning and for every other + * resident's open turn. Returns whether anything is still queued for the + * agent. + */ + private flushForStart(agent: Agent, occasion: 'turn-start' | 'assembly'): boolean { + const stored = this.writeDrainedInOrder(this.drainDeferredFor(agent.name), `${occasion} flush for ${agent.name}`, (msg) => + agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); + for (const [msg, { id }] of stored) { + this.emitTrace({ type: 'message:added', messageId: id, source: `deferred-flush:${occasion}` }); + this.noteArrival(agent.name, msg.metadata, id); + this.noteForOtherResidents(agent, msg.metadata, id); + } + this.ackDeferredWrites(); + return this.queuedFor(agent.name); + } + + /** Whether anything is queued for `agentName`, by drainDeferredFor's + * target rule. */ + private queuedFor(agentName: string): boolean { + return this.deferredMessages.some((msg) => (msg.forAgent ?? this.primaryAgentName) === agentName); + } + /** Re-add drained entries through addMessage, which stores each or defers * it again, in queue order and cut per target (writeDrainedInOrder). With * `bypassDeferral`, each is stored past its target's deferral (stop()). A @@ -20693,6 +20808,33 @@ export class AgentFramework { return cm.queryMessages({ metadata: { deferredWriteId: writeId } }).messages[0] ?? null; } + /** Whether the agent has nothing left to read of a deferred write, on + * evidence only: the row it landed as is at or below the agent's consumed + * watermark on its current branch, or it's in neither the queue (drained + * entries not yet stored included) nor the store's current branch: + * removed, or left on a branch the agent has switched from, where a turn + * now would find nothing of it (switching back brings it back unread, + * with no wake, as main's wake spent on this branch leaves it). Any doubt + * reads as unread, such as a watermark that a branch switch or an undo + * left on another branch, or a store that can't be read: a turn that + * finds nothing new costs less than a write no wake brings the agent to + * read. */ + private deferredWriteReadOrGone(agent: Agent, writeId: string): boolean { + if (this.deferredMessages.some((m) => m.id === writeId)) return false; + try { + const cm = agent.getContextManager(); + const row = this.landedRow(cm, writeId); + // No row yet for a write drained from the queue and still to be stored + // (un-acked); one stored and awaiting a sync is judged by its row. + if (!row) return !this.unackedDeferredWrites.some((m) => m.id === writeId); + const watermark = agent.getConsumedWatermark(); + return !!watermark && watermark.branch === cm.currentBranch().name && row.sequence <= watermark.sequence; + } catch (error) { + console.error(`[coalescing] ${agent.name}: could not read whether deferred write ${writeId} was read, so its wake stays:`, error); + return false; + } + } + // A module context's reads, edits and removals, and a process answer's, // go to the store addMessage writes by default, and take a held write's // handle as well as a stored id. A handle reaches the write while it's diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index fae7d094b..156b4a561 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -139,10 +139,11 @@ export interface AssemblyResult { * 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. */ + * include the agent, a render placed in a deferred queue (another + * agent's, or this agent's own behind a write held there), 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; } @@ -222,9 +223,9 @@ export interface CoalescerHost { 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; + * no request consumed, at `deferredId`. It carries no subject, so the + * subject's other wakes, and later withdrawals of them, leave it. */ + wakeToRead(occurrence: CoalescedOccurrence, agent: string, deferredId?: 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. */ @@ -917,13 +918,17 @@ export class PushCoalescer { const run = state.occupant?.rendered ? state.occupant : undefined; const placement = await this.host.deliver({ ...occurrence, timestamp, content }, content, assemblingFor); if (placement?.deferredId) { - // 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, 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. + // Landed in a deferred queue: another agent's, whose turn is alive, + // or the assembling agent's own, behind a write held there (the + // host's bypassDeferralFor), or after its assembly ended, as any + // write during its turn. 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, + // 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 (the wake starts no turn + // once the agent has read it, or it's removed). // The run's renders keep their own records, whatever else an // occupant carries. let before: Occupant['before']; @@ -937,7 +942,7 @@ export class PushCoalescer { await this.host.wakeForBatch(state.batch.latest); } state.audienceAgent = placement.agent; - await this.host.wakeToRead(occurrence, placement.agent); + await this.host.wakeToRead(occurrence, placement.agent, placement.deferredId); return; } // The materialized occurrence is consumed by the request being diff --git a/src/mcpl/push-handler.ts b/src/mcpl/push-handler.ts index 803789212..c6fd5fcab 100644 --- a/src/mcpl/push-handler.ts +++ b/src/mcpl/push-handler.ts @@ -48,7 +48,9 @@ export interface McplPushEvent { targetAgents?: string[]; /** RFC-006: set when the coalescer delivers; rides the wake it queues. */ coalescingSubject?: string; - /** RFC-006 assembly: materialized for this agent's turn (store directly). */ + /** RFC-006 assembly: materialized for this agent's turn, stored directly + * while its assembly runs, after what is queued for it (the framework's + * addMessage, bypassDeferralFor). */ assemblingFor?: string; /** Host acceptance time (epoch ms), stamped where the push is admitted. */ acceptedAt?: number; diff --git a/src/types/agent.ts b/src/types/agent.ts index 58c2bebac..576bd91e1 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -371,6 +371,10 @@ export interface InferenceRequest { * all of them without materializing content, the turn has no cause left * and does not run. */ coalescingBatchSubjects?: string[]; + /** Host-owned: this wake's sole cause is reading this deferred write, a + * coalesced render placed in its agent's queue (wakeToRead). Once the + * agent has read the write, or it was removed, the wake starts no turn. */ + readsDeferredWrite?: string; agentName: string; reason: string; source: string; diff --git a/test/deferred-wake-integrity.test.ts b/test/deferred-wake-integrity.test.ts index 6aa5cdc5d..f6e65e71b 100644 --- a/test/deferred-wake-integrity.test.ts +++ b/test/deferred-wake-integrity.test.ts @@ -16,7 +16,10 @@ * it (turn-start flush runs before the locus announcement and compile). * 2. addMessage defers on turn-alive (activeTurnTokens), not merely * stream-alive — the compile window between dequeue and stream - * registration is closed to cross-turn writers. + * registration is closed to cross-turn writers: what they write is + * deferred, and what is deferred before the turn's checkpoint lands as + * the turn's input through its own flushes (flushForStart), unless the + * host is quiesced by then, when it waits for resume(). * 3. endTurn edge: a message stored at an endTurn tool boundary (collected * for injection but never delivered — the stream is cancelled) is seen by * the follow-up turn its own wake starts, positioned AFTER the turn that diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index b0c5ffb83..49fd21081 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -1,7 +1,10 @@ /** * What the window stores mid-turn is what the live stream carried. * - * A message deferred during a turn is stored at the agent's next tool + * A message deferred during a turn's start, until its coalescing assembly + * ends, is stored before the turn's checkpoint, in that request, as the + * turn-start flush stores what came before. One deferred after that is + * stored at the agent's next tool * boundary and injected into the resumed stream (hear-while-acting), so the * next compile's prefix is the request the turn's later replies were minted * under. Two kinds of message can't be injected: one carrying tool blocks @@ -331,11 +334,18 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { pendingAssistantBlocks: Map; activeTurnTokens: Map; activeStreams: Map; - deferredMessages: Array<{ id: string; seq: number; participant: string; forAgent?: string }>; + deferredMessages: Array<{ id: string; seq: number; participant: string; forAgent?: string; content: ContentBlock[] }>; + unackedDeferredWrites: Array<{ id: string }>; deferredWritesPersisted: boolean; + quiesced: boolean; + assembling: Set; drainDeferredFor: (name: string, stopAt?: unknown) => unknown[]; readdDrained: (entries: unknown[], where: string) => unknown[]; - addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string; fromQueue?: { id: string; seq: number } }): string; + ackDeferredWrites: () => void; + addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { + forAgent?: string; fromQueue?: { id: string; seq: number }; bypassDeferralFor?: string; + placement?: { agent?: string; messageId?: string; deferredId?: string }; + }): string; }; } const otherCycle = (): ContentBlock[] => [{ type: 'tool_use', id: 'b1', name: 'robot--move', input: {} } as ContentBlock]; @@ -1087,4 +1097,164 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { } await framework.stop(); }); + + // The coalescing assembly writes past its agent's deferral, in the window + // the turn-start flush writes in (bypassDeferralFor). What was deferred + // for the agent before such a write came first, so it lands first. + const textsOf = (entries: Array<{ content: ContentBlock[] }>): string[] => + entries.map((m) => m.content.map((b) => (b as { text?: string }).text ?? `<${b.type}>`).join(' ')); + + it("a write assembled for a turn's start lands after what was deferred for its agent before it, and acknowledges it", async () => { + const framework = await createFramework(); + const internals = internalsOf(framework); + internals.deferredWritesPersisted = true; + internals.activeTurnTokens.set('assistant', 1); + internals.assembling.add('assistant'); + try { + assert.equal(internals.addMessage('Antra', text('first')), '', 'deferred: the turn is alive'); + // Another target's entry: it neither flushes nor holds this agent's write. + internals.deferredMessages.push({ id: 'ghost-1', seq: 0, participant: 'Antra', forAgent: 'ghost', content: text('elsewhere') }); + const placement: { agent?: string; messageId?: string; deferredId?: string } = {}; + const id = internals.addMessage('Antra', text('second'), undefined, { bypassDeferralFor: 'assistant', placement }); + assert.notEqual(id, '', 'the assembled write is stored'); + assert.equal(placement.deferredId, undefined); + assert.deepEqual(windowOf(framework).slice(-2), ['Antra: first', 'Antra: second'], 'what was deferred first lands first'); + assert.deepEqual(textsOf(internals.deferredMessages), ['elsewhere'], "only the other target's entry is still queued"); + assert.deepEqual(internals.unackedDeferredWrites, [], 'what the flush stored is acknowledged'); + } finally { + internals.assembling.delete('assistant'); + internals.activeTurnTokens.delete('assistant'); + internals.deferredMessages.splice(0); + } + await framework.stop(); + }); + + it("a write naming its agent's assembly after that assembly has ended is deferred like any write during the turn", async () => { + const framework = await createFramework(); + const internals = internalsOf(framework); + internals.activeTurnTokens.set('assistant', 1); + try { + internals.addMessage('Antra', text('first')); + // An assembly that throws can leave a render running, which delivers + // after the window has closed. + const placement: { agent?: string; messageId?: string; deferredId?: string } = {}; + assert.equal(internals.addMessage('Antra', text('late'), undefined, { bypassDeferralFor: 'assistant', placement }), '', 'deferred'); + assert.ok(placement.deferredId); + assert.deepEqual(textsOf([...internals.deferredMessages].sort((a, b) => a.seq - b.seq)), ['first', 'late'], 'nothing was flushed into the live turn'); + assert.ok(!windowOf(framework).some((line) => line === 'Antra: first' || line === 'Antra: late')); + } finally { + internals.activeTurnTokens.delete('assistant'); + } + await framework.stop(); + }); + + it("a write assembled for a turn's start waits behind a deferred write that can't be stored, and lands after it", async () => { + const framework = await createFramework(); + const internals = internalsOf(framework); + const cm = framework.getAgent('assistant')!.getContextManager(); + const add = cm.addMessage.bind(cm); + let failing = true; + (cm as unknown as { addMessage: typeof add }).addMessage = (...args) => { + if (failing && (args[1] as Array<{ text?: string }>).some((b) => b.text === 'first')) throw new Error('disk full'); + return add(...args); + }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + internals.activeTurnTokens.set('assistant', 1); + internals.assembling.add('assistant'); + try { + internals.addMessage('Antra', text('first')); + const placement: { agent?: string; messageId?: string; deferredId?: string } = {}; + assert.equal(internals.addMessage('Antra', text('second'), undefined, { bypassDeferralFor: 'assistant', placement }), '', 'deferred'); + assert.ok(placement.deferredId, 'its placement names the queued entry'); + assert.ok(errors.some((e) => e.includes('assembly flush for assistant') && e.includes('disk full')), errors.join('\n')); + assert.deepEqual(textsOf([...internals.deferredMessages].sort((a, b) => a.seq - b.seq)), ['first', 'second'], 'it waits behind the held write'); + assert.ok(!windowOf(framework).includes('Antra: second')); + } finally { + internals.assembling.delete('assistant'); + internals.activeTurnTokens.delete('assistant'); + console.error = error; + } + failing = false; + internals.readdDrained(internals.drainDeferredFor('assistant'), 'the turn ends'); + internals.ackDeferredWrites(); + assert.deepEqual(windowOf(framework).slice(-2), ['Antra: first', 'Antra: second'], 'both land, in the order they arrived'); + await framework.stop(); + }); + + it("a write assembled for a turn's start waits behind what is queued for its agent while the host is quiesced", async () => { + const framework = await createFramework(); + const internals = internalsOf(framework); + internals.activeTurnTokens.set('assistant', 1); + internals.assembling.add('assistant'); + try { + internals.addMessage('Antra', text('first')); + // A quiesce that begins while the turn's assembly awaits a render lets + // the turn drain: what is queued waits for resume(), and so does what + // the assembly writes after it. + internals.quiesced = true; + const placement: { agent?: string; messageId?: string; deferredId?: string } = {}; + assert.equal(internals.addMessage('Antra', text('second'), undefined, { bypassDeferralFor: 'assistant', placement }), '', 'deferred'); + assert.ok(placement.deferredId); + assert.deepEqual(textsOf([...internals.deferredMessages].sort((a, b) => a.seq - b.seq)), ['first', 'second']); + assert.ok(!windowOf(framework).some((line) => line === 'Antra: first' || line === 'Antra: second'), 'nothing queued is written while quiesced'); + } finally { + internals.quiesced = false; + internals.assembling.delete('assistant'); + internals.activeTurnTokens.delete('assistant'); + } + internals.readdDrained(internals.drainDeferredFor('assistant'), 'resume'); + internals.ackDeferredWrites(); + assert.deepEqual(windowOf(framework).slice(-2), ['Antra: first', 'Antra: second']); + await framework.stop(); + }); + + it("a write assembled for a turn's start is stored while the host is quiesced when nothing is queued for its agent", async () => { + const framework = await createFramework(); + const internals = internalsOf(framework); + internals.activeTurnTokens.set('assistant', 1); + internals.assembling.add('assistant'); + internals.quiesced = true; + try { + // Another target's entry doesn't count as queued for this agent. + internals.deferredMessages.push({ id: 'ghost-1', seq: 0, participant: 'Antra', forAgent: 'ghost', content: text('elsewhere') }); + assert.notEqual(internals.addMessage('Antra', text('assembled'), undefined, { bypassDeferralFor: 'assistant' }), '', 'stored, as the turn\'s own input'); + assert.deepEqual(windowOf(framework).slice(-1), ['Antra: assembled']); + assert.deepEqual(textsOf(internals.deferredMessages), ['elsewhere'], 'and nothing of the queue was written'); + } finally { + internals.quiesced = false; + internals.assembling.delete('assistant'); + internals.activeTurnTokens.delete('assistant'); + internals.deferredMessages.splice(0); + } + await framework.stop(); + }); + + it("stop()'s drain, which writes past each agent's deferral, keeps its entries un-acked until it has stored them", async () => { + const framework = await createFramework(); + const internals = internalsOf(framework); + internals.deferredWritesPersisted = true; + internals.activeTurnTokens.set('assistant', 1); + internals.addMessage('Antra', text('first')); + internals.addMessage('Antra', text('second')); + internals.activeTurnTokens.delete('assistant'); + // An entry for a target that isn't registered stays queued through the + // drain, so the queue isn't empty while the drain writes. + internals.deferredMessages.push({ id: 'ghost-1', seq: 0, participant: 'Antra', forAgent: 'ghost', content: text('elsewhere') }); + const cm = framework.getAgent('assistant')!.getContextManager(); + const add = cm.addMessage.bind(cm); + const unackedAt: Record = {}; + (cm as unknown as { addMessage: typeof add }).addMessage = (...args) => { + const written = (args[1] as Array<{ text?: string }>).map((b) => b.text).join(' '); + unackedAt[written] = internals.unackedDeferredWrites.map((m) => m.id); + return add(...args); + }; + const secondId = internals.deferredMessages.find((m) => textsOf([m])[0] === 'second')!.id; + const error = console.error; + console.error = () => {}; + try { await framework.stop(); } finally { console.error = error; } + assert.ok(unackedAt.first?.includes(secondId), `'second' is still un-acked while 'first' is stored: ${JSON.stringify(unackedAt)}`); + assert.ok(unackedAt.second?.includes(secondId), `and while 'second' itself is stored: ${JSON.stringify(unackedAt)}`); + }); }); diff --git a/test/mcpl-coalescer-render-run.test.ts b/test/mcpl-coalescer-render-run.test.ts index 5e4838bae..414181248 100644 --- a/test/mcpl-coalescer-render-run.test.ts +++ b/test/mcpl-coalescer-render-run.test.ts @@ -18,7 +18,8 @@ interface Options { * 'other', an agent whose turn is alive, until it's removed, read, or gone, * whichever agent assembled it. CoalescerHost leaves placement to the host, * so this is one it allows; the framework stores the assembling agent's own - * render directly, which mcpl-coalescing's tests run. Its removal, like the + * render directly while its assembly runs, after what is queued for it, + * which mcpl-coalescing's tests run. Its removal, like the * framework's, doesn't refuse a message a request has read. Each render says * which notices it covers, as a notice-only server's does. */ function makeHost(options: Options = {}) { diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index 2805142c2..dc1c7121a 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; coalescingBatchSubjects?: string[] }; +type Wake = { agentName: string; reason?: string; source?: string; timestamp?: number; coalescingSubject?: string; coalescingBatch?: boolean; coalescingBatchSubjects?: string[]; readsDeferredWrite?: string }; type TurnInternals = { startAgentStream: (...a: unknown[]) => Promise; @@ -1370,6 +1370,507 @@ test("a render placed in a live turn's deferred queue takes a newer batch of its assert(f.lastRequest().includes('first_render'), "the primary's turn read the render"); }); +test("a render whose write failed at its agent's turn start is read before the next render of its subject, in that turn's request", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + let renders = 0; + f.renderer(() => Promise.resolve({ content: [{ type: 'text', text: ++renders > 1 ? 'second_render' : 'first_render' }] })); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // As in the test above: 'other''s assembly places the render in the + // primary's deferred queue while the primary's turn is alive, and a notice + // opens a newer batch, which follows the primary. That turn ends without + // a flush, so the render is still queued when the primary's next turn + // starts, and that start's flush fails to store it, once. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'the scheduler pass'); + assert.equal(renders, 1, "'other''s assembly rendered the first batch"); + await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); + } finally { + internals.activeTurnTokens.delete('agent'); + } + const cm = f.framework.getAgent('agent')!.getContextManager(); + const add = cm.addMessage.bind(cm); + let failed = false; + (cm as unknown as { addMessage: typeof add }).addMessage = (...args) => { + if (!failed && JSON.stringify(args[1]).includes('first_render')) { + failed = true; + throw new Error('transient'); + } + return add(...args); + }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + await settles(f.framework.runUntilIdle(), "the primary's turn"); + } finally { + console.error = error; + } + assert.ok(errors.some((e) => e.includes('turn-start flush for agent') && e.includes('transient')), "the start's flush couldn't store the first render"); + assert.equal(renders, 2); + assert.equal(f.membrane.calls.length, 1); + const request = f.lastRequest(); + assert(request.includes('first_render') && request.indexOf('first_render') < request.indexOf('second_render'), 'one call read the first render, then the second'); +}); + +test("a render an assembly joins reaches the joining agent's request when it lands in that agent's queue (vector 25)", 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 joined!: () => void; + const notesStarted = new Promise((resolve) => { joined = resolve; }); + f.renderer((params) => { + if (JSON.stringify(params).includes('notes')) { + joined(); + return Promise.resolve({ content: [{ type: 'text', text: 'notes_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; + // While the primary's turn is alive, 'other''s assembly freezes the + // document's batch and awaits its render. + internals.activeTurnTokens.set('agent', -1); + const first = internals.processInferenceRequests(); + await settles(started, "the document's render for 'other'"); + // A batch of another subject wakes the primary, whose assembly joins the + // document's render (it is in that batch's audience) and renders the notes' + // batch. Its iteration reaches the document first, so the notes' render + // starting means the join is made. + await f.send('push/event', f.params('2', 'notes_fallback', { deferred: true, initial: true, key: 'notes' })); + internals.activeTurnTokens.delete('agent'); + const second = internals.processInferenceRequests(); + await settles(notesStarted, "the notes' render for the primary"); + // The document's render lands in the primary's window, written for + // 'other''s assembly, so the primary's live turn defers it into its queue. + const flushed: string[] = []; + f.framework.onTrace((e) => { + const event = e as { type: string; source?: string; messageId?: string }; + if (event.type === 'message:added' && event.source === 'deferred-flush:assembly' && event.messageId) flushed.push(event.messageId); + }); + release({ content: [{ type: 'text', text: 'first_render' }] }); + await settles(Promise.all([first, second]), 'both passes'); + await settles(f.framework.runUntilIdle(), 'the drain'); + const primaryCalls = f.membrane.calls.filter((c) => JSON.stringify((c as { messages?: unknown }).messages).includes('notes_render')); + assert.equal(primaryCalls.length, 1, "one primary turn: the render's wake to read it was retired once its request read it"); + assert(JSON.stringify((primaryCalls[0] as { messages?: unknown }).messages).includes('first_render'), "the primary's request read the render its assembly joined"); + const store = f.framework.getAgent('agent')!.getContextManager().getAllMessages(); + assert.ok(store.some((m) => flushed.includes(m.id) && JSON.stringify(m.content).includes('first_render')), "the render was deferred into the primary's queue and written by its assembly's flush"); + // Stored before the turn's checkpoint, it is the turn's input: undoing the + // turn keeps it. + f.framework.undoLastTurn('agent'); + assert(f.context().includes('first_render'), 'the undo of the turn that read it keeps it'); +}); + +test("a render deferred behind a write the store keeps refusing stays tracked, and a retraction removes both", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + let renders = 0; + f.renderer(() => Promise.resolve({ content: [{ type: 'text', text: ++renders > 1 ? 'second_render' : 'first_render' }] })); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals & { deferredMessages: Array<{ content: unknown }> }; + // As above, the first render waits in the primary's queue when its next + // turn starts, and the newer batch follows the primary. Here the store + // refuses the first render every time. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'the scheduler pass'); + await f.send('push/event', f.params('2', 'second_fallback', { deferred: true })); + } finally { + internals.activeTurnTokens.delete('agent'); + } + const cm = f.framework.getAgent('agent')!.getContextManager(); + const add = cm.addMessage.bind(cm); + (cm as unknown as { addMessage: typeof add }).addMessage = (...args) => { + if (JSON.stringify(args[1]).includes('first_render')) throw new Error('disk full'); + return add(...args); + }; + const error = console.error; + console.error = () => {}; + try { + await settles(f.framework.runUntilIdle(), "the primary's turns"); + } finally { + console.error = error; + } + assert.equal(renders, 2, "the primary's assembly rendered the newer batch"); + const queued = () => internals.deferredMessages.map((m) => JSON.stringify(m.content)); + assert.ok(queued().some((c) => c.includes('second_render')), 'the newer render waits behind the refused one'); + assert.ok(f.membrane.calls.every((c) => !JSON.stringify((c as { messages?: unknown }).messages).includes('second_render')), 'no request read it without the render it builds on'); + const r = await f.send('push/event', f.params('3', '', { retract: true })); + assert.equal(r.result.coalesce.outcome, 'retracted', 'both renders were unread, and both are withdrawn'); + assert.ok(!queued().some((c) => c.includes('first_render') || c.includes('second_render')), 'neither is queued any more'); +}); + +test("a wake to read a render that was retracted unread starts no turn", async (t) => { + const f = await fixture({ agents: TWO_RESIDENTS }); t.after(f.close); + f.alwaysRespond(); + f.renderer(() => Promise.resolve({ content: [{ type: 'text', text: 'first_render' }] })); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals; + // While the primary's turn is alive, 'other''s assembly places the render + // in the primary's queue, with a wake for the primary to read it. The + // server retracts it before the primary is free. + internals.activeTurnTokens.set('agent', -1); + try { + await settles(internals.processInferenceRequests(), 'the scheduler pass'); + assert.ok(internals.pendingRequests.some((r) => r.agentName === 'agent'), "the primary's wake to read the render"); + const r = await f.send('push/event', f.params('2', '', { retract: true })); + assert.equal(r.result.coalesce.outcome, 'retracted'); + } finally { + endStandInTurn(internals); + } + await settles(f.framework.runUntilIdle(), 'the drain'); + assert.equal(f.membrane.calls.length, 0, 'nothing is left to read, so no turn runs'); +}); + +test("a wake to read a render neither carries its mark into a merge with other causes nor absorbs another wake while parked", async (t) => { + const f = await fixture(); t.after(f.close); + const internals = f.framework as unknown as TurnInternals & { + quiesced: boolean; + deferredMessages: Array<{ id: string; seq: number; participant: string; content: unknown[]; heldAt: number }>; + }; + // The renders they read wait, unread, in the agent's queue. + internals.deferredMessages.push( + { id: 'render-1', seq: 1, participant: 'editor', content: [{ type: 'text', text: 'queued_render' }], heldAt: Date.now() }, + { id: 'render-2', seq: 2, participant: 'editor', content: [{ type: 'text', text: 'other_render' }], heldAt: Date.now() }); + const wake = (extra: Partial = {}): Wake => ({ agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), ...extra }); + // Parked while quiesced, wakes are kept newest by key; a wake to read the + // render keys apart from a plain push wake, so neither absorbs the other. + internals.quiesced = true; + try { + internals.pendingRequests.push(wake({ readsDeferredWrite: 'render-1' }), wake(), wake({ readsDeferredWrite: 'render-2' })); + await settles(internals.processInferenceRequests(), 'the parked pass'); + assert.deepEqual(internals.pendingRequests.map((r) => r.readsDeferredWrite), ['render-1', undefined, 'render-2'], 'each wake stays parked, those reading different renders included'); + } finally { + internals.quiesced = false; + } + // Merged into one turn, the trigger keeps the mark only when every wake + // has it. The wake to read the render comes first, so it is the trigger. + const triggers: Array<{ readsDeferredWrite?: string }> = []; + const start = internals.startAgentStream; + internals.startAgentStream = async (_agent: unknown, trigger: unknown) => { triggers.push(trigger as { readsDeferredWrite?: string }); return undefined; }; + try { + await settles(internals.processInferenceRequests(), 'the merged pass'); + internals.pendingRequests.push(wake({ readsDeferredWrite: 'render-1' }), wake({ readsDeferredWrite: 'render-2' })); + await settles(internals.processInferenceRequests(), 'the pass of two renders'); + internals.pendingRequests.push(wake({ readsDeferredWrite: 'render-1' })); + await settles(internals.processInferenceRequests(), 'the lone pass'); + } finally { + internals.startAgentStream = start; + internals.deferredMessages.splice(0); + } + assert.deepEqual(triggers.map((r) => r.readsDeferredWrite), [undefined, undefined, 'render-1'], 'merged with a plain wake, or one reading another render, it drops the mark; alone it keeps it'); +}); + +test("a wake to read a render starts its turn until the render's row is at or below the agent's watermark on the current branch", async (t) => { + const f = await fixture({ server: { shouldTriggerInference: () => true } }); t.after(f.close); + const internals = f.framework as unknown as TurnInternals & { + deferredMessages: Array<{ id: string; seq: number; participant: string; content: unknown[]; heldAt: number }>; + }; + const agent = f.framework.getAgent('agent')! as unknown as { + getContextManager(): { getAllMessages(): Array<{ sequence: number; metadata?: { deferredWriteId?: string } }>; currentBranch(): { name: string } }; + getConsumedWatermark(): { branch: string; sequence: number } | null; + consumedWatermark: { branch: string; sequence: number } | null; + }; + // A render placed in the agent's queue lands in its store, as a flush + // stores it, while its wake to read it waits. + const land = (id: string): { sequence: number } => { + internals.deferredMessages.push({ id, seq: 1, participant: 'editor', content: [{ type: 'text', text: `${id}_text` }], heldAt: Date.now() }); + internals.readdDrained(internals.drainDeferredFor('agent'), `${id} lands`); + internals.ackDeferredWrites(); + const row = agent.getContextManager().getAllMessages().find((m) => m.metadata?.deferredWriteId === id); + assert(row, `${id} landed on the current branch`); + return row; + }; + // The pass starts whatever turns are left. + const pass = async (id: string): Promise => { + internals.pendingRequests.push({ agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: id }); + let started = 0; + const start = internals.startAgentStream; + internals.startAgentStream = async () => { started++; return undefined; }; + try { + await settles(internals.processInferenceRequests(), `the pass after ${id}`); + } finally { + internals.startAgentStream = start; + } + return started; + }; + // Above the watermark the agent's creation set: unread. + land('render_0'); + assert.equal(await pass('render_0'), 1, 'above the watermark on its branch, the wake starts the turn'); + // At the watermark itself: read, as a compile whose newest row it was. + const atMark = land('render_at'); + agent.consumedWatermark = { branch: agent.getContextManager().currentBranch().name, sequence: atMark.sequence }; + assert.equal(await pass('render_at'), 0, 'at the watermark on its branch, the wake starts no turn'); + // Below it: read, by a compile that read what followed it too. + land('render_below'); + const after = land('render_after'); + agent.consumedWatermark = { branch: agent.getContextManager().currentBranch().name, sequence: after.sequence }; + assert.equal(await pass('render_below'), 0, 'below the watermark on its branch, the wake starts no turn'); + // Drained from the queue but not yet stored: still queued. + const pendingWrite = { id: 'render_drained', seq: 1, participant: 'editor', content: [{ type: 'text', text: 'render_drained_text' }], heldAt: Date.now() }; + (internals as unknown as { unackedDeferredWrites: unknown[] }).unackedDeferredWrites.push(pendingWrite); + assert.equal(await pass('render_drained'), 1, 'drained and not yet stored, the wake starts the turn'); + (internals as unknown as { unackedDeferredWrites: unknown[] }).unackedDeferredWrites.splice(0); + // Stored and read, but still un-acked while the chronicle's sync fails: + // judged by its row. + const unsynced = land('render_unsynced'); + (internals as unknown as { unackedDeferredWrites: unknown[] }).unackedDeferredWrites.push({ ...pendingWrite, id: 'render_unsynced' }); + agent.consumedWatermark = { branch: agent.getContextManager().currentBranch().name, sequence: unsynced.sequence }; + assert.equal(await pass('render_unsynced'), 0, 'stored and read, though not yet synced, the wake starts no turn'); + (internals as unknown as { unackedDeferredWrites: unknown[] }).unackedDeferredWrites.splice(0); + // With no watermark (a store that couldn't be read when it was taken): in doubt. + const noMark = land('render_null'); + agent.consumedWatermark = null; + assert.equal(await pass('render_null'), 1, 'with no watermark, the wake starts the turn'); + agent.consumedWatermark = { branch: agent.getContextManager().currentBranch().name, sequence: noMark.sequence }; + await f.send('push/event', f.params('a', 'turn_one', { key: 'a', initial: true })); + await f.framework.runUntilIdle(); + // An operator's undo switches the store to a new branch; the agent's + // watermark stays on the old one until its next committing compile. The + // render lands at or below that watermark's sequence, so only the branch + // tells it unread. + f.framework.undoLastTurn('agent'); + const afterUndo = land('render_1'); + const mark = agent.getConsumedWatermark(); + assert(mark && mark.branch !== agent.getContextManager().currentBranch().name && afterUndo.sequence <= mark.sequence, + `the watermark ${JSON.stringify(mark)} is on another branch, at or past the render's row (${afterUndo.sequence})`); + assert.equal(await pass('render_1'), 1, 'with the watermark on another branch, the wake starts the turn'); +}); + +test("a turn a wake to read a render started, held by a provider hold, is retried once the hold ends, though its compile read the render", async (t) => { + const f = await fixture(); t.after(f.close); + 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])); + }; + const internals = f.framework as unknown as TurnInternals & { + deferredMessages: Array<{ id: string; seq: number; participant: string; content: unknown[]; heldAt: number }>; + }; + // A render waits in the agent's queue with its wake to read it; the turn + // the wake starts stores it at its start and compiles it. + internals.deferredMessages.push({ id: 'render-1', seq: 1, participant: 'editor', content: [{ type: 'text', text: 'queued_render' }], heldAt: Date.now() }); + internals.pendingRequests.push({ agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: 'render-1' }); + await settles(f.framework.runUntilIdle(), 'the turn that meets the limit'); + assert.equal(f.membrane.calls.length, 1); + assert(f.lastRequest().includes('queued_render'), 'its request read the render'); + const hold = internals.providerAccelerationCooldowns.get('agent'); + assert.equal(hold?.heldRequests.length, 1, "the turn's trigger is held"); + assert.equal(hold!.heldRequests[0].readsDeferredWrite, undefined, 'as an ordinary wake, without the mark'); + 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'); +}); + +test("a wake to read a render starts its turn when the store can't say whether the agent read it", async (t) => { + const f = await fixture(); t.after(f.close); + const internals = f.framework as unknown as TurnInternals; + const cm = f.framework.getAgent('agent')!.getContextManager() as unknown as { queryMessages: (...a: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('a blob the store names is missing'); }; + internals.pendingRequests.push({ agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: 'render-1' }); + const triggers: unknown[] = []; + const start = internals.startAgentStream; + internals.startAgentStream = async (_agent: unknown, trigger: unknown) => { triggers.push(trigger); return undefined; }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + await settles(internals.processInferenceRequests(), 'the pass'); + } finally { + console.error = error; + internals.startAgentStream = start; + cm.queryMessages = query; + } + assert.equal(triggers.length, 1, 'in doubt, the wake starts the turn'); + assert(errors.some((e) => e.includes('render-1') && e.includes('a blob the store names is missing')), `the failed read is logged: ${JSON.stringify(errors)}`); +}); + +test("a wake to read a render that's gone is dropped from a pass whose other wakes still start the turn", async (t) => { + const f = await fixture(); t.after(f.close); + const internals = f.framework as unknown as TurnInternals; + internals.pendingRequests.push( + { agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: 'render-gone' }, + { agentName: 'agent', reason: 'heartbeat', source: 'framework', timestamp: Date.now() }); + const triggers: Wake[] = []; + const start = internals.startAgentStream; + internals.startAgentStream = async (_agent: unknown, trigger: unknown) => { triggers.push(trigger as Wake); return undefined; }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + await settles(internals.processInferenceRequests(), 'the pass'); + } finally { + console.error = error; + internals.startAgentStream = start; + } + assert.deepEqual(triggers.map((r) => [r.reason, r.readsDeferredWrite]), [['heartbeat', undefined]], 'the turn starts for the heartbeat alone'); + assert(errors.some((e) => e.includes('1 wake(s) to read a render') && e.includes('dropped')), `the drop is logged: ${JSON.stringify(errors)}`); +}); + +test("a provider recovery whose released wakes start no turn ends and reopens auxiliary admission: a gone render's wake under any policy, and a wake the policy drops", async (t) => { + const f = await fixture(); t.after(f.close); + const internals = f.framework as unknown as TurnInternals & { + inferencePolicy: { shouldInfer(...a: unknown[]): boolean }; + providerAccelerationRecoveries: Map; + }; + const gate = internals.providerGate('agent') as unknown as { primaryPending: boolean; auxiliaryWaiters: Array<() => void> }; + // A hold's release leaves a primary turn pending, which holds auxiliary + // calls back until it runs; the pass then finds nothing to start. + const releaseThenPass = async (shouldInfer: boolean, wake: Wake): Promise => { + internals.inferencePolicy = { shouldInfer: () => shouldInfer }; + gate.primaryPending = true; + internals.providerAccelerationRecoveries.set('agent', { startedAt: Date.now(), failures: 1, heldRequests: 1, reason: 'held', releasedAt: Date.now() }); + let admitted = false; + gate.auxiliaryWaiters.push(() => { admitted = true; }); + internals.pendingRequests.push(wake); + const error = console.error; + console.error = () => {}; + try { + await settles(internals.processInferenceRequests(), 'the pass'); + await settles(f.framework.runUntilIdle(), 'the drain'); + } finally { + console.error = error; + } + assert.equal(gate.primaryPending, false, 'the pending primary turn is given up'); + assert.equal(internals.providerAccelerationRecoveries.has('agent'), false, 'and the recovery ends'); + return admitted; + }; + // Under a policy that would start a turn for anything, even no cause. + assert.equal(await releaseThenPass(true, { agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: 'render-gone' }), + true, "a gone render's wake: the parked auxiliary call is admitted"); + assert.equal(await releaseThenPass(false, { agentName: 'agent', reason: 'heartbeat', source: 'framework', timestamp: Date.now() }), + true, 'a wake the policy drops: the parked auxiliary call is admitted'); + assert.equal(f.membrane.calls.length, 0, 'no turn runs'); + // A gone render's wake dropped beside a wake that starts the turn leaves + // the recovery to that turn. + internals.inferencePolicy = { shouldInfer: () => true }; + gate.primaryPending = true; + internals.providerAccelerationRecoveries.set('agent', { startedAt: Date.now(), failures: 1, heldRequests: 2, reason: 'held', releasedAt: Date.now() }); + internals.pendingRequests.push( + { agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: 'render-gone' }, + { agentName: 'agent', reason: 'heartbeat', source: 'framework', timestamp: Date.now() }); + let started = 0; + const start = internals.startAgentStream; + internals.startAgentStream = async () => { started++; return undefined; }; + const error = console.error; + console.error = () => {}; + try { + await settles(internals.processInferenceRequests(), 'the pass with a turn to start'); + } finally { + console.error = error; + internals.startAgentStream = start; + } + assert.equal(started, 1, "the heartbeat's turn starts"); + assert.equal(internals.providerAccelerationRecoveries.has('agent'), true, 'and the recovery stays for that turn to end'); +}); + +test("a provider hold keeps a wake to read a render, a batch wake and an ordinary wake apart, though they share a millisecond", async (t) => { + const f = await fixture(); t.after(f.close); + const internals = f.framework as unknown as TurnInternals; + const timer = setTimeout(() => {}, 60_000); + t.after(() => clearTimeout(timer)); + internals.providerAccelerationCooldowns.set('agent', { startedAt: Date.now(), until: Date.now() + 60_000, timer, heldRequests: [], reason: 'held', failures: 1 }); + const at = Date.now(); + const wake = (extra: Partial = {}): Wake => ({ agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: at, ...extra }); + internals.pendingRequests.push(wake(), wake({ coalescingSubject: 'doc', coalescingBatch: true }), wake({ readsDeferredWrite: 'render-1' }), wake({ readsDeferredWrite: 'render-2' })); + await settles(internals.processInferenceRequests(), 'the held pass'); + const held = internals.providerAccelerationCooldowns.get('agent')?.heldRequests ?? []; + assert.deepEqual(held.map((r) => [r.coalescingSubject, r.readsDeferredWrite]), [[undefined, undefined], ['doc', undefined], [undefined, 'render-1'], [undefined, 'render-2']], 'each is held: none stands for another, which a withdrawal or a read could take with it'); +}); + +test("an assembly that throws closes the window in which its writes bypass the turn's deferral", async (t) => { + const f = await fixture(); t.after(f.close); + f.alwaysRespond(); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + const internals = f.framework as unknown as TurnInternals & { + assembling: Set; + pushCoalescer: { assemble(agent: string, causes?: readonly string[]): Promise }; + }; + const assemble = internals.pushCoalescer.assemble.bind(internals.pushCoalescer); + let during: boolean | undefined; + internals.pushCoalescer.assemble = async () => { + during = internals.assembling.has('agent'); + throw new Error('a freeze failed'); + }; + const error = console.error; + console.error = () => {}; + try { + await settles(f.framework.runUntilIdle(), "the agent's turn"); + } finally { + console.error = error; + internals.pushCoalescer.assemble = assemble; + } + assert.equal(during, true, 'the window was open while the assembly ran'); + assert.equal(internals.assembling.size, 0, 'and closed when it threw'); +}); + +test("a write deferred for an agent while its assembly awaits a render lands in that turn's request, unless the host is quiesced by then", async (t) => { + for (const quiesce of [false, true]) { + const f = await fixture(); + try { + 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; }); + }); + const internals = f.framework as unknown as TurnInternals & { + quiesced: boolean; + assembling: Set; + deferredMessages: unknown[]; + addMessage(participant: string, content: Array<{ type: 'text'; text: string }>): string; + }; + // A plain push wakes the agent and a notice opens a batch while another + // turn of its holds it, so one turn starts for both: it has a cause + // besides the batch, and the batch's empty render doesn't withdraw it. + internals.activeTurnTokens.set('agent', -1); + try { + await f.send('push/event', { featureSet: 'doc', eventId: 'p1', timestamp: TS, payload: { content: [{ type: 'text', text: 'plain_before' }] } }); + await f.send('push/event', f.params('1', 'first_fallback', { deferred: true, initial: true })); + await eventually(() => internals.pendingRequests.length > 0, 'the wakes'); + } finally { + internals.activeTurnTokens.delete('agent'); + } + const turn = f.framework.runUntilIdle(); + await settles(started, 'the render'); + // A module's write while the assembly awaits the render is deferred: + // the agent's turn is alive. The render then gives nothing, so no write + // of the assembly's comes after it. + assert.equal(internals.addMessage('Antra', [{ type: 'text', text: 'module_write' }]), '', 'deferred'); + if (quiesce) internals.quiesced = true; + release({ content: [] }); + await settles(turn, "the agent's turn"); + const first = JSON.stringify((f.membrane.calls[0] as { messages?: unknown } | undefined)?.messages); + assert(first.includes('plain_before'), `quiesced ${quiesce}: the turn ran`); + assert.equal(internals.assembling.size, 0, `quiesced ${quiesce}: the window closed with the assembly`); + assert.equal(first.includes('module_write'), !quiesce, quiesce + ? 'a quiesced host keeps it queued for resume()' + : "the turn's request read the write that arrived while its assembly awaited the render"); + if (quiesce) { + assert.ok(internals.deferredMessages.length > 0, 'still queued'); + internals.quiesced = false; + } + } finally { + await f.close(); + } + } +}); + 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(); diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index 7e5d7c8ad..c27ed3e6f 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -1115,6 +1115,7 @@ describe('present while acting', () => { const internals = framework as unknown as { turnRouting: Map; openLogical: Map; pendingRequests: InferenceRequest[]; activeTurnTokens: Map; + assembling: Set; deferredMessages: Array>; addMessage(participant: string, content: ContentBlock[], metadata: Record, opts?: { bypassDeferralFor?: string }): string; }; @@ -1128,7 +1129,9 @@ describe('present while acting', () => { // A delivery assembled while a start of the turn holds it (the // coalescing assembly writes past the deferral) is an arrival too. internals.activeTurnTokens.set('assistant', -1); + internals.assembling.add('assistant'); internals.addMessage('Antra', [{ type: 'text', text: 'batched' }] as ContentBlock[], addressed('discord:dm:q'), { bypassDeferralFor: 'assistant' }); + internals.assembling.delete('assistant'); internals.activeTurnTokens.delete('assistant'); assert.deepEqual(open.arrivals.map((a) => a.channelId), ['discord:dm:q']); // A message still deferred when the turn's next start begins lands in diff --git a/test/silent-heartbeat-rows.test.ts b/test/silent-heartbeat-rows.test.ts index 72aa7bc9a..28be2aee3 100644 --- a/test/silent-heartbeat-rows.test.ts +++ b/test/silent-heartbeat-rows.test.ts @@ -607,6 +607,23 @@ describe('replies to a [Continue] turn', () => { } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); + it("stamps the reply of a [Continue] turn that a wake to read a render started, whose trigger the turn copies without the wake's mark", async () => { + const x = await make({ deliver: false }); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + // A render still queued (for another resident) keeps its wake to read + // it, which starts a turn with nothing new for this agent: [Continue]. + const internals = x.framework as any; + internals.deferredMessages.push({ id: 'render-x', seq: 1, participant: 'editor', forAgent: 'ghost', content: [{ type: 'text', text: 'elsewhere' }], heldAt: Date.now() }); + x.membrane.pushResponse(createMockResponse(signed('sig-c', 'continued reply'), 'end_turn')); + internals.pendingRequests.push({ agentName: 'assistant', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: 'render-x' }); + await x.framework.runUntilIdle(); + internals.deferredMessages.splice(0); + assert.equal(textOf(x.membrane.calls[1]!.messages.at(-1)!), '[Continue]'); + assert.equal(rowWith(x.framework, 'continued reply').metadata?.promptedBy?.text, '[Continue]', 'the reply carries the prompt it followed'); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + it('a row that is both a tick row and a reply to [Continue] gets both turns, separator first', () => { const tick = { eventId: 'hb-x', serverId: 'heartbeat', agentName: 'assistant', separator: SILENT_HEARTBEAT_SEPARATOR }; const prompt = { turn: 'p-1', text: '[Continue]' }; From babd0bf5696321bc5e942ae00bb281278349bf75 Mon Sep 17 00:00:00 2001 From: Crispin-2713 Date: Mon, 19 Oct 2026 09:44:29 -0400 Subject: [PATCH 2/2] fix(deferral): a provider hold tells a merge of several batches from an ordinary wake sameInferenceRequest compared a batch wake's coalescingSubject and a render wake's readsDeferredWrite, but a trigger merged from batches of several subjects carries no single subject, only coalescingBatchSubjects. Held by startAgentStream's provider wait, it matched an ordinary wake from the same millisecond, which was discarded; once every batch was retracted, the gate withdrew the turn, and the ordinary wake's cause was lost. The comparison now includes the merge's subjects. (coalescingBatch is set only beside a subject or a merge's subjects, so those two tell every batch-marked wake apart.) Tests: mcpl-coalescing's provider-hold test also holds a merge of two batches, and a merge of two other batches, apart from an ordinary wake from the same millisecond. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 12 ++++++++---- test/mcpl-coalescing.test.ts | 13 ++++++++++--- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index 6d52b0b40..974fd7586 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -3960,13 +3960,17 @@ export class AgentFramework { return { delayMs: ownMs, ownMs, stated: false }; } /** Whether two requests are one wake, for a provider hold's dedup: the - * same agent, reason, source, time and channel, and the same - * coalescingSubject and readsDeferredWrite, since a wake with either can be - * withdrawn or dropped on its own, and must not stand for one without. */ + * same agent, reason, source, time and channel, and the same coalescing + * marks: a batch wake's subject, a batch-only merge's subjects (it has no + * single subject), and the write a wake reads. A wake with any of them + * can be withdrawn or dropped on its own, and must not stand for one + * without. (coalescingBatch is set only beside one of the first two.) */ private sameInferenceRequest(a: InferenceRequest, b: InferenceRequest): boolean { + const batchSubjects = (r: InferenceRequest) => (r.coalescingBatchSubjects ?? []).join('\n'); 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.readsDeferredWrite === b.readsDeferredWrite; + a.coalescingSubject === b.coalescingSubject && batchSubjects(a) === batchSubjects(b) && + a.readsDeferredWrite === b.readsDeferredWrite; } private consultProviderHold(error: Error, agentName: string): { holdMs: number; reason: string } | undefined { if (!this.providerHoldHook) return undefined; diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index dc1c7121a..5e7144bbd 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1777,7 +1777,7 @@ test("a provider recovery whose released wakes start no turn ends and reopens au assert.equal(internals.providerAccelerationRecoveries.has('agent'), true, 'and the recovery stays for that turn to end'); }); -test("a provider hold keeps a wake to read a render, a batch wake and an ordinary wake apart, though they share a millisecond", async (t) => { +test("a provider hold keeps a wake to read a render, a batch wake, a batch-only merge and an ordinary wake apart, though they share a millisecond", async (t) => { const f = await fixture(); t.after(f.close); const internals = f.framework as unknown as TurnInternals; const timer = setTimeout(() => {}, 60_000); @@ -1785,10 +1785,17 @@ test("a provider hold keeps a wake to read a render, a batch wake and an ordinar internals.providerAccelerationCooldowns.set('agent', { startedAt: Date.now(), until: Date.now() + 60_000, timer, heldRequests: [], reason: 'held', failures: 1 }); const at = Date.now(); const wake = (extra: Partial = {}): Wake => ({ agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: at, ...extra }); - internals.pendingRequests.push(wake(), wake({ coalescingSubject: 'doc', coalescingBatch: true }), wake({ readsDeferredWrite: 'render-1' }), wake({ readsDeferredWrite: 'render-2' })); + // A trigger merged from batches of two subjects carries no single + // subject: only its subjects tell it from an ordinary wake, or from a + // merge of other batches. + const merged = wake({ coalescingBatch: true, coalescingBatchSubjects: ['doc', 'notes'] }); + internals.pendingRequests.push(merged, wake(), wake({ coalescingSubject: 'doc', coalescingBatch: true }), wake({ readsDeferredWrite: 'render-1' }), wake({ readsDeferredWrite: 'render-2' }), + wake({ coalescingBatch: true, coalescingBatchSubjects: ['doc', 'other'] })); await settles(internals.processInferenceRequests(), 'the held pass'); const held = internals.providerAccelerationCooldowns.get('agent')?.heldRequests ?? []; - assert.deepEqual(held.map((r) => [r.coalescingSubject, r.readsDeferredWrite]), [[undefined, undefined], ['doc', undefined], [undefined, 'render-1'], [undefined, 'render-2']], 'each is held: none stands for another, which a withdrawal or a read could take with it'); + assert.deepEqual(held.map((r) => [r.coalescingSubject, r.coalescingBatchSubjects?.join(','), r.readsDeferredWrite]), + [[undefined, 'doc,notes', undefined], [undefined, undefined, undefined], ['doc', undefined, undefined], [undefined, undefined, 'render-1'], [undefined, undefined, 'render-2'], [undefined, 'doc,other', undefined]], + 'each is held: none stands for another, which a withdrawal or a read could take with it'); }); test("an assembly that throws closes the window in which its writes bypass the turn's deferral", async (t) => {