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..974fd7586 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,18 @@ 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 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.timestamp === b.timestamp && a.channelId === b.channelId && + 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; @@ -10841,9 +10868,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 +12288,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 +12683,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 +12701,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 +13111,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 +13181,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 +13210,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 +13288,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 +15287,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 +15486,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 +15497,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 +15514,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 +15561,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 +15899,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 +20077,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 +20149,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 +20291,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 +20435,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 +20812,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..5e7144bbd 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,514 @@ 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, 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); + 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 }); + // 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.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) => { + 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]' };