diff --git a/changelog.d/fix-inject-render-null-as-stored.fixed.md b/changelog.d/fix-inject-render-null-as-stored.fixed.md new file mode 100644 index 000000000..455c1c369 --- /dev/null +++ b/changelog.d/fix-inject-render-null-as-stored.fixed.md @@ -0,0 +1,16 @@ +- A message that arrives during a turn's tool call, and that the strategy + renders as `null`, is injected into the live stream as stored again. The + memory strategy gives `null` for a message before its recent window (a + batch bigger than that window), and for every message while a head + window a topic change reset hasn't filled yet. Since 0.24.1, such a + message was left out of the live round, and a `[not shown]` system line + was stored in its place. But the next compile sends it raw, in order, + after the tool result, so the agent didn't hear it mid-turn, the line was + false, and the next request differed from the live rounds anyway. The + `[not shown]` line, and the log line naming what was left out, are gone. + A clock row the render gives `null` goes in as stored too. As stored + still differs from the next request where its per-message cap or an + image limit reaches such a message, since the render neither cuts it nor + counts its images: a batch with more images than `maxLiveImages`, for + one. It differs too for a single message too big for the recent window, + which the strategy stores in parts. diff --git a/src/framework.ts b/src/framework.ts index c0e6f9ae0..1e62c4120 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -862,13 +862,10 @@ type InboundResponder = { * For an entry whose write threw after storing it, found in the store and * counted as written, it is what landed, joined the same way, which for a * sharded body can be its first shards alone. A store that can't be read - * back (storedCopiesOf logs it) leaves a sharded body as given. `parts` is - * how many store messages hold it, as read back: a sharded body's shards, - * consecutive from `id`, and 1 where it couldn't be read back. */ + * back (storedCopiesOf logs it) leaves a sharded body as given. */ interface DrainedWrite { id: MessageId; content: ContentBlock[]; - parts: number; } /** What later compiles render from the store for the copies of one write @@ -885,22 +882,6 @@ function asCompiled(landed: ReadonlyArray<{ content: ContentBlock[] }>): Content return [...blocks.filter((b) => b.type !== 'text'), { type: 'text', text } as ContentBlock]; } -/** A message's size as a notice gives it: its text's characters and its - * images, each left out when there are none, so empty when it has - * neither. */ -function contentSize(content: readonly ContentBlock[]): string { - let chars = 0; - let images = 0; - for (const b of content) { - if (b.type === 'text') chars += b.text.length; - else if (b.type === 'image' || b.type === 'generated_image') images++; - } - const parts: string[] = []; - if (chars > 0) parts.push(`${chars.toLocaleString('en-US')} character${chars === 1 ? '' : 's'}`); - if (images > 0) parts.push(`${images} image${images === 1 ? '' : 's'}`); - return parts.join(' and '); -} - /** One stored copy of a deferred write, as the drain reads it back: a * sharded body's copies carry its `bodyGroupId`. */ type StoredCopy = { id: MessageId; content: ContentBlock[]; bodyGroupId?: string }; @@ -12352,9 +12333,8 @@ export class AgentFramework { // safe), up to what came after them: a later message can still // take an earlier one past an image limit, as for any message, and // the next compile then strips that image where the live rounds - // carried it. One the next compile won't send among its newest - // isn't injected; a notice after the batch tells the agent of it - // (recordLeftOutNotice). A write that threw after it was stored, + // carried it. One the render gives null goes in as stored (see the + // boundary's null below). A write that threw after it was stored, // whose readback can't say whether it landed, is looked up by its // id here, and carried here, where it landed. One that can be // neither read back nor looked up is the exception, a known limit: @@ -12404,9 +12384,9 @@ export class AgentFramework { // rounds the stream never showed it. const midTurnInjections: Array<{ participant: string; content: ContentBlock[]; metadata?: MessageMetadata }> = []; // What this boundary stored for the agent to hear, in store order, - // injected or left out, for a boundary that leaves it to the turn's - // next start as arrivals: that start's compile sends it. - const heard: Array<{ msg: DeferredWrite; id: MessageId; content: ContentBlock[]; parts: number }> = []; + // for a boundary that leaves it to the turn's next start as + // arrivals: that start's compile sends it. + const heard: Array<{ msg: DeferredWrite; id: MessageId; content: ContentBlock[] }> = []; if (currentState.supportsInjectedMessages !== false) { // A clock row due (storeDueClock) goes right after the round's // results, ahead of what is drained below, and joins the stream @@ -12426,7 +12406,7 @@ export class AgentFramework { // counts as stored, as what landed of it (writeDrainedInOrder). const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) => agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id))); - for (const [msg, { id, content, parts }] of stored) { + for (const [msg, { id, content }] of stored) { // Every other resident reads what this boundary stores; for its // own agent, the boundary weighs what it injects below. this.noteForOtherResidents(agent, msg.metadata, id); @@ -12435,45 +12415,51 @@ export class AgentFramework { // compiled view; injecting it here would put in the live turn // what the next compile hides (#232 gap 5). if ((msg.metadata as { tuneOut?: unknown } | undefined)?.tuneOut) continue; - heard.push({ msg, id, content, parts }); + heard.push({ msg, id, content }); } // Rendered in one call: a render prepares as a compile does, so // the batch shares one preparation. The image limits count from - // the batch's oldest message to the newest, as a compile does. + // the batch's oldest message the render places to the newest, as + // a compile does. // The clock, stored ahead of them, renders first in the same batch // and is injected first, as the next compile sends it. const rendered = this.renderForInjection(agent, [...(clock ? [clock.id] : []), ...heard.map((h) => h.id)]); const first = clock ? 1 : 0; - if (clock) { - const sent = rendered ? rendered[0]! : clock.content; - if (sent !== null) { - midTurnInjections.push({ participant: 'user', content: sent, metadata: clock.metadata }); - } else { - console.error(`[clock] ${agent.name}: clock ${clock.id} left out of the live round, since the next compile won't send it among its newest`); - } - } - const leftOut: typeof heard = []; + // A null is a message the render places before its window. Under + // the memory strategy that's one before its recent window, when + // what this boundary stored is bigger than it, or any message + // while a head window a topic change reset hasn't filled yet. The + // next compile still sends it, raw and in order, after the tool + // result, so leaving it out would change the prefix as surely as + // any form, and the agent wouldn't hear it (context-manager#275's + // review, R1-1). It goes in as stored, which is the form the next + // compile sends with two exceptions. The render doesn't count a + // null message's images or cut it, so a batch with more images + // than the limit, or a long message under a cap that doesn't + // shard, still differs where an image limit or the per-message + // cap reaches it. And a body the strategy stored in parts goes in + // whole, joined as the store reads it back, where the next + // compile sends it in more than one piece. + // That rests on two things. The only view filter this framework + // gives a store is the tune-out stamp (createAgent's; the other + // stores it opens have none), and a stamped message is skipped + // above, so no null here means a message the view hides. And the + // store and the render run with nothing between them, so nothing + // can put a summary in place of what this flush stored, or move + // windowed passthrough's anchor past it, before the render. A + // write found here that landed at an earlier flush + // (writeDrainedInOrder) is the exception: by now a summary may + // stand in for it, or the anchor be past it, and then it goes in + // where the next compile doesn't send it. + const asSent = (i: number, content: ContentBlock[]): ContentBlock[] => rendered?.[i] ?? content; + if (clock) midTurnInjections.push({ participant: 'user', content: asSent(0, clock.content), metadata: clock.metadata }); heard.forEach((h, i) => { - const sent = rendered ? rendered[i + first]! : h.content; - // The next compile won't send it among its newest: injected, it - // would put in the live turn what that compile sends elsewhere, - // or not at all. It's under any consumed mark the boundary sets - // (below), as anything stored here is. - if (sent === null) { leftOut.push(h); return; } midTurnInjections.push({ participant: h.msg.participant, - content: sent, + content: asSent(i + first, h.content), ...(h.msg.metadata ? { metadata: h.msg.metadata } : {}), }); }); - if (leftOut.length > 0) { - console.error(`[injection] ${agent.name}: left out of the live round, since the next compile won't send them among its newest: ${leftOut.map((h) => h.id).join(', ')}`); - // The agent is told what it didn't hear, after the batch, as - // the routing notice is: stored, so the next compile sends it - // where the live round heard it. - const notice = this.recordLeftOutNotice(agent, leftOut); - if (notice) midTurnInjections.push(notice); - } } // Deferred writes handed to the store above: sync, then drop them @@ -12612,8 +12598,7 @@ export class AgentFramework { // A boundary that can't re-pin now (over budget, or no live // stream to tell) leaves what it stored for the agent to the // turn's next start, as arrivals: a budget restart, or a - // re-inference on these tool results. That includes a message - // left out of the injection, since that start compiles anew. + // re-inference on these tool results. for (const { msg, id } of heard) this.noteArrival(agent.name, msg.metadata, id); } @@ -12736,21 +12721,6 @@ export class AgentFramework { // Under context-manager#160 such a write already gives the // replies stored after it stamps their requests didn't carry, // whatever the mark says; deferring those writers is the fix. - // The mark also covers a message the boundary stored and left out - // of the injection, since the next compile won't send it among its - // newest: the notice telling the agent of it is injected, so a - // boundary that leaves one out injects something, unless that - // notice couldn't be stored. The message then counts as read - // though no request carried it: a coalesced render's wake to read - // it drops once a request after the boundary is answered (and, - // for one that carries a channel, the agent has published there - // since; deferredWriteDone), and a release doesn't pass it to the - // primary. The - // notice has told the agent it came, from whom, and how to read - // it. Where the view leaves it out, no request will carry it. - // Where the memory strategy places it before its recent window, a - // batch bigger than that window, the next compile carries it, in - // whatever turn something else starts. if (midTurnInjections.length > 0) { // Possibly shown, whatever the stream declares: the first-view // mark (CoalescerHost.isUnshown) counts a doubt as shown. @@ -23539,9 +23509,8 @@ export class AgentFramework { // messages flush at the next turn's START (before its compile — riding // their own wake), at the agent's next tool boundary (where they are // ALSO injected into the live stream — hear-while-acting — all but a - // message a tune-out holds from the residents, and one the next compile - // won't send among its newest), or in driveStream's finally when the - // turn ends. + // message a tune-out holds from the residents), or in driveStream's + // finally when the turn ends. // Quiesce (issue #122): the window exists to run compression/refold // against a FROZEN context — module events and api message.send must not // append mid-surgery. Deferred here, flushed by resume(). tool_result @@ -23826,10 +23795,11 @@ export class AgentFramework { * its window (ContextManager.renderAsNewest): cut by the per-message cap, * a sharded body joined, and the image limits applied over the batch, so * the live stream hears what the next compile sends. An id comes back as - * null where that compile won't send the message among its newest. - * Undefined where the strategy has no such hook, or the render threw, - * which is logged: the boundary then injects what it stored, as it did - * before strategies could say. + * null where the render can't place the message among the newest, and the + * boundary injects that one as stored (see the boundary's null). Undefined + * where the strategy has no such hook, or the render threw, which is + * logged: the boundary then injects what it stored, as it did before + * strategies could say. */ private renderForInjection(agent: Agent, ids: MessageId[]): Array | undefined { if (ids.length === 0) return undefined; @@ -23841,53 +23811,6 @@ export class AgentFramework { } } - /** - * Tell the agent of the messages a tool boundary stored and left out of - * its injection, since the next compile won't send them among its newest: - * who sent each and its size, and how to read it, `history--extract` - * around it where the agent is shown that tool, else its id. None of a - * message's words: the line stands in for a message the live round - * doesn't carry. A system line, so it's no conversation: it neither - * re-pins the turn nor clears its send suppression. It names the agent - * it's for, since every resident sharing the slot reads it too. Returns - * the notice as stored, for the boundary to inject after the batch, or - * undefined when it couldn't be stored, which is logged. - */ - private recordLeftOutNotice( - agent: Agent, - leftOut: ReadonlyArray<{ msg: { participant: string }; id: MessageId; content: ContentBlock[]; parts: number }>, - ): { participant: string; content: ContentBlock[]; metadata: MessageMetadata } | undefined { - const canExtract = this.advertisedToolsForAgent(agent.name).some((t) => t.name === 'history--extract'); - const one = leftOut.length === 1; - const lines = leftOut.map(({ msg, id, content, parts }) => { - const size = contentSize(content); - // A sharded body is that many store messages, consecutive from its - // first, which extract reads around it one by one. (Its "after" stops - // at 100, so a body of more parts, over 300,000 tokens at the memory - // strategy's default shard size, is read only that far.) - const read = canExtract - ? `history--extract with aroundId "${id}", before 0 and after ${parts - 1} reads it` + - `${parts > 1 ? `, stored as ${parts} parts` : ''}.` - : `Its id is ${id}${parts > 1 ? `, the first of the ${parts} parts it's stored as` : ''}.`; - return `- From ${msg.participant}${size ? `, ${size}` : ''}. ${read}`; - }); - const content: ContentBlock[] = [{ - type: 'text', - text: `[not shown] For ${agent.name}: ${one ? 'A message' : `${leftOut.length} messages`} arrived during your turn that ` + - `${one ? "isn't" : "aren't"} shown here, since your context doesn't keep ${one ? 'it' : 'them'} among ` + - `your newest messages:\n${lines.join('\n')}`, - }]; - const metadata = { system: true, kind: 'left-out-notice' } as MessageMetadata; - try { - const id = agent.getContextManager().addMessage('user', content, metadata); - this.emitTrace({ type: 'message:added', messageId: id, source: 'left-out-notice' }); - return { participant: 'user', content, metadata }; - } catch (err) { - console.error(`[injection] ${agent.name}: couldn't store the notice of messages left out of the live round:`, err); - return undefined; - } - } - /** * Write drained entries in queue (seq) order, whatever their targets: * residents share one message slot, so the order entries arrived in is the @@ -24021,7 +23944,7 @@ export class AgentFramework { continue; } if (found.length > 0) { - written.set(msg, { id: found[0]!.id, content: asCompiled(found), parts: found.length }); + written.set(msg, { id: found[0]!.id, content: asCompiled(found) }); this.logDeferredWriteError(msg, where, undefined, { count: found.length, sharded: found[0]!.bodyGroupId !== undefined, foundLater: true }); onLanded?.(msg, found[0]!.id); continue; @@ -24038,7 +23961,7 @@ export class AgentFramework { const copies = agent && before !== undefined && (this.storedCountOf(agent) ?? before) - before > 1 ? this.storedCopiesOf(agent, before, msg.id) ?? [] : []; - written.set(msg, { id, content: copies.length > 1 ? asCompiled(copies) : msg.content, parts: Math.max(1, copies.length) }); + written.set(msg, { id, content: copies.length > 1 ? asCompiled(copies) : msg.content }); } catch (error) { // What landed, read back from where the store stood before the // write, or, where that can't say, before the write or after it, @@ -24050,7 +23973,7 @@ export class AgentFramework { let landed = !agent ? [] : before === undefined ? undefined : this.storedCopiesOf(agent, before, msg.id); if (landed === undefined && agent) landed = this.landedCopiesById(agent, msg.id); if (landed && landed.length > 0) { - written.set(msg, { id: landed[0]!.id, content: asCompiled(landed), parts: landed.length }); + written.set(msg, { id: landed[0]!.id, content: asCompiled(landed) }); this.logDeferredWriteError(msg, where, error, { count: landed.length, sharded: landed[0]!.bodyGroupId !== undefined }); onLanded?.(msg, landed[0]!.id); continue; diff --git a/test/clock-rows.test.ts b/test/clock-rows.test.ts index 3dfdaf168..b79215486 100644 --- a/test/clock-rows.test.ts +++ b/test/clock-rows.test.ts @@ -523,6 +523,59 @@ describe('clock rows in the framework', () => { } finally { await x.framework.stop(); } }); + it('a mid-turn clock and a message after it each go in as their own render gives them', async () => { + const x = await make(); + x.membrane.supportsInjectedMessages = true; + try { + setClock('2026-10-21T10:41:10Z'); + await wake(x, event('warm up'), [{ type: 'text', text: 'ok' }]); + const cm = x.framework.getAgent('assistant')!.getContextManager() as any; + const asked: number[] = []; + // A stand-in that marks each id's answer with its position in the batch. + cm.renderAsNewest = (ids: readonly string[]) => { asked.push(ids.length); return ids.map((_, i) => [{ type: 'text', text: `RENDERED${i}` }]); }; + setClock('2026-10-21T10:48:10Z'); + x.mod.during = () => { + setClock('2026-10-21T10:52:10Z'); + // Queued for the agent to hear, synchronously, with no wake of its own. + (x.framework as any).handleMcplPushEvent({ ...event('MIDTURN'), triggerInference: false }); + }; + // The stream that runs the tool round, kept before the message's own + // wake starts another and takes lastStream. + let boundary: typeof x.membrane.lastStream = null; + const streamYielding = x.membrane.streamYielding.bind(x.membrane); + x.membrane.streamYielding = (...args: Parameters) => { + const st = streamYielding(...args); + boundary ??= x.membrane.lastStream; + return st; + }; + await wake(x, event('go'), [ping], [{ type: 'text', text: 'done' }]); + assert.deepEqual(asked, [2], 'the premise: the boundary rendered a clock and a message together'); + const injected = (boundary!.receivedToolResultOptions[0]?.injectedMessages ?? []) as Array<{ content: unknown }>; + assert.deepEqual(injected.map((m) => m.content), [[{ type: 'text', text: 'RENDERED0' }], [{ type: 'text', text: 'RENDERED1' }]]); + } finally { await x.framework.stop(); } + }); + + it('a mid-turn clock the render gives null goes in as stored, as the next compile sends it from before its window', async () => { + const x = await make(); + x.membrane.supportsInjectedMessages = true; + try { + setClock('2026-10-21T10:41:10Z'); + await wake(x, event('warm up'), [{ type: 'text', text: 'ok' }]); + const cm = x.framework.getAgent('assistant')!.getContextManager() as any; + let rendered = 0; + cm.renderAsNewest = (ids: readonly string[]) => { rendered++; return ids.map(() => null); }; + setClock('2026-10-21T10:48:10Z'); + x.mod.during = () => setClock('2026-10-21T10:52:10Z'); + await wake(x, event('go'), [ping], [{ type: 'text', text: 'done' }]); + assert.equal(rendered, 1, 'the premise: the boundary rendered the clock'); + const all = rows(x.framework); + const clock = all[all.findIndex((r) => r.content.some((b) => b.type === 'tool_result')) + 1]!; + assert.ok(isClock(clock), 'the premise: a clock follows the tool results'); + assert.deepEqual(x.membrane.lastStream!.receivedToolResultOptions[0]?.injectedMessages, + [{ participant: 'user', content: clock.content }], 'the stream got the stored clock'); + } finally { await x.framework.stop(); } + }); + it('on a stream that can\'t carry injected messages: no mid-turn clock, and the next event takes it', async () => { const x = await make(); x.membrane.supportsInjectedMessages = false; diff --git a/test/mid-turn-injection-as-compiled.test.ts b/test/mid-turn-injection-as-compiled.test.ts index 81f757809..23a4cda98 100644 --- a/test/mid-turn-injection-as-compiled.test.ts +++ b/test/mid-turn-injection-as-compiled.test.ts @@ -2,9 +2,9 @@ * A message injected at a tool boundary goes into the live stream as the next * compile sends it: the agent's strategy renders the boundary's batch as the * newest of its window (ContextManager.renderAsNewest), cut by the per-message - * cap and with the image limits applied across the batch. One the next compile - * won't send among its newest isn't injected, which is logged, and a notice - * after the batch tells the agent of it. + * cap and with the image limits applied across the batch. One the render + * can't place among the newest goes in as stored: the next compile sends it + * from before its recent window. */ import { describe, it, beforeEach, afterEach } from 'node:test'; @@ -23,7 +23,6 @@ import type { ToolResult, } from '../src/index.js'; import { AgentFramework } from '../src/index.js'; -import { HistoryModule } from '../src/modules/history/index.js'; import { AutobiographicalStrategy, WindowedPassthroughStrategy, type ContextInjection, type MessageId } from '@animalabs/context-manager'; import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; import type { ContentBlock } from '@animalabs/membrane'; @@ -164,123 +163,114 @@ describe('a message injected at a tool boundary', () => { return seen; } - it("isn't injected where the next compile won't send it among its newest, which is logged, and a notice after the batch tells the agent of it", async () => { + it('goes in as stored where the render gives null, in its place in the batch', async () => { await create(new WindowedPassthroughStrategy({})); membrane.supportsInjectedMessages = true; - // The first of the batch comes back null, as for a message the view leaves out. + // The first of the batch comes back null, as for a message the render places before its window. const seen = renderWith((ids, real) => real(ids)!.map((c, i) => (i === 0 ? null : c))); const agent = framework.getAgent('assistant')!; - const logged: string[] = []; - const original = console.error; - console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; - let injected: Array<{ content: unknown[] }>; - try { - injected = await turnWith([[text('LEFTOUT')], [text('HEARD')]]); - } finally { - console.error = original; - } + const injected = await turnWith([[text('PLACEDBEFORE')], [text('HEARD')]]); assert.equal(seen.calls, 1, 'the batch is rendered in one call'); - const row = agent.getContextManager().getAllMessages().find((m) => JSON.stringify(m.content).includes('LEFTOUT'))!; - assert.ok(row, 'the premise: the boundary stored it'); - // Who sent it, its size, and its id where the agent has no history--extract; none of its words. - const notice = text( - "[not shown] For assistant: A message arrived during your turn that isn't shown here, since your context doesn't keep it among your newest messages:\n" + - `- From Antra, 7 characters. Its id is ${row.id}.`, - ); - assert.deepEqual(injected.map((m) => m.content), [[text('HEARD')], [notice]]); - assert.ok(logged.some((l) => l.includes('[injection]') && l.includes('left out') && l.includes(row.id)), `the log names it: ${JSON.stringify(logged)}`); - // As anything stored at a boundary that marks: the text says what that costs. - assert.ok(agent.getConsumedWatermark()!.sequence >= row.sequence, 'the consumed mark covers it'); - // Stored after the batch, so the next request carries it where the live round heard it. - const stored = agent.getContextManager().getAllMessages(); - const heardAt = stored.findIndex((m) => JSON.stringify(m.content).includes('HEARD')); - assert.equal(stored[heardAt + 1]?.metadata?.system, true, 'the notice is stored right after the batch, as a system line'); - assert.deepEqual(JSON.parse((await nextRequest())('[not shown]')), [notice]); + assert.deepEqual(injected.map((m) => m.content), [[text('PLACEDBEFORE')], [text('HEARD')]]); + const row = agent.getContextManager().getAllMessages().find((m) => JSON.stringify(m.content).includes('PLACEDBEFORE'))!; + assert.ok(agent.getConsumedWatermark()!.sequence >= row.sequence, 'the consumed mark covers it, as anything injected'); }); - it("names a left-out message's sender in its notice as the store keeps the name, a lone surrogate as U+FFFD", async () => { + it("each message goes in as its own render gives it, in its place in the batch", async () => { await create(new WindowedPassthroughStrategy({})); membrane.supportsInjectedMessages = true; - renderWith((ids, real) => real(ids)!.map((c, i) => (i === 0 ? null : c))); - module.participant = 'An \ud83d tra'; - const injected = await turnWith([[text('LEFTOUT')], [text('HEARD')]]); - const notice = JSON.parse(holding(injected, '[not shown]') || '[]') as Array<{ text?: string }>; - assert.ok(notice[0]?.text, 'the premise: the boundary injected its notice'); - assert.ok(notice[0]!.text!.isWellFormed() && notice[0]!.text!.includes('From An \ufffd tra'), `the notice names the sender well formed: ${JSON.stringify(notice[0]!.text)}`); - assert.deepEqual(JSON.parse((await nextRequest())('[not shown]')), notice, 'as the next request sends it'); + // A stand-in render that marks each id's answer with its position, so a + // message given another's render shows. + renderWith((ids) => ids.map((_, i) => [text(`RENDERED${i}`)])); + const injected = await turnWith([[text('FIRST')], [text('SECOND')]]); + assert.deepEqual(injected.map((m) => m.content), [[text('RENDERED0')], [text('RENDERED1')]]); }); - it('tells the agent of a boundary whose whole batch is left out, with each one\'s images, and names no message a tune-out holds', async () => { + it('a whole batch the render gives null goes in as stored, images and all, and nothing a tune-out holds', async () => { await create(new WindowedPassthroughStrategy({})); membrane.supportsInjectedMessages = true; - renderWith((ids) => ids.map(() => null)); + const seen = renderWith((ids) => ids.map(() => null)); const agent = framework.getAgent('assistant')!; module.interjectionMetadata = [{ tuneOut: { epochId: 'epoch-1' } }]; - const original = console.error; - console.error = () => {}; - let injected: Array<{ content: unknown[] }>; - try { - injected = await turnWith([[text('TUNEDOUT')], [text('PICTURED'), image()]]); - } finally { - console.error = original; - } - const rows = agent.getContextManager().getAllMessages(); - const tuned = rows.find((m) => JSON.stringify(m.content).includes('TUNEDOUT'))!; - const pictured = rows.find((m) => JSON.stringify(m.content).includes('PICTURED'))!; + const injected = await turnWith([[text('TUNEDOUT')], [text('PICTURED'), image()]]); + const tuned = agent.getContextManager().getAllMessages().find((m) => JSON.stringify(m.content).includes('TUNEDOUT'))!; assert.ok(tuned && (tuned.metadata as { tuneOut?: unknown } | undefined)?.tuneOut, 'the premise: the boundary stored the tuned-out message, stamped'); - assert.deepEqual(injected.map((m) => m.content), [[text( - "[not shown] For assistant: A message arrived during your turn that isn't shown here, since your context doesn't keep it among your newest messages:\n" + - `- From Antra, 8 characters and 1 image. Its id is ${pictured.id}.`, - )]]); - // The notice is what the boundary injects, so its mark covers what it left out. - assert.ok(agent.getConsumedWatermark()!.sequence >= pictured.sequence, 'the consumed mark covers it'); + assert.equal(seen.calls, 1, 'the premise: the render ran'); + assert.deepEqual(injected.map((m) => m.content), [[text('PICTURED'), image()]]); + }); + + it('under the memory strategy, a batch bigger than the recent window goes in whole, as the next request sends it (context-manager#275, R1-1)', async () => { + // No stand-in: the strategy's own render decides, and gives null for the + // screenshots before its recent window. The next compile sends those raw, + // in order, as its newest. + await create(new AutobiographicalStrategy({ + adaptiveResolution: true, foldingStrategy: 'kv-stable', recentWindowTokens: 4000, kvStableReachTokens: 8_000, + compressionModel: 'mock', targetChunkTokens: 100, autoTickOnNewMessage: false, + } as never)); + const cm = framework.getAgent('assistant')!.getContextManager(); + for (let i = 0; i < 3; i++) { + cm.addMessage(i % 2 ? 'assistant' : 'user', [text(`hist${i} ` + Array.from({ length: 40 }, (_, k) => `word${k}`).join(' '))]); + } + const nulls: number[] = []; + renderWith((ids, real) => { const r = real(ids)!; nulls.push(r.filter((c) => c === null).length); return r; }); + const shots = [1, 2, 3, 4].map((i) => `SCREENSHOT${i}`); + const injected = await turnWith(shots.map((s) => [text(s), image()])); + assert.ok(nulls[0]! >= 1, `the premise: the render placed some of the batch before its window (${nulls[0]} nulls)`); + // Every screenshot went into the live round, in the order it arrived. + const live = injected.map((m) => JSON.stringify(m.content)); + assert.deepEqual(live.map((c) => shots.findIndex((s) => c.includes(s))), [0, 1, 2, 3], JSON.stringify(live.map((c) => c.slice(0, 60)))); + // The next request carries them where the live round did, right after the + // tool result, in the same form. + membrane.pushResponse(createMockResponse([text('Again.')])); + go(); + await framework.runUntilIdle(); + const sentNext = membrane.calls[membrane.calls.length - 1]!.messages.map((m) => JSON.stringify(m.content)); + const at = shots.map((s) => sentNext.findIndex((c) => c.includes(s))); + assert.deepEqual(at, at.map((_, k) => at[0]! + k), `in order, one after another: ${JSON.stringify(at)}`); + assert.ok(sentNext[at[0]! - 1]!.includes('"tool_result"'), `right after the tool result: ${sentNext[at[0]! - 1]!.slice(0, 120)}`); + assert.deepEqual(at.map((k) => JSON.parse(sentNext[k]!)), live.map((c) => JSON.parse(c))); + }); + + it('under the memory strategy, after a head window reset, a message the head has not filled up to goes in as the next request sends it', async () => { + // A reset head window (connectome-host's /newtopic) that hasn't filled yet + // ends at the newest message, so the render gives every id null, whatever + // the batch's size. The next compile sends the head's messages raw. + await create(new AutobiographicalStrategy({ + adaptiveResolution: true, foldingStrategy: 'kv-stable', headWindowTokens: 4000, compressionModel: 'mock', + autoTickOnNewMessage: false, + } as never)); + const cm = framework.getAgent('assistant')!.getContextManager(); + for (let i = 0; i < 3; i++) cm.addMessage(i % 2 ? 'assistant' : 'user', [text(`hist${i} before the new topic`)]); + await cm.resetHeadWindow('A new topic.'); + const nulls: number[] = []; + renderWith((ids, real) => { const r = real(ids)!; nulls.push(r.filter((c) => c === null).length); return r; }); + const injected = await turnWith([[text('AFTERRESET a short message')]]); + assert.equal(nulls[0], 1, 'the premise: the render gave the message null'); + const live = holding(injected, 'AFTERRESET'); + assert.deepEqual(JSON.parse(live), [text('AFTERRESET a short message')], 'it goes in as stored'); + assert.deepEqual(JSON.parse((await nextRequest())('AFTERRESET')), JSON.parse(live), 'as the next request sends it'); }); - it('under the memory strategy, a batch bigger than the recent window: the message before it is left out, and the rest goes in as the next request carries it', async () => { - // No stand-in: the strategy's own render decides. A recent window smaller than the batch puts the - // batch's first message before it, and a small shard size stores that message as several, as a - // resident's long message is. - const history = new HistoryModule(); + it('under the memory strategy, a body bigger than the recent window goes in as stored, and the rest as the next request sends it', async () => { + // The strategy stores the long message as several shards and the render + // gives it null. As stored, it goes in joined, as the store reads it back: + // its image first, then its text as one block, which differs from the + // blocks as they were queued. await create(new AutobiographicalStrategy({ adaptiveResolution: true, foldingStrategy: 'kv-stable', recentWindowTokens: 300, kvStableReachTokens: 8_000, targetChunkTokens: 200, - }), [history]); - history.bind(framework.getAgent('assistant')!.getContextManager()); - // About 1,000 tokens, more than the recent window holds, then a short message that fits in it. + })); const long = 'FIRSTLONG ' + Array.from({ length: 600 }, (_, i) => `word${i}`).join(' '); - const logged: string[] = []; - const original = console.error; - console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; - let injected: Array<{ content: unknown[] }>; - try { - injected = await turnWith([[text(long)], [text('SECONDSHORT and a few words more')]]); - } finally { - console.error = original; - } - assert.equal(holding(injected, 'FIRSTLONG'), '', 'the message before the recent window is left out'); - const second = holding(injected, 'SECONDSHORT'); - assert.ok(second.length > 0, 'the rest of the batch goes in'); - assert.ok(logged.some((l) => l.includes('[injection]') && l.includes('left out')), `the log names it: ${JSON.stringify(logged.filter((l) => l.includes('[injection]')))}`); + const nulls: number[] = []; + renderWith((ids, real) => { const r = real(ids)!; nulls.push(r.filter((c) => c === null).length); return r; }); + const injected = await turnWith([[text('Intro. '), image(), text(long)], [text('SECONDSHORT and a few words more')]]); const rows = framework.getAgent('assistant')!.getContextManager().getAllMessages(); const row = rows.find((m) => JSON.stringify(m.content).includes('FIRSTLONG'))!; - const parts = rows.filter((m) => m.bodyGroupId !== undefined && m.bodyGroupId === row.bodyGroupId).length; - assert.ok(parts > 1, `the premise: the long message is stored as several parts (${parts})`); - // The agent is shown history--extract, so the notice gives the call that reads just that message, every part. - const call = `history--extract with aroundId "${row.id}", before 0 and after ${parts - 1} reads it, stored as ${parts} parts.`; - const notice = holding(injected, '[not shown]'); - const said = (JSON.parse(notice) as Array<{ text: string }>)[0]!.text; - assert.ok(said.endsWith(`\n- From Antra, ${long.length.toLocaleString('en-US')} characters. ${call}`), said); - assert.ok(!said.includes('FIRSTLONG') && !said.includes('word0'), 'none of its words'); - const next = await nextRequest(); - assert.deepEqual(JSON.parse(next('SECONDSHORT')), JSON.parse(second)); - assert.deepEqual(JSON.parse(next('[not shown]')), JSON.parse(notice)); - // The call the notice gives reads the message. - membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 'x1', name: 'history--extract', input: { aroundId: row.id, before: 0, after: parts - 1 } }] as ContentBlock[], 'tool_use')); - membrane.pushResponse(createMockResponse([text('Read it.')])); - go(); - await framework.runUntilIdle(); - const result = JSON.stringify(membrane.lastStream!.receivedToolResults); - assert.ok(result.includes('FIRSTLONG') && result.includes('word599') && !result.includes('SECONDSHORT'), `that message, whole, alone: ${result.slice(0, 300)}`); + assert.ok(rows.filter((m) => m.bodyGroupId !== undefined && m.bodyGroupId === row.bodyGroupId).length > 1, 'the premise: it is stored in parts'); + assert.equal(nulls[0], 1, 'the premise: the render gave it null'); + assert.deepEqual(JSON.parse(holding(injected, 'FIRSTLONG')), [image(), text(`Intro. ${long}`)], 'it goes in whole, as the store reads it back'); + const second = holding(injected, 'SECONDSHORT'); + assert.deepEqual(JSON.parse((await nextRequest())('SECONDSHORT')), JSON.parse(second)); }); it('goes in as stored where the strategy has no render', async () => { diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index 4d7fe8924..279af658f 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -1363,16 +1363,15 @@ describe('present while acting', () => { await framework.stop(); }); - it("an addressed message a boundary leaves out of the injection doesn't re-pin the turn: the live stream heard only the notice of it", async () => { - // The control is the test above that re-pins: the same DM, injected, - // moves the trailing prose to the DM. The notice is a system line, which - // moves no locus. + it("an addressed message the render gives null goes in as stored, and re-pins the turn", async () => { + // As the test above, with the render placing the DM before its window: + // the live stream hears it as stored, so the trailing prose follows it. membrane.pushResponse(createMockResponse([ { type: 'text', text: 'Station four, proceeding.' }, { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, ] as ContentBlock[], 'tool_use')); membrane.pushResponse(createMockResponse([ - { type: 'text', text: 'Carrying on where I was.' }, + { type: 'text', text: 'Answering the person who addressed me.' }, ] as ContentBlock[])); const framework = await createFramework(); @@ -1387,22 +1386,21 @@ describe('present while acting', () => { trigger(framework); await framework.runUntilIdle(); - assert.equal(rendered, 1, 'the premise: the boundary rendered the DM, and left it out'); + assert.equal(rendered, 1, 'the premise: the boundary rendered the DM'); const injected = membrane.lastStream!.receivedToolResultOptions[0]?.injectedMessages ?? []; - assert.equal(injected.length, 1, 'only the notice of what was left out is injected'); - assert.ok(JSON.stringify(injected[0]!.content).startsWith('[{"type":"text","text":"[not shown]'), JSON.stringify(injected[0])); + assert.deepEqual(injected[0], { participant: 'Antra', content: [{ type: 'text', text: 'hey, quick question over here?' }] }, 'the DM, as stored'); assert.deepEqual(routed, [ { text: 'Station four, proceeding.', locus: 'chan-live-1' }, - { text: 'Carrying on where I was.', locus: 'chan-live-1' }, + { text: 'Answering the person who addressed me.', locus: 'discord:dm:antra' }, ]); await framework.stop(); }); - it("a budget restart takes as an arrival an addressed message its boundary stored and left out of the injection", async () => { - // The boundary renders the DM as null, as for one the next compile won't - // send among its newest, so the live stream never hears it. The restart - // compiles anew, and the DM is still an arrival of the turn it continues. + it("a budget restart takes as an arrival an addressed message its boundary stored, though the render gave it null", async () => { + // The boundary renders the DM as null, and the stream is over its budget, + // so no live stream hears it. The restart compiles anew, and the DM is + // still an arrival of the turn it continues. const overBudget = createMockResponse([ { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, ] as ContentBlock[], 'tool_use'); @@ -1423,7 +1421,7 @@ describe('present while acting', () => { await framework.runUntilIdle(); assert.equal(membrane.calls.length, 2, 'control: the turn restarted for its budget'); - assert.equal(rendered, 1, 'the premise: the boundary rendered the DM, and left it out'); + assert.equal(rendered, 1, 'the premise: the boundary rendered the DM'); assert.deepEqual(routed, [{ text: 'After the restart.', locus: 'discord:dm:antra' }], 'the restart re-pinned to the DM'); await framework.stop();