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..faf8802ab --- /dev/null +++ b/changelog.d/deferred-writes-286-follow-up.fixed.md @@ -0,0 +1,20 @@ +- 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 so does its window's release when it + leaves. Before, the entry named no agent, so neither the run's flushes nor + 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 + 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. +- 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 ebb8edddb..198129727 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -582,6 +582,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; @@ -595,10 +597,12 @@ type InboundResponder = { /** 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 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[]; @@ -2167,7 +2171,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 @@ -2177,8 +2183,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; @@ -2844,9 +2851,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( @@ -5800,11 +5809,11 @@ export class AgentFramework { // The windows of forks and runs deregistered during the window are // released first, each landing its queued writes and passing on its // unread failed replies, as at any release. Then 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. + // deferred by the quiesce window (module events, api message.send) — + // with no turn alive there is no other flush point. The batch is + // stored in queue order and acknowledged once, after the chronicle + // syncs (see flushDeferredWrites); nothing here may reach the plane + // reopen below. this.runReleasesAfterResume(); await this.flushDeferredWrites('resume'); this.persistHostMode(null); @@ -6450,33 +6459,27 @@ 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, 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 @@ -6484,7 +6487,7 @@ export class AgentFramework { // receipt predates the position field. let landed: Set; try { - landed = this.landedDeferredWriteIds(new Set(candidates.map((m) => m.id)), scanFrom); + landed = this.landedDeferredWriteIds(new Set(ordered.map((m) => m.id)), scanFrom); } catch (error) { throw new Error( `the deferred context writes (${file ?? DEFERRED_WRITES_ID}) can't be checked for writes that already landed: ` + @@ -6494,7 +6497,8 @@ export class AgentFramework { { cause: error }, ); } - const restored = candidates.filter((m) => !landed.has(m.id)); + 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 ` + @@ -6505,8 +6509,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; } @@ -10588,9 +10596,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 - // 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). + // 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) { @@ -19360,7 +19369,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 window's release 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 @@ -19481,13 +19496,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. `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) 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[], @@ -19507,7 +19529,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) { @@ -19545,7 +19575,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 []; } } diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index 7fa50a936..0c57dfb7a 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -713,6 +713,154 @@ describe('Conversation routing', () => { await framework.stop(); }); + 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 lands in its window at resume', 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 = forkName('slack:D6', 1); + 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 }>; + }; + // 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; + 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.equal(framework.getAgent(fork), null, 'disposed at its closure turn\'s end, though the host is quiesced'); + assert.equal(queued(), 1, 'what was held for it waits for its window\'s release at resume'); + assert.ok(!lastStored().includes('one more thing'), `not stored while quiesced:\n${lastStored()}`); + + await framework.resume(); + assert.ok(lastStored().includes('one more thing'), `stored in the fork at resume:\n${lastStored()}`); + assert.equal(queued(), 0, 'nothing is left queued for the fork'); + + await framework.stop(); + }); + + it("a write deferred for a fork that its store refuses at the closure turn's end is tried again at the fork's release, and leaves the queue there, logged", 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 fork = forkName('slack:D5', 1); + const g1 = framework.getAgent(fork)!; + 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: fork }); + 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(fork), null, 'disposed once its closure turn finished'); + assert.deepEqual(internals.deferredMessages.filter((m) => m.forAgent === fork), [], 'nothing is left queued for the fork'); + assert.ok(errors.some((e) => e.includes('turn-end flush: deferred write ') + && e.includes(`(alice: text) for ${fork} 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(`[deferred-flush] ${fork}: failed to store a queued write before release (participant=alice)`) + && e.includes('disk full')), `the fork's release met it again:\n${errors.join('\n')}`); + + await framework.stop(); + }); + + /** 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, text: 'hi', channelType: 'im' })); + await framework.runUntilIdle(); + const fork = forkName(channelId, 1); + const g1 = framework.getAgent(fork)!; + 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 }>; + pendingRequests: Array<{ agentName: string }>; + activeTurnTokens: Map; + lastConversationSweep: number; + sweepExpiredConversations(): void; + }; + 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)); + 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, sweep, lastStored, queued }; + } + + it('the reaper disposes a closing fork on a quiesced host too, and its queued write lands in its window at resume', async () => { + const { framework, fork, sweep, lastStored, queued } = await closingForkWithAQueuedWrite('slack:D9'); + await framework.quiesce({ reason: 'test window' }); + sweep(); + assert.equal(framework.getAgent(fork), null, 'disposed, though the host is quiesced'); + assert.equal(queued(), 1, 'its write waits for its window\'s release at resume'); + assert.ok(!lastStored().includes('one more thing'), `not stored while quiesced:\n${lastStored()}`); + + await framework.resume(); + assert.ok(lastStored().includes('one more thing'), `stored in the fork at resume:\n${lastStored()}`); + assert.equal(queued(), 0, 'nothing is left queued for the fork'); + + 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/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 44145ca24..b0c5ffb83 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, @@ -330,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[]; @@ -379,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'); @@ -656,6 +660,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'); diff --git a/test/ephemeral-settle.test.ts b/test/ephemeral-settle.test.ts index 889e54474..67db8f986 100644 --- a/test/ephemeral-settle.test.ts +++ b/test/ephemeral-settle.test.ts @@ -64,12 +64,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[] { @@ -749,6 +753,105 @@ 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' }])); + }; + try { + const { agent, contextManager } = await createEphemeral(framework, 'worker'); + const result = await runStartedEphemeral(framework, agent, contextManager); + 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'); + } finally { + await stopAndRemove(framework, tempDir); + } + }); + + it('a write queued for a run that its store refuses leaves the queue at the run\'s release, logged', 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 }>; + 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' }); + }; + 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'); + 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('[deferred-flush] worker: failed to store a queued write before release (participant=Antra)') + && e.includes('disk full')), 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('resolves through the endTurn tool-result path', async () => { const { tempDir, storePath } = tempStorePath('ephemeral-endturn-'); const membrane = new MockMembrane(); diff --git a/test/host-quiesce.test.ts b/test/host-quiesce.test.ts index 578d39cf1..0144d53c0 100644 --- a/test/host-quiesce.test.ts +++ b/test/host-quiesce.test.ts @@ -1506,6 +1506,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');