From 1ff5b357706db52a48891b8e6307153deacba6f1 Mon Sep 17 00:00:00 2001 From: Nigel-2654 Date: Mon, 19 Oct 2026 19:03:55 -0400 Subject: [PATCH 1/6] fix(deferral): a drained write whose landing can't be read back isn't stored twice When a drained deferred write threw, writeDrainedInOrder read the agent's store back from where it stood before the write, for copies carrying the entry's deferredWriteId, so a write that landed before the throw counted as written. But storedCopiesOf returned [] when that read threw, and a count that couldn't be read before the write left nothing to read from, so the drain took either as "nothing landed": it cut the target and requeued the entry, and a later flush stored it a second time. An unknown landing is now kept apart from an empty one. storedCopiesOf returns undefined when the store can't be read back, and so does the catch when the count before the write couldn't be read. Such an entry is marked (DeferredWrite.landing = 'unknown') before it's requeued, so the queue persists the mark. Before a marked entry is written again, the store is searched for its id with queryMessages (landedCopiesById), which needs no position: - found, it counts as written, as when the readback works: its copies joined (asCompiled), logged, and passed to onLanded; - not found, the mark goes and it's written; - when the search throws, the target is cut and the entry requeued, still marked, holding what's behind it, as a write that keeps failing does. handOffDeferredWrites sets its agent's boot scan position to 0 for a marked entry, since an earlier copy could sit anywhere before the store's present count; boot's check scans from there. A write found later is carried at the flush that finds it, after what the turn stored since, while the store holds it earlier. The success path, where the write returned and its sharded body can't be read back, still carries the body as given. logDeferredWriteError names the three new cases. The drain's doc now says what an unknown landing does. test/deferred-window-only.test.ts: at a tool boundary, a write that throws after storing, whose readback then throws, is stored once and found by its id; and four tests of the drain itself: a count that couldn't be read before a throwing write, an unknown entry the store doesn't hold, an unknown entry whose search throws, and a hand-off of an unknown entry. Co-Authored-By: Claude Opus 5.5 --- .../deferred-write-landing-unknown.fixed.md | 14 ++ src/framework.ts | 125 +++++++++++--- test/deferred-window-only.test.ts | 156 ++++++++++++++++++ 3 files changed, 271 insertions(+), 24 deletions(-) create mode 100644 changelog.d/deferred-write-landing-unknown.fixed.md diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md new file mode 100644 index 000000000..0144707d9 --- /dev/null +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -0,0 +1,14 @@ +- A deferred context write whose write threw, and whose store then couldn't + be read back to tell whether it had landed, is no longer put back as one + that wasn't stored. Until now a later flush stored it again, so a write + that had landed appeared twice. It's now put back marked as unknown, and + before it's written again the store is searched for its id: found, it + counts as written and isn't stored again; not found, it's written; and + while the store can't be searched, it stays queued, holding what's queued + behind it for that agent, as a write that keeps failing does. A store that + couldn't say how many messages it held before the write is taken the same + way. Such an entry's mark is persisted with the queue, and boot looks for + it in its agent's whole store. + - A write found this way is carried at the flush that finds it, after + what the turn stored since, though the store holds it before those, so + later requests place it earlier. diff --git a/src/framework.ts b/src/framework.ts index 07eb7e8f8..f5efe93c0 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -735,6 +735,12 @@ interface DeferredWrite { /** When the framework first held the write, in ms since the epoch: a * module reads it as the held row's timestamp (heldRows). */ heldAt: number; + /** Set when a write of this entry threw and its store couldn't be read + * back to tell whether it landed (writeDrainedInOrder): before it's + * written again, the store is searched for its id. Persisted with the + * entry, and a hand-off of such an entry has boot scan its agent's whole + * store (handOffDeferredWrites). */ + landing?: 'unknown'; } const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; @@ -780,6 +786,12 @@ function asCompiled(landed: ReadonlyArray<{ content: ContentBlock[] }>): Content const text = blocks.filter((b) => b.type === 'text').map((b) => (b as { text: string }).text).join(''); return [...blocks.filter((b) => b.type !== 'text'), { type: 'text', text } as ContentBlock]; } +/** One stored copy of a deferred write, as the drain reads it back: a + * sharded body's copies carry its `bodyGroupId`. */ +type StoredCopy = { id: MessageId; content: ContentBlock[]; bodyGroupId?: string }; +const storedCopy = (m: { id: MessageId; content: ContentBlock[]; bodyGroupId?: string }): StoredCopy => + ({ id: m.id, content: m.content, ...(m.bodyGroupId !== undefined ? { bodyGroupId: m.bodyGroupId } : {}) }); + /** Stamp a deferred write's durable id into the metadata it is stored with * (boot recovery dedups replays by it). Idempotent for an already-stamped * message. */ @@ -6568,6 +6580,10 @@ export class AgentFramework { const cm = agent.getContextManager() as unknown as { getMessageCount?: () => number; getAllMessages: () => unknown[] }; count = typeof cm.getMessageCount === 'function' ? cm.getMessageCount() : cm.getAllMessages().length; } catch { count = 0; } + // An entry whose earlier write may have landed (landing unknown) could + // be anywhere before the store's present count, so boot scans the + // agent's whole store for it. + if (entry.landing === 'unknown') count = 0; const prev = this.deferredScanFrom.get(target!); this.deferredScanFrom.set(target!, prev === undefined ? count : Math.min(prev, count)); } @@ -21218,9 +21234,12 @@ export class AgentFramework { * 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. 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. + * back (storedCopiesOf logs it), or couldn't say how many messages it held + * before the write, whether the entry landed is unknown: it's put back + * marked so (`landing`), and before it's written again the store is + * searched for its id (landedCopiesById). Found, it counts as written, as + * above, at the flush that finds it; not found, it's written; and while + * the store can't be searched, it stays queued, holding what's behind it. * * 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 @@ -21248,28 +21267,56 @@ export class AgentFramework { const target = targetOf(msg); if (cut.has(target)) continue; const agent = this.deliveryAgent(msg.forAgent); + // An earlier write of this entry threw, and its store couldn't be read + // back to tell whether it landed. It's looked up by its id before it's + // written again: found, it counts as written, as when the readback + // works; not found, it's written; and while the store can't be + // searched, it stays queued, holding what's behind it. + if (msg.landing === 'unknown') { + const found = agent ? this.landedCopiesById(agent, msg.id) : undefined; + if (found === undefined) { + cut.add(target); + this.requeueDrained(ordered.slice(i).filter((m) => targetOf(m) === target)); + this.logDeferredWriteError(msg, where, undefined, 'still-unknown'); + continue; + } + if (found.length > 0) { + written.set(msg, { id: found[0]!.id, content: asCompiled(found) }); + this.logDeferredWriteError(msg, where, undefined, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + onLanded?.(msg, found[0]!.id); + continue; + } + delete msg.landing; + } const before = agent ? this.storedCountOf(agent) : undefined; try { 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. + // than one message. One that can't be read back is carried as given. const copies = agent && before !== undefined && (this.storedCountOf(agent) ?? before) - before > 1 - ? this.storedCopiesOf(agent, before, msg.id) + ? 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) { + // What landed, read back from where the store stood before the + // write; undefined when the store couldn't say, before the write or + // after it. With no agent there's no store it could have landed in. + const landed = !agent ? [] : before === undefined ? undefined : this.storedCopiesOf(agent, before, msg.id); + if (landed && landed.length > 0) { written.set(msg, { id: landed[0]!.id, content: asCompiled(landed) }); this.logDeferredWriteError(msg, where, error, { count: landed.length, sharded: landed[0]!.bodyGroupId !== undefined }); onLanded?.(msg, landed[0]!.id); continue; } + // Requeued as unstored, a write that had landed would be stored a + // second time; so one whose landing is unknown is marked, and looked + // up by its id before it's written again (above). + if (landed === undefined) msg.landing = 'unknown'; cut.add(target); this.requeueDrained(ordered.slice(i).filter((m) => targetOf(m) === target)); - this.logDeferredWriteError(msg, where, error); + this.logDeferredWriteError(msg, where, error, landed === undefined ? 'unknown' : undefined); } } return written; @@ -21288,29 +21335,47 @@ export class AgentFramework { /** The messages carrying `deferredWriteId` among those the agent's store * gained from position `from` on, in store order, with their blobs * resolved, as a request carries them. A sharded body's carry its - * `bodyGroupId`. */ - private storedCopiesOf(agent: Agent, from: number, deferredWriteId: string): Array<{ id: MessageId; content: ContentBlock[]; bodyGroupId?: string }> { + * `bodyGroupId`. Undefined when the store can't be read back, which is + * logged: not the same as nothing found. */ + private storedCopiesOf(agent: Agent, from: number, deferredWriteId: string): StoredCopy[] | undefined { try { const cm = agent.getContextManager(); return cm.getMessageWindow(from, Math.max(0, cm.getMessageCount() - from)).messages .filter((m) => m.metadata?.deferredWriteId === deferredWriteId) - .map((m) => ({ id: m.id, content: m.content, ...(m.bodyGroupId !== undefined ? { bodyGroupId: m.bodyGroupId } : {}) })); + .map(storedCopy); } catch (error) { console.error(`[deferral] could not read back what deferred write ${deferredWriteId} stored in ${agent.name}'s store:`, error); - return []; + return undefined; + } + } + + /** The messages carrying `deferredWriteId` anywhere in the agent's store, + * found by their id (queryMessages, as landedRow finds a held write's), + * so no position is needed. Undefined when the store can't be searched, + * which is logged. */ + private landedCopiesById(agent: Agent, deferredWriteId: string): StoredCopy[] | undefined { + try { + return agent.getContextManager().queryMessages({ metadata: { deferredWriteId } }).messages.map(storedCopy); + } catch (error) { + console.error(`[deferral] could not search ${agent.name}'s store for deferred write ${deferredWriteId}:`, error); + return undefined; } } /** The line naming a drained write that threw: one that couldn't be stored, - * which is already back in the queue, or, with `landed`, one stored before - * the throw (for a sharded body, how many of its shards). Logging mustn't - * fail the flush that met it, so a thrown value that has no text is named - * as such. */ + * which is already back in the queue; with a count, one stored before the + * throw (for a sharded body, how many of its shards), or, with + * `foundLater`, one an earlier write of which landed and was found by its + * id; with `'unknown'`, one whose store couldn't be read back to tell + * whether it landed, back in the queue to be looked up first; and with + * `'still-unknown'`, one that still can't be looked up. Logging mustn't + * fail the flush that met it, so a thrown value that has no text is + * named as such. */ private logDeferredWriteError( msg: DeferredWrite, where: string, error: unknown, - landed?: { count: number; sharded: boolean }, + outcome?: { count: number; sharded: boolean; foundLater?: boolean } | 'unknown' | 'still-unknown', ): void { let reason: string; try { @@ -21318,15 +21383,27 @@ export class AgentFramework { } catch { reason = 'a thrown value that has no text'; } + const held = 'it stays queued, holding what is queued behind it for that agent'; + const stored = (landed: { count: number; sharded: boolean }): string => landed.sharded + ? `${landed.count} shard${landed.count === 1 ? '' : 's'} of its body ${landed.count === 1 ? 'was' : 'were'} stored` + : 'it was stored'; + let what: string; + if (outcome === 'unknown') { + what = `threw, and its store couldn't be read back to tell whether it was stored; ${held}, and is looked up ` + + `by its id before it's written again: ${reason}`; + } else if (outcome === 'still-unknown') { + what = `still can't be looked up by its id, to tell whether an earlier write of it was stored; ${held}`; + } else if (outcome?.foundLater) { + what = `was found by its id: ${stored(outcome)} by an earlier write whose landing couldn't be read back; what ` + + `was stored counts as written and isn't stored again`; + } else if (outcome) { + what = `threw after ${stored(outcome)}; what was stored counts as written and isn't stored again: ${reason}`; + } else { + what = `could not be stored; ${held}: ${reason}`; + } console.error( `[deferral] ${where}: deferred write ${describeDeferredWrite(msg)} for ` + - `${msg.forAgent ?? this.primaryAgentName ?? 'the primary'} ` + - (landed - ? `threw after ${landed.sharded - ? `${landed.count} shard${landed.count === 1 ? '' : 's'} of its body ${landed.count === 1 ? 'was' : 'were'} stored` - : 'it was stored'}; what was stored counts as written and isn't stored again: ` - : `could not be stored; it stays queued, holding what is queued behind it for that agent: `) + - reason, + `${msg.forAgent ?? this.primaryAgentName ?? 'the primary'} ${what}`, ); } diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 8fe4f9145..b85609ade 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -564,6 +564,162 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); + /** Have the agent's strategy throw once its message carrying `textOf` is + * stored, and the store's next windowed read after that throw, the + * drain's readback of what landed, throw too. */ + function throwAfterStoringThenUnreadable(framework: AgentFramework, textOf: string): void { + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + strategy: { onNewMessage?: (message: { content: ContentBlock[] }, ctx: unknown) => Promise }; + getMessageWindow: (...args: unknown[]) => unknown; + }; + const onNewMessage = cm.strategy.onNewMessage; + const read = cm.getMessageWindow; + let thrown = false; + let unreadable = false; + cm.strategy.onNewMessage = function (this: unknown, message, ctx) { + if (!thrown && message.content.some((b) => (b as { text?: string }).text === textOf)) { + thrown = true; + unreadable = true; + throw new Error('strategy failed'); + } + return onNewMessage ? onNewMessage.call(this, message, ctx) : Promise.resolve(); + }; + cm.getMessageWindow = function (this: unknown, ...args: unknown[]) { + if (unreadable) { + unreadable = false; + throw new Error('readback failed'); + } + return read.apply(this, args); + }; + } + + it("at a tool boundary, a write that throws after it's stored, whose store then can't be read back, is found by its id later and isn't stored a second time", async () => { + 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(); + throwAfterStoringThenUnreadable(framework, 'look left'); + module.interjections = [ + { participant: 'Antra', content: text('look left') }, + { participant: 'Antra', content: text('look right') }, + ]; + 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: 'go' } as unknown as ProcessEvent); + await framework.runUntilIdle(); + } finally { console.error = error; } + assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); + assert.equal(windowOf(framework).filter((line) => line === 'Antra: look right').length, 1); + assert.ok(errors.some((e) => e.includes("threw, and its store couldn't be read back to tell whether it was stored") + && e.includes('strategy failed')), errors.join('\n')); + assert.ok(errors.some((e) => e.includes('was found by its id: it was stored by an earlier write')), errors.join('\n')); + await framework.stop(); + }); + + /** The internals a drained write's landing goes through. */ + function drainInternals(framework: AgentFramework) { + return framework as unknown as { + writeDrainedInOrder( + entries: DeferredEntry[], where: string, write: (msg: DeferredEntry) => string, + ): Map; + handOffDeferredWrites(entries: DeferredEntry[]): void; + deferredMessages: DeferredEntry[]; + deferredScanFrom: Map; + deferredWritesPersisted: boolean; + }; + } + type DeferredEntry = { + id: string; seq: number; participant: string; content: ContentBlock[]; heldAt: number; landing?: 'unknown'; + }; + const entry = (id: string, said: string): DeferredEntry => ({ id, seq: 1, participant: 'Antra', content: text(said), heldAt: Date.now() }); + /** A write that stores the entry as the framework does, under its id, then throws. */ + const storedThenThrows = (framework: AgentFramework) => (msg: DeferredEntry): string => { + framework.getAgent('assistant')!.getContextManager().addMessage(msg.participant, msg.content, { deferredWriteId: msg.id } as never); + throw new Error('strategy failed'); + }; + + it("a drained write that throws when its store couldn't count its messages before it is looked up by its id before it's written again", async () => { + const framework = await createFramework(); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { getMessageCount: () => number }; + const count = cm.getMessageCount; + let failing = true; + cm.getMessageCount = function (this: unknown) { + if (failing) { failing = false; throw new Error('count failed'); } + return count.call(this); + }; + const internals = drainInternals(framework); + const msg = entry('w-count', 'look left'); + const error = console.error; + console.error = () => {}; + try { + internals.writeDrainedInOrder([msg], 'test', storedThenThrows(framework)); + assert.equal(msg.landing, 'unknown'); + let writes = 0; + const written = internals.writeDrainedInOrder([msg], 'test', () => { writes++; return 'again'; }); + assert.equal(writes, 0, 'found by its id, it is not written again'); + assert.ok(written.get(msg)?.id, 'it counts as written, under the id it was stored with'); + } finally { console.error = error; } + assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1); + await framework.stop(); + }); + + it("a drained write whose landing is unknown, and which its store doesn't hold, is written when it's looked up", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const msg = { ...entry('w-absent', 'look left'), landing: 'unknown' as const }; + let writes = 0; + const written = internals.writeDrainedInOrder([msg], 'test', (m) => { + writes++; + return framework.getAgent('assistant')!.getContextManager().addMessage(m.participant, m.content, { deferredWriteId: m.id } as never); + }); + assert.equal(writes, 1); + assert.ok(written.has(msg)); + assert.equal(msg.landing, undefined, 'looked up and written, it is no longer unknown'); + await framework.stop(); + }); + + it("a drained write whose landing is unknown stays queued, holding what's behind it, while its store can't be searched", async () => { + const framework = await createFramework(); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = function () { throw new Error('search failed'); }; + const internals = drainInternals(framework); + const first = { ...entry('w-held', 'look left'), landing: 'unknown' as const }; + const second = { ...entry('w-behind', 'look right'), seq: 2 }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + let writes = 0; + try { + const written = internals.writeDrainedInOrder([first, second], 'test', () => { writes++; return 'x'; }); + assert.equal(written.size, 0); + } finally { console.error = error; } + assert.equal(writes, 0, 'neither is written'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), ['w-held', 'w-behind'], 'both are back in the queue'); + assert.equal(first.landing, 'unknown'); + assert.ok(errors.some((e) => e.includes("still can't be looked up by its id")), errors.join('\n')); + cm.queryMessages = query; + await framework.stop(); + }); + + it("a hand-off of an entry whose landing is unknown has boot scan its agent's whole store", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + internals.deferredWritesPersisted = true; + const cm = framework.getAgent('assistant')!.getContextManager(); + cm.addMessage('Antra', text('earlier')); + internals.handOffDeferredWrites([entry('w-known', 'one')]); + const from = internals.deferredScanFrom.get('assistant'); + assert.ok(from !== undefined && from > 0, `a known entry is scanned for from the store's count, ${from}`); + internals.handOffDeferredWrites([{ ...entry('w-unknown', 'two'), seq: 2, landing: 'unknown' }]); + assert.equal(internals.deferredScanFrom.get('assistant'), 0); + await framework.stop(); + }); + /** Shard a message whose first block is `marker` into one shard per block, * as a strategy's ingress chunking does. */ function shardOn(framework: AgentFramework, marker: string): void { From 1741955cc234733bd7e7cb8321fde25bd372dccd Mon Sep 17 00:00:00 2001 From: Nigel-2654 Date: Mon, 19 Oct 2026 19:35:15 -0400 Subject: [PATCH 2/6] fix(deferral): look a marked write up wherever a held write is stored, read or changed From our internal review of 1ff5b357, and a colleague's runs: - releaseWindow stored a released agent's queue with cm.addMessage, outside the drain, so a marked entry was stored twice there. It's now looked up by its id first. Found, it's neither stored again nor passed to the primary, since release's first pass has passed its stored copy if the agent hadn't read it. Where the store can't be searched, it's stored anyway, with a line saying it may now be stored twice, since no later flush will try again. - A marked entry whose agent isn't registered was held at every flush, since the lookup took a missing agent as a failed search. The lookup now runs only with an agent, so addMessage drops the entry as it does an unmarked one. - A module's handle to a marked held write reached only the queued entry: an edit was lost when the flush found the stored copy, a removal left the copy, and a query showed the write twice. The handle now reaches the stored copy too. An edit and a removal apply to both, looked up first, so a lookup that throws throws with nothing changed; a query shows the write once, as its held row. The coalescer's read and shown checks read the copy's own row, a failed search counting as read and as shown, and its removal takes the copy only where that row may never have been shown. - The hand-off no longer records 0 as boot's scan position for a marked entry: the persisted mark already has the first flush after a restart look it up, so scanning the whole store at boot bought nothing. - deferredMessages is typed DeferredWrite[], which took the inline type's docs, so the mark is in the queue's one shape. - The line for an unknown landing says the store couldn't be read, which covers a count that failed before the write. The boundary's comment and the test file's header name the found-later exception to the window matching what the stream carried. Tests: the end-to-end test checks the window's order, and the count test checks the id the write counts under and its onLanded. New tests cover a found write carried as stored, another target's entry behind a held one, an unregistered agent, a mark kept across a restart, release (a found entry, a found failed reply, a blind search), a module's edit, removal and query, both under a blind lookup, and the coalescer's removal by the copy's shown check. Co-Authored-By: Claude Opus 5.5 --- .../deferred-write-landing-unknown.fixed.md | 28 +- src/framework.ts | 148 +++++++--- test/deferred-window-only.test.ts | 255 ++++++++++++++++-- 3 files changed, 359 insertions(+), 72 deletions(-) diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md index 0144707d9..8627b2a23 100644 --- a/changelog.d/deferred-write-landing-unknown.fixed.md +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -1,14 +1,22 @@ - A deferred context write whose write threw, and whose store then couldn't - be read back to tell whether it had landed, is no longer put back as one - that wasn't stored. Until now a later flush stored it again, so a write - that had landed appeared twice. It's now put back marked as unknown, and - before it's written again the store is searched for its id: found, it - counts as written and isn't stored again; not found, it's written; and - while the store can't be searched, it stays queued, holding what's queued - behind it for that agent, as a write that keeps failing does. A store that - couldn't say how many messages it held before the write is taken the same - way. Such an entry's mark is persisted with the queue, and boot looks for - it in its agent's whole store. + be read to tell whether it had landed, is no longer put back as one that + wasn't stored. Until now a later flush stored it again, so a write that + had landed appeared twice. It's now put back marked as unknown, and before + it's written again the store is searched for its id: found, it counts as + written and isn't stored again; not found, it's written; and while the + store can't be searched, it stays queued, holding what's queued behind it + for that agent, as a write that keeps failing does. Such an entry's mark + is persisted with the queue, so after a restart it's still looked up + before it's written. - A write found this way is carried at the flush that finds it, after what the turn stored since, though the store holds it before those, so later requests place it earlier. + - A conversation fork's or ephemeral run's queue, stored when it's + released, looks such an entry up the same way: found, it's neither + stored again nor passed to the primary as a failed reply; where the + store can't be searched, it's stored anyway, and the log says it may now + be stored twice. + - A module's handle to such a held write reaches its stored copy too: an + edit or a removal applies to both, and throws with nothing changed when + the store can't be searched, and a query shows the write once. The + coalescer removes the stored copy only where nothing may have shown it. diff --git a/src/framework.ts b/src/framework.ts index f5efe93c0..e3b105cef 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -724,7 +724,11 @@ const DEFERRED_WRITES_PERSIST_CAP_CHARS = 4 * 1024 * 1024; * with no stream). */ /** One entry of the deferred-write queue (see AgentFramework.deferredMessages). */ interface DeferredWrite { + /** 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; participant: string; content: ContentBlock[]; @@ -736,10 +740,9 @@ interface DeferredWrite { * module reads it as the held row's timestamp (heldRows). */ heldAt: number; /** Set when a write of this entry threw and its store couldn't be read - * back to tell whether it landed (writeDrainedInOrder): before it's + * to tell whether it landed (writeDrainedInOrder): before it's * written again, the store is searched for its id. Persisted with the - * entry, and a hand-off of such an entry has boot scan its agent's whole - * store (handOffDeferredWrites). */ + * entry, so after a restart it's still looked up first. */ landing?: 'unknown'; } const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; @@ -2504,22 +2507,7 @@ export class AgentFramework { * could occupy and dedups against all of it — never a fixed tail. */ private deferredScanFrom = new Map(); - private deferredMessages: Array<{ - /** 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; - participant: string; - content: ContentBlock[]; - metadata?: MessageMetadata; - /** Delivery target (registry name). Absent = the primary agent — - * the historical behavior of every existing caller. */ - forAgent?: string; - /** When the framework first held the write (DeferredWrite.heldAt). */ - heldAt: number; - }> = []; + private deferredMessages: DeferredWrite[] = []; // Turn-alive markers (2026-07-31 Mythos phantom-skip incident): agentName → // token of the turn currently in progress. Set at startAgentStream ENTRY — @@ -6580,10 +6568,6 @@ export class AgentFramework { const cm = agent.getContextManager() as unknown as { getMessageCount?: () => number; getAllMessages: () => unknown[] }; count = typeof cm.getMessageCount === 'function' ? cm.getMessageCount() : cm.getAllMessages().length; } catch { count = 0; } - // An entry whose earlier write may have landed (landing unknown) could - // be anywhere before the store's present count, so boot scans the - // agent's whole store for it. - if (entry.landing === 'unknown') count = 0; const prev = this.deferredScanFrom.get(target!); this.deferredScanFrom.set(target!, prev === undefined ? count : Math.min(prev, count)); } @@ -11429,7 +11413,11 @@ export class AgentFramework { // them in the live conversation — same participant, same content // blocks, same metadata, and compile() emits recent messages // verbatim, so the next turn's compiled prefix byte-matches the - // live request (prompt-cache safe). + // live request (prompt-cache safe). A write that threw after it was + // stored, whose landing couldn't be read, is the exception: it's + // carried at the flush that finds it, after what was stored since, + // so there the store and the live request differ + // (writeDrainedInOrder). // // The queue is drained ONLY at the target agent's own boundary: // draining at another agent's boundary would store the messages @@ -13332,7 +13320,17 @@ export class AgentFramework { * in the deferred queue is both. */ private isUnreadPlacement(p: CoalescingPlacement, by: 'consumed' | 'shown' = 'consumed'): boolean { this.resolvePlacement(p); - if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId); + if (p.deferredId) { + const held = this.deferredMessages.find((m) => m.id === p.deferredId); + if (!held) return false; + if (held.landing !== 'unknown') return true; + // A held write whose landing is unknown may also be stored, and that + // copy may have been read or shown: its own row answers. A store that + // can't be searched is a doubt, which counts as read and as shown. + let copy: { agent: Agent; row: StoredMessage } | null; + try { copy = this.landedCopyOfHeld(held); } catch { return false; } + return copy ? this.isUnreadStoredMessage(copy.agent, copy.row.id, by) : true; + } if (!p.messageId) return false; const agent = this.agents.get(p.agent); return !!agent && this.isUnreadStoredMessage(agent, p.messageId, by); @@ -13389,9 +13387,28 @@ export class AgentFramework { private removePlacement(p: CoalescingPlacement): boolean { this.resolvePlacement(p); if (p.deferredId) { - const before = this.deferredMessages.length; + const held = this.deferredMessages.find((m) => m.id === p.deferredId); + if (!held) return false; + // A held write whose landing is unknown may also be stored: its stored + // copy goes too, but only where that row's own check says nothing may + // have shown it (the held entry reads as unshown whatever its copy + // went through). Where something may have, or the store can't be + // searched, nothing is removed. + if (held.landing === 'unknown') { + let copy: { agent: Agent; row: StoredMessage } | null; + try { copy = this.landedCopyOfHeld(held); } catch { return false; } + if (copy) { + if (!this.isUnreadStoredMessage(copy.agent, copy.row.id, 'shown')) return false; + try { + removeStoredMessage(copy.agent.getContextManager(), copy.row.id); + } catch (err) { + console.error(`[coalescing] could not remove unread message ${copy.row.id} for ${copy.agent.name}:`, err); + return false; + } + this.emitTrace({ type: 'message:removed', messageId: copy.row.id, source: 'mcpl:coalescing' }); + } + } this.deferredMessages = this.deferredMessages.filter((m) => m.id !== p.deferredId); - if (this.deferredMessages.length === before) return false; if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); return true; } @@ -21271,9 +21288,11 @@ export class AgentFramework { // back to tell whether it landed. It's looked up by its id before it's // written again: found, it counts as written, as when the readback // works; not found, it's written; and while the store can't be - // searched, it stays queued, holding what's behind it. - if (msg.landing === 'unknown') { - const found = agent ? this.landedCopiesById(agent, msg.id) : undefined; + // searched, it stays queued, holding what's behind it. An entry whose + // agent isn't registered has no store to look in: it's written, and + // addMessage drops it, as it does an unmarked one. + if (msg.landing === 'unknown' && agent) { + const found = this.landedCopiesById(agent, msg.id); if (found === undefined) { cut.add(target); this.requeueDrained(ordered.slice(i).filter((m) => targetOf(m) === target)); @@ -21366,7 +21385,7 @@ export class AgentFramework { * which is already back in the queue; with a count, one stored before the * throw (for a sharded body, how many of its shards), or, with * `foundLater`, one an earlier write of which landed and was found by its - * id; with `'unknown'`, one whose store couldn't be read back to tell + * id; with `'unknown'`, one whose store couldn't be read to tell * whether it landed, back in the queue to be looked up first; and with * `'still-unknown'`, one that still can't be looked up. Logging mustn't * fail the flush that met it, so a thrown value that has no text is @@ -21389,12 +21408,12 @@ export class AgentFramework { : 'it was stored'; let what: string; if (outcome === 'unknown') { - what = `threw, and its store couldn't be read back to tell whether it was stored; ${held}, and is looked up ` + + what = `threw, and its store couldn't be read to tell whether it was stored; ${held}, and is looked up ` + `by its id before it's written again: ${reason}`; } else if (outcome === 'still-unknown') { what = `still can't be looked up by its id, to tell whether an earlier write of it was stored; ${held}`; } else if (outcome?.foundLater) { - what = `was found by its id: ${stored(outcome)} by an earlier write whose landing couldn't be read back; what ` + + what = `was found by its id: ${stored(outcome)} by an earlier write whose landing was unknown; what ` + `was stored counts as written and isn't stored again`; } else if (outcome) { what = `threw after ${stored(outcome)}; what was stored counts as written and isn't stored again: ${reason}`; @@ -21494,7 +21513,11 @@ export class AgentFramework { * one still queued for it. Writes still * queued would wait for a boundary it never reaches, and the flush that * finds it gone would drop them, so they land in its own window now, which - * Chronicle keeps for investigation. Its stream is over or cancelled, so it + * Chronicle keeps for investigation. One whose earlier write may have + * landed (`landing` unknown) is looked up there by its id first: found, + * it's neither stored again nor passed to the primary; where the store + * can't be searched, it's stored anyway, with a line saying it may now be + * stored twice. Its stream is over or cancelled, so it * compiles nothing more. One path still stores after these writes: a * `complete` that arrived while the agent still waited for tools awaits * the spill of its last tool results, and stores them once that resolves, @@ -21509,6 +21532,27 @@ export class AgentFramework { const cm = agent.getContextManager(); const forPrimary: DeferredWrite[] = []; for (const msg of queued) { + // An entry whose earlier write may have landed (landing unknown) is + // looked up by its id first, as a drain does (writeDrainedInOrder). + // Found, it's neither stored again nor passed on below: the loop above + // has passed its stored copy if the agent hadn't read it. When the + // store can't be searched, it's written anyway, since no later flush + // will try again, and a copy its deferredWriteId tells apart costs + // less than a write lost without a trace. + if (msg.landing === 'unknown') { + const found = this.landedCopiesById(agent, msg.id); + if (found && found.length > 0) { + this.logDeferredWriteError(msg, `release of ${agent.name}`, undefined, + { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + continue; + } + if (found === undefined) { + console.error( + `[deferral] release of ${agent.name}: deferred write ${describeDeferredWrite(msg)} is stored though its ` + + `store couldn't be searched for an earlier copy, so it may now be stored twice`, + ); + } + } try { cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); } catch (err) { @@ -21785,6 +21829,17 @@ export class AgentFramework { return cm.queryMessages({ metadata: { deferredWriteId: writeId } }).messages[0] ?? null; } + /** The row a held write whose landing is unknown (DeferredWrite.landing) + * was already stored as, if its target's store holds one, found as + * landedRow finds it; null for any other held write. Throws when the + * store can't be searched. */ + private landedCopyOfHeld(msg: DeferredWrite): { agent: Agent; row: StoredMessage } | null { + if (msg.landing !== 'unknown') return null; + const agent = this.deliveryAgent(msg.forAgent); + const row = agent ? this.landedRow(agent.getContextManager(), msg.id) : null; + return agent && row ? { agent, row } : null; + } + /** Whether the agent has nothing left to read of a deferred write, on * evidence only: the row it landed as is at or below the agent's consumed * watermark on its current branch, or it's in neither the queue (drained @@ -21828,6 +21883,12 @@ export class AgentFramework { if (writeId !== undefined) { const held = this.heldWritesFor(agent).find((m) => m.id === writeId); if (held) { + // A held write whose landing is unknown may also be stored: its + // stored copy takes the edit too, so the flush that finds that copy + // carries the edit. It's looked up first, so a store that can't be + // searched throws here with nothing changed. + const copy = this.landedCopyOfHeld(held); + if (copy) copy.agent.getContextManager().editMessage(copy.row.id, content); held.content = content; if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); return; @@ -21843,7 +21904,13 @@ export class AgentFramework { throw new Error('No agents configured'); } const writeId = heldWriteId(id); - if (writeId !== undefined && this.heldWritesFor(agent).some((m) => m.id === writeId)) { + const held = writeId !== undefined ? this.heldWritesFor(agent).find((m) => m.id === writeId) : undefined; + if (held) { + // A held write whose landing is unknown may also be stored: its stored + // copy goes too. It's looked up first, so a store that can't be + // searched throws here with nothing changed. + const copy = this.landedCopyOfHeld(held); + if (copy) removeStoredMessage(copy.agent.getContextManager(), copy.row.id); this.deferredMessages = this.deferredMessages.filter((m) => m.id !== writeId); if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); return; @@ -21879,7 +21946,16 @@ export class AgentFramework { // the store's own query does it: it matches every row, then reverses, // then cuts. const { limit, reverse, ...match } = filter; - let messages: ModuleMessage[] = [...cm.queryMessages(match).messages.map(asModuleRow), ...held]; + // A held write whose landing is unknown may also be stored: it's shown + // once, as its held row, whose handle reaches its stored copy too. + const marked = new Set(this.heldWritesFor(agent).filter((m) => m.landing === 'unknown').map((m) => m.id)); + const shownHeld = new Set(held.map((row) => heldWriteId(row.id)).filter((id) => id !== undefined && marked.has(id))); + let messages: ModuleMessage[] = [ + ...cm.queryMessages(match).messages + .filter((m) => !shownHeld.has(m.metadata?.deferredWriteId as string)) + .map(asModuleRow), + ...held, + ]; const totalCount = messages.length; if (reverse) messages = messages.reverse(); if (limit !== undefined && limit < messages.length) messages = messages.slice(0, limit); diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index b85609ade..06cc111dc 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -15,7 +15,9 @@ * injected, since the next compile leaves it out too * (module-rows-held-from-residents). A provider that binds signed thinking * to its prefix (context-manager #155) refuses a reply whose stored prefix - * holds a message its request didn't. + * holds a message its request didn't. One write is the exception: one that + * threw after it was stored, whose landing couldn't be read, is carried at + * the flush that finds it by its id, after what the turn stored since. * * A message the stream carried has been read: RFC-006's consumed watermark * covers it, so a coalesced replacement can no longer remove it. Only a @@ -43,6 +45,7 @@ import type { ToolResult, } from '../src/index.js'; import { AgentFramework } from '../src/index.js'; +import { heldHandle } from '../src/held-writes.js'; import { AutobiographicalStrategy } from '@animalabs/context-manager'; import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; import type { ContentBlock } from '@animalabs/membrane'; @@ -612,9 +615,13 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); await framework.runUntilIdle(); } finally { console.error = error; } - assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); - assert.equal(windowOf(framework).filter((line) => line === 'Antra: look right').length, 1); - assert.ok(errors.some((e) => e.includes("threw, and its store couldn't be read back to tell whether it was stored") + assert.deepEqual(windowOf(framework).slice(-4), [ + 'user: ', + 'Antra: look left', + 'assistant: Done.', + 'Antra: look right', + ], 'stored once, where its first write landed, and what was queued behind it at the turn end'); + assert.ok(errors.some((e) => e.includes("threw, and its store couldn't be read to tell whether it was stored") && e.includes('strategy failed')), errors.join('\n')); assert.ok(errors.some((e) => e.includes('was found by its id: it was stored by an earlier write')), errors.join('\n')); await framework.stop(); @@ -625,15 +632,22 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { return framework as unknown as { writeDrainedInOrder( entries: DeferredEntry[], where: string, write: (msg: DeferredEntry) => string, + onLanded?: (msg: DeferredEntry, id: string) => void, ): Map; - handOffDeferredWrites(entries: DeferredEntry[]): void; + releaseWindow(agent: unknown): void; + addMessage(participant: string, content: ContentBlock[], metadata: unknown, opts: unknown): string; + editMessage(id: string, content: ContentBlock[]): void; + removeMessage(id: string): void; + queryMessages(filter: unknown): { messages: Array<{ content: ContentBlock[]; held?: true }> }; + isUnreadPlacement(placement: { agent: string; deferredId?: string }, by?: 'consumed' | 'shown'): boolean; + removePlacement(placement: { agent: string; deferredId?: string }): boolean; deferredMessages: DeferredEntry[]; - deferredScanFrom: Map; deferredWritesPersisted: boolean; }; } type DeferredEntry = { - id: string; seq: number; participant: string; content: ContentBlock[]; heldAt: number; landing?: 'unknown'; + id: string; seq: number; participant: string; content: ContentBlock[]; heldAt: number; + metadata?: unknown; forAgent?: string; landing?: 'unknown'; }; const entry = (id: string, said: string): DeferredEntry => ({ id, seq: 1, participant: 'Antra', content: text(said), heldAt: Date.now() }); /** A write that stores the entry as the framework does, under its id, then throws. */ @@ -655,15 +669,40 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const msg = entry('w-count', 'look left'); const error = console.error; console.error = () => {}; + let marked: string | undefined; + let writes = 0; + let written: Map; + const landed: Array<[string, string]> = []; try { internals.writeDrainedInOrder([msg], 'test', storedThenThrows(framework)); - assert.equal(msg.landing, 'unknown'); - let writes = 0; - const written = internals.writeDrainedInOrder([msg], 'test', () => { writes++; return 'again'; }); - assert.equal(writes, 0, 'found by its id, it is not written again'); - assert.ok(written.get(msg)?.id, 'it counts as written, under the id it was stored with'); + marked = msg.landing; + written = internals.writeDrainedInOrder([msg], 'test', (m) => { + writes++; + return framework.getAgent('assistant')!.getContextManager().addMessage(m.participant, m.content, { deferredWriteId: m.id } as never); + }, (m, id) => { landed.push([m.id, id]); }); } finally { console.error = error; } - assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1); + assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); + assert.equal(writes, 0, 'found by its id, it is not written again'); + const [copy] = framework.getAgent('assistant')!.getContextManager().queryMessages({ metadata: { deferredWriteId: 'w-count' } } as never).messages; + assert.equal(written.get(msg)?.id, copy!.id, 'it counts as written, under the id its copy was stored with'); + assert.deepEqual(landed, [['w-count', copy!.id]], 'what landed is noted, as for a write found by its readback'); + assert.equal(marked, 'unknown', 'the throw left its landing unknown'); + await framework.stop(); + }); + + it("a drained write whose landing is unknown, found by its id, is carried as its store holds it, not as it was queued", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + // A body whose first part landed before the throw stopped its write. + framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('part one'), { deferredWriteId: 'w-part' } as never); + const msg = { ...entry('w-part', 'part one'), content: [...text('part one'), ...text('part two')], landing: 'unknown' as const }; + const error = console.error; + console.error = () => {}; + let written: Map; + try { + written = internals.writeDrainedInOrder([msg], 'test', () => { throw new Error('written again'); }); + } finally { console.error = error; } + assert.deepEqual(written.get(msg)?.content, text('part one'), 'what the store holds'); await framework.stop(); }); @@ -689,16 +728,18 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { cm.queryMessages = function () { throw new Error('search failed'); }; const internals = drainInternals(framework); const first = { ...entry('w-held', 'look left'), landing: 'unknown' as const }; - const second = { ...entry('w-behind', 'look right'), seq: 2 }; + // Another target's entry, queued between them, isn't held by them. + const other = { ...entry('w-other', 'elsewhere'), seq: 2, forAgent: 'gone-fork' }; + const second = { ...entry('w-behind', 'look right'), seq: 3 }; const errors: string[] = []; const error = console.error; console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; - let writes = 0; + const writes: string[] = []; try { - const written = internals.writeDrainedInOrder([first, second], 'test', () => { writes++; return 'x'; }); - assert.equal(written.size, 0); + const written = internals.writeDrainedInOrder([first, other, second], 'test', (m) => { writes.push(m.id); return 'x'; }); + assert.deepEqual([...written.keys()].map((m) => m.id), ['w-other']); } finally { console.error = error; } - assert.equal(writes, 0, 'neither is written'); + assert.deepEqual(writes, ['w-other'], "only the other target's entry is written"); assert.deepEqual(internals.deferredMessages.map((m) => m.id), ['w-held', 'w-behind'], 'both are back in the queue'); assert.equal(first.landing, 'unknown'); assert.ok(errors.some((e) => e.includes("still can't be looked up by its id")), errors.join('\n')); @@ -706,20 +747,182 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); - it("a hand-off of an entry whose landing is unknown has boot scan its agent's whole store", async () => { + it("a drained write whose landing is unknown, for an agent that isn't registered, is dropped as an unmarked one is, not held", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const gone = { ...entry('w-gone', 'look left'), forAgent: 'gone-fork', landing: 'unknown' as const }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + let written: Map; + try { + written = internals.writeDrainedInOrder([gone], 'test', (m) => internals.addMessage(m.participant, m.content, m.metadata, { + fromQueue: { id: m.id, seq: m.seq, heldAt: m.heldAt }, forAgent: m.forAgent, + })); + } finally { console.error = error; } + assert.ok(written.has(gone), 'it leaves the drain'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'nothing is held'); + assert.ok(!errors.some((e) => e.includes("still can't be looked up by its id")), errors.join('\n')); + await framework.stop(); + }); + + it("a write whose landing is unknown keeps its mark across a restart, so it isn't stored a second time after it", async () => { const framework = await createFramework(); const internals = drainInternals(framework); internals.deferredWritesPersisted = true; - const cm = framework.getAgent('assistant')!.getContextManager(); - cm.addMessage('Antra', text('earlier')); - internals.handOffDeferredWrites([entry('w-known', 'one')]); - const from = internals.deferredScanFrom.get('assistant'); - assert.ok(from !== undefined && from > 0, `a known entry is scanned for from the store's count, ${from}`); - internals.handOffDeferredWrites([{ ...entry('w-unknown', 'two'), seq: 2, landing: 'unknown' }]); - assert.equal(internals.deferredScanFrom.get('assistant'), 0); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage(participant: string, content: ContentBlock[], metadata?: unknown): string; + queryMessages: (...args: unknown[]) => unknown; + }; + // Its first write landed; the hand-off at stop() records the store's + // count after it, so boot's check for landed writes starts past it. + cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-rt' }); + internals.deferredMessages.push({ ...entry('w-rt', 'look left'), landing: 'unknown' }); + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { await framework.stop(); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-rt', 'unknown']], 'held through stop()'); + + const reopened = await createFramework(); + assert.equal(windowOf(reopened).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); + await reopened.stop(); + }); + + it("a released window doesn't store a write whose landing is unknown a second time, when its store holds it", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager(); + const copies = (id: string): number => cm.queryMessages({ metadata: { deferredWriteId: id } } as never).messages.length; + cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-rel' } as never); + internals.deferredMessages.push({ ...entry('w-rel', 'look left'), landing: 'unknown' }); + internals.deferredMessages.push({ ...entry('w-rel-2', 'look right'), seq: 2, landing: 'unknown' }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.releaseWindow(agent); } finally { console.error = error; } + assert.equal(copies('w-rel'), 1, 'found by its id, it is not stored again'); + assert.equal(copies('w-rel-2'), 1, 'not found, it is stored'); + assert.ok(errors.some((e) => e.includes('release of assistant') && e.includes('was found by its id')), errors.join('\n')); + await framework.stop(); + }); + + it("a released window doesn't pass a failed reply on a second time when its store holds the reply's copy", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager(); + const failed = { kind: 'discord-send-failed', channelId: 'c1', reason: 'gone', textLen: 5, outcome: 'failed' }; + // Its earlier write landed. The release's first pass reads that copy as + // unread and passes it on; the queued entry mustn't be passed again. + cm.addMessage('assistant', text('hello'), { ...failed, deferredWriteId: 'w-fail' } as never); + internals.deferredMessages.push({ ...entry('w-fail', 'hello'), participant: 'assistant', metadata: failed, landing: 'unknown' }); + const error = console.error; + console.error = () => {}; + try { internals.releaseWindow(agent); } finally { console.error = error; } + const notices = cm.queryMessages({} as never).messages.filter((m) => (m.metadata as { speaker?: unknown } | undefined)?.speaker === 'assistant'); + assert.equal(notices.length, 1, 'the primary is told once'); await framework.stop(); }); + it("a released window stores a write whose landing is unknown while its store can't be searched, saying it may now be stored twice", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + internals.deferredMessages.push({ ...entry('w-blind', 'look left'), landing: 'unknown' }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.releaseWindow(agent); } finally { console.error = error; } + cm.queryMessages = query; + assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1, 'stored'); + assert.ok(errors.some((e) => e.includes('w-blind') && e.includes('may now be stored twice')), errors.join('\n')); + await framework.stop(); + }); + + /** A held write whose landing is unknown, whose first write did land. */ + function heldAndStored(framework: AgentFramework, id: string, said: string): void { + framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text(said), { deferredWriteId: id } as never); + drainInternals(framework).deferredMessages.push({ ...entry(id, said), landing: 'unknown' }); + } + const copiesOf = (framework: AgentFramework, id: string): string[] => framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages + .map((m) => m.content.map((b) => (b as { text?: string }).text ?? '').join('')); + + it("a module's edit of a held write whose landing is unknown reaches its stored copy, so the flush that finds it carries the edit", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-ed', 'look left'); + internals.editMessage(heldHandle('w-ed'), text('look up')); + const error = console.error; + console.error = () => {}; + let written: Map; + try { + written = internals.writeDrainedInOrder(internals.deferredMessages.splice(0), 'test', () => { throw new Error('written again'); }); + } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, 'w-ed'), ['look up'], 'stored once, edited'); + assert.deepEqual([...written.values()].map((w) => w.content), [text('look up')], 'carried as edited'); + await framework.stop(); + }); + + it("a module's removal of a held write whose landing is unknown removes its stored copy too", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-rm', 'look left'); + internals.removeMessage(heldHandle('w-rm')); + assert.deepEqual(copiesOf(framework, 'w-rm'), []); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), []); + await framework.stop(); + }); + + it("a module's edit or removal of a held write whose landing is unknown throws with nothing changed while its store can't be searched", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-blind', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + try { + assert.throws(() => internals.editMessage(heldHandle('w-blind'), text('look up')), /search failed/); + assert.throws(() => internals.removeMessage(heldHandle('w-blind')), /search failed/); + } finally { cm.queryMessages = query; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content]), [['w-blind', text('look left')]], 'still held, unedited'); + assert.deepEqual(copiesOf(framework, 'w-blind'), ['look left'], 'its copy unchanged'); + await framework.stop(); + }); + + it("a module's query shows a held write whose landing is unknown once, as its held row", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-q', 'look left'); + const rows = internals.queryMessages({}).messages + .filter((m) => m.content.some((b) => (b as { text?: string }).text === 'look left')) + .map((m) => (m.held ? 'held' : 'stored')); + assert.deepEqual(rows, ['held']); + await framework.stop(); + }); + + it("the coalescer removes a held write whose landing is unknown with its stored copy only where that copy may never have been shown", async () => { + for (const shown of [false, true]) { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + if (!shown) agent.markContextShown(); + heldAndStored(framework, 'w-co', 'look left'); + if (shown) agent.markContextShown(); + const placement = { agent: 'assistant', deferredId: 'w-co' }; + assert.equal(internals.isUnreadPlacement({ ...placement }, 'shown'), !shown, `unshown as its copy is (${shown ? 'shown' : 'not shown'})`); + assert.equal(internals.removePlacement({ ...placement }), !shown); + assert.deepEqual(copiesOf(framework, 'w-co'), shown ? ['look left'] : [], shown ? 'a copy that may have been shown stays' : 'its copy goes'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), shown ? ['w-co'] : [], shown ? 'nothing is removed' : 'and the held entry'); + await framework.stop(); + } + }); + /** Shard a message whose first block is `marker` into one shard per block, * as a strategy's ingress chunking does. */ function shardOn(framework: AgentFramework, marker: string): void { From bb1f0e3839f7efcd8c54e7e65d9236ff003c78d9 Mon Sep 17 00:00:00 2001 From: Nigel-2654 Date: Mon, 19 Oct 2026 19:58:47 -0400 Subject: [PATCH 3/6] fix(deferral): settle a marked write wherever held writes are read or changed Our internal review of 1741955c found the two copies that round kept side by side, a marked held entry and its stored copy, disagreeing wherever a path reached only one: an edit or removal by the copy's own id never reached the queued entry, so a removed write was written again; the wake check read any queued write as unread; a sharded copy was seen by its first shard only; an edit or removal that committed and then threw left the two apart; the tune-out backlog showed the write twice; and the handle and the copy's id read as different rows. So a marked entry is now settled wherever a path that reads or changes held writes meets it (settleUnknownLandings): looked up by its id in its target's store. Found, the write landed: it leaves the queue, logged by its id, and is a stored write from then on, so the existing paths for a landed write apply to it, and it isn't carried into a turn. Not found, the mark goes, and it's an ordinary held write. Where the store can't be searched, it stays marked, and each path reads that doubt by its own rule: an edit or removal throws with nothing changed (refuseUnsettled), the coalescer reads it as read and as shown and doesn't remove it, and the wake check reads it as unread. A change the settling makes to the queue is persisted, as an edit's is. The paths that settle: heldWritesFor, so a module's reads, edits and removals, by a handle or a stored id; the coalescer's isUnreadPlacement and removePlacement; deferredWriteReadOrGone; and the tune-out backlog's held writes. Release settles in its own loop, in the released agent's store, and now logs a store made without its lookup only once the store succeeded. The copy-side code of the round before goes: the edit, removal and query of both copies, and the coalescer's checks on the copy. The docs that said a queued write is unstored, or unshown, name the doubt: the module API's held rows, editMessage and removeMessage, the coalescer host's isUnshown, removeOccupant's walk and isUnreadPlacement. Tests: the handle tests now check a write that landed is met as its stored row, by its handle and by its stored id, for an edit, a removal, a query and getMessage; a held write the store doesn't hold is edited as an ordinary one; settling inside a read persists the queue; the wake check reads a copy the agent read as read; the coalescer reads and removes a settled write as its stored placement, and a doubt as read and shown without removing it; the tune-out backlog leaves a write that landed to the stored rows; the restart test runs with and without a landed copy; and release leaves a failed reply the agent had read unpassed. Co-Authored-By: Claude Opus 5.5 --- .../deferred-write-landing-unknown.fixed.md | 17 +- src/framework.ts | 207 ++++++++-------- src/mcpl/push-coalescer.ts | 12 +- src/types/module.ts | 12 +- test/deferred-window-only.test.ts | 222 +++++++++++++----- test/module-rows-held-from-residents.test.ts | 21 ++ 6 files changed, 323 insertions(+), 168 deletions(-) diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md index 8627b2a23..c2d7811fa 100644 --- a/changelog.d/deferred-write-landing-unknown.fixed.md +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -8,15 +8,18 @@ for that agent, as a write that keeps failing does. Such an entry's mark is persisted with the queue, so after a restart it's still looked up before it's written. - - A write found this way is carried at the flush that finds it, after - what the turn stored since, though the store holds it before those, so - later requests place it earlier. + - A write found this way at a flush is carried there, after what the turn + stored since, though the store holds it before those, so later requests + place it earlier. + - Wherever held writes are read or changed, by a module's context, the + coalescer, a wake's check or a tune-out's backlog, such a write is + looked up first. One that landed is a stored message from then on, read + and changed as one, and isn't carried into the turn; one that didn't is + an ordinary held write. Where the store can't be searched, a module's + edit or removal of it throws with nothing changed, and the coalescer + reads it as possibly read and shown. - A conversation fork's or ephemeral run's queue, stored when it's released, looks such an entry up the same way: found, it's neither stored again nor passed to the primary as a failed reply; where the store can't be searched, it's stored anyway, and the log says it may now be stored twice. - - A module's handle to such a held write reaches its stored copy too: an - edit or a removal applies to both, and throws with nothing changed when - the store can't be searched, and a query shows the write once. The - coalescer removes the stored copy only where nothing may have shown it. diff --git a/src/framework.ts b/src/framework.ts index e3b105cef..5f72f4acc 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -741,8 +741,10 @@ interface DeferredWrite { heldAt: number; /** Set when a write of this entry threw and its store couldn't be read * to tell whether it landed (writeDrainedInOrder): before it's - * written again, the store is searched for its id. Persisted with the - * entry, so after a restart it's still looked up first. */ + * written again, the store is searched for its id, and a path that reads + * or changes held writes settles it first (settleUnknownLandings). + * Persisted with the entry, so after a restart it's still looked up + * first. */ landing?: 'unknown'; } const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; @@ -3344,10 +3346,15 @@ export class AgentFramework { ? framework.agents.get(framework.primaryAgentName) : undefined; return primary ? primary.getContextManager().getAllMessages() : []; }, - getDeferredMessages: () => framework.deferredMessages - .filter((m) => (m.forAgent ?? framework.primaryAgentName) === framework.primaryAgentName) - .sort(bySeq) - .map((m) => ({ participant: m.participant, content: m.content, metadata: m.metadata as Record | undefined })), + getDeferredMessages: () => { + const forPrimary = (m: DeferredWrite): boolean => (m.forAgent ?? framework.primaryAgentName) === framework.primaryAgentName; + // Settled first, so a write that landed is in the stored rows + // alone (settleUnknownLandings). + framework.settleUnknownLandings(forPrimary); + return framework.deferredMessages.filter(forPrimary) + .sort(bySeq) + .map((m) => ({ participant: m.participant, content: m.content, metadata: m.metadata as Record | undefined })); + }, // Every reader of the shared slot compiles the dump with its own // strategy, so the dump fits the smallest of their bounds. residentMessageChars: () => { @@ -13317,20 +13324,13 @@ export class AgentFramework { /** Unread (the coalescer's isUnread) by the consumed watermark, or, by the * shown watermark, never possibly shown (its isUnshown). A placement still - * in the deferred queue is both. */ + * in the deferred queue is both, once settled (settleUnknownLandings): one + * whose write may already be stored, its store unsearchable, is a doubt, + * which counts as read and as shown. */ private isUnreadPlacement(p: CoalescingPlacement, by: 'consumed' | 'shown' = 'consumed'): boolean { + if (p.deferredId) this.settleUnknownLandings((m) => m.id === p.deferredId); this.resolvePlacement(p); - if (p.deferredId) { - const held = this.deferredMessages.find((m) => m.id === p.deferredId); - if (!held) return false; - if (held.landing !== 'unknown') return true; - // A held write whose landing is unknown may also be stored, and that - // copy may have been read or shown: its own row answers. A store that - // can't be searched is a doubt, which counts as read and as shown. - let copy: { agent: Agent; row: StoredMessage } | null; - try { copy = this.landedCopyOfHeld(held); } catch { return false; } - return copy ? this.isUnreadStoredMessage(copy.agent, copy.row.id, by) : true; - } + if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId && m.landing !== 'unknown'); if (!p.messageId) return false; const agent = this.agents.get(p.agent); return !!agent && this.isUnreadStoredMessage(agent, p.messageId, by); @@ -13385,30 +13385,15 @@ export class AgentFramework { } private removePlacement(p: CoalescingPlacement): boolean { + // Settled first, so a write that landed is removed as the stored + // placement it is; one still a doubt, its store unsearchable, isn't + // removed, since it may already be stored. + if (p.deferredId) this.settleUnknownLandings((m) => m.id === p.deferredId); this.resolvePlacement(p); if (p.deferredId) { const held = this.deferredMessages.find((m) => m.id === p.deferredId); - if (!held) return false; - // A held write whose landing is unknown may also be stored: its stored - // copy goes too, but only where that row's own check says nothing may - // have shown it (the held entry reads as unshown whatever its copy - // went through). Where something may have, or the store can't be - // searched, nothing is removed. - if (held.landing === 'unknown') { - let copy: { agent: Agent; row: StoredMessage } | null; - try { copy = this.landedCopyOfHeld(held); } catch { return false; } - if (copy) { - if (!this.isUnreadStoredMessage(copy.agent, copy.row.id, 'shown')) return false; - try { - removeStoredMessage(copy.agent.getContextManager(), copy.row.id); - } catch (err) { - console.error(`[coalescing] could not remove unread message ${copy.row.id} for ${copy.agent.name}:`, err); - return false; - } - this.emitTrace({ type: 'message:removed', messageId: copy.row.id, source: 'mcpl:coalescing' }); - } - } - this.deferredMessages = this.deferredMessages.filter((m) => m.id !== p.deferredId); + if (!held || held.landing === 'unknown') return false; + this.deferredMessages = this.deferredMessages.filter((m) => m !== held); if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); return true; } @@ -21533,28 +21518,27 @@ export class AgentFramework { const forPrimary: DeferredWrite[] = []; for (const msg of queued) { // An entry whose earlier write may have landed (landing unknown) is - // looked up by its id first, as a drain does (writeDrainedInOrder). - // Found, it's neither stored again nor passed on below: the loop above - // has passed its stored copy if the agent hadn't read it. When the - // store can't be searched, it's written anyway, since no later flush - // will try again, and a copy its deferredWriteId tells apart costs - // less than a write lost without a trace. - if (msg.landing === 'unknown') { - const found = this.landedCopiesById(agent, msg.id); - if (found && found.length > 0) { - this.logDeferredWriteError(msg, `release of ${agent.name}`, undefined, - { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); - continue; - } + // settled here as elsewhere (settleUnknownLandings), in the released + // agent's own store. Found, it landed: it's neither stored again nor + // passed on below, since the loop above has passed its stored copy if + // it's a failed reply the agent hadn't read. When the store can't be + // searched, it's stored anyway, since no later flush will try again, + // and a copy its deferredWriteId tells apart costs less than a write + // lost without a trace. + const found = msg.landing === 'unknown' ? this.landedCopiesById(agent, msg.id) : []; + if (found && found.length > 0) { + this.logDeferredWriteError(msg, `release of ${agent.name}`, undefined, + { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + continue; + } + try { + cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); if (found === undefined) { console.error( - `[deferral] release of ${agent.name}: deferred write ${describeDeferredWrite(msg)} is stored though its ` + + `[deferral] release of ${agent.name}: deferred write ${describeDeferredWrite(msg)} was stored though its ` + `store couldn't be searched for an earlier copy, so it may now be stored twice`, ); } - } - try { - cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); } catch (err) { console.error( `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + @@ -21787,12 +21771,12 @@ export class AgentFramework { * (queuedFailedReplyForPrimary). */ private heldWritesFor(reader: Agent): DeferredWrite[] { - return this.deferredMessages - .filter((m) => { - const target = this.deliveryAgent(m.forAgent); - return !!target && (target === reader || (this.isOrdinaryResident(reader) && this.isOrdinaryResident(target))); - }) - .sort(bySeq); + const heldFor = (m: DeferredWrite): boolean => { + const target = this.deliveryAgent(m.forAgent); + return !!target && (target === reader || (this.isOrdinaryResident(reader) && this.isOrdinaryResident(target))); + }; + this.settleUnknownLandings(heldFor); + return this.deferredMessages.filter(heldFor).sort(bySeq); } /** @@ -21829,15 +21813,49 @@ export class AgentFramework { return cm.queryMessages({ metadata: { deferredWriteId: writeId } }).messages[0] ?? null; } - /** The row a held write whose landing is unknown (DeferredWrite.landing) - * was already stored as, if its target's store holds one, found as - * landedRow finds it; null for any other held write. Throws when the - * store can't be searched. */ - private landedCopyOfHeld(msg: DeferredWrite): { agent: Agent; row: StoredMessage } | null { - if (msg.landing !== 'unknown') return null; - const agent = this.deliveryAgent(msg.forAgent); - const row = agent ? this.landedRow(agent.getContextManager(), msg.id) : null; - return agent && row ? { agent, row } : null; + /** + * Settle held writes whose landing is unknown (DeferredWrite.landing), + * among those `which` picks, before a path reads or changes held writes: + * each is looked up by its id in its target's store (landedCopiesById). + * Found, the write landed: it leaves the queue, logged by its id, and is a + * stored write from then on, so the paths for a landed write apply to it. + * It isn't carried into a turn, as a flush that finds it carries it: the + * agent reads it at its next compile, where the store holds it. Not + * found, the mark goes, and it's an ordinary held write. Where the store + * can't be searched, it stays marked, a doubt each path reads by its own + * rule. A change to the queue is persisted, as an edit's is. + */ + private settleUnknownLandings(which: (msg: DeferredWrite) => boolean): void { + let changed = false; + for (const msg of this.deferredMessages.filter((m) => m.landing === 'unknown' && which(m))) { + // No registered agent: no store it could have landed in. A flush + // writes it, and addMessage drops it. + const agent = this.deliveryAgent(msg.forAgent); + if (!agent) continue; + const found = this.landedCopiesById(agent, msg.id); + if (found === undefined) continue; + if (found.length > 0) { + this.deferredMessages = this.deferredMessages.filter((m) => m !== msg); + this.logDeferredWriteError(msg, `held writes for ${agent.name}`, undefined, + { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + } else { + delete msg.landing; + } + changed = true; + } + if (changed && (this.quiesced || this.deferredWritesPersisted)) this.persistDeferredWrites(); + } + + /** A held write whose landing stays unknown after settling, its store + * unsearchable, can't be edited or removed: it may already be stored, and + * changing only the queued entry could leave its stored copy as it was. + * Throws with nothing changed. */ + private refuseUnsettled(msg: DeferredWrite): void { + if (msg.landing !== 'unknown') return; + throw new Error( + `held write ${msg.id} may already be stored, and its store couldn't be searched to tell, ` + + `so nothing was changed; try again once the store can be read`, + ); } /** Whether the agent has nothing left to read of a deferred write, on @@ -21852,6 +21870,9 @@ export class AgentFramework { * finds nothing new costs less than a write no wake brings the agent to * read. */ private deferredWriteReadOrGone(agent: Agent, writeId: string): boolean { + // Settled first, so a write that landed is judged by its stored row; one + // still a doubt stays queued, which reads as unread. + this.settleUnknownLandings((m) => m.id === writeId); if (this.deferredMessages.some((m) => m.id === writeId)) return false; try { const cm = agent.getContextManager(); @@ -21879,20 +21900,16 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } + // The held writes are settled first (heldWritesFor), by a stored id as + // by a handle: a held write that landed is a stored row from then on, + // and the edit reaches it as one. const writeId = heldWriteId(id); - if (writeId !== undefined) { - const held = this.heldWritesFor(agent).find((m) => m.id === writeId); - if (held) { - // A held write whose landing is unknown may also be stored: its - // stored copy takes the edit too, so the flush that finds that copy - // carries the edit. It's looked up first, so a store that can't be - // searched throws here with nothing changed. - const copy = this.landedCopyOfHeld(held); - if (copy) copy.agent.getContextManager().editMessage(copy.row.id, content); - held.content = content; - if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); - return; - } + const held = this.heldWritesFor(agent).find((m) => writeId !== undefined && m.id === writeId); + if (held) { + this.refuseUnsettled(held); + held.content = content; + if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); + return; } const landed = writeId !== undefined ? this.landedRow(agent.getContextManager(), writeId) : null; agent.getContextManager().editMessage(landed?.id ?? id, content); @@ -21903,14 +21920,13 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } + // The held writes are settled first (heldWritesFor), by a stored id as + // by a handle, so a held write that landed is removed as the stored row + // it is, and isn't written again at the next flush. const writeId = heldWriteId(id); - const held = writeId !== undefined ? this.heldWritesFor(agent).find((m) => m.id === writeId) : undefined; + const held = this.heldWritesFor(agent).find((m) => writeId !== undefined && m.id === writeId); if (held) { - // A held write whose landing is unknown may also be stored: its stored - // copy goes too. It's looked up first, so a store that can't be - // searched throws here with nothing changed. - const copy = this.landedCopyOfHeld(held); - if (copy) removeStoredMessage(copy.agent.getContextManager(), copy.row.id); + this.refuseUnsettled(held); this.deferredMessages = this.deferredMessages.filter((m) => m.id !== writeId); if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); return; @@ -21946,16 +21962,7 @@ export class AgentFramework { // the store's own query does it: it matches every row, then reverses, // then cuts. const { limit, reverse, ...match } = filter; - // A held write whose landing is unknown may also be stored: it's shown - // once, as its held row, whose handle reaches its stored copy too. - const marked = new Set(this.heldWritesFor(agent).filter((m) => m.landing === 'unknown').map((m) => m.id)); - const shownHeld = new Set(held.map((row) => heldWriteId(row.id)).filter((id) => id !== undefined && marked.has(id))); - let messages: ModuleMessage[] = [ - ...cm.queryMessages(match).messages - .filter((m) => !shownHeld.has(m.metadata?.deferredWriteId as string)) - .map(asModuleRow), - ...held, - ]; + let messages: ModuleMessage[] = [...cm.queryMessages(match).messages.map(asModuleRow), ...held]; const totalCount = messages.length; if (reverse) messages = messages.reverse(); if (limit !== undefined && limit < messages.length) messages = messages.slice(0, limit); diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 45c87e0e0..b16e91003 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -260,8 +260,9 @@ export interface CoalescerHost { /** True iff the placement still exists and no model request has included it. */ isUnread(placement: CoalescingPlacement): boolean; /** True iff nothing may have shown the placement to the agent: it is still - * in the deferred queue, or stored where no committing compile has begun - * to assemble it and no tool boundary has injected past it. Unlike + * in the deferred queue, known not to be stored, or stored where no + * committing compile has begun to assemble it and no tool boundary has + * injected past it. Unlike * isUnread, a compile counts from its start, and a boundary counts * whether or not its stream says it carries injected messages: a doubt * counts as shown. */ @@ -660,9 +661,10 @@ export class PushCoalescer { /** * Remove the occupant and the run it builds on, newest first, auditing each * as `kind`. Every member is unread: refreshOccupant has just run, under the - * lock, and the host's removal doesn't itself refuse a read message. A - * member the host can't remove may be read, so the walk stops there, leaving - * it and the renders it builds on, and the history becomes `unknown`. + * lock. The host's removal may still refuse one, such as a held write the + * host can't tell isn't already stored. A member the host + * can't remove may be read, so the walk stops there, leaving it and the + * renders it builds on, and the history becomes `unknown`. * Returns false then. It doesn't touch `state.occupant`, which still names * the members it removed: each caller clears it, so what the walk left is * history from then on, as an occupant whose removal failed always was. diff --git a/src/types/module.ts b/src/types/module.ts index 30d5b1927..5a42c84f7 100644 --- a/src/types/module.ts +++ b/src/types/module.ts @@ -40,7 +40,9 @@ export type ModuleMessage = StoredMessage & { * and `timestamp` are provisional: the sequence comes after every stored * row's, the timestamp is when the framework held the write (for a write * an older release held, when its queue was read back), and both change - * when it lands. Absent on a stored row. + * when it lands. A held write whose earlier write may already be stored + * is looked up first, and answered as the stored row it is when the store + * holds it. Absent on a stored row. */ held?: true; /** @@ -311,13 +313,17 @@ export interface ModuleContext { /** * Edit a message, by its id or a held write's handle. A held write is - * stored with the new content. + * stored with the new content. One whose earlier write may already be + * stored, where the store can't be searched to tell, throws with nothing + * changed. */ editMessage(id: MessageId, content: ContentBlock[]): void; /** * Remove a message, by its id or a held write's handle. A held write is - * dropped, so the agent never reads it. + * dropped, so the agent never reads it. One whose earlier write may + * already be stored, where the store can't be searched to tell, throws + * with nothing changed. */ removeMessage(id: MessageId): void; diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 06cc111dc..8a4833462 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -639,6 +639,8 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { editMessage(id: string, content: ContentBlock[]): void; removeMessage(id: string): void; queryMessages(filter: unknown): { messages: Array<{ content: ContentBlock[]; held?: true }> }; + getMessage(id: string): { held?: true; content: ContentBlock[] } | null; + deferredWriteReadOrGone(agent: unknown, writeId: string): boolean; isUnreadPlacement(placement: { agent: string; deferredId?: string }, by?: 'consumed' | 'shown'): boolean; removePlacement(placement: { agent: string; deferredId?: string }): boolean; deferredMessages: DeferredEntry[]; @@ -766,28 +768,32 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); - it("a write whose landing is unknown keeps its mark across a restart, so it isn't stored a second time after it", async () => { - const framework = await createFramework(); - const internals = drainInternals(framework); - internals.deferredWritesPersisted = true; - const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { - addMessage(participant: string, content: ContentBlock[], metadata?: unknown): string; - queryMessages: (...args: unknown[]) => unknown; - }; - // Its first write landed; the hand-off at stop() records the store's - // count after it, so boot's check for landed writes starts past it. - cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-rt' }); - internals.deferredMessages.push({ ...entry('w-rt', 'look left'), landing: 'unknown' }); - cm.queryMessages = () => { throw new Error('search failed'); }; - const error = console.error; - console.error = () => {}; - try { await framework.stop(); } finally { console.error = error; } - assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-rt', 'unknown']], 'held through stop()'); + for (const landed of [true, false]) { + it(`a write whose landing is unknown keeps its mark across a restart, so it's stored once after it (${landed ? 'its first write landed' : 'nothing landed'})`, async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + internals.deferredWritesPersisted = true; + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage(participant: string, content: ContentBlock[], metadata?: unknown): string; + queryMessages: (...args: unknown[]) => unknown; + }; + // Where its first write landed, the hand-off at stop() records the + // store's count after it, so boot's check for landed writes starts + // past it and only the restored mark keeps it from a second store. + if (landed) cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-rt' }); + internals.deferredMessages.push({ ...entry('w-rt', 'look left'), landing: 'unknown' }); + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { await framework.stop(); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-rt', 'unknown']], 'held through stop()'); - const reopened = await createFramework(); - assert.equal(windowOf(reopened).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); - await reopened.stop(); - }); + const reopened = await createFramework(); + assert.equal(windowOf(reopened).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); + assert.deepEqual(drainInternals(reopened).deferredMessages.map((m) => m.id), [], 'and no longer queued'); + await reopened.stop(); + }); + } it("a released window doesn't store a write whose landing is unknown a second time, when its store holds it", async () => { const framework = await createFramework(); @@ -826,6 +832,26 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); + it("a released window doesn't pass on a failed reply the agent had read, when its store holds the reply's copy", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager(); + const failed = { kind: 'discord-send-failed', channelId: 'c1', reason: 'gone', textLen: 5, outcome: 'failed' }; + cm.addMessage('assistant', text('hello'), { ...failed, deferredWriteId: 'w-read' } as never); + // The agent read its copy: the release's first pass skips it, as main + // skips a failed reply the agent heard, and the queued entry mustn't + // pass it on either. + agent.markContextConsumed(); + internals.deferredMessages.push({ ...entry('w-read', 'hello'), participant: 'assistant', metadata: failed, landing: 'unknown' }); + const error = console.error; + console.error = () => {}; + try { internals.releaseWindow(agent); } finally { console.error = error; } + const notices = cm.queryMessages({} as never).messages.filter((m) => (m.metadata as { speaker?: unknown } | undefined)?.speaker === 'assistant'); + assert.equal(notices.length, 0, "the primary isn't told"); + await framework.stop(); + }); + it("a released window stores a write whose landing is unknown while its store can't be searched, saying it may now be stored twice", async () => { const framework = await createFramework(); const internals = drainInternals(framework); @@ -853,28 +879,64 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { .queryMessages({ metadata: { deferredWriteId: id } } as never).messages .map((m) => m.content.map((b) => (b as { text?: string }).text ?? '').join('')); - it("a module's edit of a held write whose landing is unknown reaches its stored copy, so the flush that finds it carries the edit", async () => { + it("a module's edit of a held write whose landing is unknown settles it first, so the edit reaches the stored row it landed as", async () => { + for (const by of ['handle', 'stored id'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-${by === 'handle' ? 'h' : 's'}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages; + const error = console.error; + console.error = () => {}; + try { internals.editMessage(by === 'handle' ? heldHandle(id) : copy!.id, text('look up')); } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, id), ['look up'], `by its ${by}: stored once, edited`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `by its ${by}: no longer held`); + await framework.stop(); + } + }); + + it("a module's removal of a held write whose landing is unknown settles it first, so the stored row goes and nothing is written again", async () => { + for (const by of ['handle', 'stored id'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-${by === 'handle' ? 'h' : 's'}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages; + const error = console.error; + console.error = () => {}; + try { internals.removeMessage(by === 'handle' ? heldHandle(id) : copy!.id); } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, id), [], `by its ${by}`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `by its ${by}: nothing left to write`); + await framework.stop(); + } + }); + + it("a module's edit of a held write whose landing is unknown, which its store doesn't hold, settles it as an ordinary held write and edits it", async () => { const framework = await createFramework(); const internals = drainInternals(framework); - heldAndStored(framework, 'w-ed', 'look left'); - internals.editMessage(heldHandle('w-ed'), text('look up')); - const error = console.error; - console.error = () => {}; - let written: Map; - try { - written = internals.writeDrainedInOrder(internals.deferredMessages.splice(0), 'test', () => { throw new Error('written again'); }); - } finally { console.error = error; } - assert.deepEqual(copiesOf(framework, 'w-ed'), ['look up'], 'stored once, edited'); - assert.deepEqual([...written.values()].map((w) => w.content), [text('look up')], 'carried as edited'); + internals.deferredMessages.push({ ...entry('w-absent-ed', 'look left'), landing: 'unknown' }); + internals.editMessage(heldHandle('w-absent-ed'), text('look up')); + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content, m.landing]), [['w-absent-ed', text('look up'), undefined]]); + assert.deepEqual(copiesOf(framework, 'w-absent-ed'), [], 'nothing stored'); await framework.stop(); }); - it("a module's removal of a held write whose landing is unknown removes its stored copy too", async () => { + it("settling a held write whose landing is unknown, inside a read, persists the queue as an edit does", async () => { const framework = await createFramework(); - const internals = drainInternals(framework); - heldAndStored(framework, 'w-rm', 'look left'); - internals.removeMessage(heldHandle('w-rm')); - assert.deepEqual(copiesOf(framework, 'w-rm'), []); + const internals = drainInternals(framework) as ReturnType & { persistDeferredWrites(): void }; + internals.deferredWritesPersisted = true; + heldAndStored(framework, 'w-persist', 'look left'); + let persisted = 0; + const persist = internals.persistDeferredWrites; + internals.persistDeferredWrites = function (this: unknown) { persisted++; return persist.call(this); }; + const error = console.error; + console.error = () => {}; + try { internals.queryMessages({}); } finally { console.error = error; } + assert.equal(persisted, 1, 'the queue without it is persisted'); assert.deepEqual(internals.deferredMessages.map((m) => m.id), []); await framework.stop(); }); @@ -886,43 +948,97 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; const query = cm.queryMessages; cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; try { - assert.throws(() => internals.editMessage(heldHandle('w-blind'), text('look up')), /search failed/); - assert.throws(() => internals.removeMessage(heldHandle('w-blind')), /search failed/); - } finally { cm.queryMessages = query; } - assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content]), [['w-blind', text('look left')]], 'still held, unedited'); + assert.throws(() => internals.editMessage(heldHandle('w-blind'), text('look up')), /couldn't be searched/); + assert.throws(() => internals.removeMessage(heldHandle('w-blind')), /couldn't be searched/); + } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content, m.landing]), [['w-blind', text('look left'), 'unknown']], 'still held, unedited'); assert.deepEqual(copiesOf(framework, 'w-blind'), ['look left'], 'its copy unchanged'); await framework.stop(); }); - it("a module's query shows a held write whose landing is unknown once, as its held row", async () => { + it("a module's query and getMessage meet a held write whose landing is unknown, settled, as the one stored row it landed as", async () => { const framework = await createFramework(); const internals = drainInternals(framework); + framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('look right')); heldAndStored(framework, 'w-q', 'look left'); - const rows = internals.queryMessages({}).messages - .filter((m) => m.content.some((b) => (b as { text?: string }).text === 'look left')) - .map((m) => (m.held ? 'held' : 'stored')); - assert.deepEqual(rows, ['held']); + const error = console.error; + console.error = () => {}; + let rows: string[]; + let byHandle: { held?: true; content: ContentBlock[] } | null; + try { + rows = internals.queryMessages({}).messages + .filter((m) => m.content.some((b) => ['look left', 'look right'].includes((b as { text?: string }).text ?? ''))) + .map((m) => `${m.held ? 'held' : 'stored'}: ${m.content.map((b) => (b as { text?: string }).text).join('')}`); + byHandle = internals.getMessage(heldHandle('w-q')); + } finally { console.error = error; } + assert.deepEqual(rows, ['stored: look right', 'stored: look left']); + assert.equal(byHandle?.held, undefined, 'its handle names the stored row'); await framework.stop(); }); - it("the coalescer removes a held write whose landing is unknown with its stored copy only where that copy may never have been shown", async () => { + it("deferredWriteReadOrGone settles a held write whose landing is unknown, so a copy the agent has read reads as read", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + heldAndStored(framework, 'w-wake', 'look left'); + agent.markContextConsumed(); + const error = console.error; + console.error = () => {}; + try { + assert.equal(internals.deferredWriteReadOrGone(agent, 'w-wake'), true); + } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), []); + await framework.stop(); + }); + + it("the coalescer reads and removes a held write whose landing is unknown, once settled, as the stored placement it landed as", async () => { for (const shown of [false, true]) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-co-${shown ? 's' : 'u'}`; const framework = await createFramework(); const internals = drainInternals(framework); const agent = framework.getAgent('assistant')!; + agent.markContextConsumed(); if (!shown) agent.markContextShown(); - heldAndStored(framework, 'w-co', 'look left'); + heldAndStored(framework, id, 'look left'); if (shown) agent.markContextShown(); - const placement = { agent: 'assistant', deferredId: 'w-co' }; - assert.equal(internals.isUnreadPlacement({ ...placement }, 'shown'), !shown, `unshown as its copy is (${shown ? 'shown' : 'not shown'})`); - assert.equal(internals.removePlacement({ ...placement }), !shown); - assert.deepEqual(copiesOf(framework, 'w-co'), shown ? ['look left'] : [], shown ? 'a copy that may have been shown stays' : 'its copy goes'); - assert.deepEqual(internals.deferredMessages.map((m) => m.id), shown ? ['w-co'] : [], shown ? 'nothing is removed' : 'and the held entry'); + const placement = { agent: 'assistant', deferredId: id }; + const error = console.error; + console.error = () => {}; + try { + assert.equal(internals.isUnreadPlacement(placement), true, 'unread, as its stored row is'); + assert.equal(internals.isUnreadPlacement(placement, 'shown'), !shown, `unshown as its stored row is (${shown ? 'shown' : 'not shown'})`); + assert.equal(internals.removePlacement(placement), true); + } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, id), [], 'its stored row goes'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'and nothing is held'); await framework.stop(); } }); + it("the coalescer reads a held write whose landing is unknown as read and shown, and doesn't remove it, while its store can't be searched", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-cblind', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { + const placement = { agent: 'assistant', deferredId: 'w-cblind' }; + assert.equal(internals.isUnreadPlacement({ ...placement }), false, 'a doubt reads as read'); + assert.equal(internals.isUnreadPlacement({ ...placement }, 'shown'), false, 'and as shown'); + assert.equal(internals.removePlacement({ ...placement }), false, "and isn't removed"); + } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), ['w-cblind'], 'still held'); + assert.deepEqual(copiesOf(framework, 'w-cblind'), ['look left']); + await framework.stop(); + }); + /** Shard a message whose first block is `marker` into one shard per block, * as a strategy's ingress chunking does. */ function shardOn(framework: AgentFramework, marker: string): void { diff --git a/test/module-rows-held-from-residents.test.ts b/test/module-rows-held-from-residents.test.ts index 5a4741df8..67e707552 100644 --- a/test/module-rows-held-from-residents.test.ts +++ b/test/module-rows-held-from-residents.test.ts @@ -496,4 +496,25 @@ describe("a message a tune-out holds stays out of what the residents are shown, await close(); } }); + + it("the cancel's backlog of held writes settles one whose landing was unknown first, so a write that landed is in the stored rows alone", async () => { + const { framework, internals, close } = await open(); + try { + // A held write whose first write landed, though its store couldn't be + // read to tell (agent-framework's DeferredWrite.landing). + framework.getAgent('scout')!.getContextManager().addMessage('Antra', [{ type: 'text', text: 'T7-LANDED' }], { deferredWriteId: 'w-tune' } as never); + (internals.deferredMessages as unknown as Array>).push({ + id: 'w-tune', seq: 1, participant: 'Antra', content: [{ type: 'text', text: 'T7-LANDED' }], heldAt: Date.now(), landing: 'unknown', + }); + const hooks = (internals.tuneOutCoordinator as unknown as { hooks: { getDeferredMessages(): Array<{ content: ContentBlock[] }> } }).hooks; + const error = console.error; + console.error = () => {}; + let held: string[]; + try { held = hooks.getDeferredMessages().map(textOf); } finally { console.error = error; } + assert.ok(!held.includes('T7-LANDED'), 'not among the held writes'); + assert.ok(!(internals.deferredMessages as unknown as Array<{ id: string }>).some((m) => m.id === 'w-tune'), 'and no longer queued'); + } finally { + await close(); + } + }); }); From 5216b3403e3cb04bbac9aa9ae8ac691c2b4e224e Mon Sep 17 00:00:00 2001 From: Nigel-2654 Date: Mon, 19 Oct 2026 20:16:56 -0400 Subject: [PATCH 4/6] fix(deferral): keep a settled write durable until its ack, and settle by a copy's row Our internal review of bb1f0e38, and a colleague's read: - Settling dropped a found write from the queue and persisted that, so a crash before the store synced could lose a write whose copy wasn't durable yet. A found write now moves to the un-acked set, as a drained write does (settleFound), so the durable queue keeps it until the next ack has synced the store. - A found write now takes the notes a store takes (noteStored), which its first write missed when it threw, as a flush that finds such a write takes them. So an addressed one is weighed for an open turn's re-pin and for the other residents, wherever it's found. - An edit, removal or getMessage by a stored id now settles the held write that row is a copy of (settleByCopy), read from the row's deferredWriteId, so it needs no search: a removal by the copy's id no longer leaves the queued entry to be written again, even where the store can't be searched. - The turn's draft notice settles the agent's held writes before it reads what they name. - Release's catch says, for an entry whose store couldn't be searched, that it may be stored once, twice or not at all; its comment no longer claims settleUnknownLandings runs there. - refuseUnsettled names the held write by its handle, and says the edit or removal is refused the same way while the store can't be searched and that the host log names the store's error. - The coalescer host's isUnshown contract reads "in the deferred queue and not already stored". Tests: settling a found write keeps it un-acked and durable, takes its notes and logs it by its id; one not found is persisted unmarked; an edit and a removal by a stored id settle by the row while the store can't be searched; getMessage settles by a handle and by a stored id, answering the stored row; a targeted settle leaves another marked write alone; the draft notice settles; the coalescer's removal settles on its own; release settles in an ephemeral run's own store, and says what it can't know when it can neither search nor store; and the tune-out backlog's test checks the stored rows too. Co-Authored-By: Claude Opus 5.5 --- .../deferred-write-landing-unknown.fixed.md | 4 +- src/framework.ts | 106 ++++++++---- src/mcpl/push-coalescer.ts | 2 +- test/deferred-window-only.test.ts | 158 ++++++++++++++++-- test/module-rows-held-from-residents.test.ts | 2 + 5 files changed, 225 insertions(+), 47 deletions(-) diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md index c2d7811fa..74ab31424 100644 --- a/changelog.d/deferred-write-landing-unknown.fixed.md +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -21,5 +21,5 @@ - A conversation fork's or ephemeral run's queue, stored when it's released, looks such an entry up the same way: found, it's neither stored again nor passed to the primary as a failed reply; where the - store can't be searched, it's stored anyway, and the log says it may now - be stored twice. + store can't be searched, it's stored anyway where the store takes it, + and the log says it may now be stored twice. diff --git a/src/framework.ts b/src/framework.ts index 5f72f4acc..30225b789 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -14252,6 +14252,9 @@ export class AgentFramework { private noticeUnnoticedDrafts(agent: Agent): void { const cm = agent.getContextManager(); const leftOut = new Set(this.openLogical.get(agent.name)?.drafts.map((d) => d.id)); + // A queued notice still waiting names its drafts; one whose write landed + // is settled first, so its drafts are judged by its stored row. + this.settleUnknownLandings((m) => this.deliveryAgent(m.forAgent) === agent); for (const msg of this.deferredMessages) { const meta = msg.metadata as { draftOwner?: unknown; draftIds?: unknown } | undefined; if (meta?.draftOwner !== agent.name || !Array.isArray(meta.draftIds)) continue; @@ -21518,13 +21521,14 @@ export class AgentFramework { const forPrimary: DeferredWrite[] = []; for (const msg of queued) { // An entry whose earlier write may have landed (landing unknown) is - // settled here as elsewhere (settleUnknownLandings), in the released - // agent's own store. Found, it landed: it's neither stored again nor - // passed on below, since the loop above has passed its stored copy if - // it's a failed reply the agent hadn't read. When the store can't be - // searched, it's stored anyway, since no later flush will try again, - // and a copy its deferredWriteId tells apart costs less than a write - // lost without a trace. + // looked up by its id, as settleUnknownLandings looks one up + // elsewhere, but in the released agent's own store. Found, it landed: + // it's neither stored again nor passed on below, since the loop above + // has passed its stored copy if it's a failed reply the agent hadn't + // read, by its window as that loop reads it. When the store can't be + // searched, it's stored anyway where the store takes it, since no + // later flush will try again, and a copy its deferredWriteId tells + // apart costs less than a write lost without a trace. const found = msg.landing === 'unknown' ? this.landedCopiesById(agent, msg.id) : []; if (found && found.length > 0) { this.logDeferredWriteError(msg, `release of ${agent.name}`, undefined, @@ -21542,7 +21546,11 @@ export class AgentFramework { } catch (err) { console.error( `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + - `(participant=${msg.participant}):`, + `(participant=${msg.participant})` + + (found === undefined + ? `; its store couldn't be searched for an earlier copy of deferred write ${describeDeferredWrite(msg)} either, ` + + `so it may be stored once, twice or not at all:` + : ':'), err, ); } @@ -21817,16 +21825,13 @@ export class AgentFramework { * Settle held writes whose landing is unknown (DeferredWrite.landing), * among those `which` picks, before a path reads or changes held writes: * each is looked up by its id in its target's store (landedCopiesById). - * Found, the write landed: it leaves the queue, logged by its id, and is a - * stored write from then on, so the paths for a landed write apply to it. - * It isn't carried into a turn, as a flush that finds it carries it: the - * agent reads it at its next compile, where the store holds it. Not - * found, the mark goes, and it's an ordinary held write. Where the store - * can't be searched, it stays marked, a doubt each path reads by its own - * rule. A change to the queue is persisted, as an edit's is. + * Found, the write landed (settleFound). Not found, the mark goes, and + * it's an ordinary held write, which is persisted as an edit is. Where the + * store can't be searched, it stays marked, a doubt each path reads by its + * own rule. */ private settleUnknownLandings(which: (msg: DeferredWrite) => boolean): void { - let changed = false; + let unmarked = false; for (const msg of this.deferredMessages.filter((m) => m.landing === 'unknown' && which(m))) { // No registered agent: no store it could have landed in. A flush // writes it, and addMessage drops it. @@ -21835,15 +21840,48 @@ export class AgentFramework { const found = this.landedCopiesById(agent, msg.id); if (found === undefined) continue; if (found.length > 0) { - this.deferredMessages = this.deferredMessages.filter((m) => m !== msg); - this.logDeferredWriteError(msg, `held writes for ${agent.name}`, undefined, - { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + this.settleFound(msg, agent, found[0]!.id, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined }); } else { delete msg.landing; + unmarked = true; } - changed = true; } - if (changed && (this.quiesced || this.deferredWritesPersisted)) this.persistDeferredWrites(); + if (unmarked && (this.quiesced || this.deferredWritesPersisted)) this.persistDeferredWrites(); + } + + /** + * A marked held write found to have landed, as `stored`: it leaves the + * pending queue for the un-acked set, as a drained write does, so it stays + * in the durable queue until the next ack has synced the store that holds + * it, and it's a stored write from then on, so the paths for a landed + * write apply to it. It takes the notes a store takes (noteStored), which + * its first write missed when it threw, as a flush that finds such a write + * takes them (readdDrained), and is logged by its id. It isn't carried + * into a turn, as a flush that finds it carries it: the agent reads it at + * its next compile, where the store holds it. + */ + private settleFound(msg: DeferredWrite, agent: Agent, stored: MessageId, landed: { count: number; sharded: boolean }): void { + this.deferredMessages = this.deferredMessages.filter((m) => m !== msg); + this.unackedDeferredWrites.push(msg); + this.noteStored(agent, msg.metadata, stored); + this.logDeferredWriteError(msg, `held writes for ${agent.name}`, undefined, { ...landed, foundLater: true }); + } + + /** + * A stored row a module names by its id: where it carries the + * deferredWriteId of a marked held write the reader's store holds, it's + * that write's copy, so the write landed, and it's settled as found + * (settleFound) before the row is read or changed. A lookup by the id + * can't be needed then, so this holds where the store can't be searched + * too. + */ + private settleByCopy(reader: Agent, row: { id: MessageId; metadata?: unknown } | null): void { + const writeId = (row?.metadata as { deferredWriteId?: unknown } | undefined)?.deferredWriteId; + if (!row || typeof writeId !== 'string') return; + const msg = this.deferredMessages.find((m) => m.id === writeId && m.landing === 'unknown'); + const target = msg ? this.deliveryAgent(msg.forAgent) : undefined; + if (!msg || !target || !(target === reader || (this.isOrdinaryResident(reader) && this.isOrdinaryResident(target)))) return; + this.settleFound(msg, target, row.id, { count: 1, sharded: false }); } /** A held write whose landing stays unknown after settling, its store @@ -21853,8 +21891,9 @@ export class AgentFramework { private refuseUnsettled(msg: DeferredWrite): void { if (msg.landing !== 'unknown') return; throw new Error( - `held write ${msg.id} may already be stored, and its store couldn't be searched to tell, ` + - `so nothing was changed; try again once the store can be read`, + `held write ${heldHandle(msg.id)} may already be stored, and its store can't be searched to tell, so ` + + `nothing was changed. It's refused the same way while the store can't be searched; the host log names ` + + `the store's error`, ); } @@ -21900,11 +21939,12 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } - // The held writes are settled first (heldWritesFor), by a stored id as - // by a handle: a held write that landed is a stored row from then on, - // and the edit reaches it as one. + // The held writes are settled first (heldWritesFor), and by a stored id + // the row it names too (settleByCopy): a held write that landed is a + // stored row from then on, and the edit reaches it as one. const writeId = heldWriteId(id); const held = this.heldWritesFor(agent).find((m) => writeId !== undefined && m.id === writeId); + if (writeId === undefined) this.settleByCopy(agent, agent.getContextManager().getMessage(id)); if (held) { this.refuseUnsettled(held); held.content = content; @@ -21920,11 +21960,13 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } - // The held writes are settled first (heldWritesFor), by a stored id as - // by a handle, so a held write that landed is removed as the stored row - // it is, and isn't written again at the next flush. + // The held writes are settled first (heldWritesFor), and by a stored id + // the row it names too (settleByCopy), so a held write that landed is + // removed as the stored row it is, and isn't written again at the next + // flush. const writeId = heldWriteId(id); const held = this.heldWritesFor(agent).find((m) => writeId !== undefined && m.id === writeId); + if (writeId === undefined) this.settleByCopy(agent, agent.getContextManager().getMessage(id)); if (held) { this.refuseUnsettled(held); this.deferredMessages = this.deferredMessages.filter((m) => m.id !== writeId); @@ -21941,8 +21983,12 @@ export class AgentFramework { return null; } const writeId = heldWriteId(id); + // By a handle, heldRows settles the held writes first; by a stored id, + // the row's own write is settled by it (settleByCopy). + const stored = writeId === undefined ? agent.getContextManager().getMessage(id) : null; + if (writeId === undefined) this.settleByCopy(agent, stored); const row = writeId === undefined - ? agent.getContextManager().getMessage(id) + ? stored : this.heldRows(agent).find((held) => held.id === id) ?? this.landedRow(agent.getContextManager(), writeId); return row ? asModuleRow(row) : null; } diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index b16e91003..ae974d03e 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -260,7 +260,7 @@ export interface CoalescerHost { /** True iff the placement still exists and no model request has included it. */ isUnread(placement: CoalescingPlacement): boolean; /** True iff nothing may have shown the placement to the agent: it is still - * in the deferred queue, known not to be stored, or stored where no + * in the deferred queue and not already stored, or it is stored where no * committing compile has begun to assemble it and no tool boundary has * injected past it. Unlike * isUnread, a compile counts from its start, and a boundary counts diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 8a4833462..44afccf5d 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -852,6 +852,40 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); + it("a released window that can neither search a write's store nor store it says it may be stored once, twice or not at all", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown; addMessage: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + const add = cm.addMessage; + cm.queryMessages = () => { throw new Error('search failed'); }; + cm.addMessage = () => { throw new Error('store failed'); }; + internals.deferredMessages.push({ ...entry('w-lost', 'look left'), landing: 'unknown' }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.releaseWindow(agent); } finally { console.error = error; cm.queryMessages = query; cm.addMessage = add; } + assert.ok(errors.some((e) => e.includes('w-lost') && e.includes('may be stored once, twice or not at all')), errors.join('\n')); + assert.ok(!errors.some((e) => e.includes('w-lost') && e.includes('may now be stored twice')), 'not the line for a store that went through'); + await framework.stop(); + }); + + it("a released ephemeral run's window settles a write whose landing is unknown in that run's own store", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const { agent } = await framework.createEphemeralAgent({ name: 'run-x', model: 'test-model', systemPrompt: 'Do the task.' }); + const own = agent.getContextManager(); + own.addMessage('Antra', text('look left'), { deferredWriteId: 'w-run' } as never); + internals.deferredMessages.push({ ...entry('w-run', 'look left'), forAgent: 'run-x', landing: 'unknown' }); + const error = console.error; + console.error = () => {}; + try { internals.releaseWindow(agent); } finally { console.error = error; } + assert.equal(own.queryMessages({ metadata: { deferredWriteId: 'w-run' } } as never).messages.length, 1, "found in the run's own store, it isn't stored again"); + assert.deepEqual(copiesOf(framework, 'w-run'), [], "and the primary's store holds none"); + await framework.stop(); + }); + it("a released window stores a write whose landing is unknown while its store can't be searched, saying it may now be stored twice", async () => { const framework = await createFramework(); const internals = drainInternals(framework); @@ -925,22 +959,74 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); - it("settling a held write whose landing is unknown, inside a read, persists the queue as an edit does", async () => { + it("settling a held write whose landing is unknown, found, keeps it in the durable queue until an ack, takes the notes a store takes, and logs it by its id", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework) as ReturnType & { + unackedDeferredWrites: DeferredEntry[]; + durableDeferredWrites(): DeferredEntry[]; + noteStored(agent: unknown, metadata: unknown, stored: string): void; + }; + internals.deferredWritesPersisted = true; + heldAndStored(framework, 'w-found', 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: 'w-found' } } as never).messages; + const noted: string[] = []; + const note = internals.noteStored; + internals.noteStored = function (this: unknown, agent: unknown, metadata: unknown, stored: string) { + noted.push(stored); + return note.call(this, agent, metadata, stored); + }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.queryMessages({}); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'no longer held'); + assert.deepEqual(internals.unackedDeferredWrites.map((m) => m.id), ['w-found'], 'un-acked, as a drained write is'); + assert.deepEqual(internals.durableDeferredWrites().map((m) => m.id), ['w-found'], 'so the durable queue keeps it until the next ack'); + assert.deepEqual(noted, [copy!.id], 'the notes a store takes, for the copy'); + assert.ok(errors.some((e) => e.includes('w-found') && e.includes('was found by its id')), errors.join('\n')); + await framework.stop(); + }); + + it("settling a held write whose landing is unknown, not found, persists the queue without its mark, as an edit does", async () => { const framework = await createFramework(); const internals = drainInternals(framework) as ReturnType & { persistDeferredWrites(): void }; internals.deferredWritesPersisted = true; - heldAndStored(framework, 'w-persist', 'look left'); + internals.deferredMessages.push({ ...entry('w-persist', 'look left'), landing: 'unknown' }); let persisted = 0; const persist = internals.persistDeferredWrites; internals.persistDeferredWrites = function (this: unknown) { persisted++; return persist.call(this); }; - const error = console.error; - console.error = () => {}; - try { internals.queryMessages({}); } finally { console.error = error; } - assert.equal(persisted, 1, 'the queue without it is persisted'); - assert.deepEqual(internals.deferredMessages.map((m) => m.id), []); + internals.queryMessages({}); + assert.equal(persisted, 1, 'the queue without its mark is persisted'); + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-persist', undefined]]); await framework.stop(); }); + it("a module's edit or removal by a stored id settles the held write that row is a copy of, even while its store can't be searched", async () => { + for (const op of ['edit', 'remove'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-row-${op}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => { messages: Array<{ id: string }> }; + }; + const [copy] = cm.queryMessages({ metadata: { deferredWriteId: id } }).messages; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { + if (op === 'edit') internals.editMessage(copy!.id, text('look up')); + else internals.removeMessage(copy!.id); + } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(copiesOf(framework, id), op === 'edit' ? ['look up'] : [], `${op}: the row is changed once`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `${op}: its write is settled by the row, so nothing writes it again`); + await framework.stop(); + } + }); + it("a module's edit or removal of a held write whose landing is unknown throws with nothing changed while its store can't be searched", async () => { const framework = await createFramework(); const internals = drainInternals(framework); @@ -951,15 +1037,15 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const error = console.error; console.error = () => {}; try { - assert.throws(() => internals.editMessage(heldHandle('w-blind'), text('look up')), /couldn't be searched/); - assert.throws(() => internals.removeMessage(heldHandle('w-blind')), /couldn't be searched/); + assert.throws(() => internals.editMessage(heldHandle('w-blind'), text('look up')), /held:w-blind may already be stored, and its store can't be searched/); + assert.throws(() => internals.removeMessage(heldHandle('w-blind')), /held:w-blind may already be stored, and its store can't be searched/); } finally { cm.queryMessages = query; console.error = error; } assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content, m.landing]), [['w-blind', text('look left'), 'unknown']], 'still held, unedited'); assert.deepEqual(copiesOf(framework, 'w-blind'), ['look left'], 'its copy unchanged'); await framework.stop(); }); - it("a module's query and getMessage meet a held write whose landing is unknown, settled, as the one stored row it landed as", async () => { + it("a module's query meets a held write whose landing is unknown, settled, as the one stored row it landed as", async () => { const framework = await createFramework(); const internals = drainInternals(framework); framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('look right')); @@ -967,30 +1053,48 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const error = console.error; console.error = () => {}; let rows: string[]; - let byHandle: { held?: true; content: ContentBlock[] } | null; try { rows = internals.queryMessages({}).messages .filter((m) => m.content.some((b) => ['look left', 'look right'].includes((b as { text?: string }).text ?? ''))) .map((m) => `${m.held ? 'held' : 'stored'}: ${m.content.map((b) => (b as { text?: string }).text).join('')}`); - byHandle = internals.getMessage(heldHandle('w-q')); } finally { console.error = error; } assert.deepEqual(rows, ['stored: look right', 'stored: look left']); - assert.equal(byHandle?.held, undefined, 'its handle names the stored row'); await framework.stop(); }); + it("a module's getMessage, by a handle or by a stored id, settles a held write whose landing is unknown and answers its stored row", async () => { + for (const by of ['handle', 'stored id'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-get-${by === 'handle' ? 'h' : 's'}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages; + const error = console.error; + console.error = () => {}; + let row: { held?: true; content: ContentBlock[] } | null; + try { row = internals.getMessage(by === 'handle' ? heldHandle(id) : copy!.id); } finally { console.error = error; } + assert.ok(row, `by its ${by}: a row`); + assert.equal(row.held, undefined, `by its ${by}: the stored row`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `by its ${by}: settled`); + await framework.stop(); + } + }); + it("deferredWriteReadOrGone settles a held write whose landing is unknown, so a copy the agent has read reads as read", async () => { const framework = await createFramework(); const internals = drainInternals(framework); const agent = framework.getAgent('assistant')!; heldAndStored(framework, 'w-wake', 'look left'); + heldAndStored(framework, 'w-other', 'look right'); agent.markContextConsumed(); const error = console.error; console.error = () => {}; try { assert.equal(internals.deferredWriteReadOrGone(agent, 'w-wake'), true); } finally { console.error = error; } - assert.deepEqual(internals.deferredMessages.map((m) => m.id), []); + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-other', 'unknown']], 'only the write asked about is settled'); await framework.stop(); }); @@ -1019,6 +1123,32 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { } }); + it("the turn's draft notice settles the agent's held writes whose landing is unknown before it reads what they name", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework) as ReturnType & { noticeUnnoticedDrafts(agent: unknown): void }; + heldAndStored(framework, 'w-draft', 'look left'); + const error = console.error; + console.error = () => {}; + try { internals.noticeUnnoticedDrafts(framework.getAgent('assistant')!); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), []); + await framework.stop(); + }); + + it("the coalescer's removal settles a held write whose landing is unknown, and removes it as the stored placement it landed as", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + framework.getAgent('assistant')!.markContextConsumed(); + heldAndStored(framework, 'w-rmp', 'look left'); + const error = console.error; + console.error = () => {}; + try { + assert.equal(internals.removePlacement({ agent: 'assistant', deferredId: 'w-rmp' }), true); + } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, 'w-rmp'), [], 'its stored row goes'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'and nothing is held'); + await framework.stop(); + }); + it("the coalescer reads a held write whose landing is unknown as read and shown, and doesn't remove it, while its store can't be searched", async () => { const framework = await createFramework(); const internals = drainInternals(framework); diff --git a/test/module-rows-held-from-residents.test.ts b/test/module-rows-held-from-residents.test.ts index 67e707552..43ef150f6 100644 --- a/test/module-rows-held-from-residents.test.ts +++ b/test/module-rows-held-from-residents.test.ts @@ -512,6 +512,8 @@ describe("a message a tune-out holds stays out of what the residents are shown, let held: string[]; try { held = hooks.getDeferredMessages().map(textOf); } finally { console.error = error; } assert.ok(!held.includes('T7-LANDED'), 'not among the held writes'); + const stored = internals.agents.get('scout')!.getContextManager().getAllMessages().filter((m) => textOf(m) === 'T7-LANDED'); + assert.equal(stored.length, 1, 'in the stored rows, once'); assert.ok(!(internals.deferredMessages as unknown as Array<{ id: string }>).some((m) => m.id === 'w-tune'), 'and no longer queued'); } finally { await close(); From 413b429a1f4f9dbff44958c1aeec2e4d3d30bea0 Mon Sep 17 00:00:00 2001 From: Nigel-2654 Date: Mon, 19 Oct 2026 20:36:33 -0400 Subject: [PATCH 5/6] docs(deferral): say what a settled write's notes, ack and release mean; pin two more cases From our internal review of 5216b340, text and two tests: - The coalescer host's isUnshown contract says a queued placement is unshown only when it's known not to be stored already, which a doubt isn't. - releaseWindow's doc names the line for a store that fails as well as the search: it may be stored once, twice or not at all. - The settle test checks the found write keeps its mark in the un-acked set, so a restart before the ack looks it up again. - A new test has a stored row settle only the held write whose id it carries, beside another marked write that stays marked. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 3 ++- src/mcpl/push-coalescer.ts | 6 +++--- test/deferred-window-only.test.ts | 21 ++++++++++++++++++++- 3 files changed, 25 insertions(+), 5 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index 30225b789..234b4ac4e 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -21505,7 +21505,8 @@ export class AgentFramework { * landed (`landing` unknown) is looked up there by its id first: found, * it's neither stored again nor passed to the primary; where the store * can't be searched, it's stored anyway, with a line saying it may now be - * stored twice. Its stream is over or cancelled, so it + * stored twice, or, where storing it fails too, that it may be stored + * once, twice or not at all. Its stream is over or cancelled, so it * compiles nothing more. One path still stores after these writes: a * `complete` that arrived while the agent still waited for tools awaits * the spill of its last tool results, and stores them once that resolves, diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index ae974d03e..5e713a8fb 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -260,9 +260,9 @@ export interface CoalescerHost { /** True iff the placement still exists and no model request has included it. */ isUnread(placement: CoalescingPlacement): boolean; /** True iff nothing may have shown the placement to the agent: it is still - * in the deferred queue and not already stored, or it is stored where no - * committing compile has begun to assemble it and no tool boundary has - * injected past it. Unlike + * in the deferred queue and known not to be stored already, or it is + * stored where no committing compile has begun to assemble it and no tool + * boundary has injected past it. Unlike * isUnread, a compile counts from its start, and a boundary counts * whether or not its stream says it carries injected messages: a doubt * counts as shown. */ diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 44afccf5d..ee34fc29d 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -981,7 +981,8 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; try { internals.queryMessages({}); } finally { console.error = error; } assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'no longer held'); - assert.deepEqual(internals.unackedDeferredWrites.map((m) => m.id), ['w-found'], 'un-acked, as a drained write is'); + assert.deepEqual(internals.unackedDeferredWrites.map((m) => [m.id, m.landing]), [['w-found', 'unknown']], + 'un-acked, as a drained write is, and still marked, so a restart before the ack looks it up'); assert.deepEqual(internals.durableDeferredWrites().map((m) => m.id), ['w-found'], 'so the durable queue keeps it until the next ack'); assert.deepEqual(noted, [copy!.id], 'the notes a store takes, for the copy'); assert.ok(errors.some((e) => e.includes('w-found') && e.includes('was found by its id')), errors.join('\n')); @@ -1062,6 +1063,24 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); + it("a stored row settles only the held write whose id it carries", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-theirs', 'look right'); + heldAndStored(framework, 'w-mine', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => { messages: Array<{ id: string }> }; + }; + const [mine] = cm.queryMessages({ metadata: { deferredWriteId: 'w-mine' } }).messages; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { internals.getMessage(mine!.id); } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-theirs', 'unknown']], "the other write stays marked"); + await framework.stop(); + }); + it("a module's getMessage, by a handle or by a stored id, settles a held write whose landing is unknown and answers its stored row", async () => { for (const by of ['handle', 'stored id'] as const) { // Each pass opens the test's one store again, so each takes its own id. From f80c0e15295c69ab35127ab48f706745d0360d09 Mon Sep 17 00:00:00 2001 From: Nigel-2654 Date: Mon, 19 Oct 2026 20:42:17 -0400 Subject: [PATCH 6/6] docs(changelog): name the refusal a module's edit or removal can now meet A module context's editMessage and removeMessage can now throw for a held write whose landing is unknown, while its store can't be searched. The fixed fragment names that with the error's words, so a module author whose code edits or removes held writes knows it can throw there. Co-Authored-By: Claude Opus 5.5 --- changelog.d/deferred-write-landing-unknown.fixed.md | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md index 74ab31424..c4c203d78 100644 --- a/changelog.d/deferred-write-landing-unknown.fixed.md +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -15,9 +15,14 @@ coalescer, a wake's check or a tune-out's backlog, such a write is looked up first. One that landed is a stored message from then on, read and changed as one, and isn't carried into the turn; one that didn't is - an ordinary held write. Where the store can't be searched, a module's - edit or removal of it throws with nothing changed, and the coalescer - reads it as possibly read and shown. + an ordinary held write. Where the store can't be searched, the + coalescer reads it as possibly read and shown. + - A module context's `editMessage` and `removeMessage` can now throw for + such a held write: where its store can't be searched to tell whether + it landed, nothing is changed, and the error reads "held write held:… + may already be stored, and its store can't be searched to tell, so + nothing was changed. It's refused the same way while the store can't be + searched; the host log names the store's error". - A conversation fork's or ephemeral run's queue, stored when it's released, looks such an entry up the same way: found, it's neither stored again nor passed to the primary as a failed reply; where the