From ea03c9a2e238a2dfb112cedf8bb2f49519b7bd77 Mon Sep 17 00:00:00 2001 From: Ansel-2428 Date: Thu, 22 Oct 2026 20:10:58 -0400 Subject: [PATCH 1/3] fix(deferral): look a write found later up where it landed Answers the review of #607 (R1-1, R1-3, R1-4, R1-5). - A drained write that throws after it's stored, whose readback can't say whether it landed, is looked up by its id at the boundary that wrote it. Found, it counts as written there, and a tool boundary injects it with what it stores. Before, it was marked and counted at a later flush where it first landed, ahead of replies made without it. - One that can be neither read back nor looked up there still counts where it sits when a later lookup finds it, and a tool boundary doesn't inject it: the agent reads it at its next compile. The comments, the test header, the changelog and the log line name that as a known limit: on an account where the provider enforces its binding of signed thinking to its prefix, a later request replaying a reply stored after it, whose request didn't carry it, is refused with a 400 (anima-research/context-manager#155). - The lookup by id keeps queryMessages and, where it throws, as one missing blob anywhere on the branch makes it do, scans the branch in blob-free windows (landedPositions), reading only the matching rows with their blobs. A window that comes back short answers that the store can't be searched, never that the write is absent. Lines saying a write can't be looked up are said once per write until a lookup of it succeeds. - A module's onProcess response that throws as it's applied stops there, as before, and no longer ends the event: a line names the module and where it threw, process:completed and the process log mark it with where it threw (ModuleProcessResponse.threw), and the other responses, a tool call's dispatch and the trace go on. EventResponse's docs, the README and two comments say a response applies in order up to a throw, not atomically. - A held write settled through one shard a module names says a shard was stored. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- changelog.d/deferral-found-later.fixed.md | 5 + src/framework.ts | 281 ++++++++--- src/module-registry.ts | 3 +- src/modules/discord/index.ts | 2 +- src/types/framework.ts | 7 + src/types/module.ts | 16 +- ...deferred-malformed-unknown-landing.test.ts | 1 + test/deferred-window-only.test.ts | 459 ++++++++++++++++-- test/mcpl-coalescing.test.ts | 5 + test/process-response-edit-refusals.test.ts | 215 ++++++++ 11 files changed, 897 insertions(+), 99 deletions(-) create mode 100644 changelog.d/deferral-found-later.fixed.md create mode 100644 test/process-response-edit-refusals.test.ts diff --git a/README.md b/README.md index f9fbfb1d4..ef49a3115 100644 --- a/README.md +++ b/README.md @@ -122,7 +122,7 @@ interface Module { } ``` -An `EventResponse` can add/edit/remove messages, request inference, signal tool changes, and atomically update module state: +An `EventResponse` can add/edit/remove messages, request inference, signal tool changes, and update module state after its message changes. The framework applies it in that order; when one step throws, what came before it stays applied, that step may have taken effect in part, and the rest of that response isn't: ```typescript return { diff --git a/changelog.d/deferral-found-later.fixed.md b/changelog.d/deferral-found-later.fixed.md new file mode 100644 index 000000000..9a584c70f --- /dev/null +++ b/changelog.d/deferral-found-later.fixed.md @@ -0,0 +1,5 @@ +- A deferred write that throws after it's stored, whose store can't be read back to tell whether it landed, is now looked up by its id at the boundary that wrote it, so it counts as written there, where it landed, and at a tool boundary it's injected with what that boundary stores, before the turn's later replies. Before, it was queued as a doubt and counted at a later flush where it first landed, ahead of later replies whose requests never held it. That's still so, as a known limit, for a write that can be neither read back nor looked up there: on an account where the provider enforces its binding of signed thinking to the prefix it was made with, a later request replaying the signed thinking of a reply stored after it, whose request didn't carry it, is refused with a 400 (context-manager #155). +- A held write found where an earlier write stored it is no longer injected into the live turn at a tool boundary, where a compile might already have carried it. The agent reads it at its next compile, from the store. +- Looking a held write up by its id no longer fails because another message's blob is missing: where context-manager's query throws, the branch is scanned without blobs, and only the rows that match are read with theirs. Before, every such write stayed a doubt, holding what was queued behind it for its agent, and each read logged it again. The lines saying a store can't be searched, and that a drain holds what's queued behind such a write, are now said once per write until a lookup succeeds. +- A module's `onProcess` response that throws as the framework applies it (at an edit of a held write that may already be stored while its store can't be searched, an edit of a message the store doesn't hold, or any store error) no longer ends the event's processing. The response stops where it threw, as before: what came before the throw stays applied, the operation that threw may have taken effect in part or whole, and the rest of it, its state update and inference request included, isn't. A line now names the module and where it threw, `process:completed` and the process log name the response with where it threw (`ModuleProcessResponse.threw`), and the other modules' responses, a `tool-call` event's dispatch, the event's trace and its process log go on. Before, the throw escaped the event's processing and all of these were lost. `EventResponse`'s docs, the README and two comments no longer say a response is applied atomically, which it wasn't: they say it's applied in order, up to a throw. +- The log line for a held write settled through a stored row a module names says when that row is one shard of a body, rather than that the whole write was stored. diff --git a/src/framework.ts b/src/framework.ts index d46cba669..94c4595d3 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -854,6 +854,10 @@ interface DrainedWrite { id: MessageId; content: ContentBlock[]; parts: number; + /** A marked entry found where an earlier write stored it: a tool + * boundary doesn't inject it, and the agent reads it at its next + * compile, where the store holds it. */ + foundLater?: true; } /** What later compiles render from the store for the copies of one write @@ -892,6 +896,18 @@ 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 } : {}) }); +/** Messages per window when a store is scanned for deferred writes by their + * ids (landedPositions). */ +const LANDED_SCAN_WINDOW = 500; + +/** The message at `position` on `cm`'s current branch, with its blobs + * resolved, as a request carries it. Throws when there's none there. */ +function messageAt(cm: ContextManager, position: number): StoredMessage { + const message = cm.getMessageWindow(position, 1).messages[0]; + if (!message) throw new Error(`the store has no message at position ${position}`); + return message; +} + /** A tool result's spill file as writeToolResultFile left it: its path, with * an error if the write failed, or null with no writable workspace. */ type SpillWrite = { path: string; error?: string; branchLeft?: true } | null; @@ -2789,6 +2805,11 @@ export class AgentFramework { */ private deferredScanFrom = new Map(); private deferredMessages: DeferredWrite[] = []; + /** The marked held writes whose store couldn't be searched, said once + * each until a lookup of them succeeds (landedCopiesById), and those a + * drain has said it holds what's queued behind (writeDrainedInOrder). */ + private unsearchableSaid = new Set(); + private stillUnknownSaid = new Set(); // Turn-alive markers (2026-07-31 Mythos phantom-skip incident): agentName → // token of the turn currently in progress. Set at startAgentStream ENTRY — @@ -11831,10 +11852,20 @@ export class AgentFramework { // the next compile then strips that image where the live rounds // carried it. One the next compile won't send among its newest // isn't injected; a notice after the batch tells the agent of it - // (recordLeftOutNotice). 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). + // (recordLeftOutNotice). A write that threw after it was stored, + // whose readback can't say whether it landed, is looked up by its + // id here, and carried here, where it landed. One that can be + // neither read back nor looked up is the exception, a known limit: + // it waits, marked, and the flush that later finds it counts it + // where it was stored, ahead of what was stored since without it. + // There the store and the live request differ, and a reply stored + // after it whose request didn't carry it was made without it. On + // an account where the provider enforces that binding, a later + // request replaying that reply's signed thinking is refused with a + // 400 (context-manager #155). Such a copy isn't injected: a compile + // may already have carried it, and injected or not, it sits ahead + // of those replies in the store. The agent reads it at its next + // compile rather than at this boundary (writeDrainedInOrder). // // The queue is drained ONLY at the target agent's own boundary: // draining at another agent's boundary would store the messages @@ -11892,10 +11923,13 @@ export class AgentFramework { // counts as stored, as what landed of it (writeDrainedInOrder). const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) => agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); - for (const [msg, { id, content, parts }] of stored) { + for (const [msg, { id, content, parts, foundLater }] of stored) { // Every other resident reads what this boundary stores; for its // own agent, the boundary weighs what it injects below. this.noteForOtherResidents(agent, msg.metadata, id); + // A copy found where an earlier write left it isn't carried into + // the live turn: the next compile carries it from the store. + if (foundLater) continue; // A message a tune-out holds from the residents is stored (the // subconscious reads it) and left out of every resident's // compiled view; injecting it here would put in the live turn @@ -12277,10 +12311,19 @@ export class AgentFramework { // and it doesn't count as accepting a surface's input. A line and a // trace name the module and where, and the other modules' responses // apply. process:completed and the process log list the responses that - // applied, so a view built on them doesn't show a refused response's - // messages as the event's. A store error while a response applies, such - // as a shard the store refuses when a strategy's chunking makes one, - // propagates as any store error does. + // passed that check, so a view built on them doesn't show a refused + // response's messages as the event's. A response that throws as it + // applies, such as at an edit of a held write while its store can't be + // searched (refuseUnsettled), a shard the store refuses when a + // strategy's chunking makes one, or any store error, stops there: what + // it asked for before the throw stays applied, the operation that threw + // may have taken effect in part or whole, and the rest of it isn't. + // A line names the module and where it threw, and process:completed and + // the process log name the response with where it threw (`threw`), so + // neither reads it as applied whole. The other modules' + // responses, a tool call's dispatch and the trace go on, since a module + // can't catch a throw on this path, and one module's mustn't cost the + // event. const applying = responses.filter(({ moduleName, response }) => { const refusal = malformedResponseContent(response); if (refusal === null) return true; @@ -12300,8 +12343,20 @@ export class AgentFramework { if (surface && applying.some(({ response }) => (response.addMessages?.length ?? 0) > 0)) { this.noteInboundAccepted(surface); } - for (const { moduleName, response } of applying) { - await this.applyProcessResponse(response, event, moduleName, surface); + for (const entry of applying) { + const { moduleName, response } = entry; + const step = { at: 'its start' }; + try { + await this.applyProcessResponse(response, event, moduleName, surface, step); + } catch (error) { + entry.threw = { at: step.at, error: error instanceof Error ? error.message : String(error) }; + console.error( + `[process] ${moduleName}'s response to ${event.type} threw at ${step.at}: what it asked for before that ` + + `was applied, that operation may have taken effect in part or whole, and the rest of it wasn't; ` + + `the event goes on:`, + error, + ); + } } // Handle tool calls specially @@ -12325,13 +12380,16 @@ export class AgentFramework { event: ProcessEvent, moduleName: string, /** The event's surface envelope, when it is console/API input. */ - surface?: InboundSource, + surface: InboundSource | undefined, + /** Names the operation under way, for the line when one throws. */ + step: { at: string }, ): Promise { // Add messages. Only the framework writes the inbound envelope: one a // module's metadata carries (echoed from a server's event, say) is // dropped, and a surface event's messages carry the framework's own. if (response.addMessages) { - for (const msg of response.addMessages) { + for (const [i, msg] of response.addMessages.entries()) { + step.at = `adding message ${i + 1} of ${response.addMessages.length}`; let metadata = msg.metadata; if (metadata && INBOUND_SOURCE_KEY in metadata) { const { [INBOUND_SOURCE_KEY]: _dropped, ...rest } = metadata; @@ -12344,26 +12402,28 @@ export class AgentFramework { } // Edit messages - if (response.editMessages) { - for (const edit of response.editMessages) { - this.editMessage(edit.messageId, edit.content); - } + for (const edit of response.editMessages ?? []) { + step.at = `the edit of ${edit.messageId}`; + this.editMessage(edit.messageId, edit.content); } // Remove messages - if (response.removeMessages) { - for (const id of response.removeMessages) { - this.removeMessage(id); - } + for (const id of response.removeMessages ?? []) { + step.at = `the removal of ${id}`; + this.removeMessage(id); } - // Apply module state update atomically with message operations + // Apply module state update after the message operations: a response + // whose operation threw stops there, so its update describes no change + // that wasn't made. if (response.stateUpdate !== undefined) { + step.at = 'its state update'; this.moduleRegistry.setModuleState(moduleName, response.stateUpdate); } // Queue inference requests if (response.requestInference) { + step.at = 'its inference request'; const source = 'source' in event ? (event as { source: string }).source : 'unknown'; // Gate non-MCPL events (MCPL events are already gated in PushHandler/ChannelRegistry) @@ -22288,11 +22348,15 @@ export class AgentFramework { * 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), or couldn't say how many messages it held - * before the write, whether the entry landed is unknown: it's put back + * before the write, the store is searched for the entry's id at once + * (landedCopiesById): found, it counts as written, as above. When that + * can't be done either, 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. + * searched for its id. Found then, it counts as written where it was + * stored, ahead of what was stored since without it, a known limit + * (context-manager #155, named at the tool boundary's drain); not found, + * it's written. 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, its entries up to the @@ -22321,23 +22385,36 @@ 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. 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. + // An earlier write of this entry threw, and its store could neither + // be read back nor searched to tell whether it landed. It's looked up + // by its id before it's written again. Found, it counts as written + // where that write stored it, as when the readback works. That's a + // known limit: the copy sits ahead of what was stored since without + // it, and a reply stored after it whose request didn't carry it was + // made without it. On an account where the provider enforces that + // binding, a later request replaying that reply's signed thinking is + // refused with a 400 (context-manager #155). A tool boundary doesn't + // inject it; the agent reads it at its next compile. Not found, it's + // written. While + // the store can't be 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)); - this.logDeferredWriteError(msg, where, undefined, 'still-unknown'); + // Said once until a lookup of it succeeds, as the search's own + // line is: every drain tries again. + if (!this.stillUnknownSaid.has(msg.id)) { + this.stillUnknownSaid.add(msg.id); + this.logDeferredWriteError(msg, where, undefined, 'still-unknown'); + } continue; } if (found.length > 0) { - written.set(msg, { id: found[0]!.id, content: asCompiled(found), parts: found.length }); + written.set(msg, { id: found[0]!.id, content: asCompiled(found), parts: found.length, foundLater: true }); this.logDeferredWriteError(msg, where, undefined, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); onLanded?.(msg, found[0]!.id); continue; @@ -22357,9 +22434,14 @@ export class AgentFramework { written.set(msg, { id, content: copies.length > 1 ? asCompiled(copies) : msg.content, parts: Math.max(1, copies.length) }); } catch (error) { // 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); + // write, or, where that can't say, before the write or after it, + // looked up by its id here, at the boundary that wrote it: found, + // it counts as written here, where it landed, so what this boundary + // carries is what's stored. Undefined when the store can be neither + // read back nor searched. With no agent there's no store it could + // have landed in. + let landed = !agent ? [] : before === undefined ? undefined : this.storedCopiesOf(agent, before, msg.id); + if (landed === undefined && agent) landed = this.landedCopiesById(agent, msg.id); if (landed && landed.length > 0) { written.set(msg, { id: landed[0]!.id, content: asCompiled(landed), parts: landed.length }); this.logDeferredWriteError(msg, where, error, { count: landed.length, sharded: landed[0]!.bodyGroupId !== undefined }); @@ -22426,17 +22508,74 @@ export class AgentFramework { } } - /** 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. */ + /** + * The positions on `cm`'s current branch of the messages carrying + * `deferredWriteId`, in store order: the whole branch, read in blob-free + * windows, so a blob missing from some other message can't make it fail, + * as it makes queryMessages fail (every blob on the branch is resolved for + * it). Throws when a window can't be read, or holds fewer messages than it + * was asked for: a scan that didn't reach every message can't say a write + * is absent, and a write taken as absent is stored again. + */ + private landedPositions(cm: ContextManager, deferredWriteId: string): number[] { + const at: number[] = []; + const total = cm.getMessageCount(); + for (let start = 0; start < total; start += LANDED_SCAN_WINDOW) { + const want = Math.min(LANDED_SCAN_WINDOW, total - start); + const { messages } = cm.getMessageWindow(start, want, { resolveBlobs: false }); + if (messages.length !== want) { + throw new Error(`the store gave ${messages.length} of the ${want} messages from position ${start} it was asked for`); + } + messages.forEach((m, i) => { + if ((m.metadata as { deferredWriteId?: unknown } | undefined)?.deferredWriteId === deferredWriteId) at.push(start + i); + }); + } + return at; + } + + /** + * The messages carrying `deferredWriteId` anywhere on the agent's current + * branch, in store order with their blobs resolved, as a request carries + * them, so no position is needed. They're found with queryMessages, over + * context-manager's cached view, as landedRow finds a held write's row. + * Where that throws, as one blob missing anywhere on the branch makes it + * do, the branch is scanned blob-free instead (landedPositions), and only + * the rows that match are read with their blobs. Undefined when the store + * can be searched neither way, or the scan finds the write's rows but one + * of them can't be read. Each write whose lookup fails is said once until + * a lookup of it succeeds, since every path that reads held writes looks + * again. + */ private landedCopiesById(agent: Agent, deferredWriteId: string): StoredCopy[] | undefined { + const cm = agent.getContextManager(); + // Said once until a lookup of it succeeds. + const say = (what: string, error: unknown): void => { + if (this.unsearchableSaid.has(deferredWriteId)) return; + this.unsearchableSaid.add(deferredWriteId); + console.error(`[deferral] ${what}; this is said once until a lookup succeeds:`, error); + }; + let found: StoredCopy[]; 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; + found = cm.queryMessages({ metadata: { deferredWriteId } }).messages.map(storedCopy); + } catch { + let at: number[]; + try { + at = this.landedPositions(cm, deferredWriteId); + } catch (error) { + say(`could not search ${agent.name}'s store for deferred write ${deferredWriteId}, to tell whether it was stored`, error); + return undefined; + } + try { + found = at.map((position) => storedCopy(messageAt(cm, position))); + } catch (error) { + // The scan found its rows; one of them can't be read whole. + say(`found deferred write ${deferredWriteId} in ${agent.name}'s store, but could not read its stored copy`, error); + return undefined; + } } + this.unsearchableSaid.delete(deferredWriteId); + this.stillUnknownSaid.delete(deferredWriteId); + return found; } /** The line naming a drained write that threw: one that couldn't be stored, @@ -22454,7 +22593,7 @@ export class AgentFramework { msg: DeferredWrite, where: string, error: unknown, - outcome?: { count: number; sharded: boolean; foundLater?: boolean } | 'unknown' | 'still-unknown' | 'dropped', + outcome?: { count?: number; sharded: boolean; foundLater?: boolean; named?: true } | 'unknown' | 'still-unknown' | 'dropped', ): void { let reason: string; try { @@ -22463,9 +22602,15 @@ export class AgentFramework { 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'; + const stored = (landed: { count?: number; sharded: boolean }): string => !landed.sharded + ? 'it was stored' + : landed.count === undefined + ? 'a shard of its body was stored' + : `${landed.count} shard${landed.count === 1 ? '' : 's'} of its body ${landed.count === 1 ? 'was' : 'were'} stored`; + // What a copy found later costs where it stays (writeDrainedInOrder). + const stays = `; it stays where that write stored it, and a reply stored after it whose request didn't ` + + `carry it was made without it: on an account where the provider enforces that binding, a later request ` + + `replaying that reply's signed thinking is refused with a 400 (context-manager #155)`; let what: string; if (outcome === 'dropped') { what = `was refused as malformed, so it's dropped rather than queued again, and what is queued behind it for that agent goes on: ${reason}`; @@ -22473,10 +22618,11 @@ export class AgentFramework { 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}`; + what = `still can't be found and read 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 was unknown; what ` + - `was stored counts as written and isn't stored again`; + what = `${outcome.named ? 'was found through the copy a module named' : '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${stays}`; } else if (outcome) { what = `threw after ${stored(outcome)}; what was stored counts as written and isn't stored again: ${reason}`; } else { @@ -23065,17 +23211,26 @@ export class AgentFramework { * as addMessage answers a stored write. Null when it hasn't landed, or is * no longer in the store. */ private landedRow(cm: ContextManager, writeId: string): StoredMessage | null { - return cm.queryMessages({ metadata: { deferredWriteId: writeId } }).messages[0] ?? null; + try { + return cm.queryMessages({ metadata: { deferredWriteId: writeId } }).messages[0] ?? null; + } catch { + // As landedCopiesById falls back: a blob missing from another + // message can't hide this write's row. Throws when the scan can't + // reach every message either. + const first = this.landedPositions(cm, writeId)[0]; + return first === undefined ? null : messageAt(cm, first); + } } /** * 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 (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. + * Found, the write landed (settleFound), where its copy sits, a known + * limit (writeDrainedInOrder). 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 unmarked = false; @@ -23107,7 +23262,7 @@ export class AgentFramework { * 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 { + private settleFound(msg: DeferredWrite, agent: Agent, stored: MessageId, landed: { count?: number; sharded: boolean; named?: true }): void { this.deferredMessages = this.deferredMessages.filter((m) => m !== msg); this.unackedDeferredWrites.push(msg); this.noteStored(agent, msg.metadata, stored); @@ -23120,15 +23275,17 @@ export class AgentFramework { * 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. + * too. One row of a sharded body says it's a shard, without a count, + * since one row doesn't tell how many landed. */ - private settleByCopy(reader: Agent, row: { id: MessageId; metadata?: unknown } | null): void { + private settleByCopy(reader: Agent, row: { id: MessageId; metadata?: unknown; bodyGroupId?: string } | 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 }); + const sharded = row.bodyGroupId !== undefined; + this.settleFound(msg, target, row.id, { ...(sharded ? {} : { count: 1 }), sharded, named: true }); } /** A held write whose landing stays unknown after settling, its store diff --git a/src/module-registry.ts b/src/module-registry.ts index daba429ca..fe13b7ae0 100644 --- a/src/module-registry.ts +++ b/src/module-registry.ts @@ -230,7 +230,8 @@ export class ModuleRegistry { /** * Set state for a module by name. - * Used by the framework to apply stateUpdate from EventResponse atomically. + * Used by the framework to apply stateUpdate from EventResponse after its + * message changes. */ setModuleState(moduleName: string, state: unknown): void { const ctx = this.moduleContexts.get(moduleName); diff --git a/src/modules/discord/index.ts b/src/modules/discord/index.ts index 6fb0b4caa..054a5b352 100644 --- a/src/modules/discord/index.ts +++ b/src/modules/discord/index.ts @@ -828,7 +828,7 @@ export class DiscordModule implements Module { const stored = this.storedRow(msg.discordMessageId); if (live) live.phase = 'writing'; - // Return declarative response - framework applies all writes atomically + // Return declarative response - the framework applies its writes return { ...(stored ? {} : { addMessages: [ diff --git a/src/types/framework.ts b/src/types/framework.ts index ccdf5628f..c1940fa8d 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -16,6 +16,13 @@ export type { TraceEvent, TraceEventListener } from './trace.js'; export interface ModuleProcessResponse { moduleName: string; response: EventResponse; + /** + * Set when applying the response threw: the operation it threw at, such + * as `the edit of `, and the error's message. What the response asked + * for before that was applied, that operation may have taken effect in + * part or whole, and the rest of it wasn't (EventResponse). + */ + threw?: { at: string; error: string }; } /** diff --git a/src/types/module.ts b/src/types/module.ts index 2421d5e78..158761b5b 100644 --- a/src/types/module.ts +++ b/src/types/module.ts @@ -512,7 +512,15 @@ export interface AgentInfo { export type AgentStatus = 'idle' | 'inferring' | 'streaming' | 'waiting_for_tools' | 'ready'; /** - * Response from a module's event handler. + * Response from a module's event handler. The framework applies it in this + * order: its messages added, edited and removed, then its stateUpdate, then + * its inference request. When one of these throws, such as an edit of a + * message the store doesn't hold, what came before it stays applied, the + * operation that threw may have taken effect in part or whole (a store can + * throw after it wrote), the rest of the response isn't, and a line names + * the module and where it threw; the event's other responses, a tool call's dispatch and its trace + * go on. Content the framework finds malformed refuses the whole response + * before any of it applies. */ export interface EventResponse { /** @@ -546,9 +554,9 @@ export interface EventResponse { toolsChanged?: boolean; /** - * Module state update. Applied atomically with message operations. - * The framework will call setState() with this value after applying - * message changes, ensuring consistent state. + * Module state update. The framework calls setState() with this value + * after applying the response's message changes, and not when one of + * them threw. */ stateUpdate?: unknown; } diff --git a/test/deferred-malformed-unknown-landing.test.ts b/test/deferred-malformed-unknown-landing.test.ts index af3e70884..0fa8dbbe0 100644 --- a/test/deferred-malformed-unknown-landing.test.ts +++ b/test/deferred-malformed-unknown-landing.test.ts @@ -17,6 +17,7 @@ function harness() { const queued: any[] = []; const outcomes: any[] = []; f.primaryAgentName = 'a'; + f.stillUnknownSaid = new Set(); f.deliveryAgent = () => agent; f.storedCountOf = () => count; f.storedCopiesOf = () => readback; diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 764fda18b..eec00a7dd 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -14,10 +14,16 @@ * tune-out holds from the residents is stored at the boundary and not * 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. 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. + * to its prefix (context-manager #155) refuses a later request that replays + * the signed thinking of a reply whose stored prefix holds a message its + * request didn't. A write that threw after it was + * stored, whose landing couldn't be read back, is looked up by its id at + * the boundary that wrote it, and carried there. One that couldn't be looked + * up either is the exception, a known limit: the flush that finds it later + * counts it where it was stored, ahead of what the turn stored since without + * it. On an account where the provider enforces that binding, a later + * request replaying the signed thinking of a reply stored after it, whose + * request didn't carry it, is refused with a 400 (context-manager #155). * * A message a stream declares it carried has been read: RFC-006's consumed * watermark covers it, so a coalesced replacement can no longer remove it. @@ -118,6 +124,15 @@ class InterjectingModule implements Module { const text = (t: string): ContentBlock[] => [{ type: 'text', text: t } as ContentBlock]; +/** Make the framework's stores unsearchable by a deferred write's id while + * context-manager's query throws: its blob-free scan, which stands in + * when that query throws (landedPositions), throws too. Returns the undo. */ +function noScan(framework: AgentFramework): () => void { + const self = framework as unknown as { landedPositions?: unknown }; + self.landedPositions = () => { throw new Error('scan failed'); }; + return () => { delete self.landedPositions; }; +} + /** One line per stored message: participant, then its text or block types. */ function windowOf(framework: AgentFramework): string[] { return framework.getAgent('assistant')!.getContextManager().getAllMessages().map((m) => { @@ -599,7 +614,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { }; } - 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 () => { + it("at a tool boundary, a write that throws after it's stored, whose store then can't be read back, is looked up by its id there, and carried there", async () => { membrane.pushResponse(createMockResponse([ { type: 'text', text: 'Moving.' }, { type: 'tool_use', id: 'c1', name: 'robot--move', input: {} }, @@ -618,15 +633,72 @@ 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.deepEqual(windowOf(framework).slice(-4), [ + 'user: ', + 'Antra: look left', + 'Antra: look right', + 'assistant: Done.', + ], 'stored once, where it landed, and both carried at the boundary, before the reply'); + assert.deepEqual(injected(), ['Antra: look left', 'Antra: look right'], 'both injected at the boundary'); + assert.ok(errors.some((e) => e.includes('threw after it was stored; what was stored counts as written') + && e.includes('failed after storing')), errors.join('\n')); + assert.ok(!errors.some((e) => e.includes("couldn't be read to tell whether it was stored")), 'never marked'); + await framework.stop(); + }); + + it("at a tool boundary, a write that throws after it's stored, whose store can then be neither read back nor searched, is found later and counts as written where it landed, ahead of the turn's reply, a known limit", 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(); + const cm = framework.getAgent('assistant')!.getContextManager(); + const addMessage = cm.addMessage; + const read = cm.getMessageWindow; + const query = cm.queryMessages; + let thrown = false; + let readsFailing = 0; + let queryFailing = false; + cm.addMessage = function (this: unknown, participant, content, ...rest) { + const id = addMessage.call(this, participant, content, ...rest); + if (!thrown && content.some((b) => (b as { text?: string }).text === 'look left')) { + thrown = true; + readsFailing = 2; + queryFailing = true; + throw new Error('failed after storing'); + } + return id; + }; + // The readback, and then the blob-free scan behind the failed query. + cm.getMessageWindow = function (this: unknown, ...args: unknown[]) { + if (readsFailing > 0) { readsFailing--; throw new Error('read failed'); } + return read.apply(this, args as Parameters); + }; + cm.queryMessages = function (this: unknown, ...args: unknown[]) { + if (queryFailing) { queryFailing = false; throw new Error('search failed'); } + return query.apply(this, args as Parameters); + }; + 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.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('failed after storing')), 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')); + assert.ok(errors.some((e) => e.includes("threw, and its store couldn't be read to tell whether it was stored")), errors.join('\n')); + assert.ok(errors.some((e) => e.includes('was found by its id: it was stored by an earlier write') + && e.includes("a later request replaying that reply's signed thinking is refused with a 400 (context-manager #155)")), errors.join('\n')); await framework.stop(); }); @@ -661,7 +733,34 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { throw new Error('failed after storing'); }; - 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 () => { + it("at a tool boundary, a marked write found where an earlier write stored it counts as written there, and isn't injected, though no request carried it", 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(); + const internals = drainInternals(framework); + module.duringTool = () => { + // An earlier write of it landed, though whether it landed wasn't + // known, and no request has carried it. + framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('look left'), { deferredWriteId: 'w-carried' } as never); + internals.deferredMessages.push({ ...entry('w-carried', 'look left'), landing: 'unknown' }); + }; + module.interjections = [{ participant: 'Antra', content: text('look right') }]; + const error = console.error; + console.error = () => {}; + try { + framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); + await framework.runUntilIdle(); + } finally { console.error = error; } + assert.deepEqual(injected(), ['Antra: look right'], 'only what the boundary stored is injected'); + assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'and not queued again'); + await framework.stop(); + }); + + it("a drained write that throws when its store couldn't count its messages before it is looked up by its id there, and counts as written", async () => { const framework = await createFramework(); const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { getMessageCount: () => number }; const count = cm.getMessageCount; @@ -674,40 +773,59 @@ 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)); - 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]); }); + written = internals.writeDrainedInOrder([msg], 'test', storedThenThrows(framework), (m, id) => { landed.push([m.id, id]); }); } finally { console.error = error; } 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'); + assert.equal(msg.landing, undefined, 'found by its id at once, it is never marked'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'and not queued again'); 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. + // A body whose first part landed before the throw stopped its write, + // where a request may have carried it, so it stays where it was stored. framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('part one'), { deferredWriteId: 'w-part' } as never); + framework.getAgent('assistant')!.markContextShown(); const msg = { ...entry('w-part', 'part one'), content: [...text('part one'), ...text('part two')], landing: 'unknown' as const }; + const errors: string[] = []; const error = console.error; - console.error = () => {}; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; 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'); + assert.ok(errors.some((e) => e.includes("it stays where that write stored it, and a reply stored after it whose request didn't carry it was made without it") + && e.includes('(context-manager #155)')), errors.join('\n')); + await framework.stop(); + }); + + it("a drained write whose landing is unknown, found by its id where no request has carried its copy, counts as written where it sits, ahead of what was stored since, and the log names the refusal", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const cm = framework.getAgent('assistant')!.getContextManager(); + cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-sits' } as never); + cm.addMessage('assistant', text('Done.')); + const msg = { ...entry('w-sits', 'look left'), 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([msg], 'test', () => { throw new Error('written again'); }); + } finally { console.error = error; } + assert.deepEqual(windowOf(framework).slice(-2), ['Antra: look left', 'assistant: Done.'], 'stored once, where it landed'); + assert.deepEqual(written.get(msg)?.content, text('look left'), 'counted as written'); + assert.equal(written.get(msg)?.foundLater, true, "a boundary doesn't inject it"); + assert.ok(errors.some((e) => e.includes('w-sits') && e.includes('(context-manager #155)')), errors.join('\n')); await framework.stop(); }); @@ -731,6 +849,7 @@ 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 = function () { throw new Error('search failed'); }; + const unscan = noScan(framework); const internals = drainInternals(framework); const first = { ...entry('w-held', 'look left'), landing: 'unknown' as const }; // Another target's entry, queued between them, isn't held by them. @@ -747,8 +866,8 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { 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')); - cm.queryMessages = query; + assert.ok(errors.some((e) => e.includes("still can't be found and read by its id")), errors.join('\n')); + cm.queryMessages = query; unscan(); await framework.stop(); }); @@ -767,7 +886,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { } 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')); + assert.ok(!errors.some((e) => e.includes("still can't be found and read by its id")), errors.join('\n')); await framework.stop(); }); @@ -849,6 +968,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { 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'); }; + noScan(framework); const error = console.error; console.error = () => {}; try { await framework.stop(); } finally { console.error = error; } @@ -937,12 +1057,13 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const query = cm.queryMessages; const add = cm.addMessage; cm.queryMessages = () => { throw new Error('search failed'); }; + const unscan = noScan(framework); 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; } + try { internals.releaseWindow(agent); } finally { console.error = error; cm.queryMessages = query; unscan(); 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(); @@ -970,12 +1091,13 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const cm = agent.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; const query = cm.queryMessages; cm.queryMessages = () => { throw new Error('search failed'); }; + const unscan = noScan(framework); 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; + cm.queryMessages = query; unscan(); 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(); @@ -1093,12 +1215,13 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const [copy] = cm.queryMessages({ metadata: { deferredWriteId: id } }).messages; const query = cm.queryMessages; cm.queryMessages = () => { throw new Error('search failed'); }; + const unscan = noScan(framework); 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; } + } finally { cm.queryMessages = query; unscan(); 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(); @@ -1112,12 +1235,13 @@ 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 unscan = noScan(framework); const error = console.error; console.error = () => {}; try { 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; } + } finally { cm.queryMessages = query; unscan(); 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(); @@ -1151,9 +1275,10 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const [mine] = cm.queryMessages({ metadata: { deferredWriteId: 'w-mine' } }).messages; const query = cm.queryMessages; cm.queryMessages = () => { throw new Error('search failed'); }; + const unscan = noScan(framework); const error = console.error; console.error = () => {}; - try { internals.getMessage(mine!.id); } finally { cm.queryMessages = query; console.error = error; } + try { internals.getMessage(mine!.id); } finally { cm.queryMessages = query; unscan(); console.error = error; } assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-theirs', 'unknown']], "the other write stays marked"); await framework.stop(); }); @@ -1178,6 +1303,279 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { } }); + it("a module's edit or removal by a stored id, while the store can't be searched, settles the write through the copy it names", async () => { + for (const op of ['edit', 'remove'] as const) { + const id = `w-named-${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 unscan = noScan(framework); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + if (op === 'edit') internals.editMessage(copy!.id, text('look up')); + else internals.removeMessage(copy!.id); + } finally { console.error = error; cm.queryMessages = query; unscan(); } + assert.deepEqual(copiesOf(framework, id), op === 'edit' ? ['look up'] : [], `${op}: the row it names is changed`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `${op}: its write is settled by the row`); + assert.ok(errors.some((e) => e.includes(id) && e.includes('was found through the copy a module named')), errors.join('\n')); + await framework.stop(); + } + }); + + it("a held write's lookup by its id falls back to a blob-free scan when context-manager's query throws, as a blob missing from another message makes it throw", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-blob', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('Blob not found: 0123'); }; + const error = console.error; + console.error = () => {}; + let settled: { held?: true; content: ContentBlock[] } | null; + let again: { held?: true; content: ContentBlock[] } | null; + try { + settled = internals.getMessage(heldHandle('w-blob')); + // Settled, the handle reaches its row by the same fallback (landedRow). + again = internals.getMessage(heldHandle('w-blob')); + } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'found, and settled'); + assert.ok(settled && settled.held === undefined && again && again.held === undefined, 'its stored row, both times'); + assert.deepEqual(again?.content, text('look left')); + assert.deepEqual(copiesOf(framework, 'w-blob'), ['look left'], 'stored once'); + await framework.stop(); + }); + + it("a blob-free scan that comes back short answers that the store can't be searched, not that the write is absent", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-short', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => unknown; + getMessageWindow: (offset: number, limit: number, opts?: { resolveBlobs?: boolean }) => { messages: unknown[] }; + }; + const query = cm.queryMessages; + const read = cm.getMessageWindow; + cm.queryMessages = () => { throw new Error('search failed'); }; + cm.getMessageWindow = function (this: unknown, offset, limit, opts) { + const window = read.call(this, offset, limit, opts); + return opts?.resolveBlobs === false ? { ...window, messages: window.messages.slice(1) } : window; + }; + const error = console.error; + console.error = () => {}; + try { internals.getMessage(heldHandle('w-short')); } finally { cm.queryMessages = query; cm.getMessageWindow = read; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-short', 'unknown']], 'still a doubt'); + assert.deepEqual(copiesOf(framework, 'w-short'), ['look left'], 'and stored once'); + await framework.stop(); + }); + + it("a held write whose store can't be searched is said to be a doubt once, until a lookup of it succeeds", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-said', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const said = () => errors.filter((e) => e.includes('could not search') && e.includes('w-said')).length; + try { + cm.queryMessages = () => { throw new Error('search failed'); }; + const unscan = noScan(framework); + internals.getMessage(heldHandle('w-said')); + internals.getMessage(heldHandle('w-said')); + internals.deferredWriteReadOrGone(framework.getAgent('assistant'), 'w-said'); + assert.equal(said(), 1, 'said once over three reads'); + cm.queryMessages = query; + unscan(); + } finally { cm.queryMessages = query; console.error = error; } + await framework.stop(); + }); + + it("a held write whose found copy is a body stored in shards counts as written where its shards sit, and says how many landed, by the query or by the scan", async () => { + for (const by of ['query', 'scan'] as const) { + const id = `w-body-${by}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + shardOn(framework, 'part one'); + const cm = framework.getAgent('assistant')!.getContextManager(); + cm.addMessage('Antra', [...text('part one'), ...text('part two')], { deferredWriteId: id } as never); + internals.deferredMessages.push({ ...entry(id, 'part one'), content: [...text('part one'), ...text('part two')], landing: 'unknown' }); + const query = cm.queryMessages; + if (by === 'scan') (cm as unknown as { queryMessages: () => never }).queryMessages = () => { throw new Error('Blob not found: 0123'); }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.getMessage(heldHandle(id)); } finally { console.error = error; cm.queryMessages = query; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `${by}: settled where it sits`); + assert.deepEqual(copiesOf(framework, id), ['part one', 'part two'], `${by}: its shards stay`); + assert.ok(errors.some((e) => e.includes(id) && e.includes('2 shards of its body were stored')), errors.join('\n')); + await framework.stop(); + } + }); + + it("a held write settled through one stored shard of its body a module names says a shard of its body was stored", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + shardOn(framework, 'part one'); + const cm = framework.getAgent('assistant')!.getContextManager(); + cm.addMessage('Antra', [...text('part one'), ...text('part two')], { deferredWriteId: 'w-shard' } as never); + internals.deferredMessages.push({ ...entry('w-shard', 'part one'), content: [...text('part one'), ...text('part two')], landing: 'unknown' }); + const [first] = cm.queryMessages({ metadata: { deferredWriteId: 'w-shard' } } as never).messages; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.getMessage(first!.id); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'settled by the row'); + assert.ok(errors.some((e) => e.includes('w-shard') && e.includes('a shard of its body was stored by an earlier write')), errors.join('\n')); + await framework.stop(); + }); + + it("one write's copy whose own blob can't be read leaves that write a doubt, and the others are answered", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-bad', 'look left'); + heldAndStored(framework, 'w-good', 'look right'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => unknown; + getMessageWindow: (offset: number, limit: number, opts?: { resolveBlobs?: boolean }) => { messages: Array<{ metadata?: { deferredWriteId?: string } }> }; + }; + const query = cm.queryMessages; + const read = cm.getMessageWindow; + cm.queryMessages = () => { throw new Error('Blob not found: 0123'); }; + cm.getMessageWindow = function (this: unknown, offset, limit, opts) { + const window = read.call(this, offset, limit, opts); + if (opts?.resolveBlobs !== false && window.messages.some((m) => m.metadata?.deferredWriteId === 'w-bad')) throw new Error('Blob not found: 4567'); + return window; + }; + const error = console.error; + console.error = () => {}; + const errors: string[] = []; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + internals.getMessage(heldHandle('w-good')); + internals.getMessage(heldHandle('w-bad')); + } finally { cm.queryMessages = query; cm.getMessageWindow = read; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-bad', 'unknown']], 'only the write whose copy can\'t be read stays a doubt'); + assert.equal(errors.filter((e) => e.includes('found deferred write w-bad') && e.includes('could not read its stored copy') && e.includes('Blob not found: 4567')).length, 1, `said once over two reads:\n${errors.join('\n')}`); + assert.ok(!errors.some((e) => e.includes('could not search')), 'the store itself could be searched'); + await framework.stop(); + }); + + it("the blob-free scan reads past its first window, and a matching row's own image comes back resolved", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage: (...args: unknown[]) => string; queryMessages: (...args: unknown[]) => unknown }; + for (let i = 0; i < 520; i++) cm.addMessage('Antra', text(`filler ${i}`)); + const image = { type: 'image', source: { type: 'base64', media_type: 'image/png', data: 'iVBORw0KGgo=' } } as unknown as ContentBlock; + cm.addMessage('Antra', [image], { deferredWriteId: 'w-far' }); + framework.getAgent('assistant')!.markContextShown(); + internals.deferredMessages.push({ ...entry('w-far', 'look left'), content: [image], landing: 'unknown' }); + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('Blob not found: 0123'); }; + const error = console.error; + console.error = () => {}; + let row: { held?: true; content: ContentBlock[] } | null; + try { row = internals.getMessage(heldHandle('w-far')); } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'found past the first window, and settled'); + assert.equal(row?.held, undefined); + assert.equal((row?.content[0] as { source?: { data?: string } } | undefined)?.source?.data, 'iVBORw0KGgo=', 'its image resolved'); + await framework.stop(); + }); + + it("a lookup whose query works doesn't scan the branch", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-fast', 'look left'); + const unscan = noScan(framework); + const error = console.error; + console.error = () => {}; + try { internals.getMessage(heldHandle('w-fast')); } finally { console.error = error; unscan(); } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'settled by the query alone'); + await framework.stop(); + }); + + it("a drain that can't search a marked write's store says it holds what's behind it once, until a lookup of it succeeds", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-twice-said', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const said = () => errors.filter((e) => e.includes('w-twice-said') && e.includes("still can't be found and read")).length; + const searchSaid = () => errors.filter((e) => e.includes('could not search') && e.includes('w-twice-said')).length; + try { + cm.queryMessages = () => { throw new Error('search failed'); }; + let unscan = noScan(framework); + internals.writeDrainedInOrder(internals.deferredMessages.splice(0), 'test', () => { throw new Error('written again'); }); + internals.writeDrainedInOrder(internals.deferredMessages.splice(0), 'test', () => { throw new Error('written again'); }); + assert.equal(said(), 1, 'said once over two drains'); + assert.equal(searchSaid(), 1); + // A lookup succeeds, and a later failure is said again. + cm.queryMessages = query; unscan(); + internals.getMessage(heldHandle('w-twice-said')); + drainInternals(framework).deferredMessages.push({ ...entry('w-twice-said', 'look left'), landing: 'unknown' }); + cm.queryMessages = () => { throw new Error('search failed'); }; + unscan = noScan(framework); + internals.getMessage(heldHandle('w-twice-said')); + assert.equal(searchSaid(), 2, 'said again after a lookup succeeded'); + internals.writeDrainedInOrder(internals.deferredMessages.splice(0), 'test', () => { throw new Error('written again'); }); + assert.equal(said(), 2, "the drain's line too"); + unscan(); + } finally { cm.queryMessages = query; console.error = error; } + await framework.stop(); + }); + + it("a marked write the blob-free scan reads the whole branch without finding is written, once", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { addMessage: (...args: unknown[]) => string; queryMessages: (...args: unknown[]) => unknown }; + cm.addMessage('Antra', text('something else')); + const msg = { ...entry('w-absent', 'look left'), landing: 'unknown' as const }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('Blob not found: 0123'); }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + internals.writeDrainedInOrder([msg], 'test', (m) => cm.addMessage(m.participant, m.content, { deferredWriteId: m.id })); + } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(copiesOf(framework, 'w-absent'), ['look left'], 'stored once'); + assert.ok(!errors.some((e) => e.includes('could not search')), errors.join('\n')); + await framework.stop(); + }); + + it("the scan reads its windows without blobs, so a blob it can't resolve elsewhere on the branch doesn't stop it", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-blobfree', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => unknown; + getMessageWindow: (offset: number, limit: number, opts?: { resolveBlobs?: boolean }) => unknown; + }; + const query = cm.queryMessages; + const read = cm.getMessageWindow; + cm.queryMessages = () => { throw new Error('Blob not found: 0123'); }; + // Any read of more than one message with its blobs meets the missing one. + cm.getMessageWindow = function (this: unknown, offset, limit, opts) { + if (opts?.resolveBlobs !== false && limit > 1) throw new Error('Blob not found: 0123'); + return read.call(this, offset, limit, opts); + }; + const error = console.error; + console.error = () => {}; + try { internals.getMessage(heldHandle('w-blobfree')); } finally { cm.queryMessages = query; cm.getMessageWindow = read; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'found, and 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); @@ -1252,6 +1650,7 @@ 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 unscan = noScan(framework); const error = console.error; console.error = () => {}; try { @@ -1259,7 +1658,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { 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; } + } finally { cm.queryMessages = query; unscan(); 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(); diff --git a/test/mcpl-coalescing.test.ts b/test/mcpl-coalescing.test.ts index f3ed3e4a2..bc7f8771f 100644 --- a/test/mcpl-coalescing.test.ts +++ b/test/mcpl-coalescing.test.ts @@ -1681,6 +1681,11 @@ test("a wake to read a render starts its turn when the store can't say whether t const cm = f.framework.getAgent('agent')!.getContextManager() as unknown as { queryMessages: (...a: unknown[]) => unknown }; const query = cm.queryMessages; cm.queryMessages = () => { throw new Error('a blob the store names is missing'); }; + // The blob-free scan that stands in when the query throws can't read the + // store either. + const scanning = f.framework as unknown as { landedPositions?: unknown }; + scanning.landedPositions = () => { throw new Error('a blob the store names is missing, and the scan failed too'); }; + t.after(() => { delete scanning.landedPositions; }); internals.pendingRequests.push({ agentName: 'agent', reason: 'mcpl:push-event', source: 'editor', timestamp: Date.now(), readsDeferredWrite: 'render-1' }); const triggers: unknown[] = []; const start = internals.startAgentStream; diff --git a/test/process-response-edit-refusals.test.ts b/test/process-response-edit-refusals.test.ts new file mode 100644 index 000000000..e9a4ec885 --- /dev/null +++ b/test/process-response-edit-refusals.test.ts @@ -0,0 +1,215 @@ +/** + * A module's process response is applied in order: its messages added, + * edited and removed, then its state update and inference request + * (applyProcessResponse). One that throws as it applies, such as at an edit + * of a held write that may already be stored while its store can't be + * searched (refuseUnsettled), or a removal the store refuses, stops there: + * what came before stays applied, the operation that threw may have taken + * effect in part or whole, and the rest of it isn't, as before. A + * line names the module and where it threw, and the other modules' + * responses, a tool call's dispatch and the event's trace go on. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { ContentBlock } from '@animalabs/membrane'; +import type { Module, ModuleContext, ProcessEvent, ProcessState, EventResponse, ToolDefinition, ToolCall, ToolResult } from '../src/index.js'; +import { AgentFramework } from '../src/index.js'; +import { heldHandle } from '../src/held-writes.js'; +import { MockMembrane } from './helpers/mock-membrane.js'; + +const text = (t: string): ContentBlock[] => [{ type: 'text', text: t } as ContentBlock]; + +/** A module that answers a `ping`, or a call of its `act` tool, with + * `onPing`, and records the calls its tool is given. */ +class Responder implements Module { + readonly name: string; + ctx!: ModuleContext; + onPing: EventResponse = {}; + calls: string[] = []; + constructor(name: string) { this.name = name; } + async start(ctx: ModuleContext): Promise { this.ctx = ctx; } + async stop(): Promise {} + getTools(): ToolDefinition[] { return [{ name: 'act', description: 'Acts.', inputSchema: { type: 'object' } }]; } + async handleToolCall(call: ToolCall): Promise { this.calls.push(call.id); return { success: true, data: {} }; } + async onProcess(event: ProcessEvent, _state: ProcessState): Promise { + const type = (event as { type: string }).type; + return type === 'ping' || type === 'tool-call' ? this.onPing : {}; + } +} + +async function harness(modules: Responder[]) { + const dir = mkdtempSync(join(tmpdir(), 'process-edit-refusals-af-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'assistant', model: 'test', systemPrompt: 'test' }], + modules, + processLogging: { persist: true }, + }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const traces: Array> = []; + framework.onTrace((e) => { traces.push(e as unknown as Record); }); + const window = () => framework.getAgent('assistant')!.getContextManager().getAllMessages() + .map((m) => `${m.participant}: ${m.content.map((b) => (b.type === 'text' ? (b as { text: string }).text : `<${b.type}>`)).join(' ')}`); + return { + framework, errors, traces, window, + close: async () => { + console.error = error; + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + }, + }; +} + +test("a response that throws at an edit of a held write that may already be stored, while its store can't be searched, stops there, and the rest of the event goes on", async (t) => { + const refused = new Responder('refused'); + const after = new Responder('after'); + const x = await harness([refused, after]); t.after(x.close); + const internals = x.framework as unknown as { + deferredMessages: Array>; + landedPositions?: unknown; + }; + const cm = x.framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage: (...args: unknown[]) => string; queryMessages: (...args: unknown[]) => unknown; + }; + // A held write whose first write landed, and whose store can't be searched + // to tell: context-manager's query throws, and so does the blob-free scan + // behind it. + cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-blind' }); + internals.deferredMessages.push({ id: 'w-blind', seq: 1, participant: 'Antra', content: text('look left'), heldAt: Date.now(), landing: 'unknown' }); + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + internals.landedPositions = () => { throw new Error('scan failed'); }; + refused.onPing = { + addMessages: [{ participant: 'Refused', content: text('added before the edit') }], + editMessages: [{ messageId: heldHandle('w-blind'), content: text('look up') }], + stateUpdate: { seen: true }, + }; + after.onPing = { addMessages: [{ participant: 'After', content: text('from after') }] }; + try { + x.framework.pushEvent({ type: 'ping', source: 'test' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + } finally { cm.queryMessages = query; delete internals.landedPositions; } + const window = x.window(); + assert.ok(window.includes('After: from after'), `the next module's response was applied:\n${window.join('\n')}`); + assert.ok(window.includes('Refused: added before the edit'), 'what came before the throw stays applied'); + assert.ok(x.errors.some((e) => e.startsWith("[process] refused's response to ping threw at the edit of held:w-blind: what it asked for before that was applied, that operation may have taken effect in part or whole, and the rest of it wasn't") + && e.includes("may already be stored, and its store can't be searched")), x.errors.join('\n')); + const registry = (x.framework as unknown as { moduleRegistry: { getModuleState(name: string): unknown } }).moduleRegistry; + assert.equal(registry.getModuleState('refused'), null, 'its state update, after the throw, was not applied'); + const completed = x.traces.filter((e) => e.type === 'process:completed' && (e.processEvent as { type: string }).type === 'ping'); + assert.equal(completed.length, 1, 'the event completed, traced'); + const listed = (completed[0]!.responses as Array<{ moduleName: string; threw?: { at: string; error: string } }>); + assert.deepEqual(listed.map((r) => [r.moduleName, r.threw?.at]), [['refused', 'the edit of held:w-blind'], ['after', undefined]], 'the trace names where the response threw, and the other as applied'); + assert.ok(listed[0]!.threw?.error.includes("can't be searched"), listed[0]!.threw?.error); + const logged = x.framework.queryProcessLogs({ eventType: 'ping' }).entries + .map(({ entry }) => (entry.responses as Array<{ moduleName: string; threw?: { at: string } }>).map((r) => [r.moduleName, r.threw?.at])); + assert.deepEqual(logged, [[['refused', 'the edit of held:w-blind'], ['after', undefined]]], 'and so does the process log'); + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content]), [['w-blind', text('look left')]], 'the held write is unedited'); +}); + +test("a response that throws at a removal the store refuses stops there, and the other modules' responses still apply", async (t) => { + const refused = new Responder('refused'); + const after = new Responder('after'); + const x = await harness([refused, after]); t.after(x.close); + refused.onPing = { removeMessages: ['no-such-message'] }; + after.onPing = { addMessages: [{ participant: 'After', content: text('from after') }] }; + x.framework.pushEvent({ type: 'ping', source: 'test' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + assert.ok(x.window().includes('After: from after'), x.window().join('\n')); + assert.ok(x.errors.some((e) => e.startsWith("[process] refused's response to ping threw at the removal of no-such-message")), x.errors.join('\n')); +}); + +test("a response whose edit throws stops there, as before: its later edits and removals and its state update aren't applied", async (t) => { + const refused = new Responder('refused'); + const x = await harness([refused]); t.after(x.close); + const cm = x.framework.getAgent('assistant')!.getContextManager(); + const kept = cm.addMessage('Antra', text('before')); + const gone = cm.addMessage('Antra', text('to remove')); + refused.onPing = { + editMessages: [ + { messageId: 'no-such-message', content: text('nowhere') }, + { messageId: kept, content: text('after') }, + ], + removeMessages: [gone], + stateUpdate: { seen: true }, + }; + x.framework.pushEvent({ type: 'ping', source: 'test' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + const window = x.window(); + assert.ok(window.includes('Antra: before') && window.includes('Antra: to remove'), `nothing after the throw applied:\n${window.join('\n')}`); + assert.ok(x.errors.some((e) => e.startsWith("[process] refused's response to ping threw at the edit of no-such-message")), x.errors.join('\n')); + const registry = (x.framework as unknown as { moduleRegistry: { getModuleState(name: string): unknown } }).moduleRegistry; + assert.equal(registry.getModuleState('refused'), null, 'its state update was not applied'); +}); + +test("a response whose removals are applied in order keeps those before a throw, and not those after it", async (t) => { + const refused = new Responder('refused'); + const x = await harness([refused]); t.after(x.close); + const cm = x.framework.getAgent('assistant')!.getContextManager(); + const first = cm.addMessage('Antra', text('first')); + const last = cm.addMessage('Antra', text('last')); + refused.onPing = { removeMessages: [first, 'no-such-message', last], stateUpdate: { seen: true } }; + x.framework.pushEvent({ type: 'ping', source: 'test' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + assert.deepEqual(x.window().filter((line) => line === 'Antra: first' || line === 'Antra: last'), ['Antra: last']); + const registry = (x.framework as unknown as { moduleRegistry: { getModuleState(name: string): unknown } }).moduleRegistry; + assert.equal(registry.getModuleState('refused'), null); +}); + +test("a response that throws while adding a message stops there, and the other modules' responses still apply", async (t) => { + const refused = new Responder('refused'); + const after = new Responder('after'); + const x = await harness([refused, after]); t.after(x.close); + const self = x.framework as unknown as { addMessage: (participant: string, content: ContentBlock[], metadata?: unknown) => string }; + const add = self.addMessage; + self.addMessage = function (this: unknown, participant, content, metadata) { + if (content.some((b) => (b as { text?: string }).text === 'second')) throw new Error('the store refused it'); + return add.call(this, participant, content, metadata); + }; + refused.onPing = { addMessages: [{ participant: 'Antra', content: text('first') }, { participant: 'Antra', content: text('second') }, { participant: 'Antra', content: text('third') }] }; + after.onPing = { addMessages: [{ participant: 'After', content: text('from after') }] }; + try { + x.framework.pushEvent({ type: 'ping', source: 'test' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + } finally { self.addMessage = add; } + const window = x.window(); + assert.ok(window.includes('Antra: first') && !window.includes('Antra: third'), window.join('\n')); + assert.ok(window.includes('After: from after'), window.join('\n')); + assert.ok(x.errors.some((e) => e.startsWith("[process] refused's response to ping threw at adding message 2 of 3") && e.includes('the store refused it')), x.errors.join('\n')); +}); + +test("a response whose edits and removals all apply has its state update applied", async (t) => { + const fine = new Responder('fine'); + const x = await harness([fine]); t.after(x.close); + const cm = x.framework.getAgent('assistant')!.getContextManager(); + const kept = cm.addMessage('Antra', text('before')); + const gone = cm.addMessage('Antra', text('to remove')); + fine.onPing = { editMessages: [{ messageId: kept, content: text('after') }], removeMessages: [gone], stateUpdate: { seen: true } }; + x.framework.pushEvent({ type: 'ping', source: 'test' } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + const registry = (x.framework as unknown as { moduleRegistry: { getModuleState(name: string): unknown } }).moduleRegistry; + assert.equal((registry.getModuleState('fine') as { seen?: boolean } | null)?.seen, true); +}); + +test("a tool-call event whose module's response throws at an edit still dispatches the call", async (t) => { + const refused = new Responder('refused'); + const x = await harness([refused]); t.after(x.close); + refused.onPing = { editMessages: [{ messageId: 'no-such-message', content: text('nowhere') }] }; + x.framework.pushEvent({ + type: 'tool-call', + callId: 'c-act', + agentName: 'assistant', + moduleName: 'refused', + toolName: 'act', + call: { id: 'c-act', name: 'refused--act', input: {} }, + } as unknown as ProcessEvent); + await x.framework.runUntilIdle(); + for (let i = 0; i < 50 && refused.calls.length === 0; i++) await new Promise((r) => setTimeout(r, 10)); + assert.deepEqual(refused.calls, ['c-act']); + assert.ok(x.errors.some((e) => e.startsWith("[process] refused's response to tool-call threw at the edit of no-such-message")), x.errors.join('\n')); +}); From 52a6f83be6655fdd9b35451efc7e7463dc305432 Mon Sep 17 00:00:00 2001 From: Ansel-2428 Date: Thu, 22 Oct 2026 23:40:01 -0400 Subject: [PATCH 2/3] fix(deferral): the automated review's round - A copy found later is injected at the tool boundary that finds it, as 0.26.0 does. Leaving it out let the boundary move the consumed mark over a copy the agent hadn't read, so a wake to read it could drop, and it skipped the copy's arrival and routing notes. A recompile before that boundary has already carried the copy, so it's carried twice there, as in 0.26.0; the boundary's comment and the changelog name that. - The said-once record's doc and the changelog say a line is said again when its failure changes. - Each failed lookup is said once per kind of failure, so a scan that can't run and a copy that can't be read are each said, and the line names the query's own error, which names the missing blob. landedRow throws an error naming both when its fallback fails. - The README gives the order a response applies in: message changes, then the state update, then the inference request. - Tests: the lookup at the boundary that finds a write absent (it's queued unmarked and stored once), the two kinds of failure said apart, landedRow's error when its store can be searched neither way, and the tool-call dispatch awaited rather than polled. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- changelog.d/deferral-found-later.fixed.md | 5 +- src/framework.ts | 80 ++++++++++--------- test/deferred-window-only.test.ts | 85 +++++++++++++++++++-- test/process-response-edit-refusals.test.ts | 17 ++++- 5 files changed, 140 insertions(+), 49 deletions(-) diff --git a/README.md b/README.md index ef49a3115..beec57c59 100644 --- a/README.md +++ b/README.md @@ -122,7 +122,7 @@ interface Module { } ``` -An `EventResponse` can add/edit/remove messages, request inference, signal tool changes, and update module state after its message changes. The framework applies it in that order; when one step throws, what came before it stays applied, that step may have taken effect in part, and the rest of that response isn't: +An `EventResponse` can add/edit/remove messages, update module state, request inference, and signal tool changes. The framework applies its message changes, then its state update, then its inference request; when one of these throws, what came before it stays applied, that step may have taken effect in part, and the rest of that response isn't: ```typescript return { diff --git a/changelog.d/deferral-found-later.fixed.md b/changelog.d/deferral-found-later.fixed.md index 9a584c70f..873fdb364 100644 --- a/changelog.d/deferral-found-later.fixed.md +++ b/changelog.d/deferral-found-later.fixed.md @@ -1,5 +1,4 @@ -- A deferred write that throws after it's stored, whose store can't be read back to tell whether it landed, is now looked up by its id at the boundary that wrote it, so it counts as written there, where it landed, and at a tool boundary it's injected with what that boundary stores, before the turn's later replies. Before, it was queued as a doubt and counted at a later flush where it first landed, ahead of later replies whose requests never held it. That's still so, as a known limit, for a write that can be neither read back nor looked up there: on an account where the provider enforces its binding of signed thinking to the prefix it was made with, a later request replaying the signed thinking of a reply stored after it, whose request didn't carry it, is refused with a 400 (context-manager #155). -- A held write found where an earlier write stored it is no longer injected into the live turn at a tool boundary, where a compile might already have carried it. The agent reads it at its next compile, from the store. -- Looking a held write up by its id no longer fails because another message's blob is missing: where context-manager's query throws, the branch is scanned without blobs, and only the rows that match are read with theirs. Before, every such write stayed a doubt, holding what was queued behind it for its agent, and each read logged it again. The lines saying a store can't be searched, and that a drain holds what's queued behind such a write, are now said once per write until a lookup succeeds. +- A deferred write that throws after it's stored, whose store can't be read back to tell whether it landed, is now looked up by its id at the boundary that wrote it, so it counts as written there, where it landed, and at a tool boundary it's injected with what that boundary stores, before the turn's later replies. Before, it was queued as a doubt and counted at a later flush where it first landed, ahead of later replies whose requests never held it. That's still so, as a known limit, for a write that can be neither read back nor looked up there: on an account where the provider enforces its binding of signed thinking to the prefix it was made with, a later request replaying the signed thinking of a reply stored after it, whose request didn't carry it, is refused with a 400 (context-manager #155). A recompile before the next tool boundary, such as a retry, carries such a copy where it was stored, and the boundary that then finds it injects it again, as before, so a later request replaying the signed thinking of a reply made after that boundary is refused the same way. +- Looking a held write up by its id no longer fails because another message's blob is missing: where context-manager's query throws, the branch is scanned without blobs, and only the rows that match are read with theirs. Before, every such write stayed a doubt, holding what was queued behind it for its agent, and each read logged it again. The lines saying a store can't be searched, and that a drain holds what's queued behind such a write, are now said once per write until a lookup succeeds. The first is said again when its failure changes, and names the query's own error, which names the missing blob. - A module's `onProcess` response that throws as the framework applies it (at an edit of a held write that may already be stored while its store can't be searched, an edit of a message the store doesn't hold, or any store error) no longer ends the event's processing. The response stops where it threw, as before: what came before the throw stays applied, the operation that threw may have taken effect in part or whole, and the rest of it, its state update and inference request included, isn't. A line now names the module and where it threw, `process:completed` and the process log name the response with where it threw (`ModuleProcessResponse.threw`), and the other modules' responses, a `tool-call` event's dispatch, the event's trace and its process log go on. Before, the throw escaped the event's processing and all of these were lost. `EventResponse`'s docs, the README and two comments no longer say a response is applied atomically, which it wasn't: they say it's applied in order, up to a throw. - The log line for a held write settled through a stored row a module names says when that row is one shard of a body, rather than that the whole write was stored. diff --git a/src/framework.ts b/src/framework.ts index 94c4595d3..f5a3ffab7 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -854,10 +854,6 @@ interface DrainedWrite { id: MessageId; content: ContentBlock[]; parts: number; - /** A marked entry found where an earlier write stored it: a tool - * boundary doesn't inject it, and the agent reads it at its next - * compile, where the store holds it. */ - foundLater?: true; } /** What later compiles render from the store for the copies of one write @@ -2805,10 +2801,11 @@ export class AgentFramework { */ private deferredScanFrom = new Map(); private deferredMessages: DeferredWrite[] = []; - /** The marked held writes whose store couldn't be searched, said once - * each until a lookup of them succeeds (landedCopiesById), and those a - * drain has said it holds what's queued behind (writeDrainedInOrder). */ - private unsearchableSaid = new Set(); + /** The marked held writes whose store couldn't be searched, each with the + * failure said, said once until a lookup of them succeeds and again when + * the failure changes (landedCopiesById), and those a drain has said it + * holds what's queued behind (writeDrainedInOrder). */ + private unsearchableSaid = new Map(); private stillUnknownSaid = new Set(); // Turn-alive markers (2026-07-31 Mythos phantom-skip incident): agentName → @@ -11862,10 +11859,11 @@ export class AgentFramework { // after it whose request didn't carry it was made without it. On // an account where the provider enforces that binding, a later // request replaying that reply's signed thinking is refused with a - // 400 (context-manager #155). Such a copy isn't injected: a compile - // may already have carried it, and injected or not, it sits ahead - // of those replies in the store. The agent reads it at its next - // compile rather than at this boundary (writeDrainedInOrder). + // 400 (context-manager #155). It's injected at the boundary that + // finds it, after what was stored since (writeDrainedInOrder). A + // recompile before that boundary, such as a retry, has already + // carried it where it was stored, so it's carried twice there, as + // in 0.26.0, and a reply made after it meets the same refusal. // // The queue is drained ONLY at the target agent's own boundary: // draining at another agent's boundary would store the messages @@ -11923,13 +11921,10 @@ export class AgentFramework { // counts as stored, as what landed of it (writeDrainedInOrder). const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) => agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); - for (const [msg, { id, content, parts, foundLater }] of stored) { + for (const [msg, { id, content, parts }] of stored) { // Every other resident reads what this boundary stores; for its // own agent, the boundary weighs what it injects below. this.noteForOtherResidents(agent, msg.metadata, id); - // A copy found where an earlier write left it isn't carried into - // the live turn: the next compile carries it from the store. - if (foundLater) continue; // A message a tune-out holds from the residents is stored (the // subconscious reads it) and left out of every resident's // compiled view; injecting it here would put in the live turn @@ -22393,13 +22388,11 @@ export class AgentFramework { // it, and a reply stored after it whose request didn't carry it was // made without it. On an account where the provider enforces that // binding, a later request replaying that reply's signed thinking is - // refused with a 400 (context-manager #155). A tool boundary doesn't - // inject it; the agent reads it at its next compile. Not found, it's - // written. While - // the store can't be 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. + // refused with a 400 (context-manager #155). Not found, it's written. + // While the store can't be 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) { @@ -22414,7 +22407,7 @@ export class AgentFramework { continue; } if (found.length > 0) { - written.set(msg, { id: found[0]!.id, content: asCompiled(found), parts: found.length, foundLater: true }); + written.set(msg, { id: found[0]!.id, content: asCompiled(found), parts: found.length }); this.logDeferredWriteError(msg, where, undefined, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); onLanded?.(msg, found[0]!.id); continue; @@ -22548,28 +22541,34 @@ export class AgentFramework { */ private landedCopiesById(agent: Agent, deferredWriteId: string): StoredCopy[] | undefined { const cm = agent.getContextManager(); - // Said once until a lookup of it succeeds. - const say = (what: string, error: unknown): void => { - if (this.unsearchableSaid.has(deferredWriteId)) return; - this.unsearchableSaid.add(deferredWriteId); - console.error(`[deferral] ${what}; this is said once until a lookup succeeds:`, error); + // Each failure is said once until a lookup of the write succeeds, and + // again when the failure changes. It names the query's own error, which + // names a missing blob, as well as the scan's. + const say = (kind: 'search' | 'read', what: string, queryError: unknown, error: unknown): void => { + if (this.unsearchableSaid.get(deferredWriteId) === kind) return; + this.unsearchableSaid.set(deferredWriteId, kind); + console.error( + `[deferral] ${what}; context-manager's query threw first (${errorMessage(queryError)}); this is said once ` + + `until a lookup succeeds:`, + error, + ); }; let found: StoredCopy[]; try { found = cm.queryMessages({ metadata: { deferredWriteId } }).messages.map(storedCopy); - } catch { + } catch (queryError) { let at: number[]; try { at = this.landedPositions(cm, deferredWriteId); } catch (error) { - say(`could not search ${agent.name}'s store for deferred write ${deferredWriteId}, to tell whether it was stored`, error); + say('search', `could not search ${agent.name}'s store for deferred write ${deferredWriteId}, to tell whether it was stored`, queryError, error); return undefined; } try { found = at.map((position) => storedCopy(messageAt(cm, position))); } catch (error) { // The scan found its rows; one of them can't be read whole. - say(`found deferred write ${deferredWriteId} in ${agent.name}'s store, but could not read its stored copy`, error); + say('read', `found deferred write ${deferredWriteId} in ${agent.name}'s store, but could not read its stored copy`, queryError, error); return undefined; } } @@ -23213,12 +23212,21 @@ export class AgentFramework { private landedRow(cm: ContextManager, writeId: string): StoredMessage | null { try { return cm.queryMessages({ metadata: { deferredWriteId: writeId } }).messages[0] ?? null; - } catch { + } catch (queryError) { // As landedCopiesById falls back: a blob missing from another // message can't hide this write's row. Throws when the scan can't - // reach every message either. - const first = this.landedPositions(cm, writeId)[0]; - return first === undefined ? null : messageAt(cm, first); + // reach every message either, or the row can't be read, naming the + // query's error too. + try { + const first = this.landedPositions(cm, writeId)[0]; + return first === undefined ? null : messageAt(cm, first); + } catch (error) { + throw new Error( + `the store couldn't be searched for deferred write ${writeId}: its query threw (${errorMessage(queryError)}), ` + + `and then ${errorMessage(error)}`, + { cause: error }, + ); + } } } diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index eec00a7dd..725627506 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -733,7 +733,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { throw new Error('failed after storing'); }; - it("at a tool boundary, a marked write found where an earlier write stored it counts as written there, and isn't injected, though no request carried it", async () => { + it("at a tool boundary, a marked write found where an earlier write stored it counts as written there, and is injected there", async () => { membrane.pushResponse(createMockResponse([ { type: 'text', text: 'Moving.' }, { type: 'tool_use', id: 'c1', name: 'robot--move', input: {} }, @@ -754,12 +754,71 @@ 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.deepEqual(injected(), ['Antra: look right'], 'only what the boundary stored is injected'); + assert.deepEqual([...injected()].sort(), ['Antra: look left', 'Antra: look right'], 'the found copy is injected with what the boundary stored'); assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'and not queued again'); await framework.stop(); }); + it("a drained write that throws before it's stored, whose readback can't say, is looked up by its id there, found absent, queued unmarked, and stored once at the next flush", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage: (...args: unknown[]) => string; + getMessageWindow: (...args: unknown[]) => unknown; + }; + const read = cm.getMessageWindow; + let readFails = true; + cm.getMessageWindow = function (this: unknown, ...args: unknown[]) { + if (readFails) { readFails = false; throw new Error('read failed'); } + return read.apply(this, args); + }; + const msg = { ...entry('w-absent-there', 'look left') }; + const error = console.error; + console.error = () => {}; + try { + internals.writeDrainedInOrder([msg], 'test', () => { throw new Error('refused before storing'); }); + } finally { console.error = error; cm.getMessageWindow = read; } + assert.equal(readFails, false, 'the readback was tried, and failed'); + assert.equal((msg as { landing?: string }).landing, undefined, 'found absent by its id, it is never marked'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), ['w-absent-there'], 'queued again, as unstored'); + internals.writeDrainedInOrder(internals.deferredMessages.splice(0), 'test', + (m) => cm.addMessage(m.participant, m.content, { deferredWriteId: m.id })); + assert.deepEqual(copiesOf(framework, 'w-absent-there'), ['look left'], 'stored once'); + await framework.stop(); + }); + + it("a write whose scan fails and then whose copy can't be read is said again, since the failure changed", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-kinds', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => unknown; + getMessageWindow: (offset: number, limit: number, opts?: { resolveBlobs?: boolean }) => { messages: Array<{ metadata?: { deferredWriteId?: string } }> }; + }; + const query = cm.queryMessages; + const read = cm.getMessageWindow; + cm.queryMessages = () => { throw new Error('Blob not found: 0123'); }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + const unscan = noScan(framework); + internals.getMessage(heldHandle('w-kinds')); + unscan(); + cm.getMessageWindow = function (this: unknown, offset, limit, opts) { + const window = read.call(this, offset, limit, opts); + if (opts?.resolveBlobs !== false && window.messages.some((m) => m.metadata?.deferredWriteId === 'w-kinds')) throw new Error('Blob not found: 4567'); + return window; + }; + internals.getMessage(heldHandle('w-kinds')); + } finally { cm.queryMessages = query; cm.getMessageWindow = read; console.error = error; } + assert.equal(errors.filter((e) => e.includes('could not search') && e.includes('w-kinds')).length, 1, errors.join('\n')); + assert.equal(errors.filter((e) => e.includes('found deferred write w-kinds') && e.includes('could not read its stored copy')).length, 1, errors.join('\n')); + assert.ok(errors.every((e) => !e.includes('w-kinds') || e.includes('Blob not found: 0123')), 'each names the query\'s own error'); + await framework.stop(); + }); + it("a drained write that throws when its store couldn't count its messages before it is looked up by its id there, and counts as written", async () => { const framework = await createFramework(); const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { getMessageCount: () => number }; @@ -818,13 +877,12 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const errors: string[] = []; const error = console.error; console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; - let written: Map; + let written: Map; try { written = internals.writeDrainedInOrder([msg], 'test', () => { throw new Error('written again'); }); } finally { console.error = error; } assert.deepEqual(windowOf(framework).slice(-2), ['Antra: look left', 'assistant: Done.'], 'stored once, where it landed'); assert.deepEqual(written.get(msg)?.content, text('look left'), 'counted as written'); - assert.equal(written.get(msg)?.foundLater, true, "a boundary doesn't inject it"); assert.ok(errors.some((e) => e.includes('w-sits') && e.includes('(context-manager #155)')), errors.join('\n')); await framework.stop(); }); @@ -1283,6 +1341,24 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); + it("a landed write's row, looked up by its handle while the store can be searched neither way, throws an error naming the query's error and the scan's", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage: (...args: unknown[]) => string; + queryMessages: (...args: unknown[]) => unknown; + }; + cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-landed-blind' }); + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('Blob not found: 0123'); }; + const unscan = noScan(framework); + try { + assert.throws(() => internals.getMessage(heldHandle('w-landed-blind')), + /deferred write w-landed-blind: its query threw \(Blob not found: 0123\), and then scan failed/); + } finally { cm.queryMessages = query; unscan(); } + 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. @@ -1454,7 +1530,6 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { return window; }; const error = console.error; - console.error = () => {}; const errors: string[] = []; console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; try { diff --git a/test/process-response-edit-refusals.test.ts b/test/process-response-edit-refusals.test.ts index e9a4ec885..a19112e3a 100644 --- a/test/process-response-edit-refusals.test.ts +++ b/test/process-response-edit-refusals.test.ts @@ -29,11 +29,21 @@ class Responder implements Module { ctx!: ModuleContext; onPing: EventResponse = {}; calls: string[] = []; - constructor(name: string) { this.name = name; } + /** Resolves with the first call its tool is given. */ + readonly called: Promise; + private resolveCalled!: (id: string) => void; + constructor(name: string) { + this.name = name; + this.called = new Promise((resolve) => { this.resolveCalled = resolve; }); + } async start(ctx: ModuleContext): Promise { this.ctx = ctx; } async stop(): Promise {} getTools(): ToolDefinition[] { return [{ name: 'act', description: 'Acts.', inputSchema: { type: 'object' } }]; } - async handleToolCall(call: ToolCall): Promise { this.calls.push(call.id); return { success: true, data: {} }; } + async handleToolCall(call: ToolCall): Promise { + this.calls.push(call.id); + this.resolveCalled(call.id); + return { success: true, data: {} }; + } async onProcess(event: ProcessEvent, _state: ProcessState): Promise { const type = (event as { type: string }).type; return type === 'ping' || type === 'tool-call' ? this.onPing : {}; @@ -209,7 +219,6 @@ test("a tool-call event whose module's response throws at an edit still dispatch call: { id: 'c-act', name: 'refused--act', input: {} }, } as unknown as ProcessEvent); await x.framework.runUntilIdle(); - for (let i = 0; i < 50 && refused.calls.length === 0; i++) await new Promise((r) => setTimeout(r, 10)); - assert.deepEqual(refused.calls, ['c-act']); + assert.equal(await refused.called, 'c-act'); assert.ok(x.errors.some((e) => e.startsWith("[process] refused's response to tool-call threw at the edit of no-such-message")), x.errors.join('\n')); }); From 4fe56f888ad87d9afabd3f9edba2454fb5f06856 Mon Sep 17 00:00:00 2001 From: Ansel-2428 Date: Fri, 23 Oct 2026 14:55:43 -0400 Subject: [PATCH 3/3] test(deferral): drop a repeated test and lines no code reads - One test of a write found by its id "where no request has carried its copy" tested a condition no code reads since moving a found copy was taken out. Its assertions repeat the found-later end-to-end test (the order) and the test of a write carried as its store holds it (counted as written, and the line naming anima-research/context-manager#155). - The latter test no longer marks the context shown, and two test comments no longer say no request carried the copy. - The drain's inline comment points to writeDrainedInOrder's doc for the known limit rather than repeating it. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 16 ++++++---------- test/deferred-window-only.test.ts | 26 ++------------------------ 2 files changed, 8 insertions(+), 34 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index f5a3ffab7..49c921e4e 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -22383,16 +22383,12 @@ export class AgentFramework { // An earlier write of this entry threw, and its store could neither // be read back nor searched to tell whether it landed. It's looked up // by its id before it's written again. Found, it counts as written - // where that write stored it, as when the readback works. That's a - // known limit: the copy sits ahead of what was stored since without - // it, and a reply stored after it whose request didn't carry it was - // made without it. On an account where the provider enforces that - // binding, a later request replaying that reply's signed thinking is - // refused with a 400 (context-manager #155). Not found, it's written. - // While the store can't be 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. + // where that write stored it, as when the readback works, a known + // limit (above, and at the tool boundary's drain). Not found, it's + // written. While the store can't be 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) { diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 725627506..f2fbb2bb1 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -743,7 +743,7 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { const internals = drainInternals(framework); module.duringTool = () => { // An earlier write of it landed, though whether it landed wasn't - // known, and no request has carried it. + // known. framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('look left'), { deferredWriteId: 'w-carried' } as never); internals.deferredMessages.push({ ...entry('w-carried', 'look left'), landing: 'unknown' }); }; @@ -849,10 +849,8 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { 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, - // where a request may have carried it, so it stays where it was stored. + // 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); - framework.getAgent('assistant')!.markContextShown(); const msg = { ...entry('w-part', 'part one'), content: [...text('part one'), ...text('part two')], landing: 'unknown' as const }; const errors: string[] = []; const error = console.error; @@ -867,26 +865,6 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); - it("a drained write whose landing is unknown, found by its id where no request has carried its copy, counts as written where it sits, ahead of what was stored since, and the log names the refusal", async () => { - const framework = await createFramework(); - const internals = drainInternals(framework); - const cm = framework.getAgent('assistant')!.getContextManager(); - cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-sits' } as never); - cm.addMessage('assistant', text('Done.')); - const msg = { ...entry('w-sits', 'look left'), 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([msg], 'test', () => { throw new Error('written again'); }); - } finally { console.error = error; } - assert.deepEqual(windowOf(framework).slice(-2), ['Antra: look left', 'assistant: Done.'], 'stored once, where it landed'); - assert.deepEqual(written.get(msg)?.content, text('look left'), 'counted as written'); - assert.ok(errors.some((e) => e.includes('w-sits') && e.includes('(context-manager #155)')), errors.join('\n')); - 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);