diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md new file mode 100644 index 000000000..c4c203d78 --- /dev/null +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -0,0 +1,30 @@ +- A deferred context write whose write threw, and whose store then couldn't + be read to tell whether it had landed, is no longer put back as one that + wasn't stored. Until now a later flush stored it again, so a write that + had landed appeared twice. It's now put back marked as unknown, and before + it's written again the store is searched for its id: found, it counts as + written and isn't stored again; not found, it's written; and while the + store can't be searched, it stays queued, holding what's queued behind it + for that agent, as a write that keeps failing does. Such an entry's mark + is persisted with the queue, so after a restart it's still looked up + before it's written. + - A write found this way at a flush is carried there, after what the turn + stored since, though the store holds it before those, so later requests + place it earlier. + - Wherever held writes are read or changed, by a module's context, the + coalescer, a wake's check or a tune-out's backlog, such a write is + looked up first. One that landed is a stored message from then on, read + and changed as one, and isn't carried into the turn; one that didn't is + an ordinary held write. Where the store can't be searched, the + coalescer reads it as possibly read and shown. + - A module context's `editMessage` and `removeMessage` can now throw for + such a held write: where its store can't be searched to tell whether + it landed, nothing is changed, and the error reads "held write held:… + may already be stored, and its store can't be searched to tell, so + nothing was changed. It's refused the same way while the store can't be + searched; the host log names the store's error". + - A conversation fork's or ephemeral run's queue, stored when it's + released, looks such an entry up the same way: found, it's neither + stored again nor passed to the primary as a failed reply; where the + store can't be searched, it's stored anyway where the store takes it, + and the log says it may now be stored twice. diff --git a/src/framework.ts b/src/framework.ts index 07eb7e8f8..234b4ac4e 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -724,7 +724,11 @@ const DEFERRED_WRITES_PERSIST_CAP_CHARS = 4 * 1024 * 1024; * with no stream). */ /** One entry of the deferred-write queue (see AgentFramework.deferredMessages). */ interface DeferredWrite { + /** Durable identity, stamped on the stored message as its + * deferredWriteId: a flush interrupted by a crash replays its batch at + * the next boot, which skips each entry that already landed by it. */ id: string; + /** Order stamp (see deferredSeq). */ seq: number; participant: string; content: ContentBlock[]; @@ -735,6 +739,13 @@ interface DeferredWrite { /** When the framework first held the write, in ms since the epoch: a * module reads it as the held row's timestamp (heldRows). */ heldAt: number; + /** Set when a write of this entry threw and its store couldn't be read + * to tell whether it landed (writeDrainedInOrder): before it's + * written again, the store is searched for its id, and a path that reads + * or changes held writes settles it first (settleUnknownLandings). + * Persisted with the entry, so after a restart it's still looked up + * first. */ + landing?: 'unknown'; } const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; @@ -780,6 +791,12 @@ function asCompiled(landed: ReadonlyArray<{ content: ContentBlock[] }>): Content const text = blocks.filter((b) => b.type === 'text').map((b) => (b as { text: string }).text).join(''); return [...blocks.filter((b) => b.type !== 'text'), { type: 'text', text } as ContentBlock]; } +/** One stored copy of a deferred write, as the drain reads it back: a + * sharded body's copies carry its `bodyGroupId`. */ +type StoredCopy = { id: MessageId; content: ContentBlock[]; bodyGroupId?: string }; +const storedCopy = (m: { id: MessageId; content: ContentBlock[]; bodyGroupId?: string }): StoredCopy => + ({ id: m.id, content: m.content, ...(m.bodyGroupId !== undefined ? { bodyGroupId: m.bodyGroupId } : {}) }); + /** Stamp a deferred write's durable id into the metadata it is stored with * (boot recovery dedups replays by it). Idempotent for an already-stamped * message. */ @@ -2492,22 +2509,7 @@ export class AgentFramework { * could occupy and dedups against all of it — never a fixed tail. */ private deferredScanFrom = new Map(); - private deferredMessages: Array<{ - /** Durable identity, stamped on the stored message as its - * deferredWriteId: a flush interrupted by a crash replays its batch at - * the next boot, which skips each entry that already landed by it. */ - id: string; - /** Order stamp (see deferredSeq). */ - seq: number; - participant: string; - content: ContentBlock[]; - metadata?: MessageMetadata; - /** Delivery target (registry name). Absent = the primary agent — - * the historical behavior of every existing caller. */ - forAgent?: string; - /** When the framework first held the write (DeferredWrite.heldAt). */ - heldAt: number; - }> = []; + private deferredMessages: DeferredWrite[] = []; // Turn-alive markers (2026-07-31 Mythos phantom-skip incident): agentName → // token of the turn currently in progress. Set at startAgentStream ENTRY — @@ -3344,10 +3346,15 @@ export class AgentFramework { ? framework.agents.get(framework.primaryAgentName) : undefined; return primary ? primary.getContextManager().getAllMessages() : []; }, - getDeferredMessages: () => framework.deferredMessages - .filter((m) => (m.forAgent ?? framework.primaryAgentName) === framework.primaryAgentName) - .sort(bySeq) - .map((m) => ({ participant: m.participant, content: m.content, metadata: m.metadata as Record | undefined })), + getDeferredMessages: () => { + const forPrimary = (m: DeferredWrite): boolean => (m.forAgent ?? framework.primaryAgentName) === framework.primaryAgentName; + // Settled first, so a write that landed is in the stored rows + // alone (settleUnknownLandings). + framework.settleUnknownLandings(forPrimary); + return framework.deferredMessages.filter(forPrimary) + .sort(bySeq) + .map((m) => ({ participant: m.participant, content: m.content, metadata: m.metadata as Record | undefined })); + }, // Every reader of the shared slot compiles the dump with its own // strategy, so the dump fits the smallest of their bounds. residentMessageChars: () => { @@ -11413,7 +11420,11 @@ export class AgentFramework { // them in the live conversation — same participant, same content // blocks, same metadata, and compile() emits recent messages // verbatim, so the next turn's compiled prefix byte-matches the - // live request (prompt-cache safe). + // live request (prompt-cache safe). A write that threw after it was + // stored, whose landing couldn't be read, is the exception: it's + // carried at the flush that finds it, after what was stored since, + // so there the store and the live request differ + // (writeDrainedInOrder). // // The queue is drained ONLY at the target agent's own boundary: // draining at another agent's boundary would store the messages @@ -13313,10 +13324,13 @@ export class AgentFramework { /** Unread (the coalescer's isUnread) by the consumed watermark, or, by the * shown watermark, never possibly shown (its isUnshown). A placement still - * in the deferred queue is both. */ + * in the deferred queue is both, once settled (settleUnknownLandings): one + * whose write may already be stored, its store unsearchable, is a doubt, + * which counts as read and as shown. */ private isUnreadPlacement(p: CoalescingPlacement, by: 'consumed' | 'shown' = 'consumed'): boolean { + if (p.deferredId) this.settleUnknownLandings((m) => m.id === p.deferredId); this.resolvePlacement(p); - if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId); + if (p.deferredId) return this.deferredMessages.some((m) => m.id === p.deferredId && m.landing !== 'unknown'); if (!p.messageId) return false; const agent = this.agents.get(p.agent); return !!agent && this.isUnreadStoredMessage(agent, p.messageId, by); @@ -13371,11 +13385,15 @@ export class AgentFramework { } private removePlacement(p: CoalescingPlacement): boolean { + // Settled first, so a write that landed is removed as the stored + // placement it is; one still a doubt, its store unsearchable, isn't + // removed, since it may already be stored. + if (p.deferredId) this.settleUnknownLandings((m) => m.id === p.deferredId); this.resolvePlacement(p); if (p.deferredId) { - const before = this.deferredMessages.length; - this.deferredMessages = this.deferredMessages.filter((m) => m.id !== p.deferredId); - if (this.deferredMessages.length === before) return false; + const held = this.deferredMessages.find((m) => m.id === p.deferredId); + if (!held || held.landing === 'unknown') return false; + this.deferredMessages = this.deferredMessages.filter((m) => m !== held); if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); return true; } @@ -14234,6 +14252,9 @@ export class AgentFramework { private noticeUnnoticedDrafts(agent: Agent): void { const cm = agent.getContextManager(); const leftOut = new Set(this.openLogical.get(agent.name)?.drafts.map((d) => d.id)); + // A queued notice still waiting names its drafts; one whose write landed + // is settled first, so its drafts are judged by its stored row. + this.settleUnknownLandings((m) => this.deliveryAgent(m.forAgent) === agent); for (const msg of this.deferredMessages) { const meta = msg.metadata as { draftOwner?: unknown; draftIds?: unknown } | undefined; if (meta?.draftOwner !== agent.name || !Array.isArray(meta.draftIds)) continue; @@ -21218,9 +21239,12 @@ export class AgentFramework { * landed shards would be stored twice, and only the rest could not be * appended later without splitting the body.) Boot's scan for landed * writes counts a landed shard the same way. When the store can't be read - * back (storedCopiesOf logs it), nothing is found, and the entry is put - * back as one that couldn't be stored: if it had landed, a later flush - * stores it again. + * back (storedCopiesOf logs it), or couldn't say how many messages it held + * before the write, whether the entry landed is unknown: it's put back + * marked so (`landing`), and before it's written again the store is + * searched for its id (landedCopiesById). Found, it counts as written, as + * above, at the flush that finds it; not found, it's written; and while + * the store can't be searched, it stays queued, holding what's behind it. * * Returns what was written (stored, or deferred again by addMessage, whose * id is then empty), in queue order: for one target, a prefix of its @@ -21248,28 +21272,58 @@ 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. + 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'); + continue; + } + if (found.length > 0) { + written.set(msg, { id: found[0]!.id, content: asCompiled(found) }); + this.logDeferredWriteError(msg, where, undefined, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + onLanded?.(msg, found[0]!.id); + continue; + } + delete msg.landing; + } const before = agent ? this.storedCountOf(agent) : undefined; try { const id = write(msg); // A body the store took as more than one message (a sharded body) is // returned as later compiles render it: its copies, joined // (asCompiled). The store is read for them only when it grew by more - // than one message. + // than one message. One that can't be read back is carried as given. const copies = agent && before !== undefined && (this.storedCountOf(agent) ?? before) - before > 1 - ? this.storedCopiesOf(agent, before, msg.id) + ? this.storedCopiesOf(agent, before, msg.id) ?? [] : []; written.set(msg, { id, content: copies.length > 1 ? asCompiled(copies) : msg.content }); } catch (error) { - const landed = agent && before !== undefined ? this.storedCopiesOf(agent, before, msg.id) : []; - if (landed.length > 0) { + // What landed, read back from where the store stood before the + // write; undefined when the store couldn't say, before the write or + // after it. With no agent there's no store it could have landed in. + const landed = !agent ? [] : before === undefined ? undefined : this.storedCopiesOf(agent, before, msg.id); + if (landed && landed.length > 0) { written.set(msg, { id: landed[0]!.id, content: asCompiled(landed) }); this.logDeferredWriteError(msg, where, error, { count: landed.length, sharded: landed[0]!.bodyGroupId !== undefined }); onLanded?.(msg, landed[0]!.id); continue; } + // Requeued as unstored, a write that had landed would be stored a + // second time; so one whose landing is unknown is marked, and looked + // up by its id before it's written again (above). + if (landed === undefined) msg.landing = 'unknown'; cut.add(target); this.requeueDrained(ordered.slice(i).filter((m) => targetOf(m) === target)); - this.logDeferredWriteError(msg, where, error); + this.logDeferredWriteError(msg, where, error, landed === undefined ? 'unknown' : undefined); } } return written; @@ -21288,29 +21342,47 @@ export class AgentFramework { /** The messages carrying `deferredWriteId` among those the agent's store * gained from position `from` on, in store order, with their blobs * resolved, as a request carries them. A sharded body's carry its - * `bodyGroupId`. */ - private storedCopiesOf(agent: Agent, from: number, deferredWriteId: string): Array<{ id: MessageId; content: ContentBlock[]; bodyGroupId?: string }> { + * `bodyGroupId`. Undefined when the store can't be read back, which is + * logged: not the same as nothing found. */ + private storedCopiesOf(agent: Agent, from: number, deferredWriteId: string): StoredCopy[] | undefined { try { const cm = agent.getContextManager(); return cm.getMessageWindow(from, Math.max(0, cm.getMessageCount() - from)).messages .filter((m) => m.metadata?.deferredWriteId === deferredWriteId) - .map((m) => ({ id: m.id, content: m.content, ...(m.bodyGroupId !== undefined ? { bodyGroupId: m.bodyGroupId } : {}) })); + .map(storedCopy); } catch (error) { console.error(`[deferral] could not read back what deferred write ${deferredWriteId} stored in ${agent.name}'s store:`, error); - return []; + return undefined; + } + } + + /** The messages carrying `deferredWriteId` anywhere in the agent's store, + * found by their id (queryMessages, as landedRow finds a held write's), + * so no position is needed. Undefined when the store can't be searched, + * which is logged. */ + private landedCopiesById(agent: Agent, deferredWriteId: string): StoredCopy[] | undefined { + try { + return agent.getContextManager().queryMessages({ metadata: { deferredWriteId } }).messages.map(storedCopy); + } catch (error) { + console.error(`[deferral] could not search ${agent.name}'s store for deferred write ${deferredWriteId}:`, error); + return undefined; } } /** The line naming a drained write that threw: one that couldn't be stored, - * which is already back in the queue, or, with `landed`, one stored before - * the throw (for a sharded body, how many of its shards). Logging mustn't - * fail the flush that met it, so a thrown value that has no text is named - * as such. */ + * which is already back in the queue; with a count, one stored before the + * throw (for a sharded body, how many of its shards), or, with + * `foundLater`, one an earlier write of which landed and was found by its + * id; with `'unknown'`, one whose store couldn't be read to tell + * whether it landed, back in the queue to be looked up first; and with + * `'still-unknown'`, one that still can't be looked up. Logging mustn't + * fail the flush that met it, so a thrown value that has no text is + * named as such. */ private logDeferredWriteError( msg: DeferredWrite, where: string, error: unknown, - landed?: { count: number; sharded: boolean }, + outcome?: { count: number; sharded: boolean; foundLater?: boolean } | 'unknown' | 'still-unknown', ): void { let reason: string; try { @@ -21318,15 +21390,27 @@ export class AgentFramework { } catch { reason = 'a thrown value that has no text'; } + const held = 'it stays queued, holding what is queued behind it for that agent'; + const stored = (landed: { count: number; sharded: boolean }): string => landed.sharded + ? `${landed.count} shard${landed.count === 1 ? '' : 's'} of its body ${landed.count === 1 ? 'was' : 'were'} stored` + : 'it was stored'; + let what: string; + if (outcome === 'unknown') { + what = `threw, and its store couldn't be read to tell whether it was stored; ${held}, and is looked up ` + + `by its id before it's written again: ${reason}`; + } else if (outcome === 'still-unknown') { + what = `still can't be looked up by its id, to tell whether an earlier write of it was stored; ${held}`; + } else if (outcome?.foundLater) { + what = `was found by its id: ${stored(outcome)} by an earlier write whose landing was unknown; what ` + + `was stored counts as written and isn't stored again`; + } else if (outcome) { + what = `threw after ${stored(outcome)}; what was stored counts as written and isn't stored again: ${reason}`; + } else { + what = `could not be stored; ${held}: ${reason}`; + } console.error( `[deferral] ${where}: deferred write ${describeDeferredWrite(msg)} for ` + - `${msg.forAgent ?? this.primaryAgentName ?? 'the primary'} ` + - (landed - ? `threw after ${landed.sharded - ? `${landed.count} shard${landed.count === 1 ? '' : 's'} of its body ${landed.count === 1 ? 'was' : 'were'} stored` - : 'it was stored'}; what was stored counts as written and isn't stored again: ` - : `could not be stored; it stays queued, holding what is queued behind it for that agent: `) + - reason, + `${msg.forAgent ?? this.primaryAgentName ?? 'the primary'} ${what}`, ); } @@ -21417,7 +21501,12 @@ export class AgentFramework { * one still queued for it. Writes still * queued would wait for a boundary it never reaches, and the flush that * finds it gone would drop them, so they land in its own window now, which - * Chronicle keeps for investigation. Its stream is over or cancelled, so it + * Chronicle keeps for investigation. One whose earlier write may have + * landed (`landing` unknown) is looked up there by its id first: found, + * it's neither stored again nor passed to the primary; where the store + * can't be searched, it's stored anyway, with a line saying it may now be + * stored twice, or, where storing it fails too, that it may be stored + * once, twice or not at all. Its stream is over or cancelled, so it * compiles nothing more. One path still stores after these writes: a * `complete` that arrived while the agent still waited for tools awaits * the spill of its last tool results, and stores them once that resolves, @@ -21432,12 +21521,37 @@ export class AgentFramework { const cm = agent.getContextManager(); const forPrimary: DeferredWrite[] = []; for (const msg of queued) { + // An entry whose earlier write may have landed (landing unknown) is + // looked up by its id, as settleUnknownLandings looks one up + // elsewhere, but in the released agent's own store. Found, it landed: + // it's neither stored again nor passed on below, since the loop above + // has passed its stored copy if it's a failed reply the agent hadn't + // read, by its window as that loop reads it. When the store can't be + // searched, it's stored anyway where the store takes it, since no + // later flush will try again, and a copy its deferredWriteId tells + // apart costs less than a write lost without a trace. + const found = msg.landing === 'unknown' ? this.landedCopiesById(agent, msg.id) : []; + if (found && found.length > 0) { + this.logDeferredWriteError(msg, `release of ${agent.name}`, undefined, + { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + continue; + } try { cm.addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); + if (found === undefined) { + console.error( + `[deferral] release of ${agent.name}: deferred write ${describeDeferredWrite(msg)} was stored though its ` + + `store couldn't be searched for an earlier copy, so it may now be stored twice`, + ); + } } catch (err) { console.error( `[deferred-flush] ${agent.name}: failed to store a queued write before release ` + - `(participant=${msg.participant}):`, + `(participant=${msg.participant})` + + (found === undefined + ? `; its store couldn't be searched for an earlier copy of deferred write ${describeDeferredWrite(msg)} either, ` + + `so it may be stored once, twice or not at all:` + : ':'), err, ); } @@ -21666,12 +21780,12 @@ export class AgentFramework { * (queuedFailedReplyForPrimary). */ private heldWritesFor(reader: Agent): DeferredWrite[] { - return this.deferredMessages - .filter((m) => { - const target = this.deliveryAgent(m.forAgent); - return !!target && (target === reader || (this.isOrdinaryResident(reader) && this.isOrdinaryResident(target))); - }) - .sort(bySeq); + const heldFor = (m: DeferredWrite): boolean => { + const target = this.deliveryAgent(m.forAgent); + return !!target && (target === reader || (this.isOrdinaryResident(reader) && this.isOrdinaryResident(target))); + }; + this.settleUnknownLandings(heldFor); + return this.deferredMessages.filter(heldFor).sort(bySeq); } /** @@ -21708,6 +21822,82 @@ export class AgentFramework { return cm.queryMessages({ metadata: { deferredWriteId: writeId } }).messages[0] ?? null; } + /** + * Settle held writes whose landing is unknown (DeferredWrite.landing), + * among those `which` picks, before a path reads or changes held writes: + * each is looked up by its id in its target's store (landedCopiesById). + * Found, the write landed (settleFound). Not found, the mark goes, and + * it's an ordinary held write, which is persisted as an edit is. Where the + * store can't be searched, it stays marked, a doubt each path reads by its + * own rule. + */ + private settleUnknownLandings(which: (msg: DeferredWrite) => boolean): void { + let unmarked = false; + for (const msg of this.deferredMessages.filter((m) => m.landing === 'unknown' && which(m))) { + // No registered agent: no store it could have landed in. A flush + // writes it, and addMessage drops it. + const agent = this.deliveryAgent(msg.forAgent); + if (!agent) continue; + const found = this.landedCopiesById(agent, msg.id); + if (found === undefined) continue; + if (found.length > 0) { + this.settleFound(msg, agent, found[0]!.id, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined }); + } else { + delete msg.landing; + unmarked = true; + } + } + if (unmarked && (this.quiesced || this.deferredWritesPersisted)) this.persistDeferredWrites(); + } + + /** + * A marked held write found to have landed, as `stored`: it leaves the + * pending queue for the un-acked set, as a drained write does, so it stays + * in the durable queue until the next ack has synced the store that holds + * it, and it's a stored write from then on, so the paths for a landed + * write apply to it. It takes the notes a store takes (noteStored), which + * its first write missed when it threw, as a flush that finds such a write + * takes them (readdDrained), and is logged by its id. It isn't carried + * into a turn, as a flush that finds it carries it: the agent reads it at + * its next compile, where the store holds it. + */ + private settleFound(msg: DeferredWrite, agent: Agent, stored: MessageId, landed: { count: number; sharded: boolean }): void { + this.deferredMessages = this.deferredMessages.filter((m) => m !== msg); + this.unackedDeferredWrites.push(msg); + this.noteStored(agent, msg.metadata, stored); + this.logDeferredWriteError(msg, `held writes for ${agent.name}`, undefined, { ...landed, foundLater: true }); + } + + /** + * A stored row a module names by its id: where it carries the + * deferredWriteId of a marked held write the reader's store holds, it's + * that write's copy, so the write landed, and it's settled as found + * (settleFound) before the row is read or changed. A lookup by the id + * can't be needed then, so this holds where the store can't be searched + * too. + */ + private settleByCopy(reader: Agent, row: { id: MessageId; metadata?: unknown } | null): void { + const writeId = (row?.metadata as { deferredWriteId?: unknown } | undefined)?.deferredWriteId; + if (!row || typeof writeId !== 'string') return; + const msg = this.deferredMessages.find((m) => m.id === writeId && m.landing === 'unknown'); + const target = msg ? this.deliveryAgent(msg.forAgent) : undefined; + if (!msg || !target || !(target === reader || (this.isOrdinaryResident(reader) && this.isOrdinaryResident(target)))) return; + this.settleFound(msg, target, row.id, { count: 1, sharded: false }); + } + + /** A held write whose landing stays unknown after settling, its store + * unsearchable, can't be edited or removed: it may already be stored, and + * changing only the queued entry could leave its stored copy as it was. + * Throws with nothing changed. */ + private refuseUnsettled(msg: DeferredWrite): void { + if (msg.landing !== 'unknown') return; + throw new Error( + `held write ${heldHandle(msg.id)} may already be stored, and its store can't be searched to tell, so ` + + `nothing was changed. It's refused the same way while the store can't be searched; the host log names ` + + `the store's error`, + ); + } + /** Whether the agent has nothing left to read of a deferred write, on * evidence only: the row it landed as is at or below the agent's consumed * watermark on its current branch, or it's in neither the queue (drained @@ -21720,6 +21910,9 @@ export class AgentFramework { * finds nothing new costs less than a write no wake brings the agent to * read. */ private deferredWriteReadOrGone(agent: Agent, writeId: string): boolean { + // Settled first, so a write that landed is judged by its stored row; one + // still a doubt stays queued, which reads as unread. + this.settleUnknownLandings((m) => m.id === writeId); if (this.deferredMessages.some((m) => m.id === writeId)) return false; try { const cm = agent.getContextManager(); @@ -21747,14 +21940,17 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } + // The held writes are settled first (heldWritesFor), and by a stored id + // the row it names too (settleByCopy): a held write that landed is a + // stored row from then on, and the edit reaches it as one. const writeId = heldWriteId(id); - if (writeId !== undefined) { - const held = this.heldWritesFor(agent).find((m) => m.id === writeId); - if (held) { - held.content = content; - if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); - return; - } + const held = this.heldWritesFor(agent).find((m) => writeId !== undefined && m.id === writeId); + if (writeId === undefined) this.settleByCopy(agent, agent.getContextManager().getMessage(id)); + if (held) { + this.refuseUnsettled(held); + held.content = content; + if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); + return; } const landed = writeId !== undefined ? this.landedRow(agent.getContextManager(), writeId) : null; agent.getContextManager().editMessage(landed?.id ?? id, content); @@ -21765,8 +21961,15 @@ export class AgentFramework { if (!agent) { throw new Error('No agents configured'); } + // The held writes are settled first (heldWritesFor), and by a stored id + // the row it names too (settleByCopy), so a held write that landed is + // removed as the stored row it is, and isn't written again at the next + // flush. const writeId = heldWriteId(id); - if (writeId !== undefined && this.heldWritesFor(agent).some((m) => m.id === writeId)) { + const held = this.heldWritesFor(agent).find((m) => writeId !== undefined && m.id === writeId); + if (writeId === undefined) this.settleByCopy(agent, agent.getContextManager().getMessage(id)); + if (held) { + this.refuseUnsettled(held); this.deferredMessages = this.deferredMessages.filter((m) => m.id !== writeId); if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites(); return; @@ -21781,8 +21984,12 @@ export class AgentFramework { return null; } const writeId = heldWriteId(id); + // By a handle, heldRows settles the held writes first; by a stored id, + // the row's own write is settled by it (settleByCopy). + const stored = writeId === undefined ? agent.getContextManager().getMessage(id) : null; + if (writeId === undefined) this.settleByCopy(agent, stored); const row = writeId === undefined - ? agent.getContextManager().getMessage(id) + ? stored : this.heldRows(agent).find((held) => held.id === id) ?? this.landedRow(agent.getContextManager(), writeId); return row ? asModuleRow(row) : null; } diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts index 45c87e0e0..5e713a8fb 100644 --- a/src/mcpl/push-coalescer.ts +++ b/src/mcpl/push-coalescer.ts @@ -260,8 +260,9 @@ export interface CoalescerHost { /** True iff the placement still exists and no model request has included it. */ isUnread(placement: CoalescingPlacement): boolean; /** True iff nothing may have shown the placement to the agent: it is still - * in the deferred queue, or stored where no committing compile has begun - * to assemble it and no tool boundary has injected past it. Unlike + * in the deferred queue and known not to be stored already, or it is + * stored where no committing compile has begun to assemble it and no tool + * boundary has injected past it. Unlike * isUnread, a compile counts from its start, and a boundary counts * whether or not its stream says it carries injected messages: a doubt * counts as shown. */ @@ -660,9 +661,10 @@ export class PushCoalescer { /** * Remove the occupant and the run it builds on, newest first, auditing each * as `kind`. Every member is unread: refreshOccupant has just run, under the - * lock, and the host's removal doesn't itself refuse a read message. A - * member the host can't remove may be read, so the walk stops there, leaving - * it and the renders it builds on, and the history becomes `unknown`. + * lock. The host's removal may still refuse one, such as a held write the + * host can't tell isn't already stored. A member the host + * can't remove may be read, so the walk stops there, leaving it and the + * renders it builds on, and the history becomes `unknown`. * Returns false then. It doesn't touch `state.occupant`, which still names * the members it removed: each caller clears it, so what the walk left is * history from then on, as an occupant whose removal failed always was. diff --git a/src/types/module.ts b/src/types/module.ts index 30d5b1927..5a42c84f7 100644 --- a/src/types/module.ts +++ b/src/types/module.ts @@ -40,7 +40,9 @@ export type ModuleMessage = StoredMessage & { * and `timestamp` are provisional: the sequence comes after every stored * row's, the timestamp is when the framework held the write (for a write * an older release held, when its queue was read back), and both change - * when it lands. Absent on a stored row. + * when it lands. A held write whose earlier write may already be stored + * is looked up first, and answered as the stored row it is when the store + * holds it. Absent on a stored row. */ held?: true; /** @@ -311,13 +313,17 @@ export interface ModuleContext { /** * Edit a message, by its id or a held write's handle. A held write is - * stored with the new content. + * stored with the new content. One whose earlier write may already be + * stored, where the store can't be searched to tell, throws with nothing + * changed. */ editMessage(id: MessageId, content: ContentBlock[]): void; /** * Remove a message, by its id or a held write's handle. A held write is - * dropped, so the agent never reads it. + * dropped, so the agent never reads it. One whose earlier write may + * already be stored, where the store can't be searched to tell, throws + * with nothing changed. */ removeMessage(id: MessageId): void; diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 8fe4f9145..ee34fc29d 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -15,7 +15,9 @@ * injected, since the next compile leaves it out too * (module-rows-held-from-residents). A provider that binds signed thinking * to its prefix (context-manager #155) refuses a reply whose stored prefix - * holds a message its request didn't. + * holds a message its request didn't. One write is the exception: one that + * threw after it was stored, whose landing couldn't be read, is carried at + * the flush that finds it by its id, after what the turn stored since. * * A message the stream carried has been read: RFC-006's consumed watermark * covers it, so a coalesced replacement can no longer remove it. Only a @@ -43,6 +45,7 @@ import type { ToolResult, } from '../src/index.js'; import { AgentFramework } from '../src/index.js'; +import { heldHandle } from '../src/held-writes.js'; import { AutobiographicalStrategy } from '@animalabs/context-manager'; import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; import type { ContentBlock } from '@animalabs/membrane'; @@ -564,6 +567,627 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { await framework.stop(); }); + /** Have the agent's strategy throw once its message carrying `textOf` is + * stored, and the store's next windowed read after that throw, the + * drain's readback of what landed, throw too. */ + function throwAfterStoringThenUnreadable(framework: AgentFramework, textOf: string): void { + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + strategy: { onNewMessage?: (message: { content: ContentBlock[] }, ctx: unknown) => Promise }; + getMessageWindow: (...args: unknown[]) => unknown; + }; + const onNewMessage = cm.strategy.onNewMessage; + const read = cm.getMessageWindow; + let thrown = false; + let unreadable = false; + cm.strategy.onNewMessage = function (this: unknown, message, ctx) { + if (!thrown && message.content.some((b) => (b as { text?: string }).text === textOf)) { + thrown = true; + unreadable = true; + throw new Error('strategy failed'); + } + return onNewMessage ? onNewMessage.call(this, message, ctx) : Promise.resolve(); + }; + cm.getMessageWindow = function (this: unknown, ...args: unknown[]) { + if (unreadable) { + unreadable = false; + throw new Error('readback failed'); + } + return read.apply(this, args); + }; + } + + it("at a tool boundary, a write that throws after it's stored, whose store then can't be read back, is found by its id later and isn't stored a second time", async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Moving.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: {} }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse(text('Done.'))); + const framework = await createFramework(); + throwAfterStoringThenUnreadable(framework, 'look left'); + module.interjections = [ + { participant: 'Antra', content: text('look left') }, + { participant: 'Antra', content: text('look right') }, + ]; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { + framework.pushEvent({ type: 'external-message', source: 'test', content: 'go' } as unknown as ProcessEvent); + await framework.runUntilIdle(); + } finally { console.error = error; } + assert.deepEqual(windowOf(framework).slice(-4), [ + 'user: ', + 'Antra: look left', + 'assistant: Done.', + 'Antra: look right', + ], 'stored once, where its first write landed, and what was queued behind it at the turn end'); + assert.ok(errors.some((e) => e.includes("threw, and its store couldn't be read to tell whether it was stored") + && e.includes('strategy failed')), errors.join('\n')); + assert.ok(errors.some((e) => e.includes('was found by its id: it was stored by an earlier write')), errors.join('\n')); + await framework.stop(); + }); + + /** The internals a drained write's landing goes through. */ + function drainInternals(framework: AgentFramework) { + return framework as unknown as { + writeDrainedInOrder( + entries: DeferredEntry[], where: string, write: (msg: DeferredEntry) => string, + onLanded?: (msg: DeferredEntry, id: string) => void, + ): Map; + releaseWindow(agent: unknown): void; + addMessage(participant: string, content: ContentBlock[], metadata: unknown, opts: unknown): string; + editMessage(id: string, content: ContentBlock[]): void; + removeMessage(id: string): void; + queryMessages(filter: unknown): { messages: Array<{ content: ContentBlock[]; held?: true }> }; + getMessage(id: string): { held?: true; content: ContentBlock[] } | null; + deferredWriteReadOrGone(agent: unknown, writeId: string): boolean; + isUnreadPlacement(placement: { agent: string; deferredId?: string }, by?: 'consumed' | 'shown'): boolean; + removePlacement(placement: { agent: string; deferredId?: string }): boolean; + deferredMessages: DeferredEntry[]; + deferredWritesPersisted: boolean; + }; + } + type DeferredEntry = { + id: string; seq: number; participant: string; content: ContentBlock[]; heldAt: number; + metadata?: unknown; forAgent?: string; landing?: 'unknown'; + }; + const entry = (id: string, said: string): DeferredEntry => ({ id, seq: 1, participant: 'Antra', content: text(said), heldAt: Date.now() }); + /** A write that stores the entry as the framework does, under its id, then throws. */ + const storedThenThrows = (framework: AgentFramework) => (msg: DeferredEntry): string => { + framework.getAgent('assistant')!.getContextManager().addMessage(msg.participant, msg.content, { deferredWriteId: msg.id } as never); + throw new Error('strategy failed'); + }; + + it("a drained write that throws when its store couldn't count its messages before it is looked up by its id before it's written again", async () => { + const framework = await createFramework(); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { getMessageCount: () => number }; + const count = cm.getMessageCount; + let failing = true; + cm.getMessageCount = function (this: unknown) { + if (failing) { failing = false; throw new Error('count failed'); } + return count.call(this); + }; + const internals = drainInternals(framework); + const msg = entry('w-count', 'look left'); + const error = console.error; + console.error = () => {}; + 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]); }); + } 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'); + await framework.stop(); + }); + + it("a drained write whose landing is unknown, found by its id, is carried as its store holds it, not as it was queued", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + // A body whose first part landed before the throw stopped its write. + framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('part one'), { deferredWriteId: 'w-part' } as never); + const msg = { ...entry('w-part', 'part one'), content: [...text('part one'), ...text('part two')], landing: 'unknown' as const }; + const error = console.error; + console.error = () => {}; + let written: Map; + try { + written = internals.writeDrainedInOrder([msg], 'test', () => { throw new Error('written again'); }); + } finally { console.error = error; } + assert.deepEqual(written.get(msg)?.content, text('part one'), 'what the store holds'); + await framework.stop(); + }); + + it("a drained write whose landing is unknown, and which its store doesn't hold, is written when it's looked up", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const msg = { ...entry('w-absent', 'look left'), landing: 'unknown' as const }; + let writes = 0; + const written = internals.writeDrainedInOrder([msg], 'test', (m) => { + writes++; + return framework.getAgent('assistant')!.getContextManager().addMessage(m.participant, m.content, { deferredWriteId: m.id } as never); + }); + assert.equal(writes, 1); + assert.ok(written.has(msg)); + assert.equal(msg.landing, undefined, 'looked up and written, it is no longer unknown'); + await framework.stop(); + }); + + it("a drained write whose landing is unknown stays queued, holding what's behind it, while its store can't be searched", async () => { + const framework = await createFramework(); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = function () { throw new Error('search failed'); }; + const internals = drainInternals(framework); + const first = { ...entry('w-held', 'look left'), landing: 'unknown' as const }; + // Another target's entry, queued between them, isn't held by them. + const other = { ...entry('w-other', 'elsewhere'), seq: 2, forAgent: 'gone-fork' }; + const second = { ...entry('w-behind', 'look right'), seq: 3 }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + const writes: string[] = []; + try { + const written = internals.writeDrainedInOrder([first, other, second], 'test', (m) => { writes.push(m.id); return 'x'; }); + assert.deepEqual([...written.keys()].map((m) => m.id), ['w-other']); + } finally { console.error = error; } + assert.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; + await framework.stop(); + }); + + it("a drained write whose landing is unknown, for an agent that isn't registered, is dropped as an unmarked one is, not held", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const gone = { ...entry('w-gone', 'look left'), forAgent: 'gone-fork', landing: 'unknown' as const }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + let written: Map; + try { + written = internals.writeDrainedInOrder([gone], 'test', (m) => internals.addMessage(m.participant, m.content, m.metadata, { + fromQueue: { id: m.id, seq: m.seq, heldAt: m.heldAt }, forAgent: m.forAgent, + })); + } finally { console.error = error; } + assert.ok(written.has(gone), 'it leaves the drain'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'nothing is held'); + assert.ok(!errors.some((e) => e.includes("still can't be looked up by its id")), errors.join('\n')); + await framework.stop(); + }); + + for (const landed of [true, false]) { + it(`a write whose landing is unknown keeps its mark across a restart, so it's stored once after it (${landed ? 'its first write landed' : 'nothing landed'})`, async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + internals.deferredWritesPersisted = true; + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage(participant: string, content: ContentBlock[], metadata?: unknown): string; + queryMessages: (...args: unknown[]) => unknown; + }; + // Where its first write landed, the hand-off at stop() records the + // store's count after it, so boot's check for landed writes starts + // past it and only the restored mark keeps it from a second store. + if (landed) cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-rt' }); + internals.deferredMessages.push({ ...entry('w-rt', 'look left'), landing: 'unknown' }); + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { await framework.stop(); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-rt', 'unknown']], 'held through stop()'); + + const reopened = await createFramework(); + assert.equal(windowOf(reopened).filter((line) => line === 'Antra: look left').length, 1, 'stored once'); + assert.deepEqual(drainInternals(reopened).deferredMessages.map((m) => m.id), [], 'and no longer queued'); + await reopened.stop(); + }); + } + + it("a released window doesn't store a write whose landing is unknown a second time, when its store holds it", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager(); + const copies = (id: string): number => cm.queryMessages({ metadata: { deferredWriteId: id } } as never).messages.length; + cm.addMessage('Antra', text('look left'), { deferredWriteId: 'w-rel' } as never); + internals.deferredMessages.push({ ...entry('w-rel', 'look left'), landing: 'unknown' }); + internals.deferredMessages.push({ ...entry('w-rel-2', 'look right'), seq: 2, landing: 'unknown' }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.releaseWindow(agent); } finally { console.error = error; } + assert.equal(copies('w-rel'), 1, 'found by its id, it is not stored again'); + assert.equal(copies('w-rel-2'), 1, 'not found, it is stored'); + assert.ok(errors.some((e) => e.includes('release of assistant') && e.includes('was found by its id')), errors.join('\n')); + await framework.stop(); + }); + + it("a released window doesn't pass a failed reply on a second time when its store holds the reply's copy", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager(); + const failed = { kind: 'discord-send-failed', channelId: 'c1', reason: 'gone', textLen: 5, outcome: 'failed' }; + // Its earlier write landed. The release's first pass reads that copy as + // unread and passes it on; the queued entry mustn't be passed again. + cm.addMessage('assistant', text('hello'), { ...failed, deferredWriteId: 'w-fail' } as never); + internals.deferredMessages.push({ ...entry('w-fail', 'hello'), participant: 'assistant', metadata: failed, landing: 'unknown' }); + const error = console.error; + console.error = () => {}; + try { internals.releaseWindow(agent); } finally { console.error = error; } + const notices = cm.queryMessages({} as never).messages.filter((m) => (m.metadata as { speaker?: unknown } | undefined)?.speaker === 'assistant'); + assert.equal(notices.length, 1, 'the primary is told once'); + await framework.stop(); + }); + + it("a released window doesn't pass on a failed reply the agent had read, when its store holds the reply's copy", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager(); + const failed = { kind: 'discord-send-failed', channelId: 'c1', reason: 'gone', textLen: 5, outcome: 'failed' }; + cm.addMessage('assistant', text('hello'), { ...failed, deferredWriteId: 'w-read' } as never); + // The agent read its copy: the release's first pass skips it, as main + // skips a failed reply the agent heard, and the queued entry mustn't + // pass it on either. + agent.markContextConsumed(); + internals.deferredMessages.push({ ...entry('w-read', 'hello'), participant: 'assistant', metadata: failed, landing: 'unknown' }); + const error = console.error; + console.error = () => {}; + try { internals.releaseWindow(agent); } finally { console.error = error; } + const notices = cm.queryMessages({} as never).messages.filter((m) => (m.metadata as { speaker?: unknown } | undefined)?.speaker === 'assistant'); + assert.equal(notices.length, 0, "the primary isn't told"); + await framework.stop(); + }); + + it("a released window that can neither search a write's store nor store it says it may be stored once, twice or not at all", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown; addMessage: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + const add = cm.addMessage; + cm.queryMessages = () => { throw new Error('search failed'); }; + cm.addMessage = () => { throw new Error('store failed'); }; + internals.deferredMessages.push({ ...entry('w-lost', 'look left'), landing: 'unknown' }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.releaseWindow(agent); } finally { console.error = error; cm.queryMessages = query; cm.addMessage = add; } + assert.ok(errors.some((e) => e.includes('w-lost') && e.includes('may be stored once, twice or not at all')), errors.join('\n')); + assert.ok(!errors.some((e) => e.includes('w-lost') && e.includes('may now be stored twice')), 'not the line for a store that went through'); + await framework.stop(); + }); + + it("a released ephemeral run's window settles a write whose landing is unknown in that run's own store", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const { agent } = await framework.createEphemeralAgent({ name: 'run-x', model: 'test-model', systemPrompt: 'Do the task.' }); + const own = agent.getContextManager(); + own.addMessage('Antra', text('look left'), { deferredWriteId: 'w-run' } as never); + internals.deferredMessages.push({ ...entry('w-run', 'look left'), forAgent: 'run-x', landing: 'unknown' }); + const error = console.error; + console.error = () => {}; + try { internals.releaseWindow(agent); } finally { console.error = error; } + assert.equal(own.queryMessages({ metadata: { deferredWriteId: 'w-run' } } as never).messages.length, 1, "found in the run's own store, it isn't stored again"); + assert.deepEqual(copiesOf(framework, 'w-run'), [], "and the primary's store holds none"); + await framework.stop(); + }); + + it("a released window stores a write whose landing is unknown while its store can't be searched, saying it may now be stored twice", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + const cm = agent.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + internals.deferredMessages.push({ ...entry('w-blind', 'look left'), landing: 'unknown' }); + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.releaseWindow(agent); } finally { console.error = error; } + cm.queryMessages = query; + assert.equal(windowOf(framework).filter((line) => line === 'Antra: look left').length, 1, 'stored'); + assert.ok(errors.some((e) => e.includes('w-blind') && e.includes('may now be stored twice')), errors.join('\n')); + await framework.stop(); + }); + + /** A held write whose landing is unknown, whose first write did land. */ + function heldAndStored(framework: AgentFramework, id: string, said: string): void { + framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text(said), { deferredWriteId: id } as never); + drainInternals(framework).deferredMessages.push({ ...entry(id, said), landing: 'unknown' }); + } + const copiesOf = (framework: AgentFramework, id: string): string[] => framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages + .map((m) => m.content.map((b) => (b as { text?: string }).text ?? '').join('')); + + it("a module's edit of a held write whose landing is unknown settles it first, so the edit reaches the stored row it landed as", async () => { + for (const by of ['handle', 'stored id'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-${by === 'handle' ? 'h' : 's'}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages; + const error = console.error; + console.error = () => {}; + try { internals.editMessage(by === 'handle' ? heldHandle(id) : copy!.id, text('look up')); } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, id), ['look up'], `by its ${by}: stored once, edited`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `by its ${by}: no longer held`); + await framework.stop(); + } + }); + + it("a module's removal of a held write whose landing is unknown settles it first, so the stored row goes and nothing is written again", async () => { + for (const by of ['handle', 'stored id'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-${by === 'handle' ? 'h' : 's'}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages; + const error = console.error; + console.error = () => {}; + try { internals.removeMessage(by === 'handle' ? heldHandle(id) : copy!.id); } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, id), [], `by its ${by}`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `by its ${by}: nothing left to write`); + await framework.stop(); + } + }); + + it("a module's edit of a held write whose landing is unknown, which its store doesn't hold, settles it as an ordinary held write and edits it", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + internals.deferredMessages.push({ ...entry('w-absent-ed', 'look left'), landing: 'unknown' }); + internals.editMessage(heldHandle('w-absent-ed'), text('look up')); + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content, m.landing]), [['w-absent-ed', text('look up'), undefined]]); + assert.deepEqual(copiesOf(framework, 'w-absent-ed'), [], 'nothing stored'); + await framework.stop(); + }); + + it("settling a held write whose landing is unknown, found, keeps it in the durable queue until an ack, takes the notes a store takes, and logs it by its id", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework) as ReturnType & { + unackedDeferredWrites: DeferredEntry[]; + durableDeferredWrites(): DeferredEntry[]; + noteStored(agent: unknown, metadata: unknown, stored: string): void; + }; + internals.deferredWritesPersisted = true; + heldAndStored(framework, 'w-found', 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: 'w-found' } } as never).messages; + const noted: string[] = []; + const note = internals.noteStored; + internals.noteStored = function (this: unknown, agent: unknown, metadata: unknown, stored: string) { + noted.push(stored); + return note.call(this, agent, metadata, stored); + }; + const errors: string[] = []; + const error = console.error; + console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }; + try { internals.queryMessages({}); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'no longer held'); + assert.deepEqual(internals.unackedDeferredWrites.map((m) => [m.id, m.landing]), [['w-found', 'unknown']], + 'un-acked, as a drained write is, and still marked, so a restart before the ack looks it up'); + assert.deepEqual(internals.durableDeferredWrites().map((m) => m.id), ['w-found'], 'so the durable queue keeps it until the next ack'); + assert.deepEqual(noted, [copy!.id], 'the notes a store takes, for the copy'); + assert.ok(errors.some((e) => e.includes('w-found') && e.includes('was found by its id')), errors.join('\n')); + await framework.stop(); + }); + + it("settling a held write whose landing is unknown, not found, persists the queue without its mark, as an edit does", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework) as ReturnType & { persistDeferredWrites(): void }; + internals.deferredWritesPersisted = true; + internals.deferredMessages.push({ ...entry('w-persist', 'look left'), landing: 'unknown' }); + let persisted = 0; + const persist = internals.persistDeferredWrites; + internals.persistDeferredWrites = function (this: unknown) { persisted++; return persist.call(this); }; + internals.queryMessages({}); + assert.equal(persisted, 1, 'the queue without its mark is persisted'); + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-persist', undefined]]); + await framework.stop(); + }); + + it("a module's edit or removal by a stored id settles the held write that row is a copy of, even while its store can't be searched", async () => { + for (const op of ['edit', 'remove'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-row-${op}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => { messages: Array<{ id: string }> }; + }; + const [copy] = cm.queryMessages({ metadata: { deferredWriteId: id } }).messages; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { + if (op === 'edit') internals.editMessage(copy!.id, text('look up')); + else internals.removeMessage(copy!.id); + } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(copiesOf(framework, id), op === 'edit' ? ['look up'] : [], `${op}: the row is changed once`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `${op}: its write is settled by the row, so nothing writes it again`); + await framework.stop(); + } + }); + + it("a module's edit or removal of a held write whose landing is unknown throws with nothing changed while its store can't be searched", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-blind', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + 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; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.content, m.landing]), [['w-blind', text('look left'), 'unknown']], 'still held, unedited'); + assert.deepEqual(copiesOf(framework, 'w-blind'), ['look left'], 'its copy unchanged'); + await framework.stop(); + }); + + it("a module's query meets a held write whose landing is unknown, settled, as the one stored row it landed as", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + framework.getAgent('assistant')!.getContextManager().addMessage('Antra', text('look right')); + heldAndStored(framework, 'w-q', 'look left'); + const error = console.error; + console.error = () => {}; + let rows: string[]; + try { + rows = internals.queryMessages({}).messages + .filter((m) => m.content.some((b) => ['look left', 'look right'].includes((b as { text?: string }).text ?? ''))) + .map((m) => `${m.held ? 'held' : 'stored'}: ${m.content.map((b) => (b as { text?: string }).text).join('')}`); + } finally { console.error = error; } + assert.deepEqual(rows, ['stored: look right', 'stored: look left']); + await framework.stop(); + }); + + it("a stored row settles only the held write whose id it carries", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-theirs', 'look right'); + heldAndStored(framework, 'w-mine', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (...args: unknown[]) => { messages: Array<{ id: string }> }; + }; + const [mine] = cm.queryMessages({ metadata: { deferredWriteId: 'w-mine' } }).messages; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { internals.getMessage(mine!.id); } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-theirs', 'unknown']], "the other write stays marked"); + await framework.stop(); + }); + + it("a module's getMessage, by a handle or by a stored id, settles a held write whose landing is unknown and answers its stored row", async () => { + for (const by of ['handle', 'stored id'] as const) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-get-${by === 'handle' ? 'h' : 's'}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, id, 'look left'); + const [copy] = framework.getAgent('assistant')!.getContextManager() + .queryMessages({ metadata: { deferredWriteId: id } } as never).messages; + const error = console.error; + console.error = () => {}; + let row: { held?: true; content: ContentBlock[] } | null; + try { row = internals.getMessage(by === 'handle' ? heldHandle(id) : copy!.id); } finally { console.error = error; } + assert.ok(row, `by its ${by}: a row`); + assert.equal(row.held, undefined, `by its ${by}: the stored row`); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], `by its ${by}: settled`); + await framework.stop(); + } + }); + + it("deferredWriteReadOrGone settles a held write whose landing is unknown, so a copy the agent has read reads as read", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + heldAndStored(framework, 'w-wake', 'look left'); + heldAndStored(framework, 'w-other', 'look right'); + agent.markContextConsumed(); + const error = console.error; + console.error = () => {}; + try { + assert.equal(internals.deferredWriteReadOrGone(agent, 'w-wake'), true); + } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => [m.id, m.landing]), [['w-other', 'unknown']], 'only the write asked about is settled'); + await framework.stop(); + }); + + it("the coalescer reads and removes a held write whose landing is unknown, once settled, as the stored placement it landed as", async () => { + for (const shown of [false, true]) { + // Each pass opens the test's one store again, so each takes its own id. + const id = `w-co-${shown ? 's' : 'u'}`; + const framework = await createFramework(); + const internals = drainInternals(framework); + const agent = framework.getAgent('assistant')!; + agent.markContextConsumed(); + if (!shown) agent.markContextShown(); + heldAndStored(framework, id, 'look left'); + if (shown) agent.markContextShown(); + const placement = { agent: 'assistant', deferredId: id }; + const error = console.error; + console.error = () => {}; + try { + assert.equal(internals.isUnreadPlacement(placement), true, 'unread, as its stored row is'); + assert.equal(internals.isUnreadPlacement(placement, 'shown'), !shown, `unshown as its stored row is (${shown ? 'shown' : 'not shown'})`); + assert.equal(internals.removePlacement(placement), true); + } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, id), [], 'its stored row goes'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'and nothing is held'); + await framework.stop(); + } + }); + + it("the turn's draft notice settles the agent's held writes whose landing is unknown before it reads what they name", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework) as ReturnType & { noticeUnnoticedDrafts(agent: unknown): void }; + heldAndStored(framework, 'w-draft', 'look left'); + const error = console.error; + console.error = () => {}; + try { internals.noticeUnnoticedDrafts(framework.getAgent('assistant')!); } finally { console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), []); + await framework.stop(); + }); + + it("the coalescer's removal settles a held write whose landing is unknown, and removes it as the stored placement it landed as", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + framework.getAgent('assistant')!.markContextConsumed(); + heldAndStored(framework, 'w-rmp', 'look left'); + const error = console.error; + console.error = () => {}; + try { + assert.equal(internals.removePlacement({ agent: 'assistant', deferredId: 'w-rmp' }), true); + } finally { console.error = error; } + assert.deepEqual(copiesOf(framework, 'w-rmp'), [], 'its stored row goes'); + assert.deepEqual(internals.deferredMessages.map((m) => m.id), [], 'and nothing is held'); + await framework.stop(); + }); + + it("the coalescer reads a held write whose landing is unknown as read and shown, and doesn't remove it, while its store can't be searched", async () => { + const framework = await createFramework(); + const internals = drainInternals(framework); + heldAndStored(framework, 'w-cblind', 'look left'); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { queryMessages: (...args: unknown[]) => unknown }; + const query = cm.queryMessages; + cm.queryMessages = () => { throw new Error('search failed'); }; + const error = console.error; + console.error = () => {}; + try { + const placement = { agent: 'assistant', deferredId: 'w-cblind' }; + assert.equal(internals.isUnreadPlacement({ ...placement }), false, 'a doubt reads as read'); + assert.equal(internals.isUnreadPlacement({ ...placement }, 'shown'), false, 'and as shown'); + assert.equal(internals.removePlacement({ ...placement }), false, "and isn't removed"); + } finally { cm.queryMessages = query; console.error = error; } + assert.deepEqual(internals.deferredMessages.map((m) => m.id), ['w-cblind'], 'still held'); + assert.deepEqual(copiesOf(framework, 'w-cblind'), ['look left']); + await framework.stop(); + }); + /** Shard a message whose first block is `marker` into one shard per block, * as a strategy's ingress chunking does. */ function shardOn(framework: AgentFramework, marker: string): void { diff --git a/test/module-rows-held-from-residents.test.ts b/test/module-rows-held-from-residents.test.ts index 5a4741df8..43ef150f6 100644 --- a/test/module-rows-held-from-residents.test.ts +++ b/test/module-rows-held-from-residents.test.ts @@ -496,4 +496,27 @@ describe("a message a tune-out holds stays out of what the residents are shown, await close(); } }); + + it("the cancel's backlog of held writes settles one whose landing was unknown first, so a write that landed is in the stored rows alone", async () => { + const { framework, internals, close } = await open(); + try { + // A held write whose first write landed, though its store couldn't be + // read to tell (agent-framework's DeferredWrite.landing). + framework.getAgent('scout')!.getContextManager().addMessage('Antra', [{ type: 'text', text: 'T7-LANDED' }], { deferredWriteId: 'w-tune' } as never); + (internals.deferredMessages as unknown as Array>).push({ + id: 'w-tune', seq: 1, participant: 'Antra', content: [{ type: 'text', text: 'T7-LANDED' }], heldAt: Date.now(), landing: 'unknown', + }); + const hooks = (internals.tuneOutCoordinator as unknown as { hooks: { getDeferredMessages(): Array<{ content: ContentBlock[] }> } }).hooks; + const error = console.error; + console.error = () => {}; + let held: string[]; + try { held = hooks.getDeferredMessages().map(textOf); } finally { console.error = error; } + assert.ok(!held.includes('T7-LANDED'), 'not among the held writes'); + const stored = internals.agents.get('scout')!.getContextManager().getAllMessages().filter((m) => textOf(m) === 'T7-LANDED'); + assert.equal(stored.length, 1, 'in the stored rows, once'); + assert.ok(!(internals.deferredMessages as unknown as Array<{ id: string }>).some((m) => m.id === 'w-tune'), 'and no longer queued'); + } finally { + await close(); + } + }); });