From 0d1db865fdc79425955c63888a5d0f4f2793cf10 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 17:18:07 -0400 Subject: [PATCH 01/13] fix(deferral): an agent that leaves the framework takes what is queued for it out of the queue, logged as lost An ephemeral run's deregistration and a conversation fork's disposal are the only places an agent leaves the agent map. At each, what is still queued for that agent now leaves the deferred-write queue with it (dropDeferredFor), each entry logged by id as lost, and a persisted queue is rewritten without it. The last flush before the agent leaves can leave an entry queued: one whose store threw, which writeDrainedInOrder puts back, or, on a quiesced host, one that flush defers again. Before, a closing fork's entry stayed queued for the disposed fork, where no flush could store it. An ephemeral run's was dropped by the run's own turn-end flush, which runs after the run is deregistered, with addMessage's line that its delivery target isn't a registered agent, after the requeue's line saying it stays queued. What the store has already taken, waiting for its ack, isn't queued and stays for the ack. Tests: a run whose store refuses a write at its last flush, a run that ends while the host is quiesced, with the recovery file read after, and a closing fork whose store refuses a write at its closure turn's end. On main, the fork's entry stays queued, and the runs' entries reach addMessage after the run is deregistered. Fixes R2-1 of the round-2 review of #286. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 10 ++ src/framework.ts | 34 +++++- test/conversation-routing.test.ts | 47 ++++++++ test/ephemeral-settle.test.ts | 110 ++++++++++++++++++ 4 files changed, 198 insertions(+), 3 deletions(-) create mode 100644 changelog.d/deferred-writes-286-follow-up.fixed.md diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md new file mode 100644 index 000000000..39173377e --- /dev/null +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -0,0 +1,10 @@ +- What is still queued for an agent when it leaves the framework now leaves + the deferred-write queue with it, at that moment, and the host log names + each entry as lost, with the agent and where it left. An agent leaves at an + ephemeral run's end and at a conversation fork's disposal. The last flush + before it leaves can leave an entry there: one whose store threw, or, on a + quiesced host, one that flush deferred again. Before, a closing fork's entry + stayed queued for the disposed fork, where no flush could store it. An + ephemeral run's, once the run's stream had ended, was dropped by the run's + own turn-end flush, which runs after the run is deregistered, with a line + saying its delivery target isn't a registered agent. diff --git a/src/framework.ts b/src/framework.ts index 3455acb3b..5b4b30ca9 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -6523,13 +6523,14 @@ export class AgentFramework { // is deregistered, a flush finds no agent to take it and drops it. // The run's namespace stays in the store for investigation, so the // messages stay with it. A quiesced host defers them again instead, - // as stop() does, and with the run gone, the flush that meets them - // later drops them. + // as stop() does. What is still queued then, deferred again or put + // back because its store threw, leaves with the run, logged as lost. const queued = this.drainDeferredFor(agent.name); if (queued.length > 0) { this.readdDrained(queued, `end of ephemeral run ${agent.name}`, { bypassDeferral: !this.quiesced }); this.ackDeferredWrites(); } + this.dropDeferredFor(agent.name, `end of ephemeral run ${agent.name}`); // Exact-generation disposal owns these name-keyed entries. Remove them // before deregistration; the late driveStream finally will see the // generation mismatch and leave any future owner untouched. @@ -10134,9 +10135,13 @@ export class AgentFramework { * Remove a closed conversation fork from the framework: its closure turn * has finished, so the home mapping has served its purpose and keeping the * agent registered would just grow every agent scan and broadcast filter - * forever. The fork's context stays in Chronicle for investigation. + * forever. The fork's context stays in Chronicle for investigation. What + * is still queued for it leaves with it, logged as lost (dropDeferredFor). + * When the end of its closure turn disposes it, that turn's flush has + * already stored what it could. */ private disposeConversationAgent(agentName: string): void { + this.dropDeferredFor(agentName, `disposal of conversation fork ${agentName}`); this.closingConversationAgents.delete(agentName); const channelId = this.conversationAgentHomes.get(agentName); const agent = this.agents.get(agentName); @@ -16102,6 +16107,29 @@ export class AgentFramework { return drained; } + /** + * An agent that leaves the framework takes what is still queued for it out + * of the queue, each entry logged as lost. Once the agent is gone, a flush + * that meets such an entry drops it as addressed to no registered agent, so + * kept, the entry would only wait for that, written to the recovery file + * meanwhile when the queue is persisted. Called where an agent leaves: an + * ephemeral run's deregistration and a conversation fork's disposal. What + * its store has already taken, waiting for its ack, isn't queued, and stays + * for the ack. + */ + private dropDeferredFor(agentName: string, where: string): void { + const lost = this.deferredMessages.filter((msg) => (msg.forAgent ?? this.primaryAgentName) === agentName); + if (lost.length === 0) return; + this.deferredMessages = this.deferredMessages.filter((msg) => !lost.includes(msg)); + for (const msg of lost.sort(bySeq)) { + console.error( + `[deferral] ${where}: deferred write ${describeDeferredWrite(msg)} for ${agentName} is lost: ` + + `its agent has left the framework, so nothing will store it`, + ); + } + if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); + } + /** * Put drained entries back into the queue under their ids and seq: out of * the un-acked set, and persisted when the queue is. A memory-only queue diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index 50db20df3..a6f1553de 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -664,6 +664,53 @@ describe('Conversation routing', () => { await framework.stop(); }); + it("a write deferred for a fork that its store refuses at the closure turn's end leaves the queue with the fork, logged as lost", async () => { + const IDLE_TTL_MS = 60_000; + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }])); + const framework = await makeFramework({ idleTtlMs: IDLE_TTL_MS }); + framework.pushEvent(incomingEvent({ channelId: 'slack:D5', text: 'hi', channelType: 'im' })); + await framework.runUntilIdle(); + const g1 = framework.getAgent('conversation-slack-D5-g1')!; + assert.ok(g1, 'g1 fork should exist after spawn'); + // The fork's store refuses 'one more thing' each time it's written. + const cm = g1.getContextManager(); + const add = cm.addMessage.bind(cm); + (cm as unknown as { addMessage: typeof add }).addMessage = (...args) => { + if ((args[1] as Array<{ text?: string }>).some((b) => b.text === 'one more thing')) throw new Error('disk full'); + return add(...args); + }; + + forceExpiry(framework, 'slack:D5', IDLE_TTL_MS); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }])); + const internals = framework as unknown as { + addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string }): string; + deferredMessages: Array<{ forAgent?: string }>; + }; + const streamYielding = membrane.streamYielding.bind(membrane); + let deferred: string | undefined; + membrane.streamYielding = (request, options) => { + deferred = internals.addMessage('alice', [{ type: 'text', text: 'one more thing' }], undefined, + { forAgent: 'conversation-slack-D5-g1' }); + return streamYielding(request, options); + }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + framework.pushEvent({ type: 'external-message', source: 'test', content: 'tick', metadata: {} } as unknown as ProcessEvent); + await framework.runUntilIdle(); + } finally { console.error = error; } + + assert.equal(deferred, '', 'deferred: the closure turn was alive'); + assert.equal(framework.getAgent('conversation-slack-D5-g1'), null, 'disposed once its closure turn finished'); + assert.deepEqual(internals.deferredMessages.filter((m) => m.forAgent === 'conversation-slack-D5-g1'), [], + 'nothing is left queued for the fork'); + assert.ok(errors.some((e) => e.includes('disposal of conversation fork conversation-slack-D5-g1: deferred write ') + && e.includes('(alice: text) for conversation-slack-D5-g1 is lost: its agent has left the framework')), errors.join('\n')); + + await framework.stop(); + }); + it('a closure turn that restarts is disposed by the frame that ends it, after storing what was held for it', async () => { const IDLE_TTL_MS = 60_000; // One tool, for the closure turn's tool round. diff --git a/test/ephemeral-settle.test.ts b/test/ephemeral-settle.test.ts index 58bb2ddcc..bd2d16a27 100644 --- a/test/ephemeral-settle.test.ts +++ b/test/ephemeral-settle.test.ts @@ -473,6 +473,116 @@ describe('runEphemeralToCompletion settle signal', () => { } }); + it('a write queued for a run that its store refuses at the run\'s last flush leaves the queue with the run, logged as lost', async () => { + const { tempDir, storePath } = tempStorePath('ephemeral-lost-'); + const membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'call_1', name: 'test--echo', input: {} }, + ], 'tool_use')); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Done.' }])); + + const tools = new DuringCallToolModule(); + const framework = await AgentFramework.create({ + storePath, + membrane: membrane.asMembrane(), + agents: [], + modules: [tools], + syncIntervalMs: 0, + }); + const internals = framework as unknown as { + addMessage(participant: string, content: Array<{ type: 'text'; text: string }>, metadata?: undefined, opts?: { forAgent?: string }): string; + deferredMessages: Array<{ forAgent?: string }>; + }; + tools.duringCall = () => { + internals.addMessage('worker', [{ type: 'text', text: 'my own words' }], undefined, { forAgent: 'worker' }); + internals.addMessage('Antra', [{ type: 'text', text: 'look left' }], undefined, { forAgent: 'worker' }); + }; + const errors: string[] = []; + const error = console.error; + try { + const { agent, contextManager } = await createEphemeral(framework, 'worker'); + // The run's store refuses 'look left' each time it's written. + const add = contextManager.addMessage.bind(contextManager); + (contextManager as unknown as { addMessage: typeof add }).addMessage = (...args) => { + if ((args[1] as Array<{ text?: string }>).some((b) => b.text === 'look left')) throw new Error('disk full'); + return add(...args); + }; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const result = await runStartedEphemeral(framework, agent, contextManager); + console.error = error; + assert.strictEqual(result.speech, 'Done.'); + assert.strictEqual(framework.getAgent(agent.name), null, 'deregistered'); + const stored = contextManager.getAllMessages().map((m) => + `${m.participant}: ${m.content.map((b) => (b.type === 'text' ? (b as { text: string }).text : `<${b.type}>`)).join(' ')}`); + assert.deepStrictEqual(stored.slice(-2), ['worker: Done.', 'worker: my own words'], 'what could be stored was'); + assert.deepStrictEqual(internals.deferredMessages.filter((m) => m.forAgent === 'worker'), [], 'nothing is left queued for the run'); + assert.ok(errors.some((e) => e.includes('end of ephemeral run worker: deferred write ') + && e.includes('(Antra: text) for worker is lost: its agent has left the framework')), errors.join('\n')); + assert.ok(!errors.some((e) => e.includes('delivery target "worker" is not a registered agent')), + `no later flush meets it:\n${errors.join('\n')}`); + } finally { + console.error = error; + await stopAndRemove(framework, tempDir); + } + }); + + it('what is queued for a run that ends while the host is quiesced leaves the persisted queue with the run, logged as lost', async () => { + const { tempDir, storePath } = tempStorePath('ephemeral-quiesced-'); + const membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'call_1', name: 'test--echo', input: {} }, + ], 'tool_use')); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Done.' }])); + + const tools = new DuringCallToolModule(); + const framework = await AgentFramework.create({ + storePath, + membrane: membrane.asMembrane(), + agents: [], + modules: [tools], + syncIntervalMs: 0, + }); + const internals = framework as unknown as { + addMessage(participant: string, content: Array<{ type: 'text'; text: string }>, metadata?: undefined, opts?: { forAgent?: string }): string; + deferredMessages: Array<{ forAgent?: string }>; + }; + // While the run's tool call is pending, two messages for it are deferred: + // one named as the run itself, which waits for its turn's end, and one + // queued behind it. Then an operator quiesces the host, which waits for + // the run to end. + let quiescing: Promise | undefined; + tools.duringCall = () => { + internals.addMessage('worker', [{ type: 'text', text: 'my own words' }], undefined, { forAgent: 'worker' }); + internals.addMessage('Antra', [{ type: 'text', text: 'look left' }], undefined, { forAgent: 'worker' }); + quiescing = framework.quiesce({ reason: 'surgery' }); + }; + const errors: string[] = []; + const error = console.error; + try { + const { agent, contextManager } = await createEphemeral(framework, 'worker'); + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const result = await runStartedEphemeral(framework, agent, contextManager); + await quiescing; + console.error = error; + assert.strictEqual(result.speech, 'Done.'); + assert.strictEqual(framework.getAgent(agent.name), null, 'deregistered'); + const texts = contextManager.getAllMessages().flatMap((m) => m.content.map((b) => (b as { text?: string }).text)); + assert.ok(!texts.includes('my own words') && !texts.includes('look left'), 'neither was stored: the host is quiesced'); + assert.deepStrictEqual(internals.deferredMessages.filter((m) => m.forAgent === 'worker'), [], 'nothing is left queued for the run'); + const file = JSON.parse(readFileSync(join(storePath, 'recovery', 'deferred-writes.json'), 'utf8')) as { pending: Array<{ forAgent?: string }> }; + assert.deepStrictEqual(file.pending.filter((m) => m.forAgent === 'worker'), [], 'the recovery file keeps nothing for it'); + for (const named of ['(worker: text)', '(Antra: text)']) { + assert.ok(errors.some((e) => e.includes('end of ephemeral run worker: deferred write ') + && e.includes(`${named} for worker is lost: its agent has left the framework`)), `${named}:\n${errors.join('\n')}`); + } + assert.ok(!errors.some((e) => e.includes('delivery target "worker" is not a registered agent')), + `no later flush meets them:\n${errors.join('\n')}`); + } finally { + console.error = error; + await stopAndRemove(framework, tempDir); + } + }); + it('resolves through the endTurn tool-result path', async () => { const { tempDir, storePath } = tempStorePath('ephemeral-endturn-'); const membrane = new MockMembrane(); From fa5ccfa582ddf5a3e850c2db398acc190ca80272 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 17:18:10 -0400 Subject: [PATCH 02/13] fix(deferral): boot recovery numbers the restored queue 1..N, so a seq at the limit can't hold the counter there Boot kept each restored entry's seq, numbered only the entries without a valid one, after the largest, and set the counter to that. A valid seq at Number.MAX_SAFE_INTEGER left the counter there, and the next deferral's ++ gave 2^53, which ++ doesn't move past, so every later deferral shared it. A seq is read only for order (bySeq, boot's sort, and a re-add's fromQueue, which carries it back into the queue). So boot now restores the queue in the order it resolved to before (valid seqs ascending, then the rest in file order) and numbers it 1..N in that order, and the counter goes on from N. The numbers are given in memory. Boot rewrites the recovery file only for the reasons it did before: landed, malformed or repeated entries, and an entry without a valid seq, so it isn't named again. Every write of the queue writes it whole, and a boot before that numbers the same entries the same way. Test: a recovery file whose one entry has seq Number.MAX_SAFE_INTEGER isn't rewritten at boot, the two writes deferred after it are persisted with the restored entry as 1, 2, 3, the next boot names no seq, and resume stores them in order. On main the file holds MAX_SAFE_INTEGER, 2^53 and 2^53. Fixes R2-2 of the round-2 review of #286. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 8 +++ src/framework.ts | 63 ++++++++++--------- test/host-quiesce.test.ts | 53 ++++++++++++++++ 3 files changed, 93 insertions(+), 31 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index 39173377e..86bb6fedd 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -8,3 +8,11 @@ ephemeral run's, once the run's stream had ended, was dropped by the run's own turn-end flush, which runs after the run is deregistered, with a line saying its delivery target isn't a registered agent. +- Boot recovery numbers the restored deferred-write queue 1, 2, 3, … in the + order it restores it, and later deferrals are numbered on from there. + Before, it kept each entry's `seq` and numbered only the entries without a + valid one, after the largest, so the counter went on from the largest valid + `seq`: one at `Number.MAX_SAFE_INTEGER`, which boot accepts, gave every + later deferral the same `seq`, 2^53. A recovery file whose `seq`s are all + valid isn't rewritten at boot for the new numbers; the queue's next write + carries them. diff --git a/src/framework.ts b/src/framework.ts index 5b4b30ca9..dd18878c8 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1938,7 +1938,9 @@ export class AgentFramework { private unackedDeferredWrites: DeferredWrite[] = []; /** Monotonic order stamp for deferred writes: the durable queue is always * written, restored, drained and flushed in `seq` order, whatever the - * pending/un-acked split — a re-deferred entry keeps its place. */ + * pending/un-acked split — a re-deferred entry keeps its place. Boot + * recovery numbers the restored queue 1..N in that order, and the stamp + * goes on from N (restorePersistedDeferredWrites). */ private deferredSeq = 0; /** * Per target agent: the store's message count when the OLDEST currently @@ -5457,40 +5459,35 @@ export class AgentFramework { seen.add(id); kept.push({ ...m, id }); } - // A seq is a positive whole number no larger than - // Number.MAX_SAFE_INTEGER, the entry's place in arrival order. - // An entry without one (a file that predates seq, which is in write - // order already) or with any other value (damage or a hand edit: - // JSON's 1e400 reads as Infinity) is numbered after every valid seq, in - // file order, so the counter stays finite and the rest keep their order. + // A seq says only where its entry came in arrival order. The queue is + // restored in the order the file gives: entries whose seq is a positive + // whole number up to Number.MAX_SAFE_INTEGER by seq, then, in file + // order, any without one (a file that predates seq, which is in write + // order already) or with another value (damage or a hand edit: JSON's + // 1e400 reads as Infinity). Then it is numbered 1..N in that order, so + // the counter starts again from N at every boot and never passes the + // largest safe integer, wherever the file's seqs had got to. const validSeq = (seq: unknown): seq is number => Number.isSafeInteger(seq) && (seq as number) > 0; - let lastSeq = 0; - for (const m of kept) if (validSeq(m.seq)) lastSeq = Math.max(lastSeq, m.seq); - let renumbered = 0; - const candidates: DeferredWrite[] = kept - .map((m) => { - if (validSeq(m.seq)) return { ...m, seq: m.seq }; - renumbered++; - if (m.seq !== undefined) { - // A value that isn't a number is named by its type: it could be - // any size. - const seq = typeof m.seq === 'number' ? String(m.seq) : m.seq === null ? 'null' : `of type ${typeof m.seq}`; - console.error( - `[host-mode] a recovered deferred context write (${m.id}) has seq ${seq}, not a positive ` + - `whole number up to Number.MAX_SAFE_INTEGER; it is queued after the others`, - ); - } - return { ...m, seq: ++lastSeq }; - }) - .sort(bySeq); - this.deferredSeq = Math.max(this.deferredSeq, lastSeq); + const unnumbered = kept.filter((m) => !validSeq(m.seq)); + for (const m of unnumbered) { + if (m.seq === undefined) continue; + // A value that isn't a number is named by its type: it could be + // any size. + const seq = typeof m.seq === 'number' ? String(m.seq) : m.seq === null ? 'null' : `of type ${typeof m.seq}`; + console.error( + `[host-mode] a recovered deferred context write (${m.id}) has seq ${seq}, not a positive ` + + `whole number up to Number.MAX_SAFE_INTEGER; it is queued after the others`, + ); + } + const ordered = [...kept.filter((m) => validSeq(m.seq)).sort((a, b) => (a.seq as number) - (b.seq as number)), ...unnumbered]; // Exactly-once: a message that landed before its ack was written is // already in the store under its deferredWriteId — do not replay it. // The scan covers every slot the interrupted batch could occupy (from // the receipt's per-agent position), or the whole store when the // receipt predates the position field. - const landed = this.landedDeferredWriteIds(new Set(candidates.map((m) => m.id)), scanFrom); - const restored = candidates.filter((m) => !landed.has(m.id)); + const landed = this.landedDeferredWriteIds(new Set(ordered.map((m) => m.id)), scanFrom); + const restored: DeferredWrite[] = ordered.filter((m) => !landed.has(m.id)).map((m, i) => ({ ...m, seq: i + 1 })); + this.deferredSeq = Math.max(this.deferredSeq, restored.length); if (landed.size > 0) { console.error( `[host-mode] ${landed.size} deferred context write(s) had already landed before their ack ` + @@ -5501,8 +5498,12 @@ export class AgentFramework { this.deferredWritesPersisted = true; // Rewrite the durable queue without the already-landed entries so a // second interrupted boot does not re-scan them, without the malformed - // and repeated ones, and with the seqs given here. - if (landed.size > 0 || malformed.length > 0 || duplicates > 0 || renumbered > 0) this.persistDeferredWrites(); + // and repeated ones, and with a seq for each entry that came without a + // valid one, so the next boot doesn't name it again. The numbers given + // here aren't a reason to rewrite it: every write of the queue writes + // it whole, with them, and a boot before that numbers the same entries + // the same way. + if (landed.size > 0 || malformed.length > 0 || duplicates > 0 || unnumbered.length > 0) this.persistDeferredWrites(); return restored.length; } catch (err) { console.error('[host-mode] failed to restore persisted deferred context writes:', err); diff --git a/test/host-quiesce.test.ts b/test/host-quiesce.test.ts index 4400588f7..21368586f 100644 --- a/test/host-quiesce.test.ts +++ b/test/host-quiesce.test.ts @@ -1156,6 +1156,59 @@ test('boot recovery restores a recovery file that predates seq in file order, an } }); +test('boot recovery numbers the restored queue 1..N, so after a seq at Number.MAX_SAFE_INTEGER each deferral still gets a seq of its own', { timeout: 60_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'host-quiesce-seq-limit-')); + const storePath = join(dir, 'store'); + const config = () => ({ + storePath, + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + modules: [], + }); + const add = (fw: AgentFramework, text: string) => + (fw as unknown as { addMessage(p: string, c: unknown[]): unknown }).addMessage('user', [{ type: 'text', text }]); + let framework = await AgentFramework.create(config()); + const lines: string[] = []; + const error = console.error; + try { + await framework.quiesce({ reason: 'surgery' }); + const before = framework.getAgent('agent')!.getContextManager().getAllMessages().length; + add(framework, 'first'); + await framework.stop(); + + // A seq at the limit, as damage or a hand edit could leave it. It is a + // valid seq, so boot names nothing and has no reason to rewrite the file. + const file = join(storePath, 'recovery', 'deferred-writes.json'); + const doc = JSON.parse(readFileSync(file, 'utf8')) as { pending: Array> }; + assert.equal(doc.pending.length, 1); + doc.pending[0]!.seq = Number.MAX_SAFE_INTEGER; + const edited = JSON.stringify(doc); + writeFileSync(file, edited); + + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + framework = await AgentFramework.create(config()); + assert.equal(readFileSync(file, 'utf8'), edited, "boot doesn't rewrite a file whose seqs are all valid"); + add(framework, 'second'); + add(framework, 'third'); + const persisted = JSON.parse(readFileSync(file, 'utf8')) as { pending: Array<{ seq: unknown; content: Array<{ text: string }> }> }; + assert.deepEqual(persisted.pending.map((m) => [m.content[0]!.text, m.seq]), [['first', 1], ['second', 2], ['third', 3]], + 'numbered 1..N at boot, and the deferrals after it go on from N'); + await framework.stop(); + + framework = await AgentFramework.create(config()); + console.error = error; + assert.ok(!lines.some((line) => line.includes('has seq')), `every seq the next boot reads is valid:\n${lines.join('\n')}`); + await framework.resume(); + const texts = framework.getAgent('agent')!.getContextManager().getAllMessages().slice(before) + .map((m) => (m.content[0] as { text: string }).text); + assert.deepEqual(texts, ['first', 'second', 'third']); + } finally { + console.error = error; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + test('context writes deferred while quiesced survive a restart and land at resume', async () => { const dir = mkdtempSync(join(tmpdir(), 'host-quiesce-deferred-')); const storePath = join(dir, 'store'); From 20a4bc5ef7abb772ab08025453cbddbae8247025 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 17:18:15 -0400 Subject: [PATCH 03/13] docs(deferral): the comments on boot recovery, resume's flush and a queued entry's id say a flush is acknowledged once, as one batch flushDeferredWrites acknowledges its batch once, after the chronicle syncs, and a boot after a crash mid-flush skips each entry that already landed, by its deferredWriteId. Two comments still said acknowledgements are per message: create()'s note on recovering deferred writes and resume()'s on its flush. A queued entry's id, documented as the identity for per-message flush acknowledgement, said the same. Fixes R2-3 of the round-2 review of #286, and the id's comment beside it. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index dd18878c8..1e837f831 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1950,8 +1950,9 @@ export class AgentFramework { */ private deferredScanFrom = new Map(); private deferredMessages: Array<{ - /** Durable identity for per-message flush acknowledgement: a flush - * interrupted by a crash replays only the messages not yet acked. */ + /** Durable identity, stamped on the stored message as its + * deferredWriteId: a flush interrupted by a crash replays its batch at + * the next boot, which skips each entry that already landed by it. */ id: string; /** Order stamp (see deferredSeq). */ seq: number; @@ -2614,9 +2615,11 @@ export class AgentFramework { const hostMode = framework.readHostMode(); // Deferred writes are recovered REGARDLESS of the mode flag: a crash // between a resume's flag clear and the end of its flush must not - // strand accepted messages. Per-message acks mean only the remainder - // replays. Serving boot with a remainder → flush it right now (no turn - // is alive yet, so this is a safe boundary). + // strand accepted messages. A flush acknowledges its batch once, after + // the chronicle syncs, and boot skips each entry that already landed + // (by its deferredWriteId), so only the remainder replays (see + // flushDeferredWrites). Serving boot with a remainder → flush it right + // now (no turn is alive yet, so this is a safe boundary). const restoredWrites = framework.restorePersistedDeferredWrites(); if (!hostMode?.quiesced && restoredWrites > 0) { console.error( @@ -5113,8 +5116,9 @@ export class AgentFramework { try { // Flush context writes deferred by the quiesce window (module events, // api message.send) — with no turn alive there is no other flush - // point. Per-message try/catch and per-message durable ack (see - // flushDeferredWrites); nothing here may reach the plane reopen below. + // point. Each write is tried on its own, and the batch is acknowledged + // once, after the chronicle syncs (see flushDeferredWrites); nothing + // here may reach the plane reopen below. await this.flushDeferredWrites('resume'); this.persistHostMode(null); } finally { From f9fd41db7c08d870228c01e944da6dc38564db45 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 17:18:15 -0400 Subject: [PATCH 04/13] fix(deferral): a sharded body injected at a tool boundary is carried as later compiles render it A deferred message that the strategy stores as more than one message, a sharded body, was injected at a tool boundary as given, while later compiles render the store's shards joined. So a body of more than one text block, or with a non-text block after a text block, reached the stream in a different shape from later requests. writeDrainedInOrder already returned what landed, joined (asCompiled), for a write that threw after storing. It now does the same for a write that succeeds, when the store grew by more than one message: it reads the write's copies back and returns them joined. The store is read back only then, so a write of one message costs one count read more than before. Test: four bodies the strategy shards, stored whole at a tool boundary, each with a one-message write after it: what the boundary injects is what the next compile renders, and the store is read back for each body alone. On main three of the four differ: two text blocks, text then an image, and text, an image and text. It runs on context-manager 0.14.1 and 0.15. Takes up the suggestion in the round-2 review of #286. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 5 ++ src/framework.ts | 36 ++++++---- test/deferred-window-only.test.ts | 70 +++++++++++++++++++ 3 files changed, 99 insertions(+), 12 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index 86bb6fedd..eb54de324 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -16,3 +16,8 @@ later deferral the same `seq`, 2^53. A recovery file whose `seq`s are all valid isn't rewritten at boot for the new numbers; the queue's next write carries them. +- At a tool boundary, a deferred message that the strategy stores as more than + one message (a sharded body) is now injected as later compiles render it, + its shards joined. Before, it was injected as given, so a body of more than + one text block, or with a non-text block after a text block, reached the + stream in a different shape from later requests. diff --git a/src/framework.ts b/src/framework.ts index 1e837f831..131d410f3 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -506,10 +506,10 @@ const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; /** A drained entry as writeDrainedInOrder wrote it: the id its store gave it * (empty when addMessage deferred it again; a sharded body's first shard) - * and what the store holds of it. For an entry whose write threw after - * storing it, found in the store and counted as written, the content is - * what landed, as later compiles render it, which for a sharded body can be - * its first shards alone. */ + * and what the store holds of it, as later compiles render it: for a + * sharded body, its shards joined. For an entry whose write threw after + * storing it, found in the store and counted as written, that is what + * landed, which for a sharded body can be its first shards alone. */ interface DrainedWrite { id: MessageId; content: ContentBlock[]; @@ -9373,10 +9373,11 @@ export class AgentFramework { const deferred = this.drainDeferredFor(agent.name, (msg) => !isInjectableMidTurn(msg, agent.name)); // Stored in queue order, cut at the first that can't be stored: it // and what follows wait for the turn's end, and only what was - // stored is injected (the round's results go on either way). A - // write that threw after storing counts as stored, and what landed - // of it is what's injected, so the request carries what the store - // holds (writeDrainedInOrder). + // stored is injected (the round's results go on either way). What + // is injected is what the store holds, as later compiles render + // it: a sharded body's shards joined, and of a write that threw + // after storing, which counts as stored, what landed + // (writeDrainedInOrder). const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) => agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); for (const [msg, { id, content }] of stored) { @@ -16190,9 +16191,12 @@ export class AgentFramework { * * Returns what was written (stored, or deferred again by addMessage, whose * id is then empty), in queue order: for one target, a prefix of its - * entries. `onLanded` hears an entry whose write threw after storing it, - * with the id it landed under, as soon as it's found: before the writes - * after it, so whatever it records keeps their order. + * entries. Each comes with its content as later compiles render it: as + * given, unless the store took it as more than one message (a sharded + * body), whose copies are joined (asCompiled), or its write threw after + * storing it (above). `onLanded` hears an entry whose write threw after + * storing it, with the id it landed under, as soon as it's found: before + * the writes after it, so whatever it records keeps their order. */ private writeDrainedInOrder( entries: DeferredWrite[], @@ -16212,7 +16216,15 @@ export class AgentFramework { const agent = this.deliveryAgent(msg.forAgent); const before = agent ? this.storedCountOf(agent) : undefined; try { - written.set(msg, { id: write(msg), content: msg.content }); + const id = write(msg); + // A body the store took as more than one message (a sharded body) is + // returned as later compiles render it: its copies, joined + // (asCompiled). The store is read for them only when it grew by more + // than one message. + const copies = agent && before !== undefined && (this.storedCountOf(agent) ?? before) - before > 1 + ? this.storedCopiesOf(agent, before, msg.id) + : []; + written.set(msg, { id, content: copies.length > 1 ? asCompiled(copies) : msg.content }); } catch (error) { const landed = agent && before !== undefined ? this.storedCopiesOf(agent, before, msg.id) : []; if (landed.length > 0) { diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 44145ca24..cb85086a4 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -25,6 +25,7 @@ import { tmpdir } from 'node:os'; import { existsSync, mkdtempSync, readFileSync, rmSync } from 'node:fs'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; +import { isDeepStrictEqual } from 'node:util'; import type { Module, ModuleContext, @@ -656,6 +657,75 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { } }); + it('at a tool boundary, a sharded body stored whole is carried as the next compile renders it', async () => { + // A body the strategy shards on ingest is stored as several messages, + // which a compile joins. A body of one text block reads the same as + // given and joined; in the other cases the two differ, so the boundary + // has to carry the join for the stream to carry what later requests do. + const long = Array.from({ length: 60 }, (_, i) => `Sentence ${i} of a long message from Antra.`).join(' '); + const ofLong = (messages: Array<{ content: unknown }>) => + messages.filter((m) => JSON.stringify(m.content).includes('Sentence 0 ')); + const PNG = 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=='; + const image = { type: 'image', source: { type: 'base64', data: PNG, mediaType: 'image/png' } } as ContentBlock; + const cases: Array<{ name: string; content: ContentBlock[] }> = [ + { name: 'one long text block', content: text(long) }, + { name: 'a short text, then the long text', content: [...text('Intro. '), ...text(long)] }, + { name: 'the long text, then an image', content: [...text(long), image] }, + { name: 'a short text, an image, then the long text', content: [...text('Intro. '), image, ...text(long)] }, + ]; + const readBacks: number[] = []; + // Every case is run, and those whose injection differs from the next + // compile are named together. + const differ: string[] = []; + for (const [i, { name, content }] of cases.entries()) { + membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Moving.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: {} }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse(text('Done.'))); + const framework = await createFramework({ + storePath: join(tempDir, `whole-${i}.chronicle`), + membrane: membrane.asMembrane(), + agents: [{ + name: 'assistant', model: 'test-model', systemPrompt: 'You are a robot pilot.', + strategy: new AutobiographicalStrategy({ + adaptiveResolution: true, foldingStrategy: 'kv-stable', recentWindowTokens: 30_000, kvStableReachTokens: 8_000, + targetChunkTokens: 50, + }), + } as never], + }); + const cm = framework.getAgent('assistant')!.getContextManager(); + // The store is read back for a write only when it took more than one + // message: here the body, and not the short message after it. + const internals = framework as unknown as { storedCopiesOf: (...args: unknown[]) => unknown[] }; + const storedCopiesOf = internals.storedCopiesOf.bind(framework); + let readBack = 0; + internals.storedCopiesOf = (...args) => { readBack++; return storedCopiesOf(...args); }; + module.interjections = [{ participant: 'Antra', content }, { participant: 'Antra', content: text('look right') }]; + try { + framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); + await framework.runUntilIdle(); + const landed = cm.getAllMessages().filter((m) => (m as { bodyGroupId?: string }).bodyGroupId); + assert.ok(landed.length > 1, `${name}: stored as ${landed.length} shards`); + readBacks.push(readBack); + const injected = ofLong(membrane.lastStream!.receivedToolResultOptions[0]?.injectedMessages ?? []); + // The next turn's request is compiled from the store. + membrane.pushResponse(createMockResponse(text('Okay.'))); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); + await framework.runUntilIdle(); + const next = ofLong(membrane.calls.at(-1)!.messages); + assert.equal(injected.length, 1, `${name}: one message injected`); + assert.equal(next.length, 1, `${name}: one message in the next request`); + if (!isDeepStrictEqual(injected[0]!.content, next[0]!.content)) differ.push(name); + } finally { + await framework.stop(); + } + } + assert.deepEqual(differ, [], 'the boundary carried what the next compile renders'); + assert.deepEqual(readBacks, [1, 1, 1, 1], 'the store is read back for each body, and not for the short message after it'); + }); + it("a write that throws after it's stored, re-added from the queue, is weighed for its agent's open turn as a stored write is", async () => { const framework = await createFramework(); throwOnceAfterStoring(framework, 'from beta'); From f1202da582b9dec73a0ec8a9111e6ebb9b1ee02e Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 17:52:48 -0400 Subject: [PATCH 05/13] fix(deferral): the fallback reaper stores what is queued for a closing fork before it disposes it The fallback reaper disposes a closing fork that has no stream and no request left, as when its closure inference never ran. Unlike the end of a closure turn and the end of an ephemeral run, nothing stored what was still queued for the fork first. On main it stayed queued for a fork that no longer existed, and with dropDeferredFor it would be logged as lost although the fork's store could take it. storeDeferredFor is the ephemeral run's store-before-leaving step, moved into a method: drain the agent's queue, re-add it past the deferral (deferred again on a quiesced host) and acknowledge. The ephemeral run's finally and the reaper both call it before the agent leaves, and dropDeferredFor takes out only what it couldn't store. Test: a closing fork whose closure request is dropped before it runs, with a write deferred for it. The reaper's sweep stores the write in the fork before disposing it. On main it isn't stored. Found in review of this change. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 6 +++ src/framework.ts | 41 +++++++++++------ test/conversation-routing.test.ts | 46 +++++++++++++++++++ 3 files changed, 80 insertions(+), 13 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index eb54de324..c1fda84a9 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -8,6 +8,12 @@ ephemeral run's, once the run's stream had ended, was dropped by the run's own turn-end flush, which runs after the run is deregistered, with a line saying its delivery target isn't a registered agent. +- A closing conversation fork that the fallback reaper disposes, with no + stream and no request left for it, as when its closure turn never ran, first + has what is still queued for it stored in its own store, past its deferral, + as the end of a closure turn and the end of an ephemeral run already do. On + a quiesced host it's deferred again and leaves with the fork, as above. + Before, nothing stored it, and it stayed queued for the disposed fork. - Boot recovery numbers the restored deferred-write queue 1, 2, 3, … in the order it restores it, and later deferrals are numbered on from there. Before, it kept each entry's `seq` and numbered only the entries without a diff --git a/src/framework.ts b/src/framework.ts index 131d410f3..f9d632be7 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -6524,17 +6524,9 @@ export class AgentFramework { if (this.ephemeralRuns.get(agent.name) === run) this.ephemeralRuns.delete(agent.name); if (this.agents.get(agent.name) === agent) { // What is still queued for the run, such as a message held for its - // turn's end, goes to its store now, past its deferral: once the run - // is deregistered, a flush finds no agent to take it and drops it. - // The run's namespace stays in the store for investigation, so the - // messages stay with it. A quiesced host defers them again instead, - // as stop() does. What is still queued then, deferred again or put - // back because its store threw, leaves with the run, logged as lost. - const queued = this.drainDeferredFor(agent.name); - if (queued.length > 0) { - this.readdDrained(queued, `end of ephemeral run ${agent.name}`, { bypassDeferral: !this.quiesced }); - this.ackDeferredWrites(); - } + // turn's end, goes to its store now (storeDeferredFor), and what that + // can't store leaves with the run, logged as lost (dropDeferredFor). + this.storeDeferredFor(agent.name, `end of ephemeral run ${agent.name}`); this.dropDeferredFor(agent.name, `end of ephemeral run ${agent.name}`); // Exact-generation disposal owns these name-keyed entries. Remove them // before deregistration; the late driveStream finally will see the @@ -10143,8 +10135,9 @@ export class AgentFramework { * agent registered would just grow every agent scan and broadcast filter * forever. The fork's context stays in Chronicle for investigation. What * is still queued for it leaves with it, logged as lost (dropDeferredFor). - * When the end of its closure turn disposes it, that turn's flush has - * already stored what it could. + * Each caller has first stored what it could: the end of the fork's + * closure turn with that turn's flush, and the fallback reaper with + * storeDeferredFor. */ private disposeConversationAgent(agentName: string): void { this.dropDeferredFor(agentName, `disposal of conversation fork ${agentName}`); @@ -10252,11 +10245,15 @@ export class AgentFramework { // Fallback reaper: a closing fork whose closure inference never ran // (request dropped as stale, inference policy veto) would otherwise stay // registered forever. No active stream + no pending request = done. + // What is still queued for it goes to its store first, as at the end of + // a closure turn (storeDeferredFor), and what can't be stored leaves + // with it (disposeConversationAgent). for (const agentName of [...this.closingConversationAgents]) { if ( !this.activeStreams.has(agentName) && !this.pendingRequests.some((r) => r.agentName === agentName) ) { + this.storeDeferredFor(agentName, `disposal of conversation fork ${agentName}`); this.disposeConversationAgent(agentName); } } @@ -16113,6 +16110,24 @@ export class AgentFramework { return drained; } + /** + * Before an agent leaves the framework, what is still queued for it goes to + * its store, past its deferral: once the agent is gone, a flush finds no + * agent to take it and drops it. The agent's namespace stays in the store + * for investigation, so the messages stay with it. A quiesced host defers + * each entry again instead, as stop() does. Called at an ephemeral run's + * deregistration and by the fallback reaper before it disposes a closing + * fork; the end of a closure turn has its own flush. What this can't store, + * put back because its store threw or deferred again, dropDeferredFor takes + * out after it. + */ + private storeDeferredFor(agentName: string, where: string): void { + const queued = this.drainDeferredFor(agentName); + if (queued.length === 0) return; + this.readdDrained(queued, where, { bypassDeferral: !this.quiesced }); + this.ackDeferredWrites(); + } + /** * An agent that leaves the framework takes what is still queued for it out * of the queue, each entry logged as lost. Once the agent is gone, a flush diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index a6f1553de..b32ba1b05 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -711,6 +711,52 @@ describe('Conversation routing', () => { await framework.stop(); }); + it('a write queued for a closing fork whose closure turn never ran is stored in the fork before the reaper disposes it', async () => { + const IDLE_TTL_MS = 60_000; + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }])); + const framework = await makeFramework({ idleTtlMs: IDLE_TTL_MS }); + framework.pushEvent(incomingEvent({ channelId: 'slack:D7', text: 'hi', channelType: 'im' })); + await framework.runUntilIdle(); + const fork = 'conversation-slack-D7-g1'; + const g1 = framework.getAgent(fork)!; + assert.ok(g1, 'g1 fork should exist after spawn'); + const internals = framework as unknown as { + addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string }): string; + deferredMessages: Array<{ forAgent?: string }>; + pendingRequests: Array<{ agentName: string }>; + activeTurnTokens: Map; + lastConversationSweep: number; + sweepExpiredConversations(): void; + }; + // The sweep closes the fork and queues its closure turn, and the request + // is then dropped before it runs, as a stale one is. + forceExpiry(framework, 'slack:D7', IDLE_TTL_MS); + internals.sweepExpiredConversations(); + assert.equal(internals.pendingRequests.filter((r) => r.agentName === fork).length, 1, 'the closure turn is queued'); + internals.pendingRequests.splice(0, internals.pendingRequests.length, + ...internals.pendingRequests.filter((r) => r.agentName !== fork)); + // A write for the fork that was deferred meanwhile. + internals.activeTurnTokens.set(fork, 1); + assert.equal(internals.addMessage('alice', [{ type: 'text', text: 'one more thing' }], undefined, { forAgent: fork }), '', 'deferred'); + internals.activeTurnTokens.delete(fork); + + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + internals.lastConversationSweep = 0; + internals.sweepExpiredConversations(); + } finally { console.error = error; } + + assert.equal(framework.getAgent(fork), null, 'the reaper disposed it: its closure turn never ran'); + const stored = g1.getContextManager().getAllMessages().map((m) => JSON.stringify(m.content)); + assert.ok(stored[stored.length - 1]!.includes('one more thing'), `stored in the fork before it was disposed:\n${stored.join('\n')}`); + assert.deepEqual(internals.deferredMessages.filter((m) => m.forAgent === fork), [], 'nothing is left queued for the fork'); + assert.ok(!errors.some((e) => e.includes('is lost')), errors.join('\n')); + + await framework.stop(); + }); + it('a closure turn that restarts is disposed by the frame that ends it, after storing what was held for it', async () => { const IDLE_TTL_MS = 60_000; // One tool, for the closure turn's tool round. From 2f3b72531f3a9ff7e660635b52b7230d2130ec71 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 17:52:51 -0400 Subject: [PATCH 06/13] fix(deferral): a log line and five comments say only what the code does - The success path's read-back calls storedCopiesOf, whose failure line said the write threw. It now says the store couldn't be read back for what the write stored. - DrainedWrite's and writeDrainedInOrder's docs, and the tool boundary's comment, no longer say a one-message write's content is as later compiles render it. It is carried as given; only a sharded body's join, or what a write that threw left, is claimed as a compile renders it. - Boot recovery's comment no longer says the counter never passes the largest safe integer: only its value at boot is bounded. - stop()'s comment no longer says a disposed fork's entries wait for a boot or resume flush. They leave with the fork, so what can still be left is a write queued after its agent left, or one restored at a boot that doesn't register its agent. Found in review of this change. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 46 ++++++++++++++++++++++++---------------------- 1 file changed, 24 insertions(+), 22 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index f9d632be7..b6bbf6b5f 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -506,10 +506,11 @@ const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; /** A drained entry as writeDrainedInOrder wrote it: the id its store gave it * (empty when addMessage deferred it again; a sharded body's first shard) - * and what the store holds of it, as later compiles render it: for a - * sharded body, its shards joined. For an entry whose write threw after - * storing it, found in the store and counted as written, that is what - * landed, which for a sharded body can be its first shards alone. */ + * and the content to carry for it: as given, or for a sharded body, its + * shards joined as later compiles render them. For an entry whose write + * threw after storing it, found in the store and counted as written, it is + * what landed, joined the same way, which for a sharded body can be its + * first shards alone. */ interface DrainedWrite { id: MessageId; content: ContentBlock[]; @@ -3080,10 +3081,12 @@ export class AgentFramework { // A quiesced host defers each entry again instead, into the persisted // queue, for boot recovery. Past the deferral, a write that throws stays // queued with what follows it for its target, and a memory-only queue - // still loses it at exit. An entry for a target no longer registered (a - // disposed fork) is left: dropped with a memory-only queue at exit, and - // logged and dropped from a persisted one by the boot or resume flush - // that meets it. + // still loses it at exit. What is queued for an agent leaves with it + // (dropDeferredFor), so an entry for a target no longer registered is one + // queued after its agent left, or restored at a boot that doesn't + // register its agent. It is left: dropped with a memory-only queue at + // exit, and logged and dropped from a persisted one by the boot or resume + // flush that meets it. if (this.deferredMessages.length > 0) { const entries = [...this.agents.keys()].flatMap((name) => this.drainDeferredFor(name)); this.readdDrained(entries, 'stop', { bypassDeferral: !this.quiesced }); @@ -5469,8 +5472,8 @@ export class AgentFramework { // order, any without one (a file that predates seq, which is in write // order already) or with another value (damage or a hand edit: JSON's // 1e400 reads as Infinity). Then it is numbered 1..N in that order, so - // the counter starts again from N at every boot and never passes the - // largest safe integer, wherever the file's seqs had got to. + // the counter starts again from N at every boot, wherever the file's + // seqs had got to. const validSeq = (seq: unknown): seq is number => Number.isSafeInteger(seq) && (seq as number) > 0; const unnumbered = kept.filter((m) => !validSeq(m.seq)); for (const m of unnumbered) { @@ -9365,11 +9368,10 @@ export class AgentFramework { const deferred = this.drainDeferredFor(agent.name, (msg) => !isInjectableMidTurn(msg, agent.name)); // Stored in queue order, cut at the first that can't be stored: it // and what follows wait for the turn's end, and only what was - // stored is injected (the round's results go on either way). What - // is injected is what the store holds, as later compiles render - // it: a sharded body's shards joined, and of a write that threw - // after storing, which counts as stored, what landed - // (writeDrainedInOrder). + // stored is injected (the round's results go on either way). A + // sharded body is injected as later compiles render it, its shards + // joined, and a write that threw after storing, which counts as + // stored, as what landed of it (writeDrainedInOrder). const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) => agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); for (const [msg, { id, content }] of stored) { @@ -16206,12 +16208,12 @@ export class AgentFramework { * * Returns what was written (stored, or deferred again by addMessage, whose * id is then empty), in queue order: for one target, a prefix of its - * entries. Each comes with its content as later compiles render it: as - * given, unless the store took it as more than one message (a sharded - * body), whose copies are joined (asCompiled), or its write threw after - * storing it (above). `onLanded` hears an entry whose write threw after - * storing it, with the id it landed under, as soon as it's found: before - * the writes after it, so whatever it records keeps their order. + * entries. Each comes with its content: as given, unless the store took it + * as more than one message (a sharded body), whose copies are returned + * joined, as later compiles render them (asCompiled), or its write threw + * after storing it (above). `onLanded` hears an entry whose write threw + * after storing it, with the id it landed under, as soon as it's found: + * before the writes after it, so whatever it records keeps their order. */ private writeDrainedInOrder( entries: DeferredWrite[], @@ -16277,7 +16279,7 @@ export class AgentFramework { .filter((m) => m.metadata?.deferredWriteId === deferredWriteId) .map((m) => ({ id: m.id, content: m.content, ...(m.bodyGroupId !== undefined ? { bodyGroupId: m.bodyGroupId } : {}) })); } catch (error) { - console.error(`[deferral] could not read ${agent.name}'s store for a deferred write that threw:`, error); + console.error(`[deferral] could not read back what deferred write ${deferredWriteId} stored in ${agent.name}'s store:`, error); return []; } } From f524b59878b47978a09f365f81cd965275d94959 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 17:52:51 -0400 Subject: [PATCH 07/13] test(deferral): the persisted queue's file and the fork's refusal are observed The run test whose store refuses a write now runs with a persisted queue, as after a boot that restored one, and reads the recovery file after: nothing is left in it for the run. Only the quiesced test read the file before, so removing the deferredWritesPersisted half of dropDeferredFor's rewrite condition passed every test. The fork test now checks that the closure turn's flush met the refusal (its "could not be stored" line for the write) before the disposal's lost line, so it can't pass with a flush that never tried the store. Found in review of this change. Co-Authored-By: Claude Opus 5.5 --- test/conversation-routing.test.ts | 3 +++ test/ephemeral-settle.test.ts | 6 ++++++ 2 files changed, 9 insertions(+) diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index b32ba1b05..8ff308817 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -705,6 +705,9 @@ describe('Conversation routing', () => { assert.equal(framework.getAgent('conversation-slack-D5-g1'), null, 'disposed once its closure turn finished'); assert.deepEqual(internals.deferredMessages.filter((m) => m.forAgent === 'conversation-slack-D5-g1'), [], 'nothing is left queued for the fork'); + assert.ok(errors.some((e) => e.includes('turn-end flush: deferred write ') + && e.includes('(alice: text) for conversation-slack-D5-g1 could not be stored; it stays queued')), + `the closure turn's flush met the refusal:\n${errors.join('\n')}`); assert.ok(errors.some((e) => e.includes('disposal of conversation fork conversation-slack-D5-g1: deferred write ') && e.includes('(alice: text) for conversation-slack-D5-g1 is lost: its agent has left the framework')), errors.join('\n')); diff --git a/test/ephemeral-settle.test.ts b/test/ephemeral-settle.test.ts index bd2d16a27..370f9af35 100644 --- a/test/ephemeral-settle.test.ts +++ b/test/ephemeral-settle.test.ts @@ -492,7 +492,11 @@ describe('runEphemeralToCompletion settle signal', () => { const internals = framework as unknown as { addMessage(participant: string, content: Array<{ type: 'text'; text: string }>, metadata?: undefined, opts?: { forAgent?: string }): string; deferredMessages: Array<{ forAgent?: string }>; + deferredWritesPersisted: boolean; }; + // The queue is persisted, as after a boot that restored one, so its + // recovery file is written as it changes. + internals.deferredWritesPersisted = true; tools.duringCall = () => { internals.addMessage('worker', [{ type: 'text', text: 'my own words' }], undefined, { forAgent: 'worker' }); internals.addMessage('Antra', [{ type: 'text', text: 'look left' }], undefined, { forAgent: 'worker' }); @@ -516,6 +520,8 @@ describe('runEphemeralToCompletion settle signal', () => { `${m.participant}: ${m.content.map((b) => (b.type === 'text' ? (b as { text: string }).text : `<${b.type}>`)).join(' ')}`); assert.deepStrictEqual(stored.slice(-2), ['worker: Done.', 'worker: my own words'], 'what could be stored was'); assert.deepStrictEqual(internals.deferredMessages.filter((m) => m.forAgent === 'worker'), [], 'nothing is left queued for the run'); + const file = JSON.parse(readFileSync(join(storePath, 'recovery', 'deferred-writes.json'), 'utf8')) as { pending: Array<{ forAgent?: string }> }; + assert.deepStrictEqual(file.pending.filter((m) => m.forAgent === 'worker'), [], 'the recovery file keeps nothing for it'); assert.ok(errors.some((e) => e.includes('end of ephemeral run worker: deferred write ') && e.includes('(Antra: text) for worker is lost: its agent has left the framework')), errors.join('\n')); assert.ok(!errors.some((e) => e.includes('delivery target "worker" is not a registered agent')), From c5a3730c304383c5e870b3454c83d8749ec0a602 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 18:23:58 -0400 Subject: [PATCH 08/13] fix(deferral): the reaper disposes a closing fork only when what is queued for it can be stored now The fallback reaper disposed a closing fork once it had no active stream and no pending request, and (since f1202da58) first stored what was queued for it past its deferral. Two holds got through that guard: - A turn token with no stream. A surgery reserves every registered agent's token while it switches the store, and a puppet call holds its agent's across the tool's await. The sweep runs from the event loop regardless, so the reaper would write a closing fork's queue into the store mid-surgery, or ahead of the puppet's tool pair. - A quiesced host. The store-first deferred the writes again, and the disposal then logged them as lost, though resume() would store them. The guard now also waits for the fork's turn token and for a quiesced host. Each holder stores the fork's queue as it lets go (the puppet's finally, the surgery's release, resume()), and a later sweep disposes the fork. A token nobody clears would keep the fork registered, as it already wedges the agent. Tests: a closing fork whose closure request is dropped, with a write deferred for it. Under a held token the sweep leaves the fork registered and the write queued, and once the token is gone it stores the write and disposes the fork. On a quiesced host the sweep leaves it registered, resume() stores the write, and the next sweep disposes it, with nothing logged as lost. The three reaper tests share their setup. Found in review of this change. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 18 +++-- src/framework.ts | 16 +++-- test/conversation-routing.test.ts | 69 ++++++++++++++++--- 3 files changed, 84 insertions(+), 19 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index c1fda84a9..8a9d3bbc5 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -8,12 +8,18 @@ ephemeral run's, once the run's stream had ended, was dropped by the run's own turn-end flush, which runs after the run is deregistered, with a line saying its delivery target isn't a registered agent. -- A closing conversation fork that the fallback reaper disposes, with no - stream and no request left for it, as when its closure turn never ran, first - has what is still queued for it stored in its own store, past its deferral, - as the end of a closure turn and the end of an ephemeral run already do. On - a quiesced host it's deferred again and leaves with the fork, as above. - Before, nothing stored it, and it stayed queued for the disposed fork. +- The fallback reaper, which disposes a closing conversation fork with no + stream and no request left for it, as when its closure turn never ran, now + first stores what is still queued for the fork in the fork's own store, past + its deferral, as the end of a closure turn and the end of an ephemeral run + do. It waits to dispose the fork until that can be done: not while a turn + token is held for it, as a puppet call or a surgery holds one, nor while the + host is quiesced. The puppet call's end, the surgery's release and + `resume()` each store the fork's queue, and a later sweep disposes it. Only + what its store throws on then leaves with the fork, as above. Before, the + reaper disposed such a fork while a token was held or the host was quiesced + too, nothing stored what was queued for it, and that stayed queued for the + disposed fork. - Boot recovery numbers the restored deferred-write queue 1, 2, 3, … in the order it restores it, and later deferrals are numbered on from there. Before, it kept each entry's `seq` and numbered only the entries without a diff --git a/src/framework.ts b/src/framework.ts index b6bbf6b5f..38becdbfe 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -10246,14 +10246,20 @@ export class AgentFramework { // Fallback reaper: a closing fork whose closure inference never ran // (request dropped as stale, inference policy veto) would otherwise stay - // registered forever. No active stream + no pending request = done. - // What is still queued for it goes to its store first, as at the end of - // a closure turn (storeDeferredFor), and what can't be stored leaves - // with it (disposeConversationAgent). + // registered forever. No active stream + no pending request = done. It + // is disposed once what is still queued for it can be stored now, so + // not while a turn token is held for it (a puppet call's, or a surgery's + // reservation) or the host is quiesced: each of those flushes the fork's + // queue as it lets go (the puppet's finally, the surgery's release, + // resume()), and a later sweep disposes the fork. What is queued goes to + // its store first (storeDeferredFor), and what can't be, put back + // because its store threw, leaves with the fork (disposeConversationAgent). for (const agentName of [...this.closingConversationAgents]) { if ( !this.activeStreams.has(agentName) && - !this.pendingRequests.some((r) => r.agentName === agentName) + !this.pendingRequests.some((r) => r.agentName === agentName) && + !this.activeTurnTokens.has(agentName) && + !this.quiesced ) { this.storeDeferredFor(agentName, `disposal of conversation fork ${agentName}`); this.disposeConversationAgent(agentName); diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index 8ff308817..ccb928524 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -714,15 +714,19 @@ describe('Conversation routing', () => { await framework.stop(); }); - it('a write queued for a closing fork whose closure turn never ran is stored in the fork before the reaper disposes it', async () => { + /** A fork spawned on `channelId` and closed by the sweep, whose closure + * request is then dropped before it runs, as a stale one is, with a write + * for it deferred meanwhile. Nothing holds it when this returns: no + * stream, no request, no turn token, and the host isn't quiesced. */ + async function closingForkWithAQueuedWrite(channelId: string) { const IDLE_TTL_MS = 60_000; membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }])); const framework = await makeFramework({ idleTtlMs: IDLE_TTL_MS }); - framework.pushEvent(incomingEvent({ channelId: 'slack:D7', text: 'hi', channelType: 'im' })); + framework.pushEvent(incomingEvent({ channelId, text: 'hi', channelType: 'im' })); await framework.runUntilIdle(); - const fork = 'conversation-slack-D7-g1'; + const fork = `conversation-${channelId.replace(':', '-')}-g1`; const g1 = framework.getAgent(fork)!; - assert.ok(g1, 'g1 fork should exist after spawn'); + assert.ok(g1, 'the fork should exist after spawn'); const internals = framework as unknown as { addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string }): string; deferredMessages: Array<{ forAgent?: string }>; @@ -731,17 +735,25 @@ describe('Conversation routing', () => { lastConversationSweep: number; sweepExpiredConversations(): void; }; - // The sweep closes the fork and queues its closure turn, and the request - // is then dropped before it runs, as a stale one is. - forceExpiry(framework, 'slack:D7', IDLE_TTL_MS); + forceExpiry(framework, channelId, IDLE_TTL_MS); internals.sweepExpiredConversations(); assert.equal(internals.pendingRequests.filter((r) => r.agentName === fork).length, 1, 'the closure turn is queued'); internals.pendingRequests.splice(0, internals.pendingRequests.length, ...internals.pendingRequests.filter((r) => r.agentName !== fork)); - // A write for the fork that was deferred meanwhile. internals.activeTurnTokens.set(fork, 1); assert.equal(internals.addMessage('alice', [{ type: 'text', text: 'one more thing' }], undefined, { forAgent: fork }), '', 'deferred'); internals.activeTurnTokens.delete(fork); + const sweep = () => { + internals.lastConversationSweep = 0; + internals.sweepExpiredConversations(); + }; + const lastStored = () => JSON.stringify(g1.getContextManager().getAllMessages().at(-1)?.content); + const queued = () => internals.deferredMessages.filter((m) => m.forAgent === fork).length; + return { framework, fork, g1, internals, sweep, lastStored, queued }; + } + + it('a write queued for a closing fork whose closure turn never ran is stored in the fork before the reaper disposes it', async () => { + const { framework, fork, g1, internals } = await closingForkWithAQueuedWrite('slack:D7'); const errors: string[] = []; const error = console.error; @@ -760,6 +772,47 @@ describe('Conversation routing', () => { await framework.stop(); }); + it('the reaper leaves a closing fork registered while a turn token is held for it, and disposes it once the token is gone', async () => { + const { framework, fork, internals, sweep, lastStored, queued } = await closingForkWithAQueuedWrite('slack:D8'); + // A token with no stream and no request, as a puppet call holds across + // its tool's await and a surgery reserves for every agent. + internals.activeTurnTokens.set(fork, 2); + sweep(); + assert.ok(framework.getAgent(fork), 'not disposed while a turn token is held for it'); + assert.equal(queued(), 1, 'its write is still queued'); + assert.ok(!lastStored().includes('one more thing'), `not stored under the token:\n${lastStored()}`); + + internals.activeTurnTokens.delete(fork); + sweep(); + assert.equal(framework.getAgent(fork), null, 'disposed once the token is gone'); + assert.ok(lastStored().includes('one more thing'), `stored in the fork before it was disposed:\n${lastStored()}`); + assert.equal(queued(), 0, 'nothing is left queued for the fork'); + + await framework.stop(); + }); + + it('the reaper leaves a closing fork registered while the host is quiesced, and disposes it after resume() has stored its write', async () => { + const { framework, fork, sweep, lastStored, queued } = await closingForkWithAQueuedWrite('slack:D9'); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + await framework.quiesce({ reason: 'test window' }); + sweep(); + assert.ok(framework.getAgent(fork), 'not disposed while the host is quiesced'); + assert.equal(queued(), 1, 'its write is still queued'); + + await framework.resume(); + assert.ok(lastStored().includes('one more thing'), `resume() stored it in the fork:\n${lastStored()}`); + sweep(); + assert.equal(framework.getAgent(fork), null, 'disposed after resume()'); + assert.equal(queued(), 0, 'nothing is left queued for the fork'); + } finally { console.error = error; } + assert.ok(!errors.some((e) => e.includes('is lost')), errors.join('\n')); + + await framework.stop(); + }); + it('a closure turn that restarts is disposed by the frame that ends it, after storing what was held for it', async () => { const IDLE_TTL_MS = 60_000; // One tool, for the closure turn's tool round. From 4ccaf3a3b8882bc85585d01bdea960fcc492264c Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 18:23:58 -0400 Subject: [PATCH 09/13] docs(deferral): the read-back's failure, and stop()'s unregistered entries, as the code has them - DrainedWrite's doc, writeDrainedInOrder's and the tool boundary's comment said a sharded body is carried joined, and a write that threw after storing is counted at most once, without saying that both rest on reading the store back. When that read fails (storedCopiesOf logs it), a sharded body is carried as given, and a write that threw is put back as one that couldn't be stored, so one that had landed is stored again at a later flush. The docs now say so. - stop()'s comment named an entry queued after its agent left as one stop() can meet. addMessage drops a write for an agent that isn't registered before it can queue it, and puppetToolCall's one direct push is drained by the puppet's own finally in the same synchronous step, so neither a flush nor stop() meets it. The comment now names only an entry restored at a boot that doesn't register its agent. Found in review of this change. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 41 +++++++++++++++++++++++------------------ 1 file changed, 23 insertions(+), 18 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index 38becdbfe..720bcb7f7 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -506,11 +506,12 @@ const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; /** A drained entry as writeDrainedInOrder wrote it: the id its store gave it * (empty when addMessage deferred it again; a sharded body's first shard) - * and the content to carry for it: as given, or for a sharded body, its - * shards joined as later compiles render them. For an entry whose write - * threw after storing it, found in the store and counted as written, it is - * what landed, joined the same way, which for a sharded body can be its - * first shards alone. */ + * and the content to carry for it: as given, or for a sharded body read + * back from its store, its shards joined as later compiles render them. + * For an entry whose write threw after storing it, found in the store and + * counted as written, it is what landed, joined the same way, which for a + * sharded body can be its first shards alone. A store that can't be read + * back (storedCopiesOf logs it) leaves a sharded body as given. */ interface DrainedWrite { id: MessageId; content: ContentBlock[]; @@ -3083,10 +3084,9 @@ export class AgentFramework { // queued with what follows it for its target, and a memory-only queue // still loses it at exit. What is queued for an agent leaves with it // (dropDeferredFor), so an entry for a target no longer registered is one - // queued after its agent left, or restored at a boot that doesn't - // register its agent. It is left: dropped with a memory-only queue at - // exit, and logged and dropped from a persisted one by the boot or resume - // flush that meets it. + // restored at a boot that doesn't register its agent. It is left in the + // persisted queue it came from, and the boot or resume flush that meets + // it logs and drops it. if (this.deferredMessages.length > 0) { const entries = [...this.agents.keys()].flatMap((name) => this.drainDeferredFor(name)); this.readdDrained(entries, 'stop', { bypassDeferral: !this.quiesced }); @@ -9369,9 +9369,10 @@ export class AgentFramework { // Stored in queue order, cut at the first that can't be stored: it // and what follows wait for the turn's end, and only what was // stored is injected (the round's results go on either way). A - // sharded body is injected as later compiles render it, its shards - // joined, and a write that threw after storing, which counts as - // stored, as what landed of it (writeDrainedInOrder). + // sharded body read back from its store is injected as later + // compiles render it, its shards joined, and a write that threw + // after storing, which counts as stored, as what landed of it + // (writeDrainedInOrder). const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) => agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); for (const [msg, { id, content }] of stored) { @@ -16210,16 +16211,20 @@ export class AgentFramework { * of the body isn't stored. (Requeued, the * landed shards would be stored twice, and only the rest could not be * appended later without splitting the body.) Boot's scan for landed - * writes counts a landed shard the same way. + * writes counts a landed shard the same way. When the store can't be read + * back (storedCopiesOf logs it), nothing is found, and the entry is put + * back as one that couldn't be stored: if it had landed, a later flush + * stores it again. * * Returns what was written (stored, or deferred again by addMessage, whose * id is then empty), in queue order: for one target, a prefix of its * entries. Each comes with its content: as given, unless the store took it - * as more than one message (a sharded body), whose copies are returned - * joined, as later compiles render them (asCompiled), or its write threw - * after storing it (above). `onLanded` hears an entry whose write threw - * after storing it, with the id it landed under, as soon as it's found: - * before the writes after it, so whatever it records keeps their order. + * as more than one message (a sharded body) and can be read back, when its + * copies are returned joined, as later compiles render them (asCompiled), + * or its write threw after storing it (above). `onLanded` hears an entry + * whose write threw after storing it, with the id it landed under, as soon + * as it's found: before the writes after it, so whatever it records keeps + * their order. */ private writeDrainedInOrder( entries: DeferredWrite[], From c7a4f56209b5577d6b265f8e7e7418dca79c3454 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 18:39:47 -0400 Subject: [PATCH 10/13] fix(deferral): a closing fork's closure turn that ends on a quiesced host leaves the fork to the reaper The end of a fork's closure turn flushes what was held for it, then disposes the fork. quiesce() sets quiesced and then waits for in-flight turns to settle (quiesce({abandon}) cancels them), so a closure turn can end while the host is quiesced. Its flush then defers the fork's queue again, and the disposal logs it as lost, though resume() would store it: the case c5a3730c3 made the reaper wait for, met at the other place a fork is disposed. The closure turn's disposal now also waits for an unquiesced host. A fork it leaves stays closing, and the fallback reaper, which waits for resume(), stores what is queued for it and disposes it at a later sweep. A turn token needs no such check there: the frame releases its own token just before, and a frame that hands the turn to a successor doesn't reach the disposal. Test: during a fork's closure turn, a write for the fork is deferred and quiesce() begins. The turn ends with the fork still registered and the write queued; resume() stores the write in the fork, the next sweep disposes it, and nothing is logged as lost. On 4ccaf3a3b the fork was disposed while quiesced, with the write logged as lost. Found in review of this change, with a probe. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 15 +++-- src/framework.ts | 10 +++- test/conversation-routing.test.ts | 56 +++++++++++++++++++ 3 files changed, 72 insertions(+), 9 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index 8a9d3bbc5..012412258 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -2,12 +2,15 @@ the deferred-write queue with it, at that moment, and the host log names each entry as lost, with the agent and where it left. An agent leaves at an ephemeral run's end and at a conversation fork's disposal. The last flush - before it leaves can leave an entry there: one whose store threw, or, on a - quiesced host, one that flush deferred again. Before, a closing fork's entry - stayed queued for the disposed fork, where no flush could store it. An - ephemeral run's, once the run's stream had ended, was dropped by the run's - own turn-end flush, which runs after the run is deregistered, with a line - saying its delivery target isn't a registered agent. + before it leaves can leave an entry there: one whose store threw, or, at an + ephemeral run's end on a quiesced host, one that flush deferred again. A + conversation fork isn't disposed while the host is quiesced: the end of its + closure turn leaves it to the fallback reaper, which waits for `resume()` to + store what is queued for it. Before, a closing fork's entry stayed queued + for the disposed fork, where no flush could store it. An ephemeral run's, + once the run's stream had ended, was dropped by the run's own turn-end + flush, which runs after the run is deregistered, with a line saying its + delivery target isn't a registered agent. - The fallback reaper, which disposes a closing conversation fork with no stream and no request left for it, as when its closure turn never ran, now first stores what is still queued for the fork in the fork's own store, past diff --git a/src/framework.ts b/src/framework.ts index 720bcb7f7..03d3aa9a3 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -10140,7 +10140,8 @@ export class AgentFramework { * is still queued for it leaves with it, logged as lost (dropDeferredFor). * Each caller has first stored what it could: the end of the fork's * closure turn with that turn's flush, and the fallback reaper with - * storeDeferredFor. + * storeDeferredFor. Neither calls it while the host is quiesced, when + * resume() would store what that leaves queued. */ private disposeConversationAgent(agentName: string): void { this.dropDeferredFor(agentName, `disposal of conversation fork ${agentName}`); @@ -14273,8 +14274,11 @@ export class AgentFramework { // frame that ends the turn disposes it, after the flush above, so what // was held for the turn's end lands in the fork's store instead of // being dropped as addressed to no agent. A frame that hands the turn - // to a successor (a guard or budget restart) leaves the fork to it. - if (frameReachedTerminal && ownsPhysicalStream && this.closingConversationAgents.has(agent.name)) { + // to a successor (a guard or budget restart) leaves the fork to it. On + // a quiesced host the flush above deferred what was held again, so the + // fork is left to the fallback reaper, which disposes it once resume() + // has stored it. + if (frameReachedTerminal && ownsPhysicalStream && this.closingConversationAgents.has(agent.name) && !this.quiesced) { this.disposeConversationAgent(agent.name); } } diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index ccb928524..478165097 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -664,6 +664,62 @@ describe('Conversation routing', () => { await framework.stop(); }); + it('a fork whose closure turn ends while the host is quiesced is left to the reaper, which disposes it once resume() has stored its write', async () => { + const IDLE_TTL_MS = 60_000; + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }])); + const framework = await makeFramework({ idleTtlMs: IDLE_TTL_MS }); + framework.pushEvent(incomingEvent({ channelId: 'slack:D6', text: 'hi', channelType: 'im' })); + await framework.runUntilIdle(); + const fork = 'conversation-slack-D6-g1'; + const g1 = framework.getAgent(fork)!; + assert.ok(g1, 'g1 fork should exist after spawn'); + + forceExpiry(framework, 'slack:D6', IDLE_TTL_MS); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }])); + const internals = framework as unknown as { + addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string }): string; + deferredMessages: Array<{ forAgent?: string }>; + lastConversationSweep: number; + sweepExpiredConversations(): void; + }; + // While the closure turn is alive, a write for the fork is deferred to + // the turn's end, and an operator quiesces the host, which waits for the + // turn to end. + const streamYielding = membrane.streamYielding.bind(membrane); + let deferred: string | undefined; + let quiescing: Promise | undefined; + membrane.streamYielding = (request, options) => { + if (deferred === undefined) { + deferred = internals.addMessage('alice', [{ type: 'text', text: 'one more thing' }], undefined, { forAgent: fork }); + quiescing = framework.quiesce({ reason: 'test window' }); + } + return streamYielding(request, options); + }; + const lastStored = () => JSON.stringify(g1.getContextManager().getAllMessages().at(-1)?.content); + const queued = () => internals.deferredMessages.filter((m) => m.forAgent === fork).length; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + framework.pushEvent({ type: 'external-message', source: 'test', content: 'tick', metadata: {} } as unknown as ProcessEvent); + await framework.runUntilIdle(); + await quiescing; + assert.equal(deferred, '', 'deferred: the closure turn was alive'); + assert.ok(framework.getAgent(fork), 'not disposed while the host is quiesced'); + assert.equal(queued(), 1, 'its write is still queued'); + + await framework.resume(); + assert.ok(lastStored().includes('one more thing'), `resume() stored it in the fork:\n${lastStored()}`); + internals.lastConversationSweep = 0; + internals.sweepExpiredConversations(); + assert.equal(framework.getAgent(fork), null, 'the reaper disposed it after resume()'); + assert.equal(queued(), 0, 'nothing is left queued for the fork'); + } finally { console.error = error; } + assert.ok(!errors.some((e) => e.includes('is lost')), errors.join('\n')); + + await framework.stop(); + }); + it("a write deferred for a fork that its store refuses at the closure turn's end leaves the queue with the fork, logged as lost", async () => { const IDLE_TTL_MS = 60_000; membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }])); From 8e0cc02ecd4b59f65f9dae98bb08040f2307601c Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 19:46:11 -0400 Subject: [PATCH 11/13] fix(deferral): a deferred write with no target names the agent it went to, when that isn't the primary addMessage sends a write with no target to the primary or, on a host with no primary (agents: []), to the first registered agent, such as an ephemeral run. Deferred, the entry kept no target, and every drain matches an entry by forAgent ?? primaryAgentName, which is null there. So the run's own flushes passed it by, and so did its leaving (storeDeferredFor, dropDeferredFor): after the run the entry was still queued, unstored and not logged as lost, and in the recovery file when the queue was persisted. main does the same. The deferral now records the agent it resolved, when that isn't the primary. The entry is then that agent's, as a targeted write's is: its flushes store it, and it leaves with the agent. A primary-bound entry still has no target, as the recovery file and a boot whose primary was renamed expect. Tests: on a host with no primary, a module's write with no target during an ephemeral run's tool call is deferred, stored once in the run's namespace, and nothing is left queued or logged as lost once the run ends. At 45784c313 the entry stayed queued and unstored after the run. And a write deferred for the primary still has no target in its entry. Found in review of this change. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 7 +++ src/framework.ts | 10 +++- test/deferred-window-only.test.ts | 5 +- test/ephemeral-settle.test.ts | 54 ++++++++++++++++++- 4 files changed, 72 insertions(+), 4 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index 012412258..4d1ee26c7 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -11,6 +11,13 @@ once the run's stream had ended, was dropped by the run's own turn-end flush, which runs after the run is deregistered, with a line saying its delivery target isn't a registered agent. +- On a host with no primary (`agents: []`), a write with no target goes to + the first registered agent, such as an ephemeral run. Deferred, it now names + that agent in its queue entry, so it is that agent's as a targeted write is: + the agent's own flushes store it, and it leaves with the agent. Before, the + entry named no agent, so neither the run's flushes nor its end took it: + after the run it was still queued, unstored and not logged as lost, and in + the recovery file when the queue was persisted. - The fallback reaper, which disposes a closing conversation fork with no stream and no request left for it, as when its closure turn never ran, now first stores what is still queued for the fork in the fork's own store, past diff --git a/src/framework.ts b/src/framework.ts index d46641eae..3e1b16287 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -500,6 +500,8 @@ interface DeferredWrite { participant: string; content: ContentBlock[]; metadata?: MessageMetadata; + /** The agent the entry is for. Absent means the primary: addMessage + * records any other agent it defers a write for. */ forAgent?: string; } const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; @@ -16228,7 +16230,13 @@ export class AgentFramework { if (opts?.fromQueue) { this.unackedDeferredWrites = this.unackedDeferredWrites.filter((m) => m.id !== id); } - this.deferredMessages.push({ id, seq: opts?.fromQueue?.seq ?? ++this.deferredSeq, participant, content, metadata, forAgent: opts?.forAgent }); + // An entry with no target is the primary's. A write with no target that + // went to another agent (on a host with no primary, the first one + // registered, such as an ephemeral run) names that agent, so the + // agent's flushes and its leaving take the entry as they take a + // targeted one. + const forAgent = opts?.forAgent ?? (agent.name === this.primaryAgentName ? undefined : agent.name); + this.deferredMessages.push({ id, seq: opts?.fromQueue?.seq ?? ++this.deferredSeq, participant, content, metadata, forAgent }); if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); if (opts?.placement) { opts.placement.agent = agent.name; opts.placement.deferredId = id; } return '' as MessageId; // Deferred — flushed at the target's next boundary diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index cb85086a4..b0c5ffb83 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -331,7 +331,7 @@ 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 }>; + deferredMessages: Array<{ id: string; seq: number; participant: string; forAgent?: string }>; deferredWritesPersisted: boolean; drainDeferredFor: (name: string, stopAt?: unknown) => unknown[]; readdDrained: (entries: unknown[], where: string) => unknown[]; @@ -380,6 +380,9 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { assert.notEqual(internals.addMessage('Antra', text('targeted'), undefined, { forAgent: 'assistant' }), '', 'stored too'); internals.pendingAssistantBlocks.set('assistant', otherCycle()); assert.equal(internals.addMessage('Antra', text('in its own cycle')), '', "deferred: the primary's own cycle is pending"); + // Its entry names no agent: one with no target is the primary's, as the + // recovery file and a boot whose primary was renamed take it. + assert.deepEqual(internals.deferredMessages.map((m) => m.forAgent), [undefined], 'a primary-bound entry names no agent'); } finally { internals.pendingAssistantBlocks.delete('builder'); internals.pendingAssistantBlocks.delete('assistant'); diff --git a/test/ephemeral-settle.test.ts b/test/ephemeral-settle.test.ts index 370f9af35..711a3dc9c 100644 --- a/test/ephemeral-settle.test.ts +++ b/test/ephemeral-settle.test.ts @@ -65,12 +65,16 @@ class EphemeralToolModule implements Module { /** A tool that, while its call runs, has `duringCall` run: there, a test * addresses the run as the event gate does, through the framework's - * addMessage with `forAgent`. */ + * addMessage with `forAgent`, or writes as a module does, through its + * context (`ctx`). */ class DuringCallToolModule implements Module { readonly name = 'test'; + ctx: ModuleContext | null = null; duringCall: (() => void) | null = null; - async start(_ctx: ModuleContext): Promise {} + async start(ctx: ModuleContext): Promise { + this.ctx = ctx; + } async stop(): Promise {} getTools(): ToolDefinition[] { @@ -473,6 +477,52 @@ describe('runEphemeralToCompletion settle signal', () => { } }); + it("a module's write with no target, which a host with no primary sends to its run, is the run's: stored in its namespace, with nothing left queued once it ends", async () => { + const { tempDir, storePath } = tempStorePath('ephemeral-untargeted-'); + const membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'call_1', name: 'test--echo', input: {} }, + ], 'tool_use')); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Done.' }])); + + const tools = new DuringCallToolModule(); + const framework = await AgentFramework.create({ + storePath, + membrane: membrane.asMembrane(), + agents: [], + modules: [tools], + syncIntervalMs: 0, + }); + const internals = framework as unknown as { deferredMessages: Array<{ forAgent?: string }> }; + // While the run's tool call is pending, a module writes with no target, + // as its events do. The host has no primary, so the write goes to the + // run, its one registered agent, and waits for the run's turn. + const returned: string[] = []; + tools.duringCall = () => { + returned.push(tools.ctx!.addMessage('Antra', [{ type: 'text', text: 'look left' }])); + }; + const errors: string[] = []; + const error = console.error; + try { + const { agent, contextManager } = await createEphemeral(framework, 'worker'); + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const result = await runStartedEphemeral(framework, agent, contextManager); + console.error = error; + assert.strictEqual(result.speech, 'Done.'); + assert.deepStrictEqual(returned, [''], 'deferred: the run was waiting on its tool'); + assert.strictEqual(framework.getAgent(agent.name), null, 'deregistered'); + const stored = contextManager.getAllMessages().map((m) => + `${m.participant}: ${m.content.map((b) => (b.type === 'text' ? (b as { text: string }).text : `<${b.type}>`)).join(' ')}`); + assert.strictEqual(stored.filter((s) => s === 'Antra: look left').length, 1, + `stored once, in the run's namespace:\n${stored.join('\n')}`); + assert.deepStrictEqual(internals.deferredMessages, [], 'nothing is left queued'); + assert.ok(!errors.some((e) => e.includes('is lost')), errors.join('\n')); + } finally { + console.error = error; + await stopAndRemove(framework, tempDir); + } + }); + it('a write queued for a run that its store refuses at the run\'s last flush leaves the queue with the run, logged as lost', async () => { const { tempDir, storePath } = tempStorePath('ephemeral-lost-'); const membrane = new MockMembrane(); From 5e6448aa7d56984d845eda7edb565a74e62bb160 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sat, 17 Oct 2026 20:14:01 -0400 Subject: [PATCH 12/13] fix(deferral): a quiesce doesn't keep a closing fork registered: its queue leaves with it, logged as lost c7a4f5620 left a fork whose closure turn ends on a quiesced host to the fallback reaper, and c5a3730c3 made the reaper wait for an unquiesced host, so that resume() would store what was queued for the fork. That kept a fork that is closing for good registered through the quiesce, where what reads a registered agent treats it as one that runs again: resume()'s feasibility gate previewed it and refused while it didn't fit, the maintenance pass ticked it, and nothing at admission refuses a closing fork, so a nudge, a delivery or a provider hold's release could still start a turn for it. A restart before resume() dropped its queue anyway, since a fork isn't registered again at boot. So on a quiesced host the end of a closure turn and the reaper dispose a closing fork as they do on a serving host. Their flush defers what is queued for it again, and that leaves with the fork, logged as lost, as an ephemeral run's queue does on a quiesced host. The reaper keeps its wait for a turn token, held by a surgery or a puppet call, so that its store-first doesn't write into the store mid-surgery or ahead of the puppet's tool pair. Tests: the two tests of a quiesced host turn round. A fork whose closure turn ends while the host quiesces is disposed at the turn's end, and the reaper disposes a closing fork on a quiesced host. In each, the queued write is logged as lost, nothing is left queued, and resume() stores nothing in the fork. Found in review of this change. Co-Authored-By: Claude Opus 5.5 --- .../deferred-writes-286-follow-up.fixed.md | 36 +++++++++---------- src/framework.ts | 32 +++++++++-------- test/conversation-routing.test.ts | 31 +++++++--------- 3 files changed, 46 insertions(+), 53 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index 4d1ee26c7..24bb9f7e9 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -2,17 +2,16 @@ the deferred-write queue with it, at that moment, and the host log names each entry as lost, with the agent and where it left. An agent leaves at an ephemeral run's end and at a conversation fork's disposal. The last flush - before it leaves can leave an entry there: one whose store threw, or, at an - ephemeral run's end on a quiesced host, one that flush deferred again. A - conversation fork isn't disposed while the host is quiesced: the end of its - closure turn leaves it to the fallback reaper, which waits for `resume()` to - store what is queued for it. Before, a closing fork's entry stayed queued - for the disposed fork, where no flush could store it. An ephemeral run's, - once the run's stream had ended, was dropped by the run's own turn-end - flush, which runs after the run is deregistered, with a line saying its - delivery target isn't a registered agent. -- On a host with no primary (`agents: []`), a write with no target goes to - the first registered agent, such as an ephemeral run. Deferred, it now names + before it leaves can leave an entry there: one whose store threw, or, on a + quiesced host, one that flush deferred again, at an ephemeral run's end and + at a fork's disposal alike: a quiesce doesn't put off an agent's leaving. + Before, a closing fork's entry stayed queued for the disposed fork, where no + flush could store it. An ephemeral run's, once the run's stream had ended, + was dropped by the run's own turn-end flush, which runs after the run is + deregistered, with a line saying its delivery target isn't a registered + agent. +- On a host with no primary (`agents: []`), a write with no target goes to the + first registered agent, such as an ephemeral run. Deferred, it now names that agent in its queue entry, so it is that agent's as a targeted write is: the agent's own flushes store it, and it leaves with the agent. Before, the entry named no agent, so neither the run's flushes nor its end took it: @@ -22,14 +21,13 @@ stream and no request left for it, as when its closure turn never ran, now first stores what is still queued for the fork in the fork's own store, past its deferral, as the end of a closure turn and the end of an ephemeral run - do. It waits to dispose the fork until that can be done: not while a turn - token is held for it, as a puppet call or a surgery holds one, nor while the - host is quiesced. The puppet call's end, the surgery's release and - `resume()` each store the fork's queue, and a later sweep disposes it. Only - what its store throws on then leaves with the fork, as above. Before, the - reaper disposed such a fork while a token was held or the host was quiesced - too, nothing stored what was queued for it, and that stayed queued for the - disposed fork. + do. It doesn't dispose the fork while a turn token is held for it, as a + puppet call or a surgery holds one: the call's end and the surgery's release + each flush what they held, as for any agent, and a later sweep stores what + is left and disposes the fork. On a quiesced host that store defers the + queue again, and it leaves with the fork, as above. Before, the reaper + disposed such a fork while a token was held too, nothing stored what was + queued for it, and that stayed queued for the disposed fork. - Boot recovery numbers the restored deferred-write queue 1, 2, 3, … in the order it restores it, and later deferrals are numbered on from there. Before, it kept each entry's `seq` and numbered only the entries without a diff --git a/src/framework.ts b/src/framework.ts index 3e1b16287..601712c5a 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -10233,8 +10233,10 @@ export class AgentFramework { * is still queued for it leaves with it, logged as lost (dropDeferredFor). * Each caller has first stored what it could: the end of the fork's * closure turn with that turn's flush, and the fallback reaper with - * storeDeferredFor. Neither calls it while the host is quiesced, when - * resume() would store what that leaves queued. + * storeDeferredFor. On a quiesced host each of those defers the fork's + * queue again, so what is queued leaves with the fork, logged as lost, as + * at an ephemeral run's end there: a quiesce doesn't keep a closing fork + * registered for resume(). */ private disposeConversationAgent(agentName: string): void { this.dropDeferredFor(agentName, `disposal of conversation fork ${agentName}`); @@ -10342,19 +10344,20 @@ export class AgentFramework { // Fallback reaper: a closing fork whose closure inference never ran // (request dropped as stale, inference policy veto) would otherwise stay // registered forever. No active stream + no pending request = done. It - // is disposed once what is still queued for it can be stored now, so - // not while a turn token is held for it (a puppet call's, or a surgery's - // reservation) or the host is quiesced: each of those flushes the fork's - // queue as it lets go (the puppet's finally, the surgery's release, - // resume()), and a later sweep disposes the fork. What is queued goes to - // its store first (storeDeferredFor), and what can't be, put back - // because its store threw, leaves with the fork (disposeConversationAgent). + // isn't disposed while a turn token is held for it (a puppet call's, or a + // surgery's reservation), since the store-first below would write its + // queue into the store mid-surgery, or ahead of the puppet's tool pair. + // Each holder flushes what it held as it lets go, as it does for any + // agent (the puppet's finally, the surgery's release), and a later sweep + // disposes the fork. What is queued goes to its store first + // (storeDeferredFor), deferred again on a quiesced host, and what that + // leaves queued, put back because its store threw or deferred again, + // leaves with the fork, logged as lost (disposeConversationAgent). for (const agentName of [...this.closingConversationAgents]) { if ( !this.activeStreams.has(agentName) && !this.pendingRequests.some((r) => r.agentName === agentName) && - !this.activeTurnTokens.has(agentName) && - !this.quiesced + !this.activeTurnTokens.has(agentName) ) { this.storeDeferredFor(agentName, `disposal of conversation fork ${agentName}`); this.disposeConversationAgent(agentName); @@ -14443,10 +14446,9 @@ export class AgentFramework { // was held for the turn's end lands in the fork's store instead of // being dropped as addressed to no agent. A frame that hands the turn // to a successor (a guard or budget restart) leaves the fork to it. On - // a quiesced host the flush above deferred what was held again, so the - // fork is left to the fallback reaper, which disposes it once resume() - // has stored it. - if (frameReachedTerminal && ownsPhysicalStream && this.closingConversationAgents.has(agent.name) && !this.quiesced) { + // a quiesced host the flush above deferred what was held again, and + // the disposal logs it as lost, as at an ephemeral run's end there. + if (frameReachedTerminal && ownsPhysicalStream && this.closingConversationAgents.has(agent.name)) { this.disposeConversationAgent(agent.name); } } diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index 478165097..651b359fc 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -664,7 +664,7 @@ describe('Conversation routing', () => { await framework.stop(); }); - it('a fork whose closure turn ends while the host is quiesced is left to the reaper, which disposes it once resume() has stored its write', async () => { + it('a fork whose closure turn ends while the host is quiesced is disposed at the turn\'s end, and what was held for it is logged as lost', async () => { const IDLE_TTL_MS = 60_000; membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }])); const framework = await makeFramework({ idleTtlMs: IDLE_TTL_MS }); @@ -679,8 +679,6 @@ describe('Conversation routing', () => { const internals = framework as unknown as { addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string }): string; deferredMessages: Array<{ forAgent?: string }>; - lastConversationSweep: number; - sweepExpiredConversations(): void; }; // While the closure turn is alive, a write for the fork is deferred to // the turn's end, and an operator quiesces the host, which waits for the @@ -705,17 +703,14 @@ describe('Conversation routing', () => { await framework.runUntilIdle(); await quiescing; assert.equal(deferred, '', 'deferred: the closure turn was alive'); - assert.ok(framework.getAgent(fork), 'not disposed while the host is quiesced'); - assert.equal(queued(), 1, 'its write is still queued'); + assert.equal(framework.getAgent(fork), null, 'disposed at its closure turn\'s end, though the host is quiesced'); + assert.equal(queued(), 0, 'nothing is left queued for the fork'); await framework.resume(); - assert.ok(lastStored().includes('one more thing'), `resume() stored it in the fork:\n${lastStored()}`); - internals.lastConversationSweep = 0; - internals.sweepExpiredConversations(); - assert.equal(framework.getAgent(fork), null, 'the reaper disposed it after resume()'); - assert.equal(queued(), 0, 'nothing is left queued for the fork'); + assert.ok(!lastStored().includes('one more thing'), `not stored in the fork:\n${lastStored()}`); } finally { console.error = error; } - assert.ok(!errors.some((e) => e.includes('is lost')), errors.join('\n')); + assert.ok(errors.some((e) => e.includes(`disposal of conversation fork ${fork}: deferred write `) + && e.includes(`(alice: text) for ${fork} is lost: its agent has left the framework`)), errors.join('\n')); await framework.stop(); }); @@ -847,7 +842,7 @@ describe('Conversation routing', () => { await framework.stop(); }); - it('the reaper leaves a closing fork registered while the host is quiesced, and disposes it after resume() has stored its write', async () => { + it('the reaper disposes a closing fork on a quiesced host too, and its queued write is logged as lost', async () => { const { framework, fork, sweep, lastStored, queued } = await closingForkWithAQueuedWrite('slack:D9'); const errors: string[] = []; const error = console.error; @@ -855,16 +850,14 @@ describe('Conversation routing', () => { try { await framework.quiesce({ reason: 'test window' }); sweep(); - assert.ok(framework.getAgent(fork), 'not disposed while the host is quiesced'); - assert.equal(queued(), 1, 'its write is still queued'); + assert.equal(framework.getAgent(fork), null, 'disposed, though the host is quiesced'); + assert.equal(queued(), 0, 'nothing is left queued for the fork'); await framework.resume(); - assert.ok(lastStored().includes('one more thing'), `resume() stored it in the fork:\n${lastStored()}`); - sweep(); - assert.equal(framework.getAgent(fork), null, 'disposed after resume()'); - assert.equal(queued(), 0, 'nothing is left queued for the fork'); + assert.ok(!lastStored().includes('one more thing'), `not stored in the fork:\n${lastStored()}`); } finally { console.error = error; } - assert.ok(!errors.some((e) => e.includes('is lost')), errors.join('\n')); + assert.ok(errors.some((e) => e.includes(`disposal of conversation fork ${fork}: deferred write `) + && e.includes(`(alice: text) for ${fork} is lost: its agent has left the framework`)), errors.join('\n')); await framework.stop(); }); From 71ccef70e12b8bacd06182ebc9f470c931fbc80b Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sun, 18 Oct 2026 11:03:34 -0400 Subject: [PATCH 13/13] docs(changelog): a write with no target was taken by neither the run's flushes nor the run's end The entry's "Before" said "nor its release", which names the window release that #204 added to main. Before #204, the run's end drained its queue by the same rule and didn't take the entry either, and since #204 the run's end leaves its queue to that release, which doesn't take it either, so "the run's end" holds for both. In a probe under main's source at 9e3e407c, the entry was still queued after the run, unstored, and in the recovery file; under this branch's source it was stored in the run's namespace, with nothing queued after the run. Co-Authored-By: Claude Opus 5.5 --- changelog.d/deferred-writes-286-follow-up.fixed.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/changelog.d/deferred-writes-286-follow-up.fixed.md b/changelog.d/deferred-writes-286-follow-up.fixed.md index 0c5971e91..faf8802ab 100644 --- a/changelog.d/deferred-writes-286-follow-up.fixed.md +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -3,8 +3,8 @@ that agent in its queue entry, so it is that agent's as a targeted write is: the agent's own flushes store it, and so does its window's release when it leaves. Before, the entry named no agent, so neither the run's flushes nor - its release took it: after the run it was still queued, unstored, and in the - recovery file when the queue was persisted. + the run's end took it: after the run it was still queued, unstored, and in + the recovery file when the queue was persisted. - Boot recovery numbers the restored deferred-write queue 1, 2, 3, … in the order it restores it, and later deferrals are numbered on from there. Before, it kept each entry's `seq` and numbered only the entries without a