From fc2b78037bfe65f0ffcb6d4234560dc945d2ea3b Mon Sep 17 00:00:00 2001 From: Beth-2709 Date: Tue, 20 Oct 2026 00:20:15 -0400 Subject: [PATCH 1/3] fix(deferral): pass a released agent's queued failed reply once, by its own id A fork's or run's failed reply can still be in the deferred-write queue when the agent is released and also have a copy in the agent's window: its earlier write threw after storing and the store couldn't be read back to tell (`landing: 'unknown'`, #607), or a lookup has since found it and it waits for its ack. The release passed that copy to the primary as a fresh write under a new id, while it skipped or kept the entry, so the notice had two identities, and each path judged on its own whether the other would pass it: - a crash between the release's sync and its rewrite of the queue, or a crash before the ack of an entry a lookup had found, left the entry to boot, which passed it again: two notices; - stop() after a quiesce postponed the release passed the copy and left the entry queued as a gone agent's, which boot passed again: two; - a release whose window couldn't be read, while the lookup worked, passed neither: none; - a release whose store couldn't be searched, while the window could be read, stored the entry a second time and passed both: two, and one where the agent had read the copy. Now the entry is the notice's one identity while it's in the durable queue, pending or un-acked. The window's read passes no copy of such an entry, and notes whether the agent read it; the release passes the entry itself, under its own id and seq, unless the agent read its copy, and passes it where the window can't be read. Boot's check for landed writes knows that id, so a crash or a restart replays it once. Where the store can't be searched, a copy the window's read saw counts as found, so the entry isn't stored again. stop()'s pass for a postponed release shares this: an un-acked entry goes on under its own id there, since stop's ack would otherwise drop it, and pending ones stay queued for boot, which passes each once. Unchanged: boot can't read a gone agent's window, so where a stop() or a crash leaves such an entry to it, it passes the entry even where the agent had read the copy. flushDeferredWrites' comment now says so. Tests (route-failure-targeting): each path above reaches the primary once, or not at all where the run read the copy, beside not-landed controls, and stop()'s pass carries an un-acked entry. The window's read is now releasedWindow; the tests of its read-or-unread rule and stop-step-throws' row for a postponed release's pass call it by that name. #607's changelog fragment says what the release now does. Co-Authored-By: Claude Opus 5.5 --- .../deferred-write-landing-unknown.fixed.md | 13 +- src/framework.ts | 174 ++++++++++++------ test/route-failure-targeting.test.ts | 170 ++++++++++++++++- test/stop-step-throws.test.ts | 10 +- 4 files changed, 296 insertions(+), 71 deletions(-) diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md index c4c203d78..3ee38c638 100644 --- a/changelog.d/deferred-write-landing-unknown.fixed.md +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -24,7 +24,12 @@ 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. + released, looks such an entry up the same way, and, where the store + can't be searched, in the window the release reads: found, it isn't + stored again; where neither can see the store, it's stored anyway where + the store takes it, and the log says it may now be stored twice. A + failed reply among such entries reaches the primary once, as the entry + itself under its own id, so a crash or a restart doesn't tell it twice. + It isn't passed on where the release finds the fork or run had read the + copy that landed. Where a `stop()` or a crash leaves the entry to the + next boot, boot passes it on anyway, since it can't read that window. diff --git a/src/framework.ts b/src/framework.ts index dcaee1d50..8e023538f 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -798,6 +798,13 @@ 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 } : {}) }); +/** A released agent's window as its pass to the primary reads it + * (AgentFramework.releasedWindow). */ +type ReleasedWindow = { + unread: MessageMetadata[]; + held: Map; +}; + /** 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. */ @@ -3986,15 +3993,20 @@ export class AgentFramework { // later stop() takes what that leaves. Until then a persisted queue keeps // them for boot, and a memory-only one loses them at exit. // A window release the quiesce postponed (releasesAfterResume) never runs - // once the host stops, so its unread pass runs here: each failed reply - // stored unread in that window goes to the primary now, deferred while - // quiesced like any write, so it waits in the queue with the rest. What - // was queued for that agent stays queued, as a gone agent's. Each + // once the host stops, so its pass to the primary runs here + // (passOnFromReleasedWindow), deferred while quiesced like any write, so + // it waits in the queue with the rest: each failed reply stored unread in + // that window, and each un-acked one queued for it whose copy it didn't + // read, which the ack below would otherwise drop. What is still pending + // for that agent stays queued, as a gone agent's, and the boot or resume + // flush that meets it passes each failed reply among it on once, under + // its own id, so a copy of one in the window isn't passed as stored. Each // release's pass is a step of its own: one that throws fails the stop() // and leaves the others, and the flush below, to run. for (const agent of this.releasesAfterResume.values()) { step(() => { - for (const metadata of this.unreadFailedReplies(agent)) this.passFailedReplyToPrimary(agent.name, metadata); + const { forPrimary } = this.passOnFromReleasedWindow(agent); + if (forPrimary.length > 0) this.readdDrained(forPrimary, `stop: release of ${agent.name}`); }); } this.releasesAfterResume.clear(); @@ -6501,11 +6513,13 @@ export class AgentFramework { this.deferredMessages = keep; this.handOffDeferredWrites(flush); // A target no longer registered (a fork or a run gone before a crash - // that left this write to boot recovery) has no window to land in. A - // failed reply it never read becomes the primary's by-name notice, under - // the entry's own id and place in the queue (queuedFailedReplyForPrimary), - // and is written with the rest; any other write is dropped, logged, by - // addMessage. + // that left this write to boot recovery, or one whose release stop() + // left it to) has no window to land in. A failed reply queued for it + // becomes the primary's by-name notice, under the entry's own id and + // place in the queue (queuedFailedReplyForPrimary), and is written with + // the rest; any other write is dropped, logged, by addMessage. Nothing + // here can read the gone agent's window, so a failed reply an earlier + // write of which landed there, and which the agent read, goes on too. let passed = 0; const batch = flush.flatMap((msg) => { if (msg.forAgent === undefined || this.agents.has(msg.forAgent)) return [msg]; @@ -21556,19 +21570,23 @@ export class AgentFramework { * (releaseWindowNowOrAfterResume), since a write could land mid-surgery. * * The agent never reads its window again, so every failed-reply notice it - * never read is passed to the primary, by name, once: each one stored in - * its window above its consumed watermark (flushed there at its turn's end, - * or stored while it was idle) and not heard at a tool boundary, and each - * 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. 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 + * never read is passed to the primary, by name, once + * (passOnFromReleasedWindow): each one stored in its window above its + * consumed watermark (flushed there at its turn's end, or stored while it + * was idle) and not heard at a tool boundary, and each one still queued + * for it, pending or un-acked. One with an entry still in the durable + * queue goes on only as that entry, under its own id and seq, so a crash + * before the ack below replays it once; a copy of it in the window decides + * only whether the agent read 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. One whose earlier write may have landed (`landing` + * unknown) is looked up there by its id first, or, where the store can't + * be searched, in the window's own read: found, it isn't stored again; + * where neither can see the store, 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, * even if a watchdog has ended the run meanwhile. A tool boundary stores @@ -21576,25 +21594,26 @@ export class AgentFramework { * prepared. */ private releaseWindow(agent: Agent): void { - for (const metadata of this.unreadFailedReplies(agent)) this.passFailedReplyToPrimary(agent.name, metadata); + const { window, passOn, forPrimary, unacked } = this.passOnFromReleasedWindow(agent); const queued = this.drainDeferredFor(agent.name); - if (queued.length === 0) return; + // Nothing of its in the durable queue: nothing to store, pass on or ack. + if (queued.length === 0 && unacked === 0) return; 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) : []; + // elsewhere, but in the released agent's own store. Where the store + // can't be searched, a copy the window's read saw tells the same. + // Found, it landed, and it isn't stored again. When neither can see + // it, 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 lookedUp = msg.landing === 'unknown' ? this.landedCopiesById(agent, msg.id) : []; + const found = lookedUp ?? window?.held.get(msg.id)?.copies; if (found && found.length > 0) { this.logDeferredWriteError(msg, `release of ${agent.name}`, undefined, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); + passOn(msg); continue; } try { @@ -21616,45 +21635,92 @@ export class AgentFramework { err, ); } - // A failed reply, now in a window no one reads: the entry itself goes - // on to the primary as its by-name notice, under its own id and seq, so - // a crash before the ack below replays it once, not twice. - if ((msg.metadata as { kind?: unknown } | undefined)?.kind !== 'discord-send-failed') continue; - const passed = this.queuedFailedReplyForPrimary(agent.name, msg); - if (passed) forPrimary.push(passed); + passOn(msg); } if (forPrimary.length > 0) this.readdDrained(forPrimary, `release of ${agent.name}`); this.ackDeferredWrites(); } /** - * The failed-reply notices stored in `agent`'s window that it never read: - * those above the lower of its consumed watermark and the point where the - * watermark's branch and the current one part (branchesPartAt), or every - * one, when it never compiled. The watermark moves at a compile, and at a - * tool boundary whose + * The part of a released agent's pass to the primary that its release and + * stop() share (stop() runs it for a release the quiesce postponed). Each + * failed-reply notice stored in its window that it never read goes on as + * stored, by name (passFailedReplyToPrimary), except a copy of an entry + * for it still in the durable queue, pending or un-acked: that entry is + * the notice's one identity until its ack, the one boot's check for landed + * writes knows, so it's the entry that goes on, as the primary's by-name + * notice under its own id and seq (`passOn`, queuedFailedReplyForPrimary), + * and a crash before the ack replays it once, not twice. A queued failed + * reply goes on unless the agent read a copy of it that landed in its + * window; where the window can't be read, it goes on, since telling the + * primary of a failure the agent may have read costs less than telling no + * one. The un-acked entries for the agent have all landed (their write + * went through, or a lookup found it: settleFound), so they go on here; + * the pending ones are the caller's. Returns the window's read, `passOn`, + * what it has queued so far for the primary, and how many un-acked entries + * the agent had. + */ + private passOnFromReleasedWindow(agent: Agent): { + window: ReleasedWindow | null; + passOn: (msg: DeferredWrite) => void; + forPrimary: DeferredWrite[]; + unacked: number; + } { + const mine = (msg: DeferredWrite): boolean => (msg.forAgent ?? this.primaryAgentName) === agent.name; + const unacked = this.unackedDeferredWrites.filter(mine); + const window = this.releasedWindow(agent, new Set([...this.deferredMessages.filter(mine), ...unacked].map((m) => m.id))); + for (const metadata of window?.unread ?? []) this.passFailedReplyToPrimary(agent.name, metadata); + const forPrimary: DeferredWrite[] = []; + const passOn = (msg: DeferredWrite): void => { + if ((msg.metadata as { kind?: unknown } | undefined)?.kind !== 'discord-send-failed') return; + if (window?.held.get(msg.id)?.unread === false) return; + const passed = this.queuedFailedReplyForPrimary(agent.name, msg); + if (passed) forPrimary.push(passed); + }; + for (const msg of unacked) passOn(msg); + return { window, passOn, forPrimary, unacked: unacked.length }; + } + + /** + * What a released agent's window holds that its pass to the primary reads + * (passOnFromReleasedWindow): `unread`, the failed-reply notices stored + * there that it never read, other than copies of the entries `held` names; + * and `held`, for each of those entries with a copy there, its copies and + * whether the agent read them. A notice is read below the lower of its + * consumed watermark and the point where the watermark's branch and the + * current one part (branchesPartAt), and unread everywhere when it never + * compiled. The watermark moves at a compile, and at a tool boundary whose * stream declares it carries the messages injected there * (markContextConsumed), so a notice the live stream heard is under it, and * one a boundary stored without sending (a turn a tool ended, a budget - * restart, a stream that may drop injections) is above it. + * restart, a stream that may drop injections) is above it. null when the + * window can't be read, which is logged. */ - private unreadFailedReplies(agent: Agent): MessageMetadata[] { + private releasedWindow(agent: Agent, held: ReadonlySet): ReleasedWindow | null { try { const cm = agent.getContextManager(); const watermark = agent.getConsumedWatermark(); const floor = watermark ? Math.min(watermark.sequence, this.branchesPartAt(watermark.branch, cm.currentBranch().name)) : Number.NEGATIVE_INFINITY; - return cm.getAllMessages() - .filter((m) => { - const metadata = m.metadata as { kind?: unknown } | undefined; - return m.sequence > floor && metadata?.kind === 'discord-send-failed'; - }) - .map((m) => m.metadata as MessageMetadata); + const read: ReleasedWindow = { unread: [], held: new Map() }; + for (const m of cm.getAllMessages()) { + const metadata = m.metadata as { kind?: unknown; deferredWriteId?: unknown } | undefined; + const id = metadata?.deferredWriteId; + if (typeof id === 'string' && held.has(id)) { + const copies = read.held.get(id) ?? { copies: [], unread: false }; + copies.copies.push(storedCopy(m)); + copies.unread ||= m.sequence > floor; + read.held.set(id, copies); + } else if (m.sequence > floor && metadata?.kind === 'discord-send-failed') { + read.unread.push(m.metadata as MessageMetadata); + } + } + return read; } catch (err) { // A release must complete whatever its window's read meets. console.error(`[route-failure] ${agent.name}: could not read its window for unread failed replies:`, err); - return []; + return null; } } diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index 91c8bf3d3..804b4b502 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -757,13 +757,13 @@ describe('notices about an agent\'s own action', () => { }); it('counts as read, across branches, only what an agent\'s window shares with the branch it read up to the point they part', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { releasedWindow(agent: unknown, held: Set): { unread: Array<{ tag?: unknown }> } | null }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.releasedWindow(agent, new Set())!.unread.map((m) => m.tag); const read = notice('read'); agent.markContextConsumed(); @@ -783,13 +783,13 @@ describe('notices about an agent\'s own action', () => { }); it('counts as unread a notice stored on a branch after the agent read past the point where it parts from another', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { releasedWindow(agent: unknown, held: Set): { unread: Array<{ tag?: unknown }> } | null }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.releasedWindow(agent, new Set())!.unread.map((m) => m.tag); notice('before the branch'); const home = store.currentBranch().name; @@ -807,13 +807,13 @@ describe('notices about an agent\'s own action', () => { }); it('counts as unread what the current branch has from a branch between it and the one the agent read on', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { releasedWindow(agent: unknown, held: Set): { unread: Array<{ tag?: unknown }> } | null }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.releasedWindow(agent, new Set())!.unread.map((m) => m.tag); notice('read at home'); const home = store.currentBranch().name; @@ -834,13 +834,13 @@ describe('notices about an agent\'s own action', () => { }); it('counts as unread, across a branch made after the agent read, what was stored after its watermark', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { releasedWindow(agent: unknown, held: Set): { unread: Array<{ tag?: unknown }> } | null }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.releasedWindow(agent, new Set())!.unread.map((m) => m.tag); notice('read'); agent.markContextConsumed(); @@ -1281,5 +1281,159 @@ describe('notices about an agent\'s own action', () => { assert.equal(calls.mock.callCount(), 0, 'its late tool call is not dispatched'); await runFramework.resume({ force: true }); }); + + describe('a failed reply whose earlier write may have landed reaches the primary once', () => { + // A failed reply queued for the run, marked as one whose earlier write + // threw and whose landing couldn't be read back (`landing: 'unknown'`). + // Where that write landed, a copy under the entry's id is in the run's + // window, as the write left it. + const FAILED = { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 5 }; + const SPEAKERS_NOTICE = `${SEND_FAILED} Your previous reply (5 chars) could not be delivered to missing-channel (no registered channel).`; + type LandingState = ReturnType & { + releasesAfterResume: Map; + agents: Map; + persistDeferredWrites(): void; + settleUnknownLandings(which: (msg: { id: string }) => boolean): void; + }; + const state = () => internals(runFramework) as unknown as LandingState; + // Each restart's framework, stopped before the run's directory goes. + const restarts: AgentFramework[] = []; + afterEach(async () => { for (const framework of restarts.splice(0)) await framework.stop(); }); + /** The primary's by-name notices of the run's failed reply, stored or queued for it. */ + const told = (framework: AgentFramework): number => + count(framework.getAgent('primary')!.getContextManager().getAllMessages(), 'A reply by worker') + + count(internals(framework).deferredMessages.filter((m) => m.forAgent === undefined) as Array<{ content: unknown }>, 'A reply by worker'); + + async function queued(landed: boolean) { + const created = await runFramework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'explicit', + }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + if (landed) created.contextManager.addMessage('user', [{ type: 'text', text: SPEAKERS_NOTICE }], { ...FAILED, deferredWriteId: 'w-e' } as never); + (state().deferredMessages as unknown as Array>).push({ + id: 'w-e', seq: 1_000_000, participant: 'user', forAgent: 'worker', + content: [{ type: 'text', text: SPEAKERS_NOTICE }], metadata: FAILED, heldAt: Date.now(), landing: 'unknown', + }); + return created; + } + + /** Not started, so its startup watchdog ends it. */ + const ending = (created: Awaited>) => + runFramework.runEphemeralToCompletion(created.agent, created.contextManager, { startupTimeoutMs: 150 }); + + /** The framework on the same store after this one stops: a restart, which replays what the persisted queue holds. */ + async function restarted(): Promise { + await runFramework.stop(); + const framework = await AgentFramework.create({ + storePath: join(runDir, 'test.chronicle'), + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'primary', model: 'test-model', systemPrompt: 'Primary.', proseRouting: 'locus' }], + modules: [new SenderModule()], + }); + restarts.push(framework); + await internals(framework).initializeMcpl([]); + return framework; + } + + for (const landed of [true, false]) { + it(`once across a crash between its release's sync and its queue's rewrite (${landed ? 'its earlier write landed' : 'a control: nothing landed'})`, async (t) => { + t.mock.method(console, 'error', () => {}); + state().deferredWritesPersisted = true; + const created = await queued(landed); + state().persistDeferredWrites(); + // The crash: the release's ack syncs the store, and the queue is never rewritten after it. + state().persistDeferredWrites = () => {}; + await assert.rejects(ending(created), /failed to start inference/); + assert.equal(told(runFramework), 1, 'told once at the release'); + + const reopened = await restarted(); + assert.equal(told(reopened), 1, 'and once after the restart replays the queue'); + }); + + it(`once across stop() and a restart when a quiesce postponed its release (${landed ? 'its earlier write landed' : 'a control: nothing landed'})`, async (t) => { + t.mock.method(console, 'error', () => {}); + const created = await queued(landed); + const done = ending(created); + state().quiesced = true; + // quiesce() writes the queue to its record as it sets the flag. + state().persistDeferredWrites(); + await assert.rejects(done, /failed to start inference/); + assert.ok(state().releasesAfterResume.has('worker'), "its window's release waits for resume"); + + const reopened = await restarted(); + assert.equal(told(reopened), 1); + }); + } + + it('once, by its own id, when its earlier write landed and the run\'s window can\'t be read', async (t) => { + t.mock.method(console, 'error', () => {}); + const created = await queued(true); + t.mock.method(created.contextManager, 'getAllMessages', () => { throw new Error('window unreadable'); }); + await assert.rejects(ending(created), /failed to start inference/); + const notices = runFramework.getAgent('primary')!.getContextManager().getAllMessages() + .filter((m) => JSON.stringify(m.content).includes('A reply by worker')); + assert.deepEqual(notices.map((m) => (m.metadata as { deferredWriteId?: unknown } | undefined)?.deferredWriteId), ['w-e']); + }); + + for (const read of [false, true]) { + it(`${read ? 'not at all, where the run read the copy' : 'once'}, stored once, when its store can't be searched and the window's read finds the copy`, async (t) => { + t.mock.method(console, 'error', () => {}); + const created = await queued(true); + if (read) created.agent.markContextConsumed(); + t.mock.method(created.contextManager, 'queryMessages', () => { throw new Error('search failed'); }); + await assert.rejects(ending(created), /failed to start inference/); + assert.equal(told(runFramework), read ? 0 : 1); + assert.equal(count(await archived(), 'Your previous reply (5 chars)'), 1, 'not stored a second time'); + }); + } + + it('once across a crash before any ack, when a lookup found its earlier write landed and it waits for an ack', async (t) => { + t.mock.method(console, 'error', () => {}); + state().deferredWritesPersisted = true; + const created = await queued(true); + const done = ending(created); + for (let i = 0; i < 100 && !state().agents.has('worker'); i++) await new Promise((r) => setImmediate(r)); + // Found by a path that reads held writes (settleFound): un-acked until the next ack. + state().settleUnknownLandings((m) => m.id === 'w-e'); + assert.deepEqual((state().unackedDeferredWrites as Array<{ id: string }>).map((m) => m.id), ['w-e']); + state().persistDeferredWrites(); + // The crash: nothing rewrites the queue after this. + state().persistDeferredWrites = () => {}; + await assert.rejects(done, /failed to start inference/); + assert.equal(told(runFramework), 1, 'told once at the release'); + + const reopened = await restarted(); + assert.equal(told(reopened), 1, 'and once after the restart replays the queue'); + }); + + it("once across stop() and a restart when a quiesce postponed its release and it's still un-acked there (a control)", async (t) => { + t.mock.method(console, 'error', () => {}); + const created = await queued(true); + const done = ending(created); + for (let i = 0; i < 100 && !state().agents.has('worker'); i++) await new Promise((r) => setImmediate(r)); + state().settleUnknownLandings((m) => m.id === 'w-e'); + assert.deepEqual((state().unackedDeferredWrites as Array<{ id: string }>).map((m) => m.id), ['w-e']); + // A write queued for the primary, so stop()'s last flush acks what's un-acked. + (state().deferredMessages as unknown as Array>).push({ + id: 'w-p', seq: 1, participant: 'user', content: [{ type: 'text', text: 'an ordinary write' }], heldAt: Date.now(), + }); + state().quiesced = true; + state().persistDeferredWrites(); + await assert.rejects(done, /failed to start inference/); + assert.ok(state().releasesAfterResume.has('worker'), "its window's release waits for resume"); + // stop()'s first ack fails to sync, so the entry is still un-acked at + // the pass for postponed releases, and the last flush's ack takes it. + const store = runFramework.getStore(); + const sync = store.sync.bind(store); + let failNext = true; + t.mock.method(store, 'sync', () => { + if (failNext) { failNext = false; throw new Error('sync failed'); } + return sync(); + }); + + const reopened = await restarted(); + assert.equal(told(reopened), 1); + }); + }); }); }); diff --git a/test/stop-step-throws.test.ts b/test/stop-step-throws.test.ts index 2860d67b5..a41021d95 100644 --- a/test/stop-step-throws.test.ts +++ b/test/stop-step-throws.test.ts @@ -178,14 +178,14 @@ const ROWS: Row[] = [ // Other agents, such as the run whose end releases its window, read // their unread replies as usual. const owner = fw as unknown as Record; - const usual = (owner.unreadFailedReplies as (agent: unknown) => unknown[]).bind(fw); - owner.unreadFailedReplies = (agent: { name: string }) => { + const usual = (owner.releasedWindow as (agent: unknown, held: Set) => unknown).bind(fw); + owner.releasedWindow = (agent: { name: string }, held: Set) => { if (agent.name === 'first') throw e; - if (agent.name !== 'second') return usual(agent); + if (agent.name !== 'second') return usual(agent, held); unreadPassed.push(agent.name); - return []; + return { unread: [], held: new Map() }; }; - return () => { delete owner.unreadFailedReplies; }; + return () => { delete owner.releasedWindow; }; }, check: (fw) => { assert.deepEqual(unreadPassed, ['second'], "the other release's unread pass still ran"); From d7ad0225df4bf3342625004ef7f474072877502c Mon Sep 17 00:00:00 2001 From: Beth-2709 Date: Wed, 21 Oct 2026 18:05:21 -0400 Subject: [PATCH 2/3] fix(deferral): a stop() pass that throws still sends its agent's un-acked notices From an internal review of the merge with main. - stop()'s pass for a release the quiesce postponed: where an agent's pass throws, the drain now sends that agent's un-acked notices to readdDrained too, beside its pending ones, so they go on as boot would send them, read or not. Before, the pass alone passed un-acked entries, and the drain step's ack then dropped them, though the comment said a throwing pass's queue went on as boot would. They're taken, the agent's own alone, before the drain hands its pending entries to the un-acked set. - Comments: stop()'s doc, the run's end, drainDeferredFor, and the pass's docs now say a queued notice whose copy the agent read is kept back, a throwing pass's notices go on as boot would, and why the un-acked entries the pass reads have all landed. #616's and #607's pending changelog fragments say so too, the latter that the release's once holds as long as its sync went through, and that a crash or a failed sync leaves the entry to boot; #616's that a channel notice the agent could act on goes on with failed replies. Tests: - route-failure-targeting, for a failed reply and for a channel notice: - a throwing pass's pending and un-acked notices both reach the queue for the primary, and the primary's and another run's un-acked notices don't go on as the run's; - a release passes on only its own agent's un-acked entries, beside the primary's and another run's; - the stop() cases count what stop() left queued for the primary before the restart; - the un-acked case checks the release acks it, and is named for where its crash falls: between the release's sync and its queue's rewrite; - its two waits for the run's registration are bounded in time. - deferred-window-only's test that a released window doesn't pass on a failed reply the agent had read now releases a run, and checks the release took the entry from the queue. Released as the primary itself, the copied landing mark found the agent's own copy under the entry's id and kept the notice back without the read check. Co-Authored-By: Claude Opus 5.5 --- .../deferred-write-landing-unknown.fixed.md | 11 +-- changelog.d/stop-ends-live-forks.fixed.md | 8 ++- src/framework.ts | 46 +++++++----- test/deferred-window-only.test.ts | 21 +++--- test/route-failure-targeting.test.ts | 71 +++++++++++++++++-- 5 files changed, 118 insertions(+), 39 deletions(-) diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md index 4a7ca28b8..76856834f 100644 --- a/changelog.d/deferred-write-landing-unknown.fixed.md +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -30,8 +30,9 @@ the store takes it, and the log says it may now be stored twice. A failed reply, or a channel notice the fork or run could act on, among such entries reaches the primary once, as the entry itself under its own - id, so a crash or a restart doesn't tell it twice. It isn't passed on - where the release finds the fork or run had read the copy that landed, - at its release or at the `stop()` that ends it. Where a crash leaves the - entry to the next boot, boot passes it on anyway, since it can't read - that window. + id, so a crash or a restart doesn't tell it twice, as long as the + release's sync went through. It isn't passed on where the release finds + the fork or run had read the copy that landed, at its release or at the + `stop()` that ends it. Where a crash or a failed sync leaves the entry + to the next boot, boot passes it on anyway, since it can't read that + window. diff --git a/changelog.d/stop-ends-live-forks.fixed.md b/changelog.d/stop-ends-live-forks.fixed.md index e4d9594ce..973cac68a 100644 --- a/changelog.d/stop-ends-live-forks.fixed.md +++ b/changelog.d/stop-ends-live-forks.fixed.md @@ -19,6 +19,8 @@ - A failed reply still queued for a fork or a run that ended while the host was quiesced is now passed to the primary when `stop()` runs, named as the registry knows its channel then, and lands when the host resumes. - Before, the next boot's flush passed it on. A memory-only queue still - loses it at exit. Any other write queued for such an agent is still - dropped, logged. + It isn't passed on where the fork or run had read the copy of it that + landed in its window. Before, the next boot's flush passed it on. A + memory-only queue still loses it at exit. Any other write queued for + such an agent, but a channel notice it could act on, is still dropped, + logged. diff --git a/src/framework.ts b/src/framework.ts index f120df25f..8d594c527 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -3716,8 +3716,9 @@ export class AgentFramework { * * Each conversation fork still live ends, as a closing fork's end does, * whatever its stream is doing, as each run does: its channel unbinds, and - * its window's release passes on to the primary the failed replies it left, - * unread in its window or still queued for it. Its end settles a + * its window's release passes on to the primary the notices it left that + * the primary takes in its place, unread in its window, or still queued + * for it unless it read the copy that landed there. Its end settles a * tool-result batch its stream still held and removes the stream's * entries, and once it has left, its stream's late events are dropped. */ @@ -4091,9 +4092,9 @@ export class AgentFramework { // does (flushForStart). Each release's pass is a step of its own, and so // is the drain: one that throws fails the stop() and leaves the others, // and the final flush, to run, and the drain sends a queue whose pass - // threw as boot would, its notices to the primary whether read or not. - // It runs again after the waits below, for a release an end they let - // settle postpones. + // threw as boot would, its notices to the primary whether read or not, + // un-acked ones included. It runs again after the waits below, for a + // release an end they let settle postpones. const passPostponedReleases = () => { const agents = [...this.releasesAfterResume.values()]; // Each pass reads its window, and what is queued for it, before the @@ -4106,11 +4107,16 @@ export class AgentFramework { const entries: DeferredWrite[] = []; for (const agent of agents) { const pass = passes.get(agent.name); + // A pass that threw passed on none of the agent's un-acked entries, + // which the ack below would drop, so their notices go too: taken + // before the drain hands its pending entries to the un-acked set. + const unacked = pass ? [] : this.unackedDeferredWrites + .filter((msg) => (msg.forAgent ?? this.primaryAgentName) === agent.name && passesToPrimary(msg.metadata)); for (const msg of this.drainDeferredFor(agent.name)) { if (pass && passesToPrimary(msg.metadata)) pass.passOn(msg); else entries.push(msg); } - if (pass) entries.push(...pass.forPrimary); + entries.push(...(pass ? pass.forPrimary : unacked)); } this.readdDrained(entries, 'stop: postponed releases'); this.ackDeferredWrites(); @@ -8550,7 +8556,8 @@ export class AgentFramework { // investigation, now or, while quiesced, at resume. A failed reply, or // a channel notice the run could act on, among it goes on to the // primary as the same queue entry, so no ack comes between the entry - // and the primary's notice. + // and the primary's notice, unless the run read the copy of it that + // landed in its window. // Exact-generation disposal owns these name-keyed entries. Remove them // before deregistration; the late driveStream finally will see the // generation mismatch and leave any future owner untouched. A fork's @@ -21584,9 +21591,11 @@ export class AgentFramework { // A fork or run deregistered while the host is quiesced leaves its queue // to its window's release at resume (runReleasesAfterResume, which clears // the name first), which stores its writes in that window, or to stop()'s - // pass for postponed releases, which clears the name and sends them as - // boot would. Drained by any other flush meanwhile, they would meet no - // registered agent and be dropped, or written into a window no one reads. + // pass for postponed releases, which clears the name, passes its notices + // on as that release would (as boot would, where its pass throws), and + // drops the rest, logged. Drained by any + // other flush meanwhile, they would meet no registered agent and be + // dropped, or written into a window no one reads. if (this.releasesAfterResume.has(agentName)) return []; const mine: typeof this.deferredMessages = []; const rest: typeof this.deferredMessages = []; @@ -22038,11 +22047,12 @@ export class AgentFramework { * twice. A queued notice goes on unless the agent read a copy of it that * landed in its window; where the window can't be read, it goes on, since * telling the primary of a notice the agent may have read costs less than - * telling no one. The un-acked entries for the agent have all landed - * (their write went through, or a lookup found it: settleFound), so they - * go on here; the pending ones are the caller's. Returns the window's - * read, `passOn`, what it has queued so far for the primary, and how many - * un-acked entries the agent had. + * telling no one. Both callers run this before their drain, which hands + * pending entries to the un-acked set, so the un-acked entries for the + * agent here have all landed (their write went through, or a lookup found + * it: settleFound), and they go on here; the pending ones are the + * caller's. Returns the window's read, `passOn`, what it has queued so far + * for the primary, and how many un-acked entries the agent had. */ private passOnFromReleasedWindow(agent: Agent): { window: ReleasedWindow | null; @@ -22071,9 +22081,9 @@ export class AgentFramework { * never read and the primary takes in its place (passesToPrimary: its * failed replies, and the channel notices it was told it could act on), * other than copies of the entries `held` names; and `held`, for each of - * those entries with a copy there, its copies and whether the agent read - * them. A notice is read below the lower of its consumed watermark and the - * point where the watermark's branch and the current one part + * those entries with a copy there, its copies and whether any of them is + * unread. A notice is read below the lower of its consumed watermark and + * the point where the watermark's branch and the current one part * (branchesPartAt), and unread everywhere when it has no watermark. The * watermark is set when the agent is created, and moves at a compile, and * at a tool boundary whose stream declares it carries the messages diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 57647ea53..2ddc39df2 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -868,19 +868,24 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { 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(); + // A run, whose window is its own, as in the test before: the primary's + // window holds no copy under the entry's id to answer for the read check. + const { agent, contextManager: cm } = await framework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', + }); 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. + cm.addMessage('worker', text('hello'), { ...failed, deferredWriteId: 'w-read' } as never); + // The run read its copy: the window's read 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' }); + internals.deferredMessages.push({ ...entry('w-read', 'hello'), participant: 'worker', forAgent: 'worker', 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.ok(!internals.deferredMessages.some((m) => m.id === 'w-read'), 'the release took the entry from the queue'); + const notices = framework.getAgent('assistant')!.getContextManager().queryMessages({} as never).messages + .filter((m) => (m.metadata as { speaker?: unknown } | undefined)?.speaker === 'worker'); assert.equal(notices.length, 0, "the primary isn't told"); await framework.stop(); }); diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index f2655090c..96fbd60fb 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -2232,10 +2232,12 @@ describe('notices about an agent\'s own action', () => { // Each restart's framework, stopped before the run's directory goes. const restarts: AgentFramework[] = []; afterEach(async () => { for (const framework of restarts.splice(0)) await framework.stop(); }); + /** The primary's by-name notices in the run's place, queued for it. */ + const queuedForPrimary = (framework: AgentFramework): number => + count(internals(framework).deferredMessages.filter((m) => m.forAgent === undefined) as Array<{ content: unknown }>, notice.toPrimary); /** The primary's by-name notices in the run's place, stored or queued for it. */ const told = (framework: AgentFramework): number => - count(framework.getAgent('primary')!.getContextManager().getAllMessages(), notice.toPrimary) + - count(internals(framework).deferredMessages.filter((m) => m.forAgent === undefined) as Array<{ content: unknown }>, notice.toPrimary); + count(framework.getAgent('primary')!.getContextManager().getAllMessages(), notice.toPrimary) + queuedForPrimary(framework); async function queued(landed: boolean) { const created = await runFramework.createEphemeralAgent({ @@ -2296,6 +2298,7 @@ describe('notices about an agent\'s own action', () => { assert.ok(state().releasesAfterResume.has('worker'), "its window's release waits for resume"); const reopened = await restarted(); + assert.equal(queuedForPrimary(runFramework), 1, 'stop() leaves one notice queued for the primary'); assert.equal(told(reopened), 1); }); } @@ -2311,9 +2314,65 @@ describe('notices about an agent\'s own action', () => { assert.ok(state().releasesAfterResume.has('worker'), "its window's release waits for resume"); const reopened = await restarted(); + assert.equal(queuedForPrimary(runFramework), 0, 'stop() leaves nothing queued for the primary'); assert.equal(told(reopened), 0); }); + it("as boot would, read or not, a pending one and an un-acked one, when stop()'s pass for a release the quiesce postponed throws", async (t) => { + t.mock.method(console, 'error', () => {}); + const created = await queued(true); + created.agent.markContextConsumed(); + const done = ending(created); + for (let i = 0; i < 400 && !state().agents.has('worker'); i++) await new Promise((r) => setTimeout(r, 5)); + state().settleUnknownLandings((m) => m.id === 'w-e'); + assert.deepEqual((state().unackedDeferredWrites as Array<{ id: string }>).map((m) => m.id), ['w-e']); + // A second notice for the run, still pending, which never landed. + (state().deferredMessages as unknown as Array>).push({ + id: 'w-f', seq: 1_000_001, participant: 'user', forAgent: 'worker', + content: [{ type: 'text', text: notice.text }], metadata: notice.metadata, heldAt: Date.now(), + }); + // Un-acked notices of other agents', the primary's own and another + // run's, which the run's fallback mustn't take as its own. + (state().unackedDeferredWrites as unknown as Array>).push( + { id: 'w-p', seq: 999_998, participant: 'user', content: [{ type: 'text', text: notice.text }], metadata: notice.metadata, heldAt: Date.now() }, + { id: 'w-o', seq: 999_999, participant: 'user', forAgent: 'other', content: [{ type: 'text', text: notice.text }], metadata: notice.metadata, heldAt: Date.now() }, + ); + state().quiesced = true; + state().persistDeferredWrites(); + await assert.rejects(done, /failed to start inference/); + assert.ok(state().releasesAfterResume.has('worker'), "its window's release waits for resume"); + // stop()'s first ack fails to sync, so the un-acked entry is still + // there when the pass for postponed releases runs, and throws. + const store = runFramework.getStore(); + const sync = store.sync.bind(store); + let failNext = true; + t.mock.method(store, 'sync', () => { + if (failNext) { failNext = false; throw new Error('sync failed'); } + return sync(); + }); + t.mock.method(state() as unknown as { passOnFromReleasedWindow(): unknown }, 'passOnFromReleasedWindow', () => { throw new Error('pass failed'); }); + + await assert.rejects(runFramework.stop(), /pass failed/); + assert.equal(told(runFramework), 2, 'both are queued for the primary, though the run read the landed one'); + assert.deepEqual( + (state().deferredMessages as Array<{ id: string }>).map((m) => m.id).filter((id) => id === 'w-p' || id === 'w-o'), [], + "the primary's and the other run's un-acked notices, which landed, aren't queued again", + ); + }); + + it("passes on only the released run's own un-acked entries, not another agent's", async (t) => { + t.mock.method(console, 'error', () => {}); + const created = await queued(false); + // Un-acked notices of other agents', the primary's own and another + // run's: written to their windows, their acks still to come. + (state().unackedDeferredWrites as unknown as Array>).push( + { id: 'w-p', seq: 999_998, participant: 'user', content: [{ type: 'text', text: notice.text }], metadata: notice.metadata, heldAt: Date.now() }, + { id: 'w-o', seq: 999_999, participant: 'user', forAgent: 'other', content: [{ type: 'text', text: notice.text }], metadata: notice.metadata, heldAt: Date.now() }, + ); + await assert.rejects(ending(created), /failed to start inference/); + assert.equal(told(runFramework), 1, "told once, of the run's own notice"); + }); + it('once, by its own id, when its earlier write landed and the run\'s window can\'t be read', async (t) => { t.mock.method(console, 'error', () => {}); const created = await queued(true); @@ -2336,12 +2395,12 @@ describe('notices about an agent\'s own action', () => { }); } - it('once across a crash before any ack, when a lookup found its earlier write landed and it waits for an ack', async (t) => { + it("once across a crash between its release's sync and its queue's rewrite, when a lookup found its earlier write landed and it waits for an ack", async (t) => { t.mock.method(console, 'error', () => {}); state().deferredWritesPersisted = true; const created = await queued(true); const done = ending(created); - for (let i = 0; i < 100 && !state().agents.has('worker'); i++) await new Promise((r) => setImmediate(r)); + for (let i = 0; i < 400 && !state().agents.has('worker'); i++) await new Promise((r) => setTimeout(r, 5)); // Found by a path that reads held writes (settleFound): un-acked until the next ack. state().settleUnknownLandings((m) => m.id === 'w-e'); assert.deepEqual((state().unackedDeferredWrites as Array<{ id: string }>).map((m) => m.id), ['w-e']); @@ -2350,6 +2409,7 @@ describe('notices about an agent\'s own action', () => { state().persistDeferredWrites = () => {}; await assert.rejects(done, /failed to start inference/); assert.equal(told(runFramework), 1, 'told once at the release'); + assert.deepEqual(state().unackedDeferredWrites, [], 'and acked there'); const reopened = await restarted(); assert.equal(told(reopened), 1, 'and once after the restart replays the queue'); @@ -2359,7 +2419,7 @@ describe('notices about an agent\'s own action', () => { t.mock.method(console, 'error', () => {}); const created = await queued(true); const done = ending(created); - for (let i = 0; i < 100 && !state().agents.has('worker'); i++) await new Promise((r) => setImmediate(r)); + for (let i = 0; i < 400 && !state().agents.has('worker'); i++) await new Promise((r) => setTimeout(r, 5)); state().settleUnknownLandings((m) => m.id === 'w-e'); assert.deepEqual((state().unackedDeferredWrites as Array<{ id: string }>).map((m) => m.id), ['w-e']); // A write queued for the primary, so stop()'s last flush acks what's un-acked. @@ -2381,6 +2441,7 @@ describe('notices about an agent\'s own action', () => { }); const reopened = await restarted(); + assert.equal(queuedForPrimary(runFramework), 1, 'stop() leaves one notice queued for the primary'); assert.equal(told(reopened), 1); }); }); From 90030190d3b41979f7aef4f2ebc356f512292a35 Mon Sep 17 00:00:00 2001 From: Beth-2709 Date: Wed, 21 Oct 2026 18:46:49 -0400 Subject: [PATCH 3/3] docs(changelog): name the stop() pass that fails in #616's fragment The fragment said a queued failed reply isn't passed on where the fork or run had read its landed copy. Where stop()'s pass over that window throws, the agent's notices go on as boot would send them, read or not, as the test "as boot would, read or not, ... when stop()'s pass ... throws" pins. The fragment now says so. From the automated review on #680. Co-Authored-By: Claude Opus 5.5 --- changelog.d/stop-ends-live-forks.fixed.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/changelog.d/stop-ends-live-forks.fixed.md b/changelog.d/stop-ends-live-forks.fixed.md index 973cac68a..f529016a7 100644 --- a/changelog.d/stop-ends-live-forks.fixed.md +++ b/changelog.d/stop-ends-live-forks.fixed.md @@ -20,7 +20,8 @@ host was quiesced is now passed to the primary when `stop()` runs, named as the registry knows its channel then, and lands when the host resumes. It isn't passed on where the fork or run had read the copy of it that - landed in its window. Before, the next boot's flush passed it on. A - memory-only queue still loses it at exit. Any other write queued for - such an agent, but a channel notice it could act on, is still dropped, - logged. + landed in its window, unless `stop()`'s pass over that window fails, + when it goes on as boot would send it, read or not. Before, the next + boot's flush passed it on. A memory-only queue still loses it at exit. + Any other write queued for such an agent, but a channel notice it could + act on, is still dropped, logged.