Skip to content

fix(injection): a message the render gives null goes in as stored #720

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nissa-seru merged 8 commits into from
Oct 24, 2026
Merged

fix(injection): a message the render gives null goes in as stored #720

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
dc23baf
fix(injection): a message the render gives null goes in as stored
Oct 22, 2026
bf2e740
fix(injection): name a reset head window as the other null, and drop …
Oct 22, 2026
87871d9
fix(injection): say where as-stored still differs, and pin order, pla…
Oct 22, 2026
9bfe7f2
Merge main (0.25.0, #705) into fix/inject-render-null-as-stored
Oct 22, 2026
7d10708
docs(changelog): say the line came in 0.24.1, true on both release lines
Oct 22, 2026
ae627bf
test(clock-rows): the mid-turn message needs no beat, and `during` is…
Oct 23, 2026
fd4b22a
docs(injection): the null branch's comment names the sharded body too
Oct 23, 2026
649b3b9
Merge main (a4a74b48) into fix/inject-render-null-as-stored
Oct 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions changelog.d/fix-inject-render-null-as-stored.fixed.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
177 changes: 50 additions & 127 deletions src/framework.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 };
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand All @@ -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);
Expand All @@ -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
Expand Down Expand Up @@ -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);
}

Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<ContentBlock[] | null> | undefined {
if (ids.length === 0) return undefined;
Expand All @@ -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
Expand Down Expand Up @@ -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;
Expand All @@ -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,
Expand All @@ -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;
Expand Down
Loading
Loading