diff --git a/changelog.d/deferred-write-landing-unknown.fixed.md b/changelog.d/deferred-write-landing-unknown.fixed.md index c4c203d78..76856834f 100644 --- a/changelog.d/deferred-write-landing-unknown.fixed.md +++ b/changelog.d/deferred-write-landing-unknown.fixed.md @@ -24,7 +24,15 @@ 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, 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, 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..f529016a7 100644 --- a/changelog.d/stop-ends-live-forks.fixed.md +++ b/changelog.d/stop-ends-live-forks.fixed.md @@ -19,6 +19,9 @@ - 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, 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. diff --git a/src/framework.ts b/src/framework.ts index 9bea1fd49..8d594c527 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -848,6 +848,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. */ @@ -3709,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. */ @@ -4070,31 +4078,47 @@ export class AgentFramework { }); } // A window release the quiesce postponed (releasesAfterResume) never runs - // once the host stops, so its unread pass runs here: each notice stored - // unread in that window that the primary takes in its place (a failed - // reply, or a channel notice its reader could act on) 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 goes as boot would - // send it (readdDrained): such a notice to the primary, deferred again - // under its own id, and anything else dropped, logged, since nothing - // lands while quiesced. So no flush that comes later meets such a notice - // of an agent that has gone, not even one that writes into that agent's - // own window, as a turn of the fork's still assembling or at a tool - // boundary 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. It runs again after the waits - // below, for a release an end they let settle postpones. + // once the host stops, so its pass to the primary runs here, deferred + // while quiesced like any write, so it waits in the queue with the rest: + // each notice stored unread in that window that the primary takes in its + // place (a failed reply, or a channel notice its reader could act on), + // and each such notice still queued for that agent, under its own id, + // unless the agent read a copy of it that landed in its window + // (passOnFromReleasedWindow). What else was queued for it goes as boot + // would send it (readdDrained): dropped, logged, since nothing lands + // while quiesced. So no flush that comes later meets such a notice of an + // agent that has gone, not even one that writes into that agent's own + // window, as a turn of the fork's still assembling or at a tool boundary + // 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, + // un-acked ones included. It runs again after the waits below, for a + // release an end they let settle postpones. const passPostponedReleases = () => { - for (const agent of this.releasesAfterResume.values()) { - step(() => { - for (const metadata of this.unreadNoticesForPrimary(agent)) this.passNoticeToPrimary(agent.name, metadata); - }); - } - const released = [...this.releasesAfterResume.keys()]; + const agents = [...this.releasesAfterResume.values()]; + // Each pass reads its window, and what is queued for it, before the + // drain below moves its pending entries into the un-acked set. + const passes = new Map>(); + for (const agent of agents) step(() => { passes.set(agent.name, this.passOnFromReleasedWindow(agent)); }); this.releasesAfterResume.clear(); - if (released.length === 0) return; + if (agents.length === 0) return; step(() => { - this.readdDrained(released.flatMap((name) => this.drainDeferredFor(name)), 'stop: postponed releases'); + 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); + } + entries.push(...(pass ? pass.forPrimary : unacked)); + } + this.readdDrained(entries, 'stop: postponed releases'); this.ackDeferredWrites(); }); }; @@ -8532,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 @@ -21566,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 = []; @@ -21866,11 +21893,13 @@ export class AgentFramework { * landed in, and its newest stays the newest (remainingArrival). * * A target no longer registered (a fork or a run that has ended, here or - * before a crash) has no window to land in: a failed reply it never read, + * before a crash) has no window to land in: a failed reply queued for it, * or a channel notice it could have acted on, becomes the primary's * by-name notice, under the entry's own id and place in the queue * (queuedNoticeForPrimary), and is written with the rest; any other write - * is dropped, logged, by addMessage. */ + * is dropped, logged, by addMessage. Nothing here reads the gone agent's + * window, so such a notice an earlier write of which landed there, and + * which the agent read, goes on too. */ private readdDrained(drained: DeferredWrite[], where: string, opts?: { bypassDeferral?: boolean }): DeferredWrite[] { let passed = 0; const entries = drained.flatMap((msg) => { @@ -21930,49 +21959,53 @@ export class AgentFramework { * * The agent never reads its window again, so every failed-reply notice it * never read, and every channel notice it could act on and never read, is - * passed to the primary, by name, once (passesToPrimary): 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, as - * unreadNoticesForPrimary counts them, and each one still queued for it. - * Writes still queued would wait for a boundary it never reaches, and a - * flush that finds it gone would drop all but the notices it passes on, 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, or, when stop() ends it, may still run - * past its cancel, and then drops each event that comes after the agent - * has left (driveStream); a turn of its still assembling or compiling, as - * when stop() ends it, drops before it sends anything. 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 nothing for a round whose turn was - * stopped while its results were prepared. + * passed to the primary, by name, once (passesToPrimary, + * 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 a flush that finds it gone would drop all + * but the notices it passes on, 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, or, when stop() ends it, may still run past its cancel, and + * then drops each event that comes after the agent has left (driveStream); + * a turn of its still assembling or compiling, as when stop() ends it, + * drops before it sends anything. 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 nothing for a round whose turn was stopped while its results were + * prepared. */ private releaseWindow(agent: Agent): void { - for (const metadata of this.unreadNoticesForPrimary(agent)) this.passNoticeToPrimary(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 notice the primary takes in its - // place that 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 { @@ -21994,35 +22027,75 @@ export class AgentFramework { err, ); } - // A failed reply, or a channel notice its reader could act on, 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 (!passesToPrimary(msg.metadata)) continue; - const passed = this.queuedNoticeForPrimary(agent.name, msg); - if (passed) forPrimary.push(passed); + passOn(msg); } if (forPrimary.length > 0) this.readdDrained(forPrimary, `release of ${agent.name}`); this.ackDeferredWrites(); } /** - * The notices stored in `agent`'s window that it never read and the primary - * takes in its place (passesToPrimary): its failed replies, and the channel - * notices it was told it could act on. Those are the ones 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 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 injected there (markContextConsumed). So a notice such a stream - * carried is under it. - * One a boundary stored without sending (a turn a tool ended, a budget - * restart) is above it, and so is one injected into a stream that doesn't - * declare it carries injections, which may have dropped it: counted as - * unread, it goes to the primary even if the agent heard it, rather than to - * no one where the stream dropped it. - */ - private unreadNoticesForPrimary(agent: Agent): MessageMetadata[] { + * 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 + * notice stored in its window that it never read and the primary takes in + * its place (passesToPrimary: a failed reply, or a channel notice its + * reader could act on) goes on as stored, by name (passNoticeToPrimary), + * 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`, + * queuedNoticeForPrimary), and a crash before the ack replays it once, not + * 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. 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; + 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.passNoticeToPrimary(agent.name, metadata); + const forPrimary: DeferredWrite[] = []; + const passOn = (msg: DeferredWrite): void => { + if (!passesToPrimary(msg.metadata)) return; + if (window?.held.get(msg.id)?.unread === false) return; + const passed = this.queuedNoticeForPrimary(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 notices stored there that it + * 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 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 + * injected there (markContextConsumed). So a notice such a stream carried + * is under it. One a boundary stored without sending (a turn a tool ended, + * a budget restart) is above it, and so is one injected into a stream that + * doesn't declare it carries injections, which may have dropped it: + * counted as unread, it goes to the primary even if the agent heard it, + * rather than to no one where the stream dropped it. null when the window + * can't be read, which is logged. + */ + private releasedWindow(agent: Agent, held: ReadonlySet): ReleasedWindow | null { try { const cm = agent.getContextManager(); const watermark = agent.getConsumedWatermark(); @@ -22034,15 +22107,26 @@ export class AgentFramework { // so a shard that continues the body before it isn't taken again. Which // messages make up a body is context-manager's rule (continuesBody, // pairwise in store order): a group id names the text, so the same text - // stored twice is two bodies under one id. + // stored twice is two bodies under one id. A held entry's copies are + // kept whole, each shard among them, as its lookup by id finds them. + const read: ReleasedWindow = { unread: [], held: new Map() }; const messages = cm.getAllMessages(); - return messages - .filter((m, i) => m.sequence > floor && passesToPrimary(m.metadata) && !continuesBody(messages[i - 1], m)) - .map((m) => m.metadata as MessageMetadata); + messages.forEach((m, i) => { + const id = (m.metadata as { deferredWriteId?: unknown } | undefined)?.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 && passesToPrimary(m.metadata) && !continuesBody(messages[i - 1], m)) { + 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 notices the primary would take:`, err); - return []; + return null; } } diff --git a/test/deferred-window-only.test.ts b/test/deferred-window-only.test.ts index 415d56fed..2ddc39df2 100644 --- a/test/deferred-window-only.test.ts +++ b/test/deferred-window-only.test.ts @@ -844,17 +844,23 @@ describe('mid-turn deliveries keep the window what the stream carried', () => { 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(); + // A run, whose window is its own: the primary's notice of its failed + // reply goes to another window, which the notice is looked up in by its + // id before it's written (queuedNoticeForPrimary keeps the mark). + 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' }; - // 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' }); + // Its earlier write landed. The release passes the entry on under its + // own id, and the copy in the run's window, which it reads as unread, + // isn't passed on as well. + cm.addMessage('worker', text('hello'), { ...failed, deferredWriteId: 'w-fail' } as never); + internals.deferredMessages.push({ ...entry('w-fail', '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'); + const notices = framework.getAgent('assistant')!.getContextManager().queryMessages({} as never).messages + .filter((m) => (m.metadata as { speaker?: unknown } | undefined)?.speaker === 'worker'); assert.equal(notices.length, 1, 'the primary is told once'); await framework.stop(); }); @@ -862,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 569cc13a4..96fbd60fb 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -1585,7 +1585,7 @@ describe('notices about an agent\'s own action', () => { }); it('counts a body the store shards as one notice, and the same text stored again as a notice of its own', async () => { - const state = internals(framework) as unknown as { unreadNoticesForPrimary(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(); // Shard every notice into one message per block, as a strategy's ingress @@ -1610,17 +1610,17 @@ describe('notices about an agent\'s own action', () => { const stored = cm.getAllMessages().filter((m) => (m.metadata as { kind?: unknown } | undefined)?.kind === 'discord-send-failed'); assert.deepEqual(stored.map((m) => m.shardIndex), [0, 1, 0, 1, 0, 1], 'each body is stored as two shards'); assert.equal(stored[2]!.bodyGroupId, stored[0]!.bodyGroupId, 'the same text is stored again under the same group id'); - assert.deepEqual(state.unreadNoticesForPrimary(agent).map((m) => m.tag), ['first', 'second', 'third']); + assert.deepEqual(state.releasedWindow(agent, new Set())!.unread.map((m) => m.tag), ['first', 'second', 'third']); }); 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 { unreadNoticesForPrimary(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.unreadNoticesForPrimary(agent).map((m) => m.tag); + const tags = () => state.releasedWindow(agent, new Set())!.unread.map((m) => m.tag); const read = notice('read'); agent.markContextConsumed(); @@ -1640,13 +1640,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 { unreadNoticesForPrimary(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.unreadNoticesForPrimary(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; @@ -1664,13 +1664,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 { unreadNoticesForPrimary(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.unreadNoticesForPrimary(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; @@ -1691,13 +1691,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 { unreadNoticesForPrimary(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.unreadNoticesForPrimary(agent).map((m) => m.tag); + const tags = () => state.releasedWindow(agent, new Set())!.unread.map((m) => m.tag); notice('read'); agent.markContextConsumed(); @@ -2196,5 +2196,254 @@ 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 }); }); + + // Each case runs for a failed reply, and for a channel notice the run + // could act on, which the primary takes in its place too (passesToPrimary). + const takenInItsPlace = [ + { + what: 'a failed reply', + metadata: { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 5 }, + text: `${SEND_FAILED} Your previous reply (5 chars) could not be delivered to missing-channel (no registered channel).`, + inRun: 'Your previous reply (5 chars)', + toPrimary: 'A reply by worker', + opened: null, + }, + { + what: 'a channel notice the run could act on', + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-delivery', channels: [{ channelId: 'c-7', label: '#seven' }] } }, + text: '[channels] Now open: #seven (c-7): your own message into the closed channel opened it.', + inRun: 'Now open: #seven', + toPrimary: 'a message from worker into the closed channel opened it; worker has ended', + opened: 'c-7', + }, + ]; + for (const notice of takenInItsPlace) describe(`${notice.what} whose earlier write may have landed reaches the primary once`, () => { + // The notice 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. + 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 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) + queuedForPrimary(framework); + + 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.' }]); + // The channel the notice says opened is still decided open, so the primary is told of it. + if (notice.opened) decide(notice.opened, 'open', 'opened-by-delivery', runFramework); + if (landed) created.contextManager.addMessage('user', [{ type: 'text', text: notice.text }], { ...notice.metadata, 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: notice.text }], metadata: notice.metadata, 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(queuedForPrimary(runFramework), 1, 'stop() leaves one notice queued for the primary'); + assert.equal(told(reopened), 1); + }); + } + + it('not at all across stop() and a restart, when a quiesce postponed its release and the run read the copy', async (t) => { + t.mock.method(console, 'error', () => {}); + const created = await queued(true); + created.agent.markContextConsumed(); + const done = ending(created); + 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"); + + 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); + 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(notice.toPrimary)); + 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(), notice.inRun), 1, 'not stored a second time'); + }); + } + + 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 < 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']); + 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'); + assert.deepEqual(state().unackedDeferredWrites, [], 'and acked there'); + + 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 < 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. + (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(queuedForPrimary(runFramework), 1, 'stop() leaves one notice queued for the primary'); + assert.equal(told(reopened), 1); + }); + }); }); }); diff --git a/test/stop-step-throws.test.ts b/test/stop-step-throws.test.ts index 27d0f8644..c1f657c4a 100644 --- a/test/stop-step-throws.test.ts +++ b/test/stop-step-throws.test.ts @@ -129,14 +129,14 @@ function postponedReleases(framework: AgentFramework, error: Error, when: 'befor if (when === 'before stop()') postpone(); else watchdog.stop = function (this: unknown) { postpone(); return stop.call(this); }; const owner = framework as unknown as Record; - const usual = (owner.unreadNoticesForPrimary as (agent: unknown) => unknown[]).bind(framework); - owner.unreadNoticesForPrimary = (agent: { name: string }) => { + const usual = (owner.releasedWindow as (agent: unknown, held: Set) => unknown).bind(framework); + owner.releasedWindow = (agent: { name: string }, held: Set) => { if (agent.name === 'first') throw error; - 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.unreadNoticesForPrimary; watchdog.stop = stop; }; + return () => { delete owner.releasedWindow; watchdog.stop = stop; }; } function checkPostponedReleases(framework: AgentFramework): void {