From ee43daf65614ac9f03d070a97e532ebe9a59c90d Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Sat, 10 Oct 2026 18:11:23 -0400 Subject: [PATCH 01/11] fix(requests): request-only turns stay where the reply's request had them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Context-manager #155 (thinking binding): a provider that binds signed thinking to the prefix it was minted under refuses that thinking once a later request changes anything before it. Two agent-framework request-only constructs changed the prefix between the request that minted a reply and every later request. 1. Silent heartbeat: the tick's instruction rode an ephemeral system-position injection (InferenceRequest.ephemeralSystemPrompt -> ContextInjection 'framework:silent-heartbeat'), so every heartbeat request's system differed from every other request's. That request missed the prompt cache from the system on, every retained thinking block failed its binding there, and the tick reply's own thinking failed at the next ordinary turn. Now the instruction is the separator's text (SILENT_HEARTBEAT_SEPARATOR), which every compile already renders byte-identically before the tick's first surviving row and which the tick's own request ends on. Every past tick reads it too, telling the agent those turns were private self-checks. Steady-state cost: 199 characters where the separator was 16, at each tick whose rows survive in the window (roughly 50 tokens a tick). One-time cost: the first compile after upgrading renders past ticks with the new text, one prompt-cache rewrite (and no strip if this ships before agent-framework engages context-manager #160's pass). ephemeralSystemPrompt is removed from InferenceRequest (types/agent.ts), the push-event wake, the batched wake, the budget restart and the system injection: the heartbeat was its only producer; connectome-host's main neither sets nor reads it (git grep), nor does anything else in the anima checkouts here. It is an exported interface, so the removal has its own breaking fragment naming who would need to act; the behaviour is in the changed fragment. 2. [Continue]: a compile that ends on the agent's own message gets a trailing request-only [Continue] turn, and later compiles didn't render it, so the reply it prompted sat straight after the previous assistant message: the formatter merged them (signed thinking from two responses in one message, the "cannot be modified" 400 when it is the latest), and the reply's binding failed from the next request. Context-manager's chain never sees request-only turns (it chains its own compiled messages), so it would send that thinking. Now the reply's first stored row is stamped metadata.promptedBy: 'continue' and request builds render [Continue] immediately before it, found the way tick rows are found (tool_use ids, thinking signatures; text never). Same known limit as ticks: a reply with neither isn't found, and carries no thinking to fail. Mechanics: - silent-heartbeat.ts is now the request-only-rows module. indexRequestOnlyRows returns ticks (identity key -> eventId, this agent's tick rows) and continued (the identity keys of stamped rows), independently, so a row can be both. renderRequestOnlyRows puts a tick's separator before the tick's first surviving row and [Continue] immediately before each stamped row, after any separator, as the reply's request had them. A stamped row is one stored row of the agent's own reply, which a compile never splits, so [Continue] needs no once-per-group check. (Basil-1007 asked about a row carrying both stamps: no path found today opens a stream inside a tick on [Continue] — every same-turn restart or retry compiles past stored tool results or an unchanged opening — but the first version would have rendered only the separator for such a row.) Internal names only; the public exports (SILENT_HEARTBEAT_SEPARATOR, silentHeartbeatOf, the tick types) keep their names. - Agent.buildActivationRequest's body is now assembleActivationRequest (private), which also reports endsOnContinue; buildActivationRequest wraps it for previews. startStreamWithInjections sets continuePromptPending from it; addAssistantResponse stamps the first row it stores while that's set. Every stream start resets it (a budget restart, guard retry or error retry compiles afresh); the other addAssistantResponse callers store the stream's own rows. The direct Agent API stores through contextManager.addMessage and never adds [Continue]. Tests: silent-heartbeat-rows (+4: the tick's system is the agent's own; the reply to [Continue] is re-rendered byte-stable; only its first row is marked; a row with both stamps gets both turns, separator first, also when it is the tick's first surviving row) and silent-heartbeat (the instruction is the request's last turn and the system is unchanged). The separator literal in the rows test is now the exported constant. Red at 02adf27 with only these test files: 4 of the integration tests fail (the both-stamps unit test needs the new exports). Reviewed by Basil-1007. --- .../request-only-rows-stay.breaking.md | 8 + changelog.d/request-only-rows-stay.changed.md | 23 +++ src/agent.ts | 89 +++++++---- src/framework.ts | 19 +-- src/silent-heartbeat.ts | 150 ++++++++++++------ src/types/agent.ts | 6 +- test/silent-heartbeat-rows.test.ts | 132 +++++++++++++-- test/silent-heartbeat.test.ts | 6 +- 8 files changed, 318 insertions(+), 115 deletions(-) create mode 100644 changelog.d/request-only-rows-stay.breaking.md create mode 100644 changelog.d/request-only-rows-stay.changed.md diff --git a/changelog.d/request-only-rows-stay.breaking.md b/changelog.d/request-only-rows-stay.breaking.md new file mode 100644 index 000000000..4fd6cbd18 --- /dev/null +++ b/changelog.d/request-only-rows-stay.breaking.md @@ -0,0 +1,8 @@ +- **Code that sets or reads `InferenceRequest.ephemeralSystemPrompt`:** the + field is removed. It carried a system-position prompt for one turn; the + silent heartbeat was its only producer, and its instruction now rides the + tick's request-only separator. There is no replacement: a system prompt + that changes for one turn costs that request its prompt cache from the + system on, and changes what every retained signed-thinking block is bound + to. Nothing in this package or in connectome-host sets or reads it. + Unchanged: `suppressProse` and `silentHeartbeat` on `InferenceRequest`. diff --git a/changelog.d/request-only-rows-stay.changed.md b/changelog.d/request-only-rows-stay.changed.md new file mode 100644 index 000000000..067641c96 --- /dev/null +++ b/changelog.d/request-only-rows-stay.changed.md @@ -0,0 +1,23 @@ +- A silent heartbeat tick's instruction now rides the tick's request-only + separator instead of a system-position injection, so a tick's request + carries the agent's own system prompt, like every other turn. A system + prompt that changed for one turn cost the heartbeat's request its prompt + cache from the system on, and on a provider that binds signed thinking to + the prefix it was minted under (context-manager #155) it failed every + retained thinking block in that request, and the tick reply's own thinking + at the next ordinary turn. `SILENT_HEARTBEAT_SEPARATOR` is now that + instruction, so every past tick reads it too: 199 characters where it was + 16, at each tick whose rows survive in the window (roughly 50 tokens a + tick). The first compile after upgrading renders past ticks with the new + text, one prompt-cache rewrite. (`InferenceRequest.ephemeralSystemPrompt` + is removed; see the breaking entry.) +- When a compile ends on the agent's own message and the request gets a + trailing `[Continue]` turn, the reply it prompts now has its first stored + row marked `metadata.promptedBy: 'continue'`, and every later request + renders `[Continue]` again just before that row, the way a silent tick's + separator is rendered before its first row. Without it, later requests + merged the reply into the previous assistant message (one message carrying + signed thinking from two responses, a 400 when it is the latest), and + dropped a turn the reply's request held, which a provider that binds + thinking refuses. A reply with neither a tool call nor signed thinking + can't be found in a compiled window and merges as before. diff --git a/src/agent.ts b/src/agent.ts index 412707a33..7a01884b9 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -4,11 +4,13 @@ import { createHash } from 'node:crypto'; import type { CacheWireReceipt, KvUnifiedRequestHooks } from './kv-unified-wire.js'; import { ToolResultGuard, TOOL_RESULT_GUARD_NOTICE } from './tool-result-guard.js'; import { - indexSilentTickRows, - separateSilentHeartbeatTicks, + PROMPTED_BY_CONTINUE, + continueTurn, + indexRequestOnlyRows, + renderRequestOnlyRows, silentHeartbeatSeparatorTurn, + type RequestOnlyRowIndex, type SilentHeartbeatTick, - type SilentTickIndex, type StoredRowLike, } from './silent-heartbeat.js'; import { @@ -794,13 +796,13 @@ export class Agent { * `ContextManager.compile` is itself side-effect-free (compression runs in * the background). Safe to call regardless of the agent's current status. */ - async buildActivationRequest( + private async assembleActivationRequest( availableTools: ToolDefinition[], injections?: ContextInjection[], budget?: TokenBudget, compressionTools: ToolDefinition[] = availableTools, options: ActivationRequestOptions = {}, - ): Promise { + ): Promise<{ request: NormalizedRequest; endsOnContinue: boolean }> { // Compression may revisit hidden tools in recorded history. Keep its full // definition set separate from the resident's advertised tools: the // autobiographical strategy must declare the same tools on its @@ -854,31 +856,32 @@ export class Agent { })) .filter((m) => m.content.length > 0); - // Silent heartbeat ticks store no prompt row. Render a request-only - // separator before each tick's first surviving row so the wire formatter - // cannot merge the tick into the previous assistant message (see - // silent-heartbeat.ts for what can and cannot be matched). This agent's - // own first request for a tick ends on that same turn, so the prefix it - // caches is the one every later compile renders. "First" is per agent: a - // broadcast tick shares its eventId with every resident. - const ticks = this.indexSilentHeartbeatTicks(); - messages = separateSilentHeartbeatTicks(messages, ticks.rows, this.name); + // Silent heartbeat ticks and replies to a [Continue] turn store no + // prompt row. Render each request-only turn before its response's first + // surviving row, so the wire formatter cannot merge the response into the + // previous assistant message and every later request keeps the prefix + // the response was minted under (see silent-heartbeat.ts for what can + // and cannot be matched). The request that minted the response ended on + // that same turn, so the prefix it caches is the one every later compile + // renders. "First" is per agent: a broadcast tick shares its eventId with + // every resident. + const requestOnly = this.indexRequestOnlyRows(); + messages = renderRequestOnlyRows(messages, requestOnly, this.name); const openingTick = options.silentHeartbeat !== undefined - && !ticks.started.has(options.silentHeartbeat.eventId); + && !requestOnly.started.has(options.silentHeartbeat.eventId); + let endsOnContinue = false; if (openingTick) { messages = [...messages, silentHeartbeatSeparatorTurn()]; } else if (messages.length > 0 && messages[messages.length - 1]!.participant === this.name) { // Safety: ensure messages don't end with an assistant message. // Some models reject trailing assistant messages ("prefill not supported"), // and after context compression a stale assistant turn can end up last. - messages = [...messages, { - participant: 'user', - content: [{ type: 'text', text: '[Continue]' }], - }]; + messages = [...messages, continueTurn()]; + endsOnContinue = true; } - return { + const request: NormalizedRequest = { messages: this.toolResultGuard.prepareRequest(messages), system: this.buildSystemPrompt(systemInjections), config: { @@ -895,28 +898,47 @@ export class Agent { ...(this.providerParams && { providerParams: this.providerParams }), assistantParticipant: this.name, }; + return { request, endsOnContinue }; + } + + /** The request `assembleActivationRequest` builds (see there), for preview + * and debug tooling: what this agent's next activation would send. */ + async buildActivationRequest( + availableTools: ToolDefinition[], + injections?: ContextInjection[], + budget?: TokenBudget, + compressionTools: ToolDefinition[] = availableTools, + options: ActivationRequestOptions = {}, + ): Promise { + return (await this.assembleActivationRequest(availableTools, injections, budget, compressionTools, options)).request; } - /** Stored silent-tick rows, memoized on the store's cached message array - * (ContextManager returns the same array until the store changes). */ - private silentTickIndex: { source: readonly unknown[]; index: SilentTickIndex } | null = null; + /** Stored rows that follow a request-only turn, memoized on the store's + * cached message array (ContextManager returns the same array until the + * store changes). */ + private requestOnlyIndex: { source: readonly unknown[]; index: RequestOnlyRowIndex } | null = null; - private indexSilentHeartbeatTicks(): SilentTickIndex { - const empty: SilentTickIndex = { rows: new Map(), started: new Set() }; + private indexRequestOnlyRows(): RequestOnlyRowIndex { + const empty: RequestOnlyRowIndex = { ticks: new Map(), continued: new Set(), started: new Set() }; const cm = this.contextManager as Partial; if (typeof cm.getAllMessages !== 'function') return empty; try { const stored = cm.getAllMessages() as unknown as readonly StoredRowLike[]; - if (this.silentTickIndex?.source === stored) return this.silentTickIndex.index; - const index = indexSilentTickRows(stored, this.name); - this.silentTickIndex = { source: stored, index }; + if (this.requestOnlyIndex?.source === stored) return this.requestOnlyIndex.index; + const index = indexRequestOnlyRows(stored, this.name); + this.requestOnlyIndex = { source: stored, index }; return index; } catch (error) { - console.error(`[silent-heartbeat] ${this.name}: could not index tick rows; separators omitted:`, error); + console.error(`[request-only-rows] ${this.name}: could not index stored rows; request-only turns omitted:`, error); return empty; } } + /** The current stream's request ended on a [Continue] turn, and the reply + * it prompted hasn't stored a row yet: its first row is stamped + * `promptedBy: 'continue'`, so later requests render the turn before it. */ + private continuePromptPending = false; + /** * Start a yielding stream with context injections. * Same as startStream but passes injections through to compile. @@ -961,8 +983,11 @@ export class Agent { this.lastStreamRealInputTokens = 0; this.lastStreamOutputTokens = 0; - const request = await this.buildActivationRequest(availableTools, injections, budget, compressionTools, options); + const { request, endsOnContinue } = await this.assembleActivationRequest( + availableTools, injections, budget, compressionTools, options, + ); request.messages = this.toolResultGuard.prepareRequest(request.messages, true); + this.continuePromptPending = endsOnContinue; const receiptAware = (this.contextManager as unknown as { getStrategy?: () => unknown }) .getStrategy?.() as { @@ -1088,6 +1113,10 @@ export class Agent { ); return; } + if (this.continuePromptPending) { + this.continuePromptPending = false; + metadata = { ...metadata, promptedBy: PROMPTED_BY_CONTINUE } as MessageMetadata; + } this.contextManager.addMessage(this.name, content, metadata); } diff --git a/src/framework.ts b/src/framework.ts index 349bdd693..c8b211542 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -6906,7 +6906,6 @@ export class AgentFramework { source: 'framework', timestamp: Date.now(), suppressProse: this.activeTurnTriggers.get(agent.name)?.suppressProse, - ephemeralSystemPrompt: this.activeTurnTriggers.get(agent.name)?.ephemeralSystemPrompt, silentHeartbeat: this.activeTurnTriggers.get(agent.name)?.silentHeartbeat, }); } else if (currentState.stream) { @@ -8159,9 +8158,11 @@ export class AgentFramework { addressed: isAddressedMessage(event.tags, event.origin), ...(silentHeartbeat ? { suppressProse: true, - ephemeralSystemPrompt: - '[silent heartbeat] Scheduled private self-check. Review pending matters privately. ' + - 'Do not narrate or publish plain prose; use an explicit send tool only if you deliberately choose to contact someone.', + // The tick's instruction rides its request-only separator + // (SILENT_HEARTBEAT_SEPARATOR), never the system prompt: a + // per-turn system change would cost the heartbeat's request + // its prompt cache from the system on, and fail the binding of + // every retained thinking block (context-manager #155). silentHeartbeat: { eventId: event.eventId, serverId: event.serverId }, } : {}), ...(event.coalescingSubject ? { coalescingSubject: event.coalescingSubject } : {}), @@ -8498,7 +8499,6 @@ export class AgentFramework { ? [...new Set(requests.flatMap((r) => r.coalescingBatchSubjects ?? [r.coalescingSubject!]))] : undefined, suppressProse: silentOnly ? trigger?.suppressProse : undefined, - ephemeralSystemPrompt: silentOnly ? trigger?.ephemeralSystemPrompt : undefined, silentHeartbeat: silentOnly ? trigger?.silentHeartbeat : undefined, channelId: channelReq?.channelId, addressed: addressedReq !== undefined, @@ -9336,15 +9336,6 @@ export class AgentFramework { } } - if (trigger?.ephemeralSystemPrompt) { - const silentInjection: ContextInjection = { - namespace: 'framework:silent-heartbeat', - position: 'system', - content: [{ type: 'text', text: trigger.ephemeralSystemPrompt }], - }; - injections = injections ? [...injections, silentInjection] : [silentInjection]; - } - // An ephemeral watchdog may dispose this Agent while hooks/compile await. // Do not create a provider stream after its generation lost ownership. if (this.agents.get(agent.name) !== agent) { diff --git a/src/silent-heartbeat.ts b/src/silent-heartbeat.ts index 9d4601375..f9e4bd0d3 100644 --- a/src/silent-heartbeat.ts +++ b/src/silent-heartbeat.ts @@ -1,40 +1,62 @@ /** - * Stored-row handling for silent heartbeat ticks. + * Request-only rows: user turns a request carries before a stored response + * that the store never holds, rendered again by every later compile. * - * A silent tick stores no prompt row (#216): its reply is written straight - * after whatever the window held before it — often the previous assistant - * message, since a suppressed tick leaves no delivery receipt behind. The - * wire formatter merges consecutive same-role messages, so without a turn - * between them the tick would read as an unprompted continuation of the - * earlier response and one assistant message would carry signed thinking - * from two separate responses (a provider 400 when it is the latest - * assistant message; see `Agent.addAssistantResponse`). + * Two kinds exist, and both are anchored to the response they prompted: * - * Every row a tick writes is therefore stamped with `metadata.silentHeartbeat` - * and, at request-build time only, a terse user turn is rendered immediately - * before the first of the tick's rows that survives compilation. The - * separator is never stored and its position is a function of the stamped - * rows alone, so every compile renders it in the same place (the tick's own - * request ends on the same turn), which keeps the prompt-cache prefix - * stable — unlike a per-compile injection that re-anchors to the newest user - * message. + * - A silent heartbeat tick stores no prompt row (#216): its reply is written + * straight after whatever the window held before it — often the previous + * assistant message, since a suppressed tick leaves no delivery receipt + * behind. Every row a tick writes is stamped with `metadata.silentHeartbeat`, + * and a separator carrying the tick's instruction is rendered before the + * first of the tick's rows that survives compilation. + * - A compile that ends on the agent's own message gets a trailing + * `[Continue]` turn (some models reject a trailing assistant message). The + * reply it prompts is stored straight after that message, so the reply's + * first stored row is stamped `metadata.promptedBy: 'continue'`, and the + * `[Continue]` turn is rendered before it. + * + * Without the row, the wire formatter merges consecutive same-role messages: + * the reply would read as an unprompted continuation of the earlier response, + * and one assistant message would carry signed thinking from two separate + * responses (a provider 400 when it is the latest assistant message; see + * `Agent.addAssistantResponse`). And the request that minted the reply held + * the row: a provider that binds signed thinking to the prefix it was minted + * under (context-manager #155) refuses that thinking once a later request + * drops it. Each row's text is constant and its position is a function of the + * stamped rows alone, so every compile renders it in the same place (the + * request that minted the reply ends on that same turn), which keeps the + * prompt-cache prefix and the binding stable — unlike a per-compile injection + * that re-anchors to the newest user message, or a system prompt that changes + * for one turn. * * Known limit: compiled messages carry no store provenance, so a stored row * is recognised only by structural identity the provider made unique - * (tool_use ids, thinking signatures). A tick reply with neither — unsigned - * and tool-free — gets no separator and still merges as before. Without - * signed thinking that merge carries no provider-rejection risk; it is only - * a misreading. Reply text is deliberately never used as identity: identical - * replies ("all quiet") would collide, and an ordinary reply with the same - * text could take a tick's separator and rewrite cached history. Exact - * matching for every row would need the compile result to expose each - * message's source rows. + * (tool_use ids, thinking signatures). A reply with neither — unsigned and + * tool-free — gets no row and still merges as before. Without signed thinking + * that merge carries no provider-rejection risk; it is only a misreading. + * Reply text is deliberately never used as identity: identical replies ("all + * quiet") would collide, and an ordinary reply with the same text could take + * another reply's row and rewrite cached history. Exact matching for every + * row would need the compile result to expose each message's source rows. */ import type { ContentBlock, NormalizedMessage } from '@animalabs/membrane'; -/** Request-only separator text. Constant on purpose: it carries no time or - * id, so it is byte-identical on every compile. */ -export const SILENT_HEARTBEAT_SEPARATOR = '[heartbeat tick]'; +/** Request-only separator before a silent tick's first surviving row. It + * carries the tick's instruction, so the tick's own request reads it as its + * prompt and every later compile shows it at each past tick. Constant on + * purpose: it carries no time or id, so it is byte-identical on every + * compile. */ +export const SILENT_HEARTBEAT_SEPARATOR = + '[silent heartbeat] Scheduled private self-check. Review pending matters privately. ' + + 'Do not narrate or publish plain prose; use an explicit send tool only if you deliberately choose to contact someone.'; + +/** Request-only turn a compile that ends on the agent's own message gets. */ +export const CONTINUE_ROW = '[Continue]'; + +/** `metadata.promptedBy` on the first row a reply to a `[Continue]` turn + * stores. */ +export const PROMPTED_BY_CONTINUE = 'continue'; /** Identity of one authenticated silent heartbeat tick. */ export interface SilentHeartbeatTick { @@ -63,7 +85,20 @@ export function silentHeartbeatOf(metadata: unknown): SilentHeartbeatStamp | und } export function silentHeartbeatSeparatorTurn(): NormalizedMessage { - return { participant: 'user', content: [{ type: 'text', text: SILENT_HEARTBEAT_SEPARATOR }] }; + return requestOnlyTurn(SILENT_HEARTBEAT_SEPARATOR); +} + +export function continueTurn(): NormalizedMessage { + return requestOnlyTurn(CONTINUE_ROW); +} + +function requestOnlyTurn(text: string): NormalizedMessage { + return { participant: 'user', content: [{ type: 'text', text }] }; +} + +/** Whether a stored row is a reply's first row after a `[Continue]` turn. */ +export function isPromptedByContinue(metadata: unknown): boolean { + return (metadata as { promptedBy?: unknown } | undefined)?.promptedBy === PROMPTED_BY_CONTINUE; } /** @@ -92,48 +127,63 @@ export interface StoredRowLike { metadata?: unknown; } -export interface SilentTickIndex { - /** Identity key of every assistant row this agent stored during a tick → tick event id. */ - rows: Map; +export interface RequestOnlyRowIndex { + /** Identity key of every assistant row this agent stored during a silent + * tick → the tick's event id. */ + ticks: Map; + /** Identity keys of the first row of each reply to a `[Continue]` turn. */ + continued: Set; /** Ticks this agent has already stored rows for (its own, not another resident's). */ started: Set; } -/** Index the rows this agent stored during silent ticks. */ -export function indexSilentTickRows(stored: readonly StoredRowLike[], agentName: string): SilentTickIndex { - const rows = new Map(); +/** Index the rows that follow request-only turns: every row this agent stored + * during its silent ticks, and the first row of each reply to a `[Continue]`. + * The two are independent, so a row can be both. */ +export function indexRequestOnlyRows(stored: readonly StoredRowLike[], agentName: string): RequestOnlyRowIndex { + const ticks = new Map(); + const continued = new Set(); const started = new Set(); for (const row of stored) { const stamp = silentHeartbeatOf(row.metadata); - if (!stamp || stamp.agentName !== agentName) continue; - started.add(stamp.eventId); - if (row.participant !== agentName) continue; - for (const key of identityKeys(row.content)) rows.set(key, stamp.eventId); + if (stamp && stamp.agentName === agentName) { + started.add(stamp.eventId); + if (row.participant === agentName) { + for (const key of identityKeys(row.content)) ticks.set(key, stamp.eventId); + } + } + if (isPromptedByContinue(row.metadata)) { + for (const key of identityKeys(row.content)) continued.add(key); + } } - return { rows, started }; + return { ticks, continued, started }; } /** - * Insert the separator turn immediately before the first surviving row of - * each tick in a compiled window — the tick's first row when it is rendered, - * otherwise whichever later row a strategy kept. Rows folded into summaries - * simply do not match; ordinary messages pass through untouched. + * Insert the request-only turns into a compiled window: a tick's separator + * immediately before the first of the tick's rows that survives compilation + * (its first row when rendered, otherwise whichever later row a strategy + * kept), and `[Continue]` immediately before each reply it prompted, after + * any separator, as the reply's request had them. Rows folded into + * summaries simply do not match; ordinary messages pass through untouched. */ -export function separateSilentHeartbeatTicks( +export function renderRequestOnlyRows( messages: NormalizedMessage[], - rows: ReadonlyMap, + index: Pick, agentName: string, ): NormalizedMessage[] { - if (rows.size === 0) return messages; + if (index.ticks.size === 0 && index.continued.size === 0) return messages; const separated = new Set(); const out: NormalizedMessage[] = []; for (const message of messages) { if (message.participant === agentName) { - const eventId = identityKeys(message.content).map((key) => rows.get(key)).find((id) => id !== undefined); - if (eventId !== undefined && !separated.has(eventId)) { - separated.add(eventId); + const keys = identityKeys(message.content); + const tick = keys.map((key) => index.ticks.get(key)).find((id) => id !== undefined); + if (tick !== undefined && !separated.has(tick)) { + separated.add(tick); out.push(silentHeartbeatSeparatorTurn()); } + if (keys.some((key) => index.continued.has(key))) out.push(continueTurn()); } out.push(message); } diff --git a/src/types/agent.ts b/src/types/agent.ts index bbf4971ca..3fe5509e2 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -360,11 +360,9 @@ export interface InferenceRequest { /** Suppress every automatic plain-prose delivery for this logical turn. * Explicit tool calls remain available. Used by authenticated silent wakes. */ suppressProse?: boolean; - /** Ephemeral system-position prompt for this turn only. Never written to - * Chronicle; callers must supply bounded non-secret control text. */ - ephemeralSystemPrompt?: string; /** The authenticated silent heartbeat tick this turn answers. Every row the * turn stores is stamped `metadata.silentHeartbeat` with it, and request - * builds render a request-only separator before the tick's first row. */ + * builds render a request-only separator carrying the tick's instruction + * before the tick's first row (`SILENT_HEARTBEAT_SEPARATOR`). */ silentHeartbeat?: SilentHeartbeatTick; } diff --git a/test/silent-heartbeat-rows.test.ts b/test/silent-heartbeat-rows.test.ts index 618d45257..c9a8fff26 100644 --- a/test/silent-heartbeat-rows.test.ts +++ b/test/silent-heartbeat-rows.test.ts @@ -15,7 +15,8 @@ import assert from 'node:assert/strict'; import { mkdtempSync, rmSync } from 'node:fs'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; -import { AgentFramework } from '../src/index.js'; +import { AgentFramework, SILENT_HEARTBEAT_SEPARATOR } from '../src/index.js'; +import { indexRequestOnlyRows, renderRequestOnlyRows } from '../src/silent-heartbeat.js'; import type { AgentConfig, Module, ModuleContext, ToolCall, ToolDefinition, ToolResult } from '../src/index.js'; import { TOOL_RESULT_GUARD_NOTICE } from '../src/tool-result-guard.js'; import { PassthroughStrategy } from '@animalabs/context-manager'; @@ -189,10 +190,10 @@ describe('silent heartbeat stored rows', () => { } const later = x.membrane.calls[2]!.messages; const tickAt = later.findIndex((m) => textOf(m) === 'tick note'); - assert.equal(textOf(later[tickAt - 1]!), '[heartbeat tick]'); + assert.equal(textOf(later[tickAt - 1]!), SILENT_HEARTBEAT_SEPARATOR); assert.equal(textOf(later[tickAt - 2]!), 'ordinary reply'); // The tick's own request saw the separator in place of '[Continue]'. - assert.equal(textOf(x.membrane.calls[1]!.messages.at(-1)!), '[heartbeat tick]'); + assert.equal(textOf(x.membrane.calls[1]!.messages.at(-1)!), SILENT_HEARTBEAT_SEPARATOR); assert.deepEqual(later.slice(0, x.membrane.calls[1]!.messages.length), x.membrane.calls[1]!.messages); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); @@ -219,7 +220,7 @@ describe('silent heartbeat stored rows', () => { for (const row of all.slice(0, before)) { assert.equal(row.metadata?.silentHeartbeat, undefined, 'ordinary rows are not stamped'); } - assert.ok(!JSON.stringify(all).includes('[heartbeat tick]'), 'the separator is never stored'); + assert.ok(!JSON.stringify(all).includes(SILENT_HEARTBEAT_SEPARATOR), 'the separator is never stored'); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); @@ -234,16 +235,16 @@ describe('silent heartbeat stored rows', () => { const tickRequest = x.membrane.calls[1]!.messages; const later = x.membrane.calls[2]!.messages; // The tick's own request ends on the separator turn, not '[Continue]'. - assert.equal(textOf(tickRequest.at(-1)!), '[heartbeat tick]'); + assert.equal(textOf(tickRequest.at(-1)!), SILENT_HEARTBEAT_SEPARATOR); assert.equal(tickRequest.at(-1)!.participant, 'user'); assert.ok(!tickRequest.some((m) => textOf(m) === '[Continue]')); // Every later compile renders the same turn in the same place, so the // tick request is a byte-exact prefix of the next request. assert.deepEqual(later.slice(0, tickRequest.length), tickRequest); - const separatorAt = later.findIndex((m) => textOf(m) === '[heartbeat tick]'); + const separatorAt = later.findIndex((m) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR); assert.equal(separatorAt, tickRequest.length - 1); assert.equal(textOf(later[separatorAt + 1]!), 'checking'); - assert.equal(later.filter((m) => textOf(m) === '[heartbeat tick]').length, 1, + assert.equal(later.filter((m) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR).length, 1, 'one separator per tick, before its first row only'); const provider = wire(x.membrane.calls[2]!); @@ -279,7 +280,7 @@ describe('silent heartbeat stored rows', () => { const opening = membrane.calls[1]!.messages; const restarted = membrane.calls[2]!.messages; assert.deepEqual(restarted.slice(0, opening.length), opening, 'restart compiles the same prefix'); - assert.equal(restarted.filter((m) => textOf(m) === '[heartbeat tick]').length, 1); + assert.equal(restarted.filter((m) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR).length, 1); assert.ok(!restarted.some((m) => textOf(m) === '[Continue]'), 'the restart window ends on the tool_result'); assert.match(JSON.stringify(membrane.calls[2]), /Scheduled private self-check/, 'still a silent turn'); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } @@ -299,8 +300,8 @@ describe('silent heartbeat stored rows', () => { const later = x.membrane.calls[2]!.messages; assert.ok(!later.some((m) => textOf(m) === 'checking'), 'the strategy dropped the tool round'); const finalAt = later.findIndex((m) => textOf(m) === 'all quiet'); - assert.equal(textOf(later[finalAt - 1]!), '[heartbeat tick]'); - assert.equal(later.filter((m) => textOf(m) === '[heartbeat tick]').length, 1); + assert.equal(textOf(later[finalAt - 1]!), SILENT_HEARTBEAT_SEPARATOR); + assert.equal(later.filter((m) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR).length, 1); assert.deepEqual(later.slice(0, x.membrane.calls[1]!.messages.length), x.membrane.calls[1]!.messages); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); @@ -315,7 +316,7 @@ describe('silent heartbeat stored rows', () => { await wake(x, silentEvent(), signed('sig-b', 'all quiet')); await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); const later = x.membrane.calls[4]!.messages; - const separators = later.flatMap((m, i) => textOf(m) === '[heartbeat tick]' ? [i] : []); + const separators = later.flatMap((m, i) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR ? [i] : []); // Only the two signed ticks are separated, each directly before its // own reply. The unsigned, tool-free tick has no structural identity // and merges as it did before (no signed thinking, so no 400 risk); @@ -340,7 +341,7 @@ describe('silent heartbeat stored rows', () => { await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); const later = x.membrane.calls[2]!.messages; const tickAt = later.findIndex((m) => textOf(m) === 'quiet'); - assert.equal(textOf(later[tickAt - 1]!), '[heartbeat tick]'); + assert.equal(textOf(later[tickAt - 1]!), SILENT_HEARTBEAT_SEPARATOR); for (const message of wire(x.membrane.calls[2]!)) assert.ok(signatures(message).length <= 1); assert.deepEqual(later.slice(0, x.membrane.calls[1]!.messages.length), x.membrane.calls[1]!.messages); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } @@ -362,11 +363,11 @@ describe('silent heartbeat stored rows', () => { const betaCalls = x.membrane.calls.filter((c) => c.assistantParticipant === 'beta'); assert.equal(betaCalls.length, 3); const [, betaTick, betaLater] = betaCalls; - assert.equal(textOf(betaTick!.messages.at(-1)!), '[heartbeat tick]', "beta opens its own tick"); + assert.equal(textOf(betaTick!.messages.at(-1)!), SILENT_HEARTBEAT_SEPARATOR, "beta opens its own tick"); assert.deepEqual(betaLater!.messages.slice(0, betaTick!.messages.length), betaTick!.messages, "beta's tick request stays a byte-exact prefix of its next request"); const betaTickAt = betaLater!.messages.findIndex((m) => textOf(m) === 'beta tick'); - assert.equal(textOf(betaLater!.messages[betaTickAt - 1]!), '[heartbeat tick]'); + assert.equal(textOf(betaLater!.messages[betaTickAt - 1]!), SILENT_HEARTBEAT_SEPARATOR); const owners = stored(x.framework, 'beta') .filter((m) => m.metadata?.silentHeartbeat?.eventId === tick.eventId) .map((m) => [m.participant, m.metadata!.silentHeartbeat.agentName]); @@ -417,7 +418,7 @@ describe('silent heartbeat stored rows', () => { const opening = membrane.calls[1]!.messages; const retry = membrane.calls[2]!.messages; assert.deepEqual(retry.slice(0, opening.length), opening, 'the retry compiles the same prefix'); - assert.equal(retry.filter((m) => textOf(m) === '[heartbeat tick]').length, 1); + assert.equal(retry.filter((m) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR).length, 1); assert.match(JSON.stringify(membrane.calls[2]), /Scheduled private self-check/, 'still a silent turn'); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); @@ -437,10 +438,109 @@ describe('silent heartbeat stored rows', () => { await x.framework.runUntilIdle(); assert.equal(x.membrane.calls.length, 3); for (const call of x.membrane.calls) { - assert.ok(!call.messages.some((m) => textOf(m) === '[heartbeat tick]')); + assert.ok(!call.messages.some((m) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR)); } assert.equal(textOf(x.membrane.calls[2]!.messages.at(-1)!), '[Continue]'); for (const row of stored(x.framework)) assert.equal(row.metadata?.silentHeartbeat, undefined); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); + + it('sends the agent\'s own system prompt on a tick: a heartbeat changes nothing before the messages', async () => { + const x = await make(); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-ordinary', 'ordinary reply')); + await wake(x, silentEvent(), signed('sig-tick', 'tick note')); + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + assert.deepEqual(x.membrane.calls.map((call) => call.system), ['test', 'test', 'test'], + 'the instruction rides the separator, not a one-turn system prompt'); + assert.match(textOf(x.membrane.calls[1]!.messages.at(-1)!), /Scheduled private self-check/); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); +}); + +/** A wake with no stored row (a module or timer wake), answered with `content`. */ +async function timerWake(x: { membrane: MockMembrane; framework: AgentFramework }, ...responses: ContentBlock[][]) { + for (const content of responses) { + x.membrane.pushResponse(createMockResponse(content, content.some((b) => b.type === 'tool_use') ? 'tool_use' : 'end_turn')); + } + (x.framework as any).pendingRequests.push({ + agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now(), + }); + await x.framework.runUntilIdle(); +} + +describe('replies to a [Continue] turn', () => { + it('renders [Continue] again before the reply it prompted, byte-stable from that request on', async () => { + // Undelivered prose leaves no receipt, so the window ends on the reply. + const x = await make({ deliver: false }); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + await timerWake(x, signed('sig-c', 'continued reply')); + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + + const continued = x.membrane.calls[1]!.messages; + const later = x.membrane.calls[2]!.messages; + assert.equal(textOf(continued.at(-1)!), '[Continue]'); + // The request that minted the reply is a byte-exact prefix of the next. + assert.deepEqual(later.slice(0, continued.length), continued); + const replyAt = later.findIndex((m) => textOf(m) === 'continued reply'); + assert.equal(textOf(later[replyAt - 1]!), '[Continue]'); + + const provider = wire(x.membrane.calls[2]!); + for (const message of provider) { + assert.ok(signatures(message).length <= 1, + `one ${message.role} message carries signed thinking from: ${signatures(message).join(', ')}`); + } + + const rows = stored(x.framework); + const reply = rows.find((m) => m.content.some((b) => (b as { text?: string }).text === 'continued reply'))!; + assert.equal(reply.metadata?.promptedBy, 'continue'); + for (const row of rows.filter((m) => m !== reply)) assert.equal(row.metadata?.promptedBy, undefined); + assert.ok(!rows.some((m) => m.content.some((b) => (b as { text?: string }).text === '[Continue]')), + 'the [Continue] turn is never stored'); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('a row that is both a tick row and a reply to [Continue] gets both turns, separator first', () => { + const tick = { eventId: 'hb-x', serverId: 'heartbeat', agentName: 'assistant' }; + const stored = [ + { participant: 'assistant', content: signed('sig-a', 'earlier reply') }, + { participant: 'assistant', content: signed('sig-1', 'tick round'), metadata: { silentHeartbeat: tick } }, + { participant: 'assistant', content: signed('sig-2', 'tick continued'), + metadata: { silentHeartbeat: tick, promptedBy: 'continue' } }, + ]; + const index = indexRequestOnlyRows(stored, 'assistant'); + const rendered = renderRequestOnlyRows(stored.map(({ participant, content }) => ({ participant, content })), index, 'assistant'); + assert.deepEqual(rendered.map(textOf), [ + 'earlier reply', SILENT_HEARTBEAT_SEPARATOR, 'tick round', '[Continue]', 'tick continued', + ]); + // With the tick's first row folded away, the stamped row is the tick's + // first surviving row: both turns precede it, the separator first. + const folded = renderRequestOnlyRows( + [stored[0]!, stored[2]!].map(({ participant, content }) => ({ participant, content })), index, 'assistant'); + assert.deepEqual(folded.map(textOf), ['earlier reply', SILENT_HEARTBEAT_SEPARATOR, '[Continue]', 'tick continued']); + }); + + it('marks only the first row the reply stores, and renders one [Continue] before it', async () => { + const x = await make({ deliver: false }); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + await timerWake(x, + signed('sig-c', 'checking', [{ type: 'tool_use', id: 'c-t1', name: 'tools--ping', input: {} } as ContentBlock]), + signed('sig-c2', 'all done')); + assert.equal(x.mod.calls.length, 1); + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + + const rows = stored(x.framework); + const marked = rows.filter((m) => m.metadata?.promptedBy === 'continue'); + assert.equal(marked.length, 1); + assert.ok(marked[0]!.content.some((b) => b.type === 'tool_use'), 'the first row: the tool round'); + + const later = x.membrane.calls[2]!.messages; + assert.equal(later.filter((m) => textOf(m) === '[Continue]').length, 1); + const roundAt = later.findIndex((m) => textOf(m) === 'checking'); + assert.equal(textOf(later[roundAt - 1]!), '[Continue]'); + assert.deepEqual(later.slice(0, x.membrane.calls[1]!.messages.length), x.membrane.calls[1]!.messages); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); }); diff --git a/test/silent-heartbeat.test.ts b/test/silent-heartbeat.test.ts index 328f60de8..eae5301ef 100644 --- a/test/silent-heartbeat.test.ts +++ b/test/silent-heartbeat.test.ts @@ -55,7 +55,7 @@ async function make() { } describe('silent heartbeat', () => { - it('wakes once with ephemeral control context, stores no push message, and routes no prose', async () => { + it('wakes once with its instruction in a request-only turn, stores no push message, and routes no prose', async () => { const x = await make(); try { x.membrane.pushResponse(createMockResponse([{ type: 'text', text: 'private self-check prose' }] as ContentBlock[])); @@ -72,6 +72,10 @@ describe('silent heartbeat', () => { assert.match(wire, /silent heartbeat/); assert.match(wire, /Scheduled private self-check/); assert.doesNotMatch(wire, /private self-check prose/); + // The instruction is the request's last turn; the system prompt is the + // agent's own, as on every other turn. + assert.equal(x.membrane.calls[0]!.system, 'test'); + assert.match(JSON.stringify(x.membrane.calls[0]!.messages.at(-1)), /Scheduled private self-check/); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); From 6f08783c663eb60495e0a7f893489ecaca807d5d Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Sat, 10 Oct 2026 19:25:14 -0400 Subject: [PATCH 02/11] fix(requests): request-only turns render the text they were sent with, for every row their reply stores Greptile's three P1 findings on #284 (discussion_r4239514606, _611, _615), all valid: 1. Dropped rounds lost [Continue]. Only the reply's FIRST row was stamped, so when a strategy dropped that tool exchange but kept the signed final reply (the shape OmitToolExchange tests for ticks), the surviving reply sat straight after the earlier assistant message: the merge this PR set out to stop. Now the [Continue] prompt is turn-level, as a tick is. At each stream start the Agent decides the prompt the stream's reply follows (StartStreamResult.continuePrompt, { turn: uuid, text }, optional on the public type): a new one when the request ends on [Continue]; else the turn's carried prompt (ActivationRequestOptions.continuePrompt) only if this request renders it (renderRequestOnlyRows reports promptedTurns, as it does separatedTicks); else none. The framework sets the turn's trigger (InferenceRequest.continuePrompt) to that at each stream start, carries it into the budget restart's request (the guard and error retries carry it on their trigger), and silentTurnRowMetadata (the per-turn row metadata, named for its first use) stamps every row the turn stores with it as metadata.promptedBy. Request builds render the prompt's text before the first of its rows that survives. The keep-only-if-rendered rule is Basil-1007's, from their review of this commit: a trigger the provider cooldown held (holdProviderAcceleration) carried the prompt its first stream start set, so a resumed compile that ended on a message stored meanwhile rendered no [Continue], yet the reply's rows were stamped, and later compiles put [Continue] before a reply whose request never had it. The Agent no longer stamps rows (continuePromptPending is gone); a direct runInference caller never adds [Continue]. 2. A tick's restart could rewrite earlier turns. Whether a tick "opens" (its request ends on the separator) was decided by whether the tick had ever stored a row (`started`). A restart whose compile folded away all the tick's rows therefore got no separator, ended on [Continue] when the window ended on the agent's message, and later compiles put the separator before a reply whose request never held it. Now renderRequestOnlyRows reports which ticks it separated, and a tick opens wherever its separator wasn't rendered. `started` is gone. 3. Upgrades would have invalidated saved thinking. Changing SILENT_HEARTBEAT_SEPARATOR's text re-rendered every past tick with it, after compile, where context-manager can't see it; replies stored after a past tick would have their thinking refused (or 400 without drop_block). Now each tick's rows record the separator their request rendered (SilentHeartbeatStamp.separator, set by silentTurnRowMetadata), and rendering uses the recorded text; rows stamped without it render the legacy '[heartbeat tick]' byte for byte. The [Continue] stamp records its text the same way. So no stored reply's prefix changes on upgrade, and a future change to either constant changes only rows stored after it. Mechanics: indexRequestOnlyRows now indexes only the agent's own rows with identity keys: ticks (key -> { eventId, text }) and prompts (key -> { turn, text }); renderRequestOnlyRows returns { messages, separatedTicks }. silentTurnRowMetadata keeps its name (renaming it would touch seven call sites other open PRs change). RequestOnlyPrompt is exported as a type. Out of scope, noted: the non-streaming fallback's continuation request ('tool_results_ready') carries none of the turn's properties (suppressProse, silentHeartbeat, and now continuePrompt), as before. Tests (silent-heartbeat-rows): the tick stamp carries its separator and no row stores the separator; recorded texts render as recorded and legacy tick stamps render '[heartbeat tick]'; a restart whose compile left none of the tick's rows ends on the separator and the next request keeps it as a prefix; every row of a [Continue] reply's turn is marked, one [Continue] before the first; with the reply's tool exchange dropped, [Continue] precedes the surviving final reply and no wire message carries two signatures; a reply to [Continue] that restarts at its tool boundary carries the prompt into the restart's rows, so a window that later folds the first round renders [Continue] before the restart's reply; a carried prompt whose request doesn't render it is dropped (Basil-1007's probe, failing at 5188d31); the both-stamps unit test on the new shapes. At ee43daf's source 7 of them fail (the restart and dropped-round tests on behaviour, the rest on the new stamp shapes). Neighbours touching ticks, [Continue] or stream starts: 236/236, one file at a time. --- changelog.d/request-only-rows-stay.changed.md | 39 ++-- src/agent.ts | 53 +++--- src/framework.ts | 42 +++-- src/index.ts | 1 + src/silent-heartbeat.ts | 154 +++++++++------- src/types/agent.ts | 9 +- test/silent-heartbeat-rows.test.ts | 169 ++++++++++++++++-- 7 files changed, 343 insertions(+), 124 deletions(-) diff --git a/changelog.d/request-only-rows-stay.changed.md b/changelog.d/request-only-rows-stay.changed.md index 067641c96..5365f9f38 100644 --- a/changelog.d/request-only-rows-stay.changed.md +++ b/changelog.d/request-only-rows-stay.changed.md @@ -6,18 +6,29 @@ the prefix it was minted under (context-manager #155) it failed every retained thinking block in that request, and the tick reply's own thinking at the next ordinary turn. `SILENT_HEARTBEAT_SEPARATOR` is now that - instruction, so every past tick reads it too: 199 characters where it was - 16, at each tick whose rows survive in the window (roughly 50 tokens a - tick). The first compile after upgrading renders past ticks with the new - text, one prompt-cache rewrite. (`InferenceRequest.ephemeralSystemPrompt` - is removed; see the breaking entry.) + instruction (199 characters where it was 16, roughly 50 tokens at each + tick whose rows survive in the window). Each tick's rows now record the + separator their request rendered (`SilentHeartbeatStamp.separator`), and + later compiles render that text, so ticks stored before upgrading keep + their `[heartbeat tick]` marker byte for byte: no stored reply's prefix + changes. (`InferenceRequest.ephemeralSystemPrompt` is removed; see the + breaking entry.) +- A tick whose restart compiles none of the tick's rows (a strategy folded + them all) opens again: the restart's request ends on the separator, where + later compiles render it before the restart's reply. It used to end on + `[Continue]`, and later compiles put the separator before a reply whose + request never held it. - When a compile ends on the agent's own message and the request gets a - trailing `[Continue]` turn, the reply it prompts now has its first stored - row marked `metadata.promptedBy: 'continue'`, and every later request - renders `[Continue]` again just before that row, the way a silent tick's - separator is rendered before its first row. Without it, later requests - merged the reply into the previous assistant message (one message carrying - signed thinking from two responses, a 400 when it is the latest), and - dropped a turn the reply's request held, which a provider that binds - thinking refuses. A reply with neither a tool call nor signed thinking - can't be found in a compiled window and merges as before. + trailing `[Continue]` turn, every row the reply's turn stores now records + that prompt (`metadata.promptedBy`, a `{ turn, text }` pair), and every + later request renders `[Continue]` again just before the first of them + that survives, the way a silent tick's separator is rendered. A restart or + retry of the turn keeps the prompt only while its own request renders it. + Without it, later requests merged the reply into the previous assistant + message (one message carrying signed thinking from two responses, a 400 + when it is the latest), and dropped a turn the reply's request held, which + a provider that binds thinking refuses. A reply with neither a tool call + nor signed thinking can't be found in a compiled window and merges as + before. `StartStreamResult.continuePrompt` and + `InferenceRequest.continuePrompt` carry it (new, optional); the agent and + the framework set them. diff --git a/src/agent.ts b/src/agent.ts index 7a01884b9..24200ba12 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -4,15 +4,17 @@ import { createHash } from 'node:crypto'; import type { CacheWireReceipt, KvUnifiedRequestHooks } from './kv-unified-wire.js'; import { ToolResultGuard, TOOL_RESULT_GUARD_NOTICE } from './tool-result-guard.js'; import { - PROMPTED_BY_CONTINUE, + CONTINUE_ROW, continueTurn, indexRequestOnlyRows, renderRequestOnlyRows, silentHeartbeatSeparatorTurn, + type RequestOnlyPrompt, type RequestOnlyRowIndex, type SilentHeartbeatTick, type StoredRowLike, } from './silent-heartbeat.js'; +import { randomUUID } from 'node:crypto'; import { toolResultDataToHistoryString, truncateForHistory, @@ -20,6 +22,9 @@ import { } from './tool-result-history.js'; export interface ActivationRequestOptions { + /** The `[Continue]` prompt this turn's earlier streams replied to, if any: + * kept for this request's reply only while this request renders it. */ + continuePrompt?: RequestOnlyPrompt; /** The silent heartbeat tick this request answers, if any. */ silentHeartbeat?: SilentHeartbeatTick; } @@ -27,6 +32,12 @@ export interface ActivationRequestOptions { export interface StartStreamResult { stream: YieldingStream; request: NormalizedRequest; + /** The request-only `[Continue]` prompt this request's reply follows, if + * any: a new one when the request ends on `[Continue]`, else the turn's + * prompt (`ActivationRequestOptions.continuePrompt`) only if this request + * renders it. The framework stamps the reply's rows with it + * (`metadata.promptedBy`), for later compiles to render it before them. */ + continuePrompt?: RequestOnlyPrompt; takeKvSubmission?: () => { submissionId: string; wireReceipt: CacheWireReceipt } | undefined; drainKvSubmissionIds?: () => string[]; } @@ -802,7 +813,7 @@ export class Agent { budget?: TokenBudget, compressionTools: ToolDefinition[] = availableTools, options: ActivationRequestOptions = {}, - ): Promise<{ request: NormalizedRequest; endsOnContinue: boolean }> { + ): Promise<{ request: NormalizedRequest; continuePrompt?: RequestOnlyPrompt }> { // Compression may revisit hidden tools in recorded history. Keep its full // definition set separate from the resident's advertised tools: the // autobiographical strategy must declare the same tools on its @@ -865,12 +876,20 @@ export class Agent { // that same turn, so the prefix it caches is the one every later compile // renders. "First" is per agent: a broadcast tick shares its eventId with // every resident. - const requestOnly = this.indexRequestOnlyRows(); - messages = renderRequestOnlyRows(messages, requestOnly, this.name); + const rendered = renderRequestOnlyRows(messages, this.indexRequestOnlyRows(), this.name); + messages = rendered.messages; + // A tick opens wherever its separator isn't already rendered: on its first + // request, and on a restart whose compile left none of the tick's rows, + // so the restart's reply follows the separator, where every later compile + // renders it. const openingTick = options.silentHeartbeat !== undefined - && !requestOnly.started.has(options.silentHeartbeat.eventId); + && !rendered.separatedTicks.has(options.silentHeartbeat.eventId); - let endsOnContinue = false; + // The [Continue] prompt this request's reply follows: a new one when it + // ends on [Continue]; else the turn's prompt only where this request + // renders it (a stream resumed after a cooldown, say, whose compile now + // ends on a message that arrived meanwhile, renders none); else none. + let continuePrompt: RequestOnlyPrompt | undefined; if (openingTick) { messages = [...messages, silentHeartbeatSeparatorTurn()]; } else if (messages.length > 0 && messages[messages.length - 1]!.participant === this.name) { @@ -878,7 +897,10 @@ export class Agent { // Some models reject trailing assistant messages ("prefill not supported"), // and after context compression a stale assistant turn can end up last. messages = [...messages, continueTurn()]; - endsOnContinue = true; + continuePrompt = { turn: randomUUID(), text: CONTINUE_ROW }; + } + if (!continuePrompt && options.continuePrompt && rendered.promptedTurns.has(options.continuePrompt.turn)) { + continuePrompt = options.continuePrompt; } const request: NormalizedRequest = { @@ -898,7 +920,7 @@ export class Agent { ...(this.providerParams && { providerParams: this.providerParams }), assistantParticipant: this.name, }; - return { request, endsOnContinue }; + return { request, ...(continuePrompt ? { continuePrompt } : {}) }; } /** The request `assembleActivationRequest` builds (see there), for preview @@ -919,7 +941,7 @@ export class Agent { private requestOnlyIndex: { source: readonly unknown[]; index: RequestOnlyRowIndex } | null = null; private indexRequestOnlyRows(): RequestOnlyRowIndex { - const empty: RequestOnlyRowIndex = { ticks: new Map(), continued: new Set(), started: new Set() }; + const empty: RequestOnlyRowIndex = { ticks: new Map(), prompts: new Map() }; const cm = this.contextManager as Partial; if (typeof cm.getAllMessages !== 'function') return empty; try { @@ -934,11 +956,6 @@ export class Agent { } } - /** The current stream's request ended on a [Continue] turn, and the reply - * it prompted hasn't stored a row yet: its first row is stamped - * `promptedBy: 'continue'`, so later requests render the turn before it. */ - private continuePromptPending = false; - /** * Start a yielding stream with context injections. * Same as startStream but passes injections through to compile. @@ -983,11 +1000,10 @@ export class Agent { this.lastStreamRealInputTokens = 0; this.lastStreamOutputTokens = 0; - const { request, endsOnContinue } = await this.assembleActivationRequest( + const { request, continuePrompt } = await this.assembleActivationRequest( availableTools, injections, budget, compressionTools, options, ); request.messages = this.toolResultGuard.prepareRequest(request.messages, true); - this.continuePromptPending = endsOnContinue; const receiptAware = (this.contextManager as unknown as { getStrategy?: () => unknown }) .getStrategy?.() as { @@ -1035,6 +1051,7 @@ export class Agent { return { stream, request, + ...(continuePrompt ? { continuePrompt } : {}), ...(kvEnabled ? { takeKvSubmission: () => kvQueue.shift(), @@ -1113,10 +1130,6 @@ export class Agent { ); return; } - if (this.continuePromptPending) { - this.continuePromptPending = false; - metadata = { ...metadata, promptedBy: PROMPTED_BY_CONTINUE } as MessageMetadata; - } this.contextManager.addMessage(this.name, content, metadata); } diff --git a/src/framework.ts b/src/framework.ts index c8b211542..8c9a0d891 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -8,7 +8,7 @@ import type { Membrane, ContentBlock, NormalizedRequest, YieldingStream, ToolRes import { MembraneError } from '@animalabs/membrane'; import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; -import type { SilentHeartbeatStamp } from './silent-heartbeat.js'; +import { SILENT_HEARTBEAT_SEPARATOR, type SilentHeartbeatStamp } from './silent-heartbeat.js'; import { SUBCONSCIOUS_TOOLS, SUBCONSCIOUS_TOOL_NAMES, type SubconsciousConfig } from './tune-out/tools.js'; import { TuneOutCoordinator, TUNE_OUT_DEFAULTS } from './tune-out/coordinator.js'; import type { @@ -6905,6 +6905,7 @@ export class AgentFramework { reason: 'context_budget_restart', source: 'framework', timestamp: Date.now(), + continuePrompt: this.activeTurnTriggers.get(agent.name)?.continuePrompt, suppressProse: this.activeTurnTriggers.get(agent.name)?.suppressProse, silentHeartbeat: this.activeTurnTriggers.get(agent.name)?.silentHeartbeat, }); @@ -8514,20 +8515,32 @@ export class AgentFramework { } /** - * Metadata for rows the agent's current turn stores: a silent heartbeat - * tick stamps each of them (assistant responses, tool_result rows) so they - * stay identifiable after the turn — request builds key the tick separator - * on it, and a later retention policy can find or collapse them. The stamp - * names the agent: residents share one message slot and a broadcast tick - * reaches each of them with the same eventId. Undefined for ordinary - * turns, whose rows are stored exactly as before. + * Metadata for rows the agent's current turn stores (assistant responses, + * tool_result rows), for the request-only turns that came before them: + * - a silent heartbeat tick stamps each row, with the separator text its + * request rendered, so they stay identifiable after the turn — request + * builds render the separator before the first that survives, and a later + * retention policy can find or collapse them. The stamp names the agent: + * residents share one message slot and a broadcast tick reaches each of + * them with the same eventId; + * - a turn whose stream opened on `[Continue]` stamps each row from then on + * with that prompt (`promptedBy`), so request builds render it before the + * first that survives. + * Undefined for ordinary turns, whose rows are stored exactly as before. + * (Named for its first use, the silent tick.) */ private silentTurnRowMetadata(agentName: string): MessageMetadata | undefined { - const tick = this.activeTurnTriggers.get(agentName)?.silentHeartbeat; + const trigger = this.activeTurnTriggers.get(agentName); + const tick = trigger?.silentHeartbeat; const stamp: SilentHeartbeatStamp | undefined = tick - ? { eventId: tick.eventId, serverId: tick.serverId, agentName } + ? { eventId: tick.eventId, serverId: tick.serverId, agentName, separator: SILENT_HEARTBEAT_SEPARATOR } : undefined; - return stamp ? { silentHeartbeat: stamp } : undefined; + const prompt = trigger?.continuePrompt; + if (!stamp && !prompt) return undefined; + return { + ...(stamp ? { silentHeartbeat: stamp } : {}), + ...(prompt ? { promptedBy: prompt } : {}), + } as MessageMetadata; } /** Record prose segments suppressed by explicit-send silencing. */ @@ -9360,9 +9373,16 @@ export class AgentFramework { request: compiledRequest, takeKvSubmission, drainKvSubmissionIds, + continuePrompt, } = await agent.startStreamWithInjections(tools, injections, undefined, compressionTools, { silentHeartbeat: trigger?.silentHeartbeat, + continuePrompt: trigger?.continuePrompt, }); + // The [Continue] prompt this stream's reply follows, if any: every row + // the turn stores from here on carries it (silentTurnRowMetadata), and the + // turn's restarts and retries carry it in the trigger to the next + // stream start, which keeps it only if its request renders it. + if (trigger) trigger.continuePrompt = continuePrompt; if (this.agents.get(agent.name) !== agent) { stream.cancel(); agent.cancelStream(); diff --git a/src/index.ts b/src/index.ts index adfab80b2..0696aad78 100644 --- a/src/index.ts +++ b/src/index.ts @@ -153,6 +153,7 @@ export * from "./tool-presentation.js"; export { SILENT_HEARTBEAT_SEPARATOR, silentHeartbeatOf, + type RequestOnlyPrompt, type SilentHeartbeatStamp, type SilentHeartbeatTick, } from './silent-heartbeat.js'; diff --git a/src/silent-heartbeat.ts b/src/silent-heartbeat.ts index f9e4bd0d3..1a2211c31 100644 --- a/src/silent-heartbeat.ts +++ b/src/silent-heartbeat.ts @@ -12,9 +12,9 @@ * first of the tick's rows that survives compilation. * - A compile that ends on the agent's own message gets a trailing * `[Continue]` turn (some models reject a trailing assistant message). The - * reply it prompts is stored straight after that message, so the reply's - * first stored row is stamped `metadata.promptedBy: 'continue'`, and the - * `[Continue]` turn is rendered before it. + * reply it prompts is stored straight after that message, so every row the + * reply's turn stores is stamped `metadata.promptedBy` (`RequestOnlyPrompt`), + * and the turn is rendered before the first of them that survives. * * Without the row, the wire formatter merges consecutive same-role messages: * the reply would read as an unprompted continuation of the earlier response, @@ -23,12 +23,14 @@ * `Agent.addAssistantResponse`). And the request that minted the reply held * the row: a provider that binds signed thinking to the prefix it was minted * under (context-manager #155) refuses that thinking once a later request - * drops it. Each row's text is constant and its position is a function of the - * stamped rows alone, so every compile renders it in the same place (the - * request that minted the reply ends on that same turn), which keeps the - * prompt-cache prefix and the binding stable — unlike a per-compile injection - * that re-anchors to the newest user message, or a system prompt that changes - * for one turn. + * drops it. Each row's text is read from the stamp (what the request that + * minted the reply rendered, so a later change to the constant changes no + * stored reply's prefix) and its position is a function of the stamped rows + * alone, so every compile renders it in the same place (the request that + * minted the reply ends on that same turn), which keeps the prompt-cache + * prefix and the binding stable — unlike a per-compile injection that + * re-anchors to the newest user message, or a system prompt that changes for + * one turn. * * Known limit: compiled messages carry no store provenance, so a stored row * is recognised only by structural identity the provider made unique @@ -42,22 +44,23 @@ */ import type { ContentBlock, NormalizedMessage } from '@animalabs/membrane'; -/** Request-only separator before a silent tick's first surviving row. It - * carries the tick's instruction, so the tick's own request reads it as its - * prompt and every later compile shows it at each past tick. Constant on - * purpose: it carries no time or id, so it is byte-identical on every - * compile. */ +/** Request-only separator a silent tick's request ends on. It carries the + * tick's instruction, so the tick's own request reads it as its prompt, and + * every later compile renders it again before the tick's first surviving row. + * Constant on purpose: it carries no time or id. Each tick's rows record the + * text their request rendered (`SilentHeartbeatStamp.separator`), so changing + * this constant changes only ticks that open afterwards. */ export const SILENT_HEARTBEAT_SEPARATOR = '[silent heartbeat] Scheduled private self-check. Review pending matters privately. ' + 'Do not narrate or publish plain prose; use an explicit send tool only if you deliberately choose to contact someone.'; +/** The separator ticks rendered before their rows recorded its text: rows + * stamped without `separator` still render it, byte for byte. */ +const LEGACY_SILENT_HEARTBEAT_SEPARATOR = '[heartbeat tick]'; + /** Request-only turn a compile that ends on the agent's own message gets. */ export const CONTINUE_ROW = '[Continue]'; -/** `metadata.promptedBy` on the first row a reply to a `[Continue]` turn - * stores. */ -export const PROMPTED_BY_CONTINUE = 'continue'; - /** Identity of one authenticated silent heartbeat tick. */ export interface SilentHeartbeatTick { /** The MCPL push event that woke the tick. A broadcast tick reaches every @@ -72,18 +75,38 @@ export interface SilentHeartbeatTick { * (tool_result rows are stored as `user`). */ export interface SilentHeartbeatStamp extends SilentHeartbeatTick { agentName: string; + /** The separator text the tick's request rendered, which later compiles + * render again. Absent on rows stamped before it was recorded; they render + * the bare marker those requests carried. */ + separator?: string; } /** The tick stamp on a stored row's metadata, if any. */ export function silentHeartbeatOf(metadata: unknown): SilentHeartbeatStamp | undefined { const stamp = (metadata as { silentHeartbeat?: unknown } | undefined)?.silentHeartbeat; if (!stamp || typeof stamp !== 'object') return undefined; - const { eventId, serverId, agentName } = stamp as Record; + const { eventId, serverId, agentName, separator } = stamp as Record; return typeof eventId === 'string' && typeof serverId === 'string' && typeof agentName === 'string' - ? { eventId, serverId, agentName } + ? { eventId, serverId, agentName, ...(typeof separator === 'string' ? { separator } : {}) } : undefined; } +/** A request-only turn a request ended on before the agent's reply, as + * `metadata.promptedBy` on every row the reply's turn stores: `turn` names + * the one prompt, and `text` is what the request rendered. */ +export interface RequestOnlyPrompt { + turn: string; + text: string; +} + +/** The request-only prompt a stored row replies to, if any. */ +export function promptedByOf(metadata: unknown): RequestOnlyPrompt | undefined { + const prompt = (metadata as { promptedBy?: unknown } | undefined)?.promptedBy; + if (!prompt || typeof prompt !== 'object') return undefined; + const { turn, text } = prompt as Record; + return typeof turn === 'string' && typeof text === 'string' ? { turn, text } : undefined; +} + export function silentHeartbeatSeparatorTurn(): NormalizedMessage { return requestOnlyTurn(SILENT_HEARTBEAT_SEPARATOR); } @@ -96,11 +119,6 @@ function requestOnlyTurn(text: string): NormalizedMessage { return { participant: 'user', content: [{ type: 'text', text }] }; } -/** Whether a stored row is a reply's first row after a `[Continue]` turn. */ -export function isPromptedByContinue(metadata: unknown): boolean { - return (metadata as { promptedBy?: unknown } | undefined)?.promptedBy === PROMPTED_BY_CONTINUE; -} - /** * Structural identity of one stored response wherever a compile renders it: * tool_use ids and thinking signatures, which the provider made unique and @@ -129,63 +147,77 @@ export interface StoredRowLike { export interface RequestOnlyRowIndex { /** Identity key of every assistant row this agent stored during a silent - * tick → the tick's event id. */ - ticks: Map; - /** Identity keys of the first row of each reply to a `[Continue]` turn. */ - continued: Set; - /** Ticks this agent has already stored rows for (its own, not another resident's). */ - started: Set; + * tick → the tick, and the separator text to render before its first + * surviving row. */ + ticks: Map; + /** Identity key of every assistant row this agent stored in reply to a + * request-only prompt → that prompt. */ + prompts: Map; } -/** Index the rows that follow request-only turns: every row this agent stored - * during its silent ticks, and the first row of each reply to a `[Continue]`. - * The two are independent, so a row can be both. */ +/** Index the agent's own rows that follow request-only turns: the rows of its + * silent ticks, and the rows of each reply to a `[Continue]`. The two are + * independent, so a row can be both. */ export function indexRequestOnlyRows(stored: readonly StoredRowLike[], agentName: string): RequestOnlyRowIndex { - const ticks = new Map(); - const continued = new Set(); - const started = new Set(); + const ticks = new Map(); + const prompts = new Map(); for (const row of stored) { + if (row.participant !== agentName) continue; + const keys = identityKeys(row.content); + if (keys.length === 0) continue; const stamp = silentHeartbeatOf(row.metadata); if (stamp && stamp.agentName === agentName) { - started.add(stamp.eventId); - if (row.participant === agentName) { - for (const key of identityKeys(row.content)) ticks.set(key, stamp.eventId); - } - } - if (isPromptedByContinue(row.metadata)) { - for (const key of identityKeys(row.content)) continued.add(key); + const tick = { eventId: stamp.eventId, text: stamp.separator ?? LEGACY_SILENT_HEARTBEAT_SEPARATOR }; + for (const key of keys) ticks.set(key, tick); } + const prompt = promptedByOf(row.metadata); + if (prompt) for (const key of keys) prompts.set(key, prompt); } - return { ticks, continued, started }; + return { ticks, prompts }; +} + +export interface RenderedRequestOnlyRows { + messages: NormalizedMessage[]; + /** The ticks whose separator this window renders. A tick none of whose + * rows survive compilation renders none. */ + separatedTicks: Set; + /** The prompts (`RequestOnlyPrompt.turn`) this window renders, the same + * way. */ + promptedTurns: Set; } /** - * Insert the request-only turns into a compiled window: a tick's separator - * immediately before the first of the tick's rows that survives compilation - * (its first row when rendered, otherwise whichever later row a strategy - * kept), and `[Continue]` immediately before each reply it prompted, after - * any separator, as the reply's request had them. Rows folded into - * summaries simply do not match; ordinary messages pass through untouched. + * Insert the request-only turns into a compiled window, each immediately + * before the first of its rows that survives compilation (the first row when + * it is rendered, otherwise whichever later row a strategy kept): a tick's + * separator, and the prompt each reply to `[Continue]` followed, after any + * separator, as the reply's request had them. Rows folded into summaries + * simply do not match; ordinary messages pass through untouched. */ export function renderRequestOnlyRows( messages: NormalizedMessage[], - index: Pick, + index: RequestOnlyRowIndex, agentName: string, -): NormalizedMessage[] { - if (index.ticks.size === 0 && index.continued.size === 0) return messages; - const separated = new Set(); +): RenderedRequestOnlyRows { + const separatedTicks = new Set(); + const promptedTurns = new Set(); + if (index.ticks.size === 0 && index.prompts.size === 0) return { messages, separatedTicks, promptedTurns }; const out: NormalizedMessage[] = []; for (const message of messages) { if (message.participant === agentName) { const keys = identityKeys(message.content); - const tick = keys.map((key) => index.ticks.get(key)).find((id) => id !== undefined); - if (tick !== undefined && !separated.has(tick)) { - separated.add(tick); - out.push(silentHeartbeatSeparatorTurn()); + const tick = keys.map((key) => index.ticks.get(key)).find((t) => t !== undefined); + if (tick !== undefined && !separatedTicks.has(tick.eventId)) { + separatedTicks.add(tick.eventId); + out.push(requestOnlyTurn(tick.text)); + } + const prompt = keys.map((key) => index.prompts.get(key)).find((p) => p !== undefined); + if (prompt !== undefined && !promptedTurns.has(prompt.turn)) { + promptedTurns.add(prompt.turn); + out.push(requestOnlyTurn(prompt.text)); } - if (keys.some((key) => index.continued.has(key))) out.push(continueTurn()); } out.push(message); } - return out; + return { messages: out, separatedTicks, promptedTurns }; } diff --git a/src/types/agent.ts b/src/types/agent.ts index 3fe5509e2..a4fc4b9e1 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -1,7 +1,7 @@ import type { ContentBlock, YieldingStream } from '@animalabs/membrane'; import type { ContextStrategy } from '@animalabs/context-manager'; import type { ToolCallId, ToolResult, ToolCall } from './events.js'; -import type { SilentHeartbeatTick } from '../silent-heartbeat.js'; +import type { RequestOnlyPrompt, SilentHeartbeatTick } from '../silent-heartbeat.js'; export type SameRoundThinkTextPolicy = 'public' | 'private'; export type SameRoundThinkTextPolicySource = @@ -360,6 +360,13 @@ export interface InferenceRequest { /** Suppress every automatic plain-prose delivery for this logical turn. * Explicit tool calls remain available. Used by authenticated silent wakes. */ suppressProse?: boolean; + /** Set by the framework at each stream start of this turn: the request-only + * `[Continue]` prompt the stream's reply follows (`StartStreamResult. + * continuePrompt`), or none. Every row the turn stores is stamped + * `metadata.promptedBy` with it, so request builds render the turn before + * the first that survives. Carried into the turn's restarts and retries, + * whose stream start keeps it only if its own request renders it. */ + continuePrompt?: RequestOnlyPrompt; /** The authenticated silent heartbeat tick this turn answers. Every row the * turn stores is stamped `metadata.silentHeartbeat` with it, and request * builds render a request-only separator carrying the tick's instruction diff --git a/test/silent-heartbeat-rows.test.ts b/test/silent-heartbeat-rows.test.ts index c9a8fff26..a0fd91fae 100644 --- a/test/silent-heartbeat-rows.test.ts +++ b/test/silent-heartbeat-rows.test.ts @@ -76,11 +76,13 @@ class QueuedStreamsMembrane extends MockMembrane { } } -/** A strategy that leaves every tool exchange out of the compiled window. */ +/** A strategy that leaves every tool exchange out of the compiled window + * (from the moment `omit` is set, for a window that folds them later). */ class OmitToolExchange extends PassthroughStrategy { + omit = true; select(...args: Parameters) { - return super.select(...args).filter((entry) => - !entry.content.some((block) => block.type === 'tool_use' || block.type === 'tool_result')); + return super.select(...args).filter((entry) => !this.omit + || !entry.content.some((block) => block.type === 'tool_use' || block.type === 'tool_result')); } } @@ -213,14 +215,15 @@ describe('silent heartbeat stored rows', () => { 'tool round assistant row, its tool_result row, trailing assistant row — nothing else'); for (const row of tickRows) { assert.deepEqual(row.metadata?.silentHeartbeat, - { eventId: tick.eventId, serverId: 'heartbeat', agentName: 'assistant' }, + { eventId: tick.eventId, serverId: 'heartbeat', agentName: 'assistant', separator: SILENT_HEARTBEAT_SEPARATOR }, `row stamped: ${JSON.stringify(row.content)}`); } assert.ok(tickRows[1]!.content.every((b) => b.type === 'tool_result'), 'the only user row is the tool_result'); for (const row of all.slice(0, before)) { assert.equal(row.metadata?.silentHeartbeat, undefined, 'ordinary rows are not stamped'); } - assert.ok(!JSON.stringify(all).includes(SILENT_HEARTBEAT_SEPARATOR), 'the separator is never stored'); + assert.ok(!all.some((m) => JSON.stringify(m.content).includes(SILENT_HEARTBEAT_SEPARATOR)), + 'the separator is never stored as a row'); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); @@ -286,6 +289,34 @@ describe('silent heartbeat stored rows', () => { } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); + it('opens the tick again on a restart whose compile left none of its rows', async () => { + const membrane = new QueuedStreamsMembrane(); + const x = await make({ membrane, physicalWindowTokens: 200_000, deliver: false, agent: { strategy: new OmitToolExchange() } }); + try { + membrane.queues.push([createMockResponse(signed('sig-ordinary', 'ordinary reply'))]); + await wake(x, ordinaryEvent('hello')); + const nearCap = createMockResponse(TICK_TOOL_ROUND, 'tool_use'); + (nearCap as { usage: unknown }).usage = { inputTokens: 5_000, outputTokens: 5, cacheReadTokens: 195_000 }; + membrane.queues.push([nearCap], [createMockResponse(signed('sig-tick-final', 'all quiet'))]); + await wake(x, silentEvent()); + membrane.queues.push([createMockResponse(signed('sig-later', 'later reply'))]); + await wake(x, ordinaryEvent('later')); + assert.equal(membrane.calls.length, 4, 'ordinary, tick, its restart, later'); + + // The strategy dropped the tick's only rows (its tool exchange), so the + // restart's compile ends on the ordinary reply: the tick opens again. + const restarted = membrane.calls[2]!.messages; + assert.ok(!restarted.some((m) => textOf(m) === 'checking')); + assert.equal(textOf(restarted.at(-1)!), SILENT_HEARTBEAT_SEPARATOR); + assert.ok(!restarted.some((m) => textOf(m) === '[Continue]')); + // The reply it minted follows the separator in every later request. + const later = membrane.calls[3]!.messages; + assert.deepEqual(later.slice(0, restarted.length), restarted); + assert.equal(later.filter((m) => textOf(m) === SILENT_HEARTBEAT_SEPARATOR).length, 1); + assert.equal(textOf(later[later.findIndex((m) => textOf(m) === 'all quiet') - 1]!), SILENT_HEARTBEAT_SEPARATOR); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + it('separates the first SURVIVING tick row when a strategy drops the tick\'s tool exchange', async () => { const x = await make({ deliver: false, agent: { strategy: new OmitToolExchange() } }); try { @@ -494,7 +525,8 @@ describe('replies to a [Continue] turn', () => { const rows = stored(x.framework); const reply = rows.find((m) => m.content.some((b) => (b as { text?: string }).text === 'continued reply'))!; - assert.equal(reply.metadata?.promptedBy, 'continue'); + assert.equal(reply.metadata?.promptedBy?.text, '[Continue]'); + assert.equal(typeof reply.metadata?.promptedBy?.turn, 'string'); for (const row of rows.filter((m) => m !== reply)) assert.equal(row.metadata?.promptedBy, undefined); assert.ok(!rows.some((m) => m.content.some((b) => (b as { text?: string }).text === '[Continue]')), 'the [Continue] turn is never stored'); @@ -502,26 +534,43 @@ describe('replies to a [Continue] turn', () => { }); it('a row that is both a tick row and a reply to [Continue] gets both turns, separator first', () => { - const tick = { eventId: 'hb-x', serverId: 'heartbeat', agentName: 'assistant' }; + const tick = { eventId: 'hb-x', serverId: 'heartbeat', agentName: 'assistant', separator: SILENT_HEARTBEAT_SEPARATOR }; + const prompt = { turn: 'p-1', text: '[Continue]' }; const stored = [ { participant: 'assistant', content: signed('sig-a', 'earlier reply') }, { participant: 'assistant', content: signed('sig-1', 'tick round'), metadata: { silentHeartbeat: tick } }, { participant: 'assistant', content: signed('sig-2', 'tick continued'), - metadata: { silentHeartbeat: tick, promptedBy: 'continue' } }, + metadata: { silentHeartbeat: tick, promptedBy: prompt } }, ]; const index = indexRequestOnlyRows(stored, 'assistant'); - const rendered = renderRequestOnlyRows(stored.map(({ participant, content }) => ({ participant, content })), index, 'assistant'); - assert.deepEqual(rendered.map(textOf), [ + const bare = (rows: typeof stored) => rows.map(({ participant, content }) => ({ participant, content })); + assert.deepEqual(renderRequestOnlyRows(bare(stored), index, 'assistant').messages.map(textOf), [ 'earlier reply', SILENT_HEARTBEAT_SEPARATOR, 'tick round', '[Continue]', 'tick continued', ]); // With the tick's first row folded away, the stamped row is the tick's // first surviving row: both turns precede it, the separator first. - const folded = renderRequestOnlyRows( - [stored[0]!, stored[2]!].map(({ participant, content }) => ({ participant, content })), index, 'assistant'); - assert.deepEqual(folded.map(textOf), ['earlier reply', SILENT_HEARTBEAT_SEPARATOR, '[Continue]', 'tick continued']); + assert.deepEqual(renderRequestOnlyRows(bare([stored[0]!, stored[2]!]), index, 'assistant').messages.map(textOf), + ['earlier reply', SILENT_HEARTBEAT_SEPARATOR, '[Continue]', 'tick continued']); }); - it('marks only the first row the reply stores, and renders one [Continue] before it', async () => { + it('renders the text each stamp recorded, and the bare marker for ticks stamped before it was recorded', () => { + const stored = [ + { participant: 'assistant', content: signed('sig-old', 'old tick reply'), + metadata: { silentHeartbeat: { eventId: 'hb-old', serverId: 'heartbeat', agentName: 'assistant' } } }, + { participant: 'assistant', content: signed('sig-new', 'new tick reply'), + metadata: { silentHeartbeat: { eventId: 'hb-new', serverId: 'heartbeat', agentName: 'assistant', separator: 'recorded separator' } } }, + { participant: 'assistant', content: signed('sig-c', 'continued reply'), + metadata: { promptedBy: { turn: 'p-1', text: 'recorded prompt' } } }, + ]; + const index = indexRequestOnlyRows(stored, 'assistant'); + const rendered = renderRequestOnlyRows(stored.map(({ participant, content }) => ({ participant, content })), index, 'assistant'); + assert.deepEqual(rendered.messages.map(textOf), [ + '[heartbeat tick]', 'old tick reply', 'recorded separator', 'new tick reply', 'recorded prompt', 'continued reply', + ]); + assert.deepEqual([...rendered.separatedTicks].sort(), ['hb-new', 'hb-old']); + }); + + it('marks every row the reply\'s turn stores, and renders one [Continue] before the first', async () => { const x = await make({ deliver: false }); try { await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); @@ -532,9 +581,10 @@ describe('replies to a [Continue] turn', () => { await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); const rows = stored(x.framework); - const marked = rows.filter((m) => m.metadata?.promptedBy === 'continue'); - assert.equal(marked.length, 1); - assert.ok(marked[0]!.content.some((b) => b.type === 'tool_use'), 'the first row: the tool round'); + const marked = rows.filter((m) => m.metadata?.promptedBy !== undefined); + assert.deepEqual(marked.map((m) => m.participant), ['assistant', 'user', 'assistant'], + 'the tool round, its tool_result row and the final reply'); + assert.equal(new Set(marked.map((m) => m.metadata!.promptedBy.turn)).size, 1, 'one prompt'); const later = x.membrane.calls[2]!.messages; assert.equal(later.filter((m) => textOf(m) === '[Continue]').length, 1); @@ -543,4 +593,89 @@ describe('replies to a [Continue] turn', () => { assert.deepEqual(later.slice(0, x.membrane.calls[1]!.messages.length), x.membrane.calls[1]!.messages); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); + + it('carries the prompt into the reply\'s budget restart, for a window that later folds the first round', async () => { + const membrane = new QueuedStreamsMembrane(); + const strategy = new OmitToolExchange(); + strategy.omit = false; + const x = await make({ membrane, physicalWindowTokens: 200_000, deliver: false, agent: { strategy } }); + try { + membrane.queues.push([createMockResponse(signed('sig-1', 'first reply'))]); + await wake(x, ordinaryEvent('hello')); + // The reply to [Continue] restarts at its tool boundary; the restart's + // compile ends on the tool_result, so it opens on no prompt of its own. + const nearCap = createMockResponse( + signed('sig-c', 'checking', [{ type: 'tool_use', id: 'c-t1', name: 'tools--ping', input: {} } as ContentBlock]), + 'tool_use'); + (nearCap as { usage: unknown }).usage = { inputTokens: 5_000, outputTokens: 5, cacheReadTokens: 195_000 }; + membrane.queues.push([nearCap], [createMockResponse(signed('sig-c2', 'all done'))]); + (x.framework as any).pendingRequests.push({ agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now() }); + await x.framework.runUntilIdle(); + assert.equal(membrane.calls.length, 3, 'first, the continued reply, its restart'); + assert.equal(textOf(membrane.calls[1]!.messages.at(-1)!), '[Continue]'); + assert.notEqual(textOf(membrane.calls[2]!.messages.at(-1)!), '[Continue]'); + + const marked = stored(x.framework).filter((m) => m.metadata?.promptedBy !== undefined); + assert.equal(marked.length, 3, 'the round, its result and the restart\'s reply'); + assert.equal(new Set(marked.map((m) => m.metadata!.promptedBy.turn)).size, 1); + + // Later, the window folds the first round: [Continue] moves to the + // restart's reply, the first of the prompt's rows that survives. + strategy.omit = true; + membrane.queues.push([createMockResponse(signed('sig-later', 'later reply'))]); + await wake(x, ordinaryEvent('later')); + const later = membrane.calls[3]!.messages; + assert.ok(!later.some((m) => textOf(m) === 'checking')); + assert.equal(textOf(later[later.findIndex((m) => textOf(m) === 'all done') - 1]!), '[Continue]'); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('drops a carried prompt its request doesn\'t render (a stream resumed after a cooldown)', async () => { + // The shape a provider cooldown leaves (Basil-1007's probe): the held + // trigger carries the prompt its first stream start set, and a message + // lands meanwhile, so the resumed compile ends on that message and + // renders no [Continue]. + const x = await make({ deliver: false }); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + x.framework.getAgent('assistant')!.getContextManager() + .addMessage('nissa', [{ type: 'text', text: 'arrived during the cooldown' }] as ContentBlock[]); + x.membrane.pushResponse(createMockResponse(signed('sig-r', 'resumed reply'), 'end_turn')); + (x.framework as any).pendingRequests.push({ + agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now(), + continuePrompt: { turn: 'stale-p1', text: '[Continue]' }, + }); + await x.framework.runUntilIdle(); + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + + const minting = x.membrane.calls[1]!.messages; + assert.equal(textOf(minting.at(-1)!), 'arrived during the cooldown'); + const reply = stored(x.framework).find((m) => m.content.some((b) => (b as { text?: string }).text === 'resumed reply'))!; + assert.equal(reply.metadata?.promptedBy, undefined); + const later = x.membrane.calls[2]!.messages; + const replyAt = later.findIndex((m) => textOf(m) === 'resumed reply'); + assert.deepEqual(later.slice(0, replyAt), minting, 'what precedes the reply is what its request held'); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('renders [Continue] before the first SURVIVING row when a strategy drops the reply\'s tool exchange', async () => { + const x = await make({ deliver: false, agent: { strategy: new OmitToolExchange() } }); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + await timerWake(x, + signed('sig-c', 'checking', [{ type: 'tool_use', id: 'c-t1', name: 'tools--ping', input: {} } as ContentBlock]), + signed('sig-c2', 'all done')); + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + + const later = x.membrane.calls[2]!.messages; + assert.ok(!later.some((m) => textOf(m) === 'checking'), 'the strategy dropped the tool round'); + const finalAt = later.findIndex((m) => textOf(m) === 'all done'); + assert.equal(textOf(later[finalAt - 1]!), '[Continue]'); + assert.equal(later.filter((m) => textOf(m) === '[Continue]').length, 1); + for (const message of wire(x.membrane.calls[2]!)) { + assert.ok(signatures(message).length <= 1, + `one ${message.role} message carries signed thinking from: ${signatures(message).join(', ')}`); + } + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); }); From 1dd57c1c4df283e722650d79046e3ffd753f044a Mon Sep 17 00:00:00 2001 From: Basil-1007 Date: Sun, 11 Oct 2026 16:26:35 -0400 Subject: [PATCH 03/11] fix(requests): a reply follows [Continue] only when what it is sent ends on it or on its tool loop The prompt a reply follows was set at each stream start and carried on the turn's trigger, kept for a restart only if the restart's request rendered it. Two shapes slipped through: - A round inside one stream. At a tool boundary, hear-while-acting stores the messages that arrived while the tool ran after the tool results, and hands them to the live stream, so the next round follows them. Its rows still took the stream-start prompt, and once the tool exchange folded, later requests rendered [Continue] between those messages and a reply that followed them. - A stream resumed after a provider cooldown, whose compile ends on a message that arrived meanwhile, kept the carried prompt whenever the turn's earlier rows were still in the window, with the same result. And the carried prompt lived only in memory, so a stream after a process restart lost it while the stored rows still carried it. The prompt is now read from what the reply is sent, and never carried: - Agent: a new prompt when the request ends on [Continue]; else, when it ends on tool results, the prompt stamped on the reply that called those tools (toolLoopPrompt, through the index the render already builds, so the store is the only source); else none. - Framework: each stream start sets the turn's holder from that, and a tool boundary whose deferred flush stores anything after the results clears it, so the next round's rows carry no prompt. The holder is private turn state keyed by the trigger (turnPrompts), not a field on InferenceRequest: nothing passes a prompt in any more, and a value a host set would only be overwritten. ActivationRequestOptions loses continuePrompt, and RenderedRequestOnlyRows loses promptedTurns, which only the old rule read. A tick's stamp stays turn-level: its instruction is true of every round of the tick. Also: - A tick is keyed by its server and event id together (tickKey), in the index, the separator dedupe and the opening check, since two servers' events can share an id. - indexRequestOnlyRows reads the stamps before computing identity keys, so a history walk costs a metadata read per unstamped row. - The stamps have their own type, RequestOnlyStamps, a type alias that fits MessageMetadata without a cast, so a misspelled key fails to compile. - The non-streaming tool_results_ready continuation carries the turn's suppressProse and silentHeartbeat, as the budget restart does. The framework's own streams never reach it; an agent the direct API left waiting on tools does. - Docs and changelog say which rows are stamped (the reply's own: assistant responses, tool results and the contained wrapper), that a stamp on a user-role row marks it as the turn's rather than anchoring a render, and that the direct runInference paths render no request-only turns. The unreleased tick fragment now describes the recorded separator text and the server-and-event key. Tests (silent-heartbeat-rows): - a later round after a message injected at its tool boundary follows the message: its rows carry no prompt, and once the round folds, the message precedes it, not [Continue]; - a reply resumed after a real provider cooldown (a 429 holds the restart, and a message arrives during it) carries no prompt, and no [Continue] renders before it once the round folds; - a reply after the process restarts (a new framework over the same store) reads the prompt from the stored rows; - a reply the error policy retries past its first round follows the prompt; - a request that ends on the agent's own message mints a new [Continue], even with an older one rendered earlier in it; - a reply whose only identity is a redacted_thinking block gets its [Continue]; - two servers' ticks that share an event id each get their separator; - the budget-restart test now has the restart read the prompt from its own request. At the parent's source, five of these fail: the injected round, the cooldown, the process restart and the shared event id on behaviour, and the recorded-text test on the tick key's new form. --- changelog.d/request-only-rows-stay.changed.md | 32 +- .../silent-heartbeat-turn-rows.fixed.md | 24 +- src/agent.ts | 36 ++- src/framework.ts | 88 +++-- src/silent-heartbeat.ts | 94 ++++-- src/types/agent.ts | 18 +- test/silent-heartbeat-rows.test.ts | 303 +++++++++++++++--- 7 files changed, 450 insertions(+), 145 deletions(-) diff --git a/changelog.d/request-only-rows-stay.changed.md b/changelog.d/request-only-rows-stay.changed.md index 5365f9f38..a57eaceb2 100644 --- a/changelog.d/request-only-rows-stay.changed.md +++ b/changelog.d/request-only-rows-stay.changed.md @@ -19,16 +19,22 @@ `[Continue]`, and later compiles put the separator before a reply whose request never held it. - When a compile ends on the agent's own message and the request gets a - trailing `[Continue]` turn, every row the reply's turn stores now records - that prompt (`metadata.promptedBy`, a `{ turn, text }` pair), and every - later request renders `[Continue]` again just before the first of them - that survives, the way a silent tick's separator is rendered. A restart or - retry of the turn keeps the prompt only while its own request renders it. - Without it, later requests merged the reply into the previous assistant - message (one message carrying signed thinking from two responses, a 400 - when it is the latest), and dropped a turn the reply's request held, which - a provider that binds thinking refuses. A reply with neither a tool call - nor signed thinking can't be found in a compiled window and merges as - before. `StartStreamResult.continuePrompt` and - `InferenceRequest.continuePrompt` carry it (new, optional); the agent and - the framework set them. + trailing `[Continue]` turn, the reply's rows (its assistant responses and + tool results) now record that prompt (`metadata.promptedBy`, a + `{ turn, text }` pair), and every later request renders `[Continue]` + again just before the first of its assistant rows that survives, the way a + silent tick's separator is rendered. A later reply follows the same prompt + when what it is sent ends on the tool results of a reply that did: a next + round of the stream, a budget restart, an error retry, or a new process + reading the stored rows. A reply sent anything else after those results, + such as a message that arrived at the tool boundary or during a provider + cooldown, follows that instead and records no prompt. Without the turn, + later requests merged the reply into the previous assistant message (one + message carrying signed thinking from two responses, a 400 when it is the + latest), and dropped a turn the reply's request held, which a provider + that binds thinking refuses. A reply with neither a tool call nor signed or + redacted thinking can't be found in a compiled window and merges as + before. `StartStreamResult.continuePrompt` carries the prompt a stream's + request ends on (new, optional); the agent sets it. The direct + `runInference` paths compile without these request-only turns, as they + did without the tick separator, so their replies render none. diff --git a/changelog.d/silent-heartbeat-turn-rows.fixed.md b/changelog.d/silent-heartbeat-turn-rows.fixed.md index f56239fa8..7c294acae 100644 --- a/changelog.d/silent-heartbeat-turn-rows.fixed.md +++ b/changelog.d/silent-heartbeat-turn-rows.fixed.md @@ -1,14 +1,18 @@ - Silent heartbeat ticks no longer merge into the previous assistant message - on the wire. Every row a tick's turn stores (assistant responses, - tool_result rows) now carries - `metadata.silentHeartbeat: { eventId, serverId, agentName }`, and request - builds render a request-only `[heartbeat tick]` user turn immediately - before the first of the tick's rows that survives compilation (the agent's - own opening request for a tick ends on that same turn instead of - `[Continue]`). Nothing new is stored, and the separator's position follows - from the stamped rows alone, so it is identical on every compile and keeps - the cached prefix stable. Rows are recognised by tool_use ids and thinking - signatures only: an unsigned, tool-free tick reply still merges as before + on the wire. The rows a tick's reply stores (assistant responses, + tool_result rows) now carry + `metadata.silentHeartbeat: { eventId, serverId, agentName, separator }`, + and request builds render a request-only separator user turn, carrying + the tick's instruction, immediately before the first of the tick's rows + that survives compilation (the agent's own opening request for a tick + ends on that same turn instead of `[Continue]`). The separator's text is + the one the tick's request rendered, recorded in the stamp; rows stamped + without it render the bare `[heartbeat tick]` their requests carried. + Nothing new is stored, and the separator's position follows from the + stamped rows alone, so it is identical on every compile and keeps the + cached prefix stable. A tick is told apart by its server and event id + together. Rows are recognised by tool_use ids and signed or redacted + thinking only: an unsigned, tool-free tick reply still merges as before (no signed thinking, so no provider rejection). Previously back-to-back ticks, or a tick after a reply that left no delivery receipt, were sent as one assistant message carrying signed thinking from two responses. diff --git a/src/agent.ts b/src/agent.ts index 24200ba12..2348fb520 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -9,6 +9,8 @@ import { indexRequestOnlyRows, renderRequestOnlyRows, silentHeartbeatSeparatorTurn, + tickKey, + toolLoopPrompt, type RequestOnlyPrompt, type RequestOnlyRowIndex, type SilentHeartbeatTick, @@ -22,9 +24,6 @@ import { } from './tool-result-history.js'; export interface ActivationRequestOptions { - /** The `[Continue]` prompt this turn's earlier streams replied to, if any: - * kept for this request's reply only while this request renders it. */ - continuePrompt?: RequestOnlyPrompt; /** The silent heartbeat tick this request answers, if any. */ silentHeartbeat?: SilentHeartbeatTick; } @@ -33,10 +32,11 @@ export interface StartStreamResult { stream: YieldingStream; request: NormalizedRequest; /** The request-only `[Continue]` prompt this request's reply follows, if - * any: a new one when the request ends on `[Continue]`, else the turn's - * prompt (`ActivationRequestOptions.continuePrompt`) only if this request - * renders it. The framework stamps the reply's rows with it - * (`metadata.promptedBy`), for later compiles to render it before them. */ + * any, read from the request: a new one when it ends on `[Continue]`; + * else, when it ends on tool results, the prompt of the reply that called + * those tools (`toolLoopPrompt`); else none. The framework stamps the + * reply's rows with it (`metadata.promptedBy`), for later compiles to + * render it before them. */ continuePrompt?: RequestOnlyPrompt; takeKvSubmission?: () => { submissionId: string; wireReceipt: CacheWireReceipt } | undefined; drainKvSubmissionIds?: () => string[]; @@ -876,19 +876,22 @@ export class Agent { // that same turn, so the prefix it caches is the one every later compile // renders. "First" is per agent: a broadcast tick shares its eventId with // every resident. - const rendered = renderRequestOnlyRows(messages, this.indexRequestOnlyRows(), this.name); + const index = this.indexRequestOnlyRows(); + const rendered = renderRequestOnlyRows(messages, index, this.name); messages = rendered.messages; // A tick opens wherever its separator isn't already rendered: on its first // request, and on a restart whose compile left none of the tick's rows, // so the restart's reply follows the separator, where every later compile // renders it. const openingTick = options.silentHeartbeat !== undefined - && !rendered.separatedTicks.has(options.silentHeartbeat.eventId); - - // The [Continue] prompt this request's reply follows: a new one when it - // ends on [Continue]; else the turn's prompt only where this request - // renders it (a stream resumed after a cooldown, say, whose compile now - // ends on a message that arrived meanwhile, renders none); else none. + && !rendered.separatedTicks.has(tickKey(options.silentHeartbeat)); + + // The [Continue] prompt this request's reply follows, read from the + // request itself: a new one when it ends on [Continue]; else, when it + // ends on tool results, the prompt of the reply that called those tools, + // whose tool loop this reply continues; else none. A request that ends + // on anything else (a message that arrived during a cooldown, say) + // prompts a reply that follows that. let continuePrompt: RequestOnlyPrompt | undefined; if (openingTick) { messages = [...messages, silentHeartbeatSeparatorTurn()]; @@ -898,9 +901,8 @@ export class Agent { // and after context compression a stale assistant turn can end up last. messages = [...messages, continueTurn()]; continuePrompt = { turn: randomUUID(), text: CONTINUE_ROW }; - } - if (!continuePrompt && options.continuePrompt && rendered.promptedTurns.has(options.continuePrompt.turn)) { - continuePrompt = options.continuePrompt; + } else { + continuePrompt = toolLoopPrompt(messages.at(-1), index, this.name); } const request: NormalizedRequest = { diff --git a/src/framework.ts b/src/framework.ts index 8c9a0d891..43a60c9c2 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -8,7 +8,7 @@ import type { Membrane, ContentBlock, NormalizedRequest, YieldingStream, ToolRes import { MembraneError } from '@animalabs/membrane'; import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; -import { SILENT_HEARTBEAT_SEPARATOR, type SilentHeartbeatStamp } from './silent-heartbeat.js'; +import { SILENT_HEARTBEAT_SEPARATOR, type RequestOnlyPrompt, type RequestOnlyStamps } from './silent-heartbeat.js'; import { SUBCONSCIOUS_TOOLS, SUBCONSCIOUS_TOOL_NAMES, type SubconsciousConfig } from './tune-out/tools.js'; import { TuneOutCoordinator, TUNE_OUT_DEFAULTS } from './tune-out/coordinator.js'; import type { @@ -1241,6 +1241,16 @@ export class AgentFramework { */ private activeTurnTriggers: Map = new Map(); + /** + * The `[Continue]` prompt the current round of each turn follows, keyed by + * the turn's trigger. Each stream start sets it from what the Agent read in + * its request (`StartStreamResult.continuePrompt`), and a tool boundary + * that stores anything after the results clears it, since the next round + * follows those messages. `silentTurnRowMetadata` stamps the round's rows + * with it. + */ + private readonly turnPrompts = new WeakMap(); + /** The trigger of `agentName`'s turn in progress, if a turn is running. */ getActiveTurnTrigger(agentName: string): InferenceRequest | undefined { return this.activeTurnTriggers.get(agentName); @@ -6698,6 +6708,10 @@ export class AgentFramework { { const deferred = this.drainDeferredFor(agent.name); if (deferred.length > 0) { + // The next round follows these messages, not the [Continue] + // the turn's earlier rounds followed: its rows carry no prompt. + const turnTrigger = this.activeTurnTriggers.get(agent.name); + if (turnTrigger) this.turnPrompts.delete(turnTrigger); for (const msg of deferred) { agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); // Injection guards: tool blocks would corrupt the tool-cycle @@ -6905,7 +6919,6 @@ export class AgentFramework { reason: 'context_budget_restart', source: 'framework', timestamp: Date.now(), - continuePrompt: this.activeTurnTriggers.get(agent.name)?.continuePrompt, suppressProse: this.activeTurnTriggers.get(agent.name)?.suppressProse, silentHeartbeat: this.activeTurnTriggers.get(agent.name)?.silentHeartbeat, }); @@ -6925,12 +6938,16 @@ export class AgentFramework { ...(midTurnInjections.length > 0 ? { injectedMessages: midTurnInjections.length } : {}), }); } else { - // Non-streaming fallback: schedule re-inference + // Non-streaming fallback: schedule re-inference. It continues the + // same turn, so it carries the turn's properties, as the budget + // restart does. this.pendingRequests.push({ agentName: agent.name, reason: 'tool_results_ready', source: 'framework', timestamp: Date.now(), + suppressProse: this.activeTurnTriggers.get(agent.name)?.suppressProse, + silentHeartbeat: this.activeTurnTriggers.get(agent.name)?.silentHeartbeat, }); } } @@ -8515,32 +8532,39 @@ export class AgentFramework { } /** - * Metadata for rows the agent's current turn stores (assistant responses, - * tool_result rows), for the request-only turns that came before them: - * - a silent heartbeat tick stamps each row, with the separator text its - * request rendered, so they stay identifiable after the turn — request - * builds render the separator before the first that survives, and a later - * retention policy can find or collapse them. The stamp names the agent: - * residents share one message slot and a broadcast tick reaches each of - * them with the same eventId; - * - a turn whose stream opened on `[Continue]` stamps each row from then on - * with that prompt (`promptedBy`), so request builds render it before the - * first that survives. + * Stamps for the rows the agent's reply stores in its current turn: its + * assistant responses, its tool_result rows (through `storeResults`), and + * the contained-wrapper row that stands in for a response. The turn's other + * stores (a delivery receipt, a routing notice, a deferred flush) aren't the + * reply's, and carry none. + * - A silent heartbeat tick stamps each of them for the whole turn, with the + * separator text its request rendered. Request builds render the + * separator before the first of the tick's assistant rows that survives, + * and the stamps keep every row of the tick identifiable after the turn, + * for a later retention policy to find or collapse. The stamp names the + * agent: residents share one message slot, and a broadcast tick reaches + * each of them with the same eventId. + * - A `[Continue]` prompt stamps the rows of each round that follows it + * (`turnPrompts`). Request builds render it before the first of its + * assistant rows that survives. On the round's tool_result rows it marks + * them as the reply's, as the tick stamp does. * Undefined for ordinary turns, whose rows are stored exactly as before. * (Named for its first use, the silent tick.) */ - private silentTurnRowMetadata(agentName: string): MessageMetadata | undefined { + private silentTurnRowMetadata(agentName: string): RequestOnlyStamps | undefined { const trigger = this.activeTurnTriggers.get(agentName); - const tick = trigger?.silentHeartbeat; - const stamp: SilentHeartbeatStamp | undefined = tick - ? { eventId: tick.eventId, serverId: tick.serverId, agentName, separator: SILENT_HEARTBEAT_SEPARATOR } - : undefined; - const prompt = trigger?.continuePrompt; - if (!stamp && !prompt) return undefined; - return { - ...(stamp ? { silentHeartbeat: stamp } : {}), - ...(prompt ? { promptedBy: prompt } : {}), - } as MessageMetadata; + if (!trigger) return undefined; + const tick = trigger.silentHeartbeat; + const prompt = this.turnPrompts.get(trigger); + if (!tick && !prompt) return undefined; + const stamps: RequestOnlyStamps = {}; + if (tick) { + stamps.silentHeartbeat = { + eventId: tick.eventId, serverId: tick.serverId, agentName, separator: SILENT_HEARTBEAT_SEPARATOR, + }; + } + if (prompt) stamps.promptedBy = prompt; + return stamps; } /** Record prose segments suppressed by explicit-send silencing. */ @@ -9376,13 +9400,15 @@ export class AgentFramework { continuePrompt, } = await agent.startStreamWithInjections(tools, injections, undefined, compressionTools, { silentHeartbeat: trigger?.silentHeartbeat, - continuePrompt: trigger?.continuePrompt, }); - // The [Continue] prompt this stream's reply follows, if any: every row - // the turn stores from here on carries it (silentTurnRowMetadata), and the - // turn's restarts and retries carry it in the trigger to the next - // stream start, which keeps it only if its request renders it. - if (trigger) trigger.continuePrompt = continuePrompt; + // The [Continue] prompt this stream's first round follows, if any, as + // the Agent read it from the request: the round's rows carry it + // (silentTurnRowMetadata), and so do later rounds' until a tool + // boundary stores something after the results. + if (trigger) { + if (continuePrompt) this.turnPrompts.set(trigger, continuePrompt); + else this.turnPrompts.delete(trigger); + } if (this.agents.get(agent.name) !== agent) { stream.cancel(); agent.cancelStream(); diff --git a/src/silent-heartbeat.ts b/src/silent-heartbeat.ts index 1a2211c31..d819adb08 100644 --- a/src/silent-heartbeat.ts +++ b/src/silent-heartbeat.ts @@ -12,9 +12,14 @@ * first of the tick's rows that survives compilation. * - A compile that ends on the agent's own message gets a trailing * `[Continue]` turn (some models reject a trailing assistant message). The - * reply it prompts is stored straight after that message, so every row the - * reply's turn stores is stamped `metadata.promptedBy` (`RequestOnlyPrompt`), - * and the turn is rendered before the first of them that survives. + * reply it prompts is stored straight after that message, so the reply's + * rows are stamped `metadata.promptedBy` (`RequestOnlyPrompt`), and the turn + * is rendered before the first of them that survives. A reply follows the + * prompt when what it was sent ends on that `[Continue]`, or on the tool + * results of a reply that followed it (`toolLoopPrompt`): a stream start + * and each round of a stream alike. A reply sent anything else after the + * results (a message that arrived meanwhile) follows that, and carries no + * prompt. * * Without the row, the wire formatter merges consecutive same-role messages: * the reply would read as an unprompted continuation of the earlier response, @@ -81,6 +86,12 @@ export interface SilentHeartbeatStamp extends SilentHeartbeatTick { separator?: string; } +/** One tick's key in an index or a rendered window: the server and its event + * id together, since two servers' events can share an id. */ +export function tickKey(tick: SilentHeartbeatTick): string { + return JSON.stringify([tick.serverId, tick.eventId]); +} + /** The tick stamp on a stored row's metadata, if any. */ export function silentHeartbeatOf(metadata: unknown): SilentHeartbeatStamp | undefined { const stamp = (metadata as { silentHeartbeat?: unknown } | undefined)?.silentHeartbeat; @@ -92,13 +103,22 @@ export function silentHeartbeatOf(metadata: unknown): SilentHeartbeatStamp | und } /** A request-only turn a request ended on before the agent's reply, as - * `metadata.promptedBy` on every row the reply's turn stores: `turn` names - * the one prompt, and `text` is what the request rendered. */ + * `metadata.promptedBy` on the rows of each reply that follows it: `turn` + * names the one prompt, and `text` is what the request rendered. */ export interface RequestOnlyPrompt { turn: string; text: string; } +/** The stamps on a row a turn stores after a request-only turn + * (`silentHeartbeatOf`, `promptedByOf` read them back). A type alias, so a + * value fits `MessageMetadata`'s index signature without a cast, and a + * misspelled key fails to compile. */ +export type RequestOnlyStamps = { + silentHeartbeat?: SilentHeartbeatStamp; + promptedBy?: RequestOnlyPrompt; +}; + /** The request-only prompt a stored row replies to, if any. */ export function promptedByOf(metadata: unknown): RequestOnlyPrompt | undefined { const prompt = (metadata as { promptedBy?: unknown } | undefined)?.promptedBy; @@ -129,7 +149,7 @@ function requestOnlyTurn(text: string): NormalizedMessage { function identityKeys(content: ContentBlock[]): string[] { const keys: string[] = []; for (const block of content) { - if (block.type === 'tool_use' && typeof block.id === 'string' && block.id) keys.push(`tool_use:${block.id}`); + if (block.type === 'tool_use' && typeof block.id === 'string' && block.id) keys.push(toolUseKey(block.id)); else if (block.type === 'thinking' && typeof block.signature === 'string' && block.signature) { keys.push(`thinking:${block.signature}`); } else if (block.type === 'redacted_thinking' && typeof block.data === 'string' && block.data) { @@ -139,6 +159,10 @@ function identityKeys(content: ContentBlock[]): string[] { return keys; } +function toolUseKey(id: string): string { + return `tool_use:${id}`; +} + export interface StoredRowLike { participant: string; content: ContentBlock[]; @@ -147,9 +171,9 @@ export interface StoredRowLike { export interface RequestOnlyRowIndex { /** Identity key of every assistant row this agent stored during a silent - * tick → the tick, and the separator text to render before its first - * surviving row. */ - ticks: Map; + * tick → the tick (`tickKey`), and the separator text to render before + * its first surviving row. */ + ticks: Map; /** Identity key of every assistant row this agent stored in reply to a * request-only prompt → that prompt. */ prompts: Map; @@ -157,33 +181,53 @@ export interface RequestOnlyRowIndex { /** Index the agent's own rows that follow request-only turns: the rows of its * silent ticks, and the rows of each reply to a `[Continue]`. The two are - * independent, so a row can be both. */ + * independent, so a row can be both. Only stamped rows are keyed, so an + * agent's history walk costs a metadata read per row. */ export function indexRequestOnlyRows(stored: readonly StoredRowLike[], agentName: string): RequestOnlyRowIndex { - const ticks = new Map(); + const ticks = new Map(); const prompts = new Map(); for (const row of stored) { if (row.participant !== agentName) continue; - const keys = identityKeys(row.content); - if (keys.length === 0) continue; const stamp = silentHeartbeatOf(row.metadata); - if (stamp && stamp.agentName === agentName) { - const tick = { eventId: stamp.eventId, text: stamp.separator ?? LEGACY_SILENT_HEARTBEAT_SEPARATOR }; + const ownTick = stamp !== undefined && stamp.agentName === agentName; + const prompt = promptedByOf(row.metadata); + if (!ownTick && !prompt) continue; + const keys = identityKeys(row.content); + if (ownTick) { + const tick = { tick: tickKey(stamp), text: stamp.separator ?? LEGACY_SILENT_HEARTBEAT_SEPARATOR }; for (const key of keys) ticks.set(key, tick); } - const prompt = promptedByOf(row.metadata); if (prompt) for (const key of keys) prompts.set(key, prompt); } return { ticks, prompts }; } +/** + * The prompt a reply continues when what it is sent ends on `last`: if + * `last` holds tool results, the prompt of the reply that called those tools + * (this reply continues that reply's tool loop); otherwise none. Read from + * the stored stamps, so it doesn't depend on anything held in memory. A + * request that ends on its own `[Continue]` mints a new prompt instead. + */ +export function toolLoopPrompt( + last: NormalizedMessage | undefined, + index: RequestOnlyRowIndex, + agentName: string, +): RequestOnlyPrompt | undefined { + if (!last || last.participant === agentName) return undefined; + for (const block of last.content) { + if (block.type !== 'tool_result') continue; + const prompt = index.prompts.get(toolUseKey(block.toolUseId)); + if (prompt) return prompt; + } + return undefined; +} + export interface RenderedRequestOnlyRows { messages: NormalizedMessage[]; - /** The ticks whose separator this window renders. A tick none of whose - * rows survive compilation renders none. */ + /** The ticks (`tickKey`) whose separator this window renders. A tick none + * of whose rows survive compilation renders none. */ separatedTicks: Set; - /** The prompts (`RequestOnlyPrompt.turn`) this window renders, the same - * way. */ - promptedTurns: Set; } /** @@ -201,14 +245,14 @@ export function renderRequestOnlyRows( ): RenderedRequestOnlyRows { const separatedTicks = new Set(); const promptedTurns = new Set(); - if (index.ticks.size === 0 && index.prompts.size === 0) return { messages, separatedTicks, promptedTurns }; + if (index.ticks.size === 0 && index.prompts.size === 0) return { messages, separatedTicks }; const out: NormalizedMessage[] = []; for (const message of messages) { if (message.participant === agentName) { const keys = identityKeys(message.content); const tick = keys.map((key) => index.ticks.get(key)).find((t) => t !== undefined); - if (tick !== undefined && !separatedTicks.has(tick.eventId)) { - separatedTicks.add(tick.eventId); + if (tick !== undefined && !separatedTicks.has(tick.tick)) { + separatedTicks.add(tick.tick); out.push(requestOnlyTurn(tick.text)); } const prompt = keys.map((key) => index.prompts.get(key)).find((p) => p !== undefined); @@ -219,5 +263,5 @@ export function renderRequestOnlyRows( } out.push(message); } - return { messages: out, separatedTicks, promptedTurns }; + return { messages: out, separatedTicks }; } diff --git a/src/types/agent.ts b/src/types/agent.ts index a4fc4b9e1..267e41ecc 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -1,7 +1,7 @@ import type { ContentBlock, YieldingStream } from '@animalabs/membrane'; import type { ContextStrategy } from '@animalabs/context-manager'; import type { ToolCallId, ToolResult, ToolCall } from './events.js'; -import type { RequestOnlyPrompt, SilentHeartbeatTick } from '../silent-heartbeat.js'; +import type { SilentHeartbeatTick } from '../silent-heartbeat.js'; export type SameRoundThinkTextPolicy = 'public' | 'private'; export type SameRoundThinkTextPolicySource = @@ -360,16 +360,10 @@ export interface InferenceRequest { /** Suppress every automatic plain-prose delivery for this logical turn. * Explicit tool calls remain available. Used by authenticated silent wakes. */ suppressProse?: boolean; - /** Set by the framework at each stream start of this turn: the request-only - * `[Continue]` prompt the stream's reply follows (`StartStreamResult. - * continuePrompt`), or none. Every row the turn stores is stamped - * `metadata.promptedBy` with it, so request builds render the turn before - * the first that survives. Carried into the turn's restarts and retries, - * whose stream start keeps it only if its own request renders it. */ - continuePrompt?: RequestOnlyPrompt; - /** The authenticated silent heartbeat tick this turn answers. Every row the - * turn stores is stamped `metadata.silentHeartbeat` with it, and request - * builds render a request-only separator carrying the tick's instruction - * before the tick's first row (`SILENT_HEARTBEAT_SEPARATOR`). */ + /** The authenticated silent heartbeat tick this turn answers. The rows the + * turn's reply stores (its assistant responses and tool results) are + * stamped `metadata.silentHeartbeat` with it, and request builds render a + * request-only separator carrying the tick's instruction before the tick's + * first surviving row (`SILENT_HEARTBEAT_SEPARATOR`). */ silentHeartbeat?: SilentHeartbeatTick; } diff --git a/test/silent-heartbeat-rows.test.ts b/test/silent-heartbeat-rows.test.ts index a0fd91fae..ecfaccff1 100644 --- a/test/silent-heartbeat-rows.test.ts +++ b/test/silent-heartbeat-rows.test.ts @@ -16,24 +16,38 @@ import { mkdtempSync, rmSync } from 'node:fs'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; import { AgentFramework, SILENT_HEARTBEAT_SEPARATOR } from '../src/index.js'; -import { indexRequestOnlyRows, renderRequestOnlyRows } from '../src/silent-heartbeat.js'; +import { indexRequestOnlyRows, renderRequestOnlyRows, tickKey } from '../src/silent-heartbeat.js'; import type { AgentConfig, Module, ModuleContext, ToolCall, ToolDefinition, ToolResult } from '../src/index.js'; import { TOOL_RESULT_GUARD_NOTICE } from '../src/tool-result-guard.js'; import { PassthroughStrategy } from '@animalabs/context-manager'; import { MockMembrane, MockYieldingStream, createMockResponse } from './helpers/mock-membrane.js'; -import { NativeFormatter } from '@animalabs/membrane'; -import type { ContentBlock, NormalizedMessage, NormalizedRequest, NormalizedResponse, YieldingStream } from '@animalabs/membrane'; +import { MembraneError, NativeFormatter } from '@animalabs/membrane'; +import type { ContentBlock, NormalizedMessage, NormalizedRequest, NormalizedResponse, StreamEvent, YieldingStream } from '@animalabs/membrane'; class ToolModule implements Module { readonly name = 'tools'; calls: ToolCall[] = []; + /** A message someone sends while the next call runs: the agent's turn is + * alive, so it is deferred to the tool boundary. */ + arrive: string | null = null; + /** The next call's result ends the turn. */ + endTurn = false; private ctx: ModuleContext | null = null; async start(ctx: ModuleContext) { this.ctx = ctx; ctx.registerSpeechHandler('*'); } async stop() { this.ctx?.unregisterSpeechHandler(); this.ctx = null; } async onProcess() { return {}; } async onAgentSpeech() {} getTools(): ToolDefinition[] { return [{ name: 'ping', description: 'explicit action', inputSchema: { type: 'object', properties: {} } }]; } - async handleToolCall(call: ToolCall): Promise { this.calls.push(call); return { success: true, data: { ok: true } }; } + async handleToolCall(call: ToolCall): Promise { + this.calls.push(call); + if (this.arrive !== null) { + this.ctx!.addMessage('nissa', [{ type: 'text', text: this.arrive }] as ContentBlock[]); + this.arrive = null; + } + const endTurn = this.endTurn; + this.endTurn = false; + return { success: true, data: { ok: true }, ...(endTurn ? { endTurn } : {}) }; + } } let seq = 0; @@ -65,17 +79,55 @@ function signed(signature: string, text: string, extra: ContentBlock[] = []): Co ]; } -/** One explicit response queue per stream, so a restart opens a fresh one. */ +/** A stream the provider fails at once. */ +class ErrorStream implements YieldingStream { + isWaitingForTools = false; + pendingToolCallIds: string[] = []; + toolDepth = 0; + constructor(private readonly error: Error) {} + provideToolResults(): void { throw new Error('not waiting'); } + cancel(): void {} + async *[Symbol.asyncIterator](): AsyncIterator { + yield { type: 'error', error: this.error } as StreamEvent; + } +} + +/** One explicit queue per stream, so a restart opens a fresh one: the + * responses of its tool rounds, or an error the stream fails with. */ class QueuedStreamsMembrane extends MockMembrane { - queues: NormalizedResponse[][] = []; + queues: Array = []; override streamYielding(request: NormalizedRequest, _options?: unknown): YieldingStream { this.calls.push(request); - const stream = new MockYieldingStream(this.queues.shift() ?? []); + const next = this.queues.shift() ?? []; + if (next instanceof Error) return new ErrorStream(next); + const stream = new MockYieldingStream(next); this.lastStream = stream; return stream; } } +/** A 429 that holds the turn for a provider cooldown. */ +const accelerationLimit = () => new MembraneError({ + type: 'rate_limit', retryable: true, httpStatus: 429, + message: "This request would exceed your organization's maximum usage increase rate for input tokens per minute", + rawError: { status: 429 }, +}); + +/** A failure the default error policy retries at once. */ +const transientFailure = () => new MembraneError({ + type: 'server', retryable: true, httpStatus: 500, retryAfterMs: 1, + message: 'internal server error', rawError: { status: 500 }, +}); + +/** A tool round whose usage puts the next round past the physical window, so + * the stream restarts at its boundary. */ +function nearCapToolRound(signature: string, text: string, id: string): NormalizedResponse { + const response = createMockResponse( + signed(signature, text, [{ type: 'tool_use', id, name: 'tools--ping', input: {} } as ContentBlock]), 'tool_use'); + (response as { usage: unknown }).usage = { inputTokens: 5_000, outputTokens: 5, cacheReadTokens: 195_000 }; + return response; +} + /** A strategy that leaves every tool exchange out of the compiled window * (from the moment `omit` is set, for a window that folds them later). */ class OmitToolExchange extends PassthroughStrategy { @@ -94,8 +146,12 @@ const refused = () => ({ async function make(opts: { deliver?: boolean; membrane?: MockMembrane; physicalWindowTokens?: number; agents?: string[]; agent?: Partial; + /** An earlier framework's directory, to open its store again. */ + dir?: string; + /** Where the agent's plain speech lands; null for no channel. */ + locus?: string | null; } = {}) { - const dir = mkdtempSync(join(tmpdir(), 'silent-hb-rows-')); + const dir = opts.dir ?? mkdtempSync(join(tmpdir(), 'silent-hb-rows-')); const membrane = opts.membrane ?? new MockMembrane(); const mod = new ToolModule(); const framework = await AgentFramework.create({ @@ -108,7 +164,8 @@ async function make(opts: { modules: [mod], }); (framework as any).channelRegistry = new Proxy({ - resolveLocus: () => 'world:commons', getDefaultPublishChannel: () => 'world:commons', + resolveLocus: () => (opts.locus === undefined ? 'world:commons' : opts.locus), + getDefaultPublishChannel: () => (opts.locus === undefined ? 'world:commons' : opts.locus), routeSpeech: async () => ({ delivered: opts.deliver ?? true, channelId: 'world:commons' }), startTyping: () => {}, stopTyping: () => {}, getChannelTools: () => [], getDescriptor: () => undefined, @@ -152,6 +209,14 @@ function stored(framework: AgentFramework, agent = 'assistant') { }>; } +function rowWith(framework: AgentFramework, text: string) { + const row = stored(framework).find((m) => m.content.some((b) => (b as { text?: string }).text === text)); + assert.ok(row, `a stored row reads ${JSON.stringify(text)}`); + return row; +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + describe('silent heartbeat stored rows', () => { it('keeps a tick reply out of the previous assistant message on the wire', async () => { const x = await make(); @@ -567,7 +632,26 @@ describe('replies to a [Continue] turn', () => { assert.deepEqual(rendered.messages.map(textOf), [ '[heartbeat tick]', 'old tick reply', 'recorded separator', 'new tick reply', 'recorded prompt', 'continued reply', ]); - assert.deepEqual([...rendered.separatedTicks].sort(), ['hb-new', 'hb-old']); + assert.deepEqual([...rendered.separatedTicks].sort(), [ + tickKey({ serverId: 'heartbeat', eventId: 'hb-new' }), tickKey({ serverId: 'heartbeat', eventId: 'hb-old' }), + ].sort()); + }); + + it('keys a tick by its server and its event id together', () => { + // Two servers' events can share an id: each tick still gets its own + // separator, before its own first row. + const a = { eventId: 'hb-1', serverId: 'heartbeat-a', agentName: 'assistant', separator: 'separator a' }; + const b = { eventId: 'hb-1', serverId: 'heartbeat-b', agentName: 'assistant', separator: 'separator b' }; + const stored = [ + { participant: 'assistant', content: signed('sig-a', 'tick a reply'), metadata: { silentHeartbeat: a } }, + { participant: 'nissa', content: [{ type: 'text', text: 'between the ticks' }] as ContentBlock[] }, + { participant: 'assistant', content: signed('sig-b', 'tick b reply'), metadata: { silentHeartbeat: b } }, + ]; + const index = indexRequestOnlyRows(stored, 'assistant'); + const rendered = renderRequestOnlyRows(stored.map(({ participant, content }) => ({ participant, content })), index, 'assistant'); + assert.deepEqual(rendered.messages.map(textOf), + ['separator a', 'tick a reply', 'between the ticks', 'separator b', 'tick b reply']); + assert.deepEqual([...rendered.separatedTicks].sort(), [tickKey(a), tickKey(b)].sort()); }); it('marks every row the reply\'s turn stores, and renders one [Continue] before the first', async () => { @@ -594,7 +678,7 @@ describe('replies to a [Continue] turn', () => { } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); - it('carries the prompt into the reply\'s budget restart, for a window that later folds the first round', async () => { + it('a budget restart that ends on the round\'s tool results reads its prompt, for a window that later folds the round', async () => { const membrane = new QueuedStreamsMembrane(); const strategy = new OmitToolExchange(); strategy.omit = false; @@ -602,13 +686,10 @@ describe('replies to a [Continue] turn', () => { try { membrane.queues.push([createMockResponse(signed('sig-1', 'first reply'))]); await wake(x, ordinaryEvent('hello')); - // The reply to [Continue] restarts at its tool boundary; the restart's - // compile ends on the tool_result, so it opens on no prompt of its own. - const nearCap = createMockResponse( - signed('sig-c', 'checking', [{ type: 'tool_use', id: 'c-t1', name: 'tools--ping', input: {} } as ContentBlock]), - 'tool_use'); - (nearCap as { usage: unknown }).usage = { inputTokens: 5_000, outputTokens: 5, cacheReadTokens: 195_000 }; - membrane.queues.push([nearCap], [createMockResponse(signed('sig-c2', 'all done'))]); + // The reply to [Continue] restarts at its tool boundary. The restart's + // compile ends on the round's tool_result, so its reply continues that + // round's tool loop and follows the same prompt. + membrane.queues.push([nearCapToolRound('sig-c', 'checking', 'c-t1')], [createMockResponse(signed('sig-c2', 'all done'))]); (x.framework as any).pendingRequests.push({ agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now() }); await x.framework.runUntilIdle(); assert.equal(membrane.calls.length, 3, 'first, the continued reply, its restart'); @@ -630,31 +711,179 @@ describe('replies to a [Continue] turn', () => { } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); - it('drops a carried prompt its request doesn\'t render (a stream resumed after a cooldown)', async () => { - // The shape a provider cooldown leaves (Basil-1007's probe): the held - // trigger carries the prompt its first stream start set, and a message - // lands meanwhile, so the resumed compile ends on that message and - // renders no [Continue]. - const x = await make({ deliver: false }); + it('a reply resumed after a provider cooldown follows the message that arrived during it, not [Continue]', async () => { + // A reply to [Continue] stores a tool round and restarts at its boundary, + // and the restart meets a 429 that holds the turn. A message arrives + // during the cooldown, so the resumed compile ends on it, while the + // round, still in the window, renders [Continue] earlier. + const membrane = new QueuedStreamsMembrane(); + const strategy = new OmitToolExchange(); + strategy.omit = false; + const x = await make({ membrane, physicalWindowTokens: 200_000, deliver: false, agent: { strategy } }); + const internal = x.framework as unknown as { + providerAccelerationDefaultCooldownMs: number; providerAccelerationJitterMs: number; + providerAccelerationCooldowns: Map; + }; + internal.providerAccelerationDefaultCooldownMs = 1_000; // the shortest hold there is + internal.providerAccelerationJitterMs = 0; + try { + membrane.queues.push([createMockResponse(signed('sig-1', 'first reply'))]); + await wake(x, ordinaryEvent('hello')); + membrane.queues.push([nearCapToolRound('sig-c', 'checking', 'c-t1')], accelerationLimit(), + [createMockResponse(signed('sig-r', 'resumed reply'))]); + (x.framework as any).pendingRequests.push({ agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now() }); + await x.framework.runUntilIdle(); + assert.equal(internal.providerAccelerationCooldowns.size, 1, 'the restart is held'); + await wake(x, ordinaryEvent('arrived during the cooldown')); + await sleep(1_100); + await x.framework.runUntilIdle(); + assert.equal(membrane.calls.length, 4, 'first, the continued reply, its held restart, the resumed attempt'); + + const resumed = membrane.calls[3]!.messages; + assert.equal(textOf(resumed.at(-1)!), 'arrived during the cooldown'); + assert.ok(resumed.some((m) => textOf(m) === '[Continue]'), 'the round renders its [Continue]'); + assert.equal(rowWith(x.framework, 'resumed reply').metadata?.promptedBy, undefined); + + // Later, the window folds the round: no [Continue] is left to render. + strategy.omit = true; + membrane.queues.push([createMockResponse(signed('sig-later', 'later reply'))]); + await wake(x, ordinaryEvent('later')); + const later = membrane.calls[4]!.messages; + const replyAt = later.findIndex((m) => textOf(m) === 'resumed reply'); + assert.equal(textOf(later[replyAt - 1]!), 'arrived during the cooldown'); + assert.ok(!later.some((m) => textOf(m) === '[Continue]')); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('a later round that follows a message injected at its tool boundary follows the message, not [Continue]', async () => { + // Hear-while-acting: a message that arrives while the round's tool runs + // is stored after the tool results and handed to the live stream with + // them, so the stream's next round follows it. + const strategy = new OmitToolExchange(); + strategy.omit = false; + const x = await make({ deliver: false, agent: { strategy } }); try { await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); - x.framework.getAgent('assistant')!.getContextManager() - .addMessage('nissa', [{ type: 'text', text: 'arrived during the cooldown' }] as ContentBlock[]); - x.membrane.pushResponse(createMockResponse(signed('sig-r', 'resumed reply'), 'end_turn')); - (x.framework as any).pendingRequests.push({ - agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now(), - continuePrompt: { turn: 'stale-p1', text: '[Continue]' }, - }); + x.mod.arrive = 'arrived mid-turn'; + await timerWake(x, + signed('sig-c', 'checking', [{ type: 'tool_use', id: 'c-t1', name: 'tools--ping', input: {} } as ContentBlock]), + signed('sig-c2', 'all done')); + const injected = x.membrane.lastStream!.receivedToolResultOptions[0]?.injectedMessages ?? []; + assert.deepEqual(injected.map((m) => textOf(m as NormalizedMessage)), ['arrived mid-turn']); + + const marked = stored(x.framework).filter((m) => m.metadata?.promptedBy !== undefined); + assert.deepEqual(marked.map((m) => m.participant), ['assistant', 'user'], + 'the round that followed [Continue], and its tool_result row'); + assert.equal(rowWith(x.framework, 'all done').metadata?.promptedBy, undefined); + + // Later, the window folds the round: the reply follows the message. + strategy.omit = true; + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + const later = x.membrane.calls[2]!.messages; + const doneAt = later.findIndex((m) => textOf(m) === 'all done'); + assert.equal(textOf(later[doneAt - 1]!), 'arrived mid-turn'); + assert.ok(!later.some((m) => textOf(m) === '[Continue]')); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('a reply retried after a failure past its first round keeps following the prompt', async () => { + const membrane = new QueuedStreamsMembrane(); + const strategy = new OmitToolExchange(); + strategy.omit = false; + const x = await make({ membrane, physicalWindowTokens: 200_000, deliver: false, agent: { strategy } }); + try { + membrane.queues.push([createMockResponse(signed('sig-1', 'first reply'))]); + await wake(x, ordinaryEvent('hello')); + // The round is stored, its restart fails, and the error policy retries: + // the retry's compile ends on the round's tool results. + membrane.queues.push([nearCapToolRound('sig-c', 'checking', 'c-t1')], transientFailure(), + [createMockResponse(signed('sig-c2', 'all done'))]); + (x.framework as any).pendingRequests.push({ agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now() }); + await x.framework.runUntilIdle(); + await sleep(50); await x.framework.runUntilIdle(); + assert.equal(membrane.calls.length, 4, 'first, the continued reply, its failed restart, the retry'); + assert.ok(membrane.calls[3]!.messages.at(-1)!.content.some((b) => b.type === 'tool_result')); + + const marked = stored(x.framework).filter((m) => m.metadata?.promptedBy !== undefined); + assert.equal(marked.length, 3, 'the round, its result and the retry\'s reply'); + assert.equal(new Set(marked.map((m) => m.metadata!.promptedBy.turn)).size, 1); + + strategy.omit = true; + membrane.queues.push([createMockResponse(signed('sig-later', 'later reply'))]); + await wake(x, ordinaryEvent('later')); + const later = membrane.calls[4]!.messages; + assert.equal(textOf(later[later.findIndex((m) => textOf(m) === 'all done') - 1]!), '[Continue]'); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('a request that ends on the agent\'s own message mints a new [Continue], with an older one rendered before it', async () => { + const x = await make({ deliver: false }); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + await timerWake(x, signed('sig-c1', 'continued once')); + await timerWake(x, signed('sig-c2', 'continued twice')); + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + + const second = x.membrane.calls[2]!.messages; + assert.equal(textOf(second.at(-1)!), '[Continue]'); + assert.equal(second.filter((m) => textOf(m) === '[Continue]').length, 2, 'the first prompt renders before its reply'); + const once = rowWith(x.framework, 'continued once').metadata?.promptedBy; + const twice = rowWith(x.framework, 'continued twice').metadata?.promptedBy; + assert.ok(once && twice); + assert.notEqual(twice.turn, once.turn); + + const later = x.membrane.calls[3]!.messages; + for (const text of ['continued once', 'continued twice']) { + assert.equal(textOf(later[later.findIndex((m) => textOf(m) === text) - 1]!), '[Continue]', text); + } + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('finds a reply whose only identity is a redacted_thinking block', async () => { + const x = await make({ deliver: false }); + try { + await wake(x, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + await timerWake(x, [ + { type: 'redacted_thinking', data: 'opaque-reasoning' } as ContentBlock, + { type: 'text', text: 'redacted reply' } as ContentBlock, + ]); await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); - const minting = x.membrane.calls[1]!.messages; - assert.equal(textOf(minting.at(-1)!), 'arrived during the cooldown'); - const reply = stored(x.framework).find((m) => m.content.some((b) => (b as { text?: string }).text === 'resumed reply'))!; - assert.equal(reply.metadata?.promptedBy, undefined); + const continued = x.membrane.calls[1]!.messages; const later = x.membrane.calls[2]!.messages; - const replyAt = later.findIndex((m) => textOf(m) === 'resumed reply'); - assert.deepEqual(later.slice(0, replyAt), minting, 'what precedes the reply is what its request held'); + assert.equal(textOf(later[later.findIndex((m) => textOf(m) === 'redacted reply') - 1]!), '[Continue]'); + assert.deepEqual(later.slice(0, continued.length), continued); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } + }); + + it('a reply after the process restarts reads its prompt from the stored rows', async () => { + // The first process ends the turn at the round's tool boundary, so its + // store ends on the tool results. A new framework over the same store + // holds nothing in memory from that turn. The agent's plain speech has no + // channel, so no turn announces one (a locus announced at the first turn + // after boot would come last, and the reply would follow it instead). + const first = await make({ deliver: false, locus: null }); + await wake(first, ordinaryEvent('hello'), signed('sig-1', 'first reply')); + first.mod.endTurn = true; + await timerWake(first, + signed('sig-c', 'checking', [{ type: 'tool_use', id: 'c-t1', name: 'tools--ping', input: {} } as ContentBlock])); + const prompt = rowWith(first.framework, 'checking').metadata?.promptedBy; + assert.ok(prompt); + await first.framework.stop(); + + const strategy = new OmitToolExchange(); + strategy.omit = false; + const x = await make({ deliver: false, locus: null, dir: first.dir, agent: { strategy } }); + try { + await timerWake(x, signed('sig-c2', 'picked up')); + assert.ok(x.membrane.calls[0]!.messages.at(-1)!.content.some((b) => b.type === 'tool_result')); + assert.deepEqual(rowWith(x.framework, 'picked up').metadata?.promptedBy, prompt); + + strategy.omit = true; + await wake(x, ordinaryEvent('later'), signed('sig-later', 'later reply')); + const later = x.membrane.calls[1]!.messages; + assert.equal(textOf(later[later.findIndex((m) => textOf(m) === 'picked up') - 1]!), '[Continue]'); } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } }); From 68f529f071dcbffc9f4310822db4c52976559129 Mon Sep 17 00:00:00 2001 From: Basil-1007 Date: Sun, 11 Oct 2026 16:30:09 -0400 Subject: [PATCH 04/11] fix(requests): end a reply's tool loop where its next round is handed messages, not where they are drained The previous commit cleared the turn's [Continue] holder whenever the tool boundary drained any deferred message. What the next round follows is what the live stream is handed after the results: the injected messages, and the routing notice that rides only an injection. A drained message that stays window-only (tool blocks, or named as the agent) isn't in the next round's request, and a drain that stores an entry without injecting it leaves the round following the results. So the holder is now cleared only when the boundary has injections, once they are collected. Placement, so other open changes still merge as text: - the clear sits after the conversational-input signal, outside the drain block (which another change restructures) and outside the streaming branch's hand-off (where others add receipts and logging); - the stream start sets the holder after the replaced-agent check rather than before it, so a stream cancelled there sets nothing, and the check itself stays as it is on main; - the test harness's channel registry line stays as on main, which other changes edit: the process-restart test gives the agent no locus in its own body instead of through a make() option. --- src/framework.ts | 35 +++++++++++++++++------------- test/silent-heartbeat-rows.test.ts | 12 +++++----- 2 files changed, 26 insertions(+), 21 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index 43a60c9c2..a62081dac 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1245,9 +1245,9 @@ export class AgentFramework { * The `[Continue]` prompt the current round of each turn follows, keyed by * the turn's trigger. Each stream start sets it from what the Agent read in * its request (`StartStreamResult.continuePrompt`), and a tool boundary - * that stores anything after the results clears it, since the next round - * follows those messages. `silentTurnRowMetadata` stamps the round's rows - * with it. + * that hands the live stream messages after the results clears it, since + * the next round follows those messages. `silentTurnRowMetadata` stamps the + * round's rows with it. */ private readonly turnPrompts = new WeakMap(); @@ -6708,10 +6708,6 @@ export class AgentFramework { { const deferred = this.drainDeferredFor(agent.name); if (deferred.length > 0) { - // The next round follows these messages, not the [Continue] - // the turn's earlier rounds followed: its rows carry no prompt. - const turnTrigger = this.activeTurnTriggers.get(agent.name); - if (turnTrigger) this.turnPrompts.delete(turnTrigger); for (const msg of deferred) { agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)); // Injection guards: tool blocks would corrupt the tool-cycle @@ -6750,6 +6746,15 @@ export class AgentFramework { this.midTurnInputSignals.add(agent.name); } + // The next round is handed these messages after the results, so it + // follows them, not the tool loop of a reply to [Continue]: its rows + // carry no prompt. (The routing notice below rides only an + // injection already here.) + if (midTurnInjections.length > 0) { + const turnTrigger = this.activeTurnTriggers.get(agent.name); + if (turnTrigger) this.turnPrompts.delete(turnTrigger); + } + // Check if any tool result requested endTurn const shouldEndTurn = currentState.toolResults.some(tc => tc.result.endTurn); @@ -9401,14 +9406,6 @@ export class AgentFramework { } = await agent.startStreamWithInjections(tools, injections, undefined, compressionTools, { silentHeartbeat: trigger?.silentHeartbeat, }); - // The [Continue] prompt this stream's first round follows, if any, as - // the Agent read it from the request: the round's rows carry it - // (silentTurnRowMetadata), and so do later rounds' until a tool - // boundary stores something after the results. - if (trigger) { - if (continuePrompt) this.turnPrompts.set(trigger, continuePrompt); - else this.turnPrompts.delete(trigger); - } if (this.agents.get(agent.name) !== agent) { stream.cancel(); agent.cancelStream(); @@ -9416,6 +9413,14 @@ export class AgentFramework { this.eventGate?.onInferenceEnded(agent.name); return false; } + // The [Continue] prompt this stream's first round follows, if any, as + // the Agent read it from the request: the round's rows carry it + // (silentTurnRowMetadata), and so do later rounds' until a tool + // boundary hands the stream messages after the results. + if (trigger) { + if (continuePrompt) this.turnPrompts.set(trigger, continuePrompt); + else this.turnPrompts.delete(trigger); + } const handle = this.driveStream( agent, diff --git a/test/silent-heartbeat-rows.test.ts b/test/silent-heartbeat-rows.test.ts index ecfaccff1..b29a78904 100644 --- a/test/silent-heartbeat-rows.test.ts +++ b/test/silent-heartbeat-rows.test.ts @@ -148,8 +148,6 @@ async function make(opts: { agents?: string[]; agent?: Partial; /** An earlier framework's directory, to open its store again. */ dir?: string; - /** Where the agent's plain speech lands; null for no channel. */ - locus?: string | null; } = {}) { const dir = opts.dir ?? mkdtempSync(join(tmpdir(), 'silent-hb-rows-')); const membrane = opts.membrane ?? new MockMembrane(); @@ -164,8 +162,7 @@ async function make(opts: { modules: [mod], }); (framework as any).channelRegistry = new Proxy({ - resolveLocus: () => (opts.locus === undefined ? 'world:commons' : opts.locus), - getDefaultPublishChannel: () => (opts.locus === undefined ? 'world:commons' : opts.locus), + resolveLocus: () => 'world:commons', getDefaultPublishChannel: () => 'world:commons', routeSpeech: async () => ({ delivered: opts.deliver ?? true, channelId: 'world:commons' }), startTyping: () => {}, stopTyping: () => {}, getChannelTools: () => [], getDescriptor: () => undefined, @@ -863,7 +860,9 @@ describe('replies to a [Continue] turn', () => { // holds nothing in memory from that turn. The agent's plain speech has no // channel, so no turn announces one (a locus announced at the first turn // after boot would come last, and the reply would follow it instead). - const first = await make({ deliver: false, locus: null }); + const noLocus = (x: { framework: AgentFramework }) => { (x.framework as any).channelRegistry.resolveLocus = () => null; }; + const first = await make({ deliver: false }); + noLocus(first); await wake(first, ordinaryEvent('hello'), signed('sig-1', 'first reply')); first.mod.endTurn = true; await timerWake(first, @@ -874,7 +873,8 @@ describe('replies to a [Continue] turn', () => { const strategy = new OmitToolExchange(); strategy.omit = false; - const x = await make({ deliver: false, locus: null, dir: first.dir, agent: { strategy } }); + const x = await make({ deliver: false, dir: first.dir, agent: { strategy } }); + noLocus(x); try { await timerWake(x, signed('sig-c2', 'picked up')); assert.ok(x.membrane.calls[0]!.messages.at(-1)!.content.some((b) => b.type === 'tool_result')); From b7447fef2a8d1cb24954a24e81781def5c29568b Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Fri, 16 Oct 2026 15:41:55 -0400 Subject: [PATCH 05/11] test(guard): the compile holds in two tests from main hook the stream's own assembler #281's "a framework compile from before stop cannot drive after restart" and #338's "a turn dropped after its compile, with ..." cases hold a stream's compile by replacing agent.buildActivationRequest. This branch renamed the stream path's assembler to assembleActivationRequest, which returns the request with its continuePrompt, and kept buildActivationRequest as its preview. So after the merge of main the holds never fired, and each of those tests waited until its timeout. A helper, holdStreamCompile, now holds the assembler the stream path calls. Co-Authored-By: Claude Opus 5.5 --- test/tool-result-guard.test.ts | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index 1f999935a..b1bf6aa27 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -1742,6 +1742,15 @@ function gate() { const opened = new Promise((resolve) => { open = resolve; }); return { opened, open }; } +/** Hold a stream's compile: opens `entered` when a turn's stream start + * assembles its request, then waits for `release`. The stream path assembles + * through the agent's own assembler (buildActivationRequest is only its + * preview), so the hold goes there. */ +function holdStreamCompile(agent: object, entered: { open(): void }, release: Promise): void { + const assembler = agent as { assembleActivationRequest: (...args: unknown[]) => Promise }; + const assemble = assembler.assembleActivationRequest.bind(agent); + assembler.assembleActivationRequest = async (...args) => { entered.open(); await release; return assemble(...args); }; +} async function until(check: () => boolean, what: string) { for (let i = 0; i < 400 && !check(); i++) await new Promise((resolve) => setTimeout(resolve, 5)); assert.ok(check(), `timed out waiting until ${what}`); @@ -3218,8 +3227,7 @@ test('a framework compile from before stop cannot drive after restart', async () const agent = framework.getAgent('assistant')!; const compile = gate(); const entered = gate(); - const build = agent.buildActivationRequest.bind(agent); - agent.buildActivationRequest = async (...args) => { entered.open(); await compile.opened; return build(...args); }; + holdStreamCompile(agent, entered, compile.opened); let driven = 0; const stream = membrane.streamYielding.bind(membrane); membrane.streamYielding = (...args) => { @@ -3290,8 +3298,7 @@ for (const phase of ['before', 'after'] as const) { if (phase === 'before') { internal.moduleRegistry.gatherContext = async () => { entered.open(); await held.opened; return []; }; } else { - const build = agent.buildActivationRequest.bind(agent); - agent.buildActivationRequest = async (...args) => { entered.open(); await held.opened; return build(...args); }; + holdStreamCompile(agent, entered, held.opened); } const turnEvents: Array<{ type: string; reason?: string; durationMs?: number }> = []; framework.onTrace((event) => { From f5f749c3f35344538f602760a3d656ebec9e191f Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sun, 18 Oct 2026 00:48:35 -0400 Subject: [PATCH 06/11] docs(requests): which stamped rows place a request-only turn; a reply to anything else carries no prompt - silentTurnRowMetadata: request builds render a tick's separator or a [Continue] before the first surviving assistant row that carries its stamp and that they can recognise, one with a tool call or with signed or redacted thinking. The tool_result rows and the contained-wrapper row are stored as `user`, so their stamps place nothing: they mark each row as the reply's. The doc said the render went before the first of the turn's assistant rows, and a text-only reply isn't found among them. - The assembler's comment on a request that ends on anything other than [Continue] or tool results read as if that request prompted its reply. Such a reply follows that message and carries no prompt. - The tool boundary's clear: a live stream handed messages after the results follows them in its next round. A stream that doesn't say whether it carries them counts as carrying them, as at the drain above it. A boundary that ends the turn, restarts the stream or falls back to a new start hands them to no live stream, and that new start reads its own prompt from what it sends. --- src/agent.ts | 4 ++-- src/framework.ts | 32 +++++++++++++++++++------------- 2 files changed, 21 insertions(+), 15 deletions(-) diff --git a/src/agent.ts b/src/agent.ts index b9cd4bf44..821a4c269 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -1277,8 +1277,8 @@ export class Agent { // request itself: a new one when it ends on [Continue]; else, when it // ends on tool results, the prompt of the reply that called those tools, // whose tool loop this reply continues; else none. A request that ends - // on anything else (a message that arrived during a cooldown, say) - // prompts a reply that follows that. + // on anything else (a message that arrived during a cooldown, say) gets + // a reply that follows that message and carries no prompt. let continuePrompt: RequestOnlyPrompt | undefined; if (openingTick) { messages = [...messages, silentHeartbeatSeparatorTurn()]; diff --git a/src/framework.ts b/src/framework.ts index 99d205c1b..8b22dd47a 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -9764,10 +9764,15 @@ export class AgentFramework { this.midTurnInputSignals.add(agent.name); } - // The next round is handed these messages after the results, so it - // follows them, not the tool loop of a reply to [Continue]: its rows - // carry no prompt. (The routing notice below rides only an - // injection already here.) + // A live stream handed these messages after the results follows + // them in its next round, not the tool loop of a reply to + // [Continue]: that round's rows carry no prompt. One that doesn't + // say whether it carries them (membrane 0.5.87) counts as carrying + // them here, as at the drain above. A boundary that ends the turn, + // restarts the stream or falls back to a new start below hands + // them to no live stream, and a new start reads its own prompt from + // what it sends. (The routing notice below rides only an injection + // already here.) if (midTurnInjections.length > 0) { const turnTrigger = this.activeTurnTriggers.get(agent.name); if (turnTrigger) this.turnPrompts.delete(turnTrigger); @@ -11732,16 +11737,17 @@ export class AgentFramework { * stores (a delivery receipt, a routing notice, a deferred flush) aren't the * reply's, and carry none. * - A silent heartbeat tick stamps each of them for the whole turn, with the - * separator text its request rendered. Request builds render the - * separator before the first of the tick's assistant rows that survives, - * and the stamps keep every row of the tick identifiable after the turn, - * for a later retention policy to find or collapse. The stamp names the - * agent: residents share one message slot, and a broadcast tick reaches - * each of them with the same eventId. + * separator text its request rendered. The stamps keep every row of the + * tick identifiable after the turn, for a later retention policy to find + * or collapse. The stamp names the agent: residents share one message + * slot, and a broadcast tick reaches each of them with the same eventId. * - A `[Continue]` prompt stamps the rows of each round that follows it - * (`turnPrompts`). Request builds render it before the first of its - * assistant rows that survives. On the round's tool_result rows it marks - * them as the reply's, as the tick stamp does. + * (`turnPrompts`). + * Request builds render each turn before the first surviving assistant row + * that carries its stamp and that they can recognise: one with a tool call, + * or with signed or redacted thinking (see silent-heartbeat.ts). The + * tool_result rows and the wrapper row are stored as `user`, so their + * stamps place nothing: they mark each row as the reply's. * Undefined for ordinary turns, whose rows are stored exactly as before. * (Named for its first use, the silent tick.) */ From 083c910cab8e31bbb3fa98bf066e9b63b93fe2d7 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sun, 18 Oct 2026 00:48:35 -0400 Subject: [PATCH 07/11] refactor(requests): toolLoopPrompt drops a check its only caller has made Its caller handles a request that ends on the agent's own message before calling it, and such a message holds no tool results, so the check gave what the loop below it gives. Its agentName parameter goes with it. The function isn't exported from the package. --- src/agent.ts | 2 +- src/silent-heartbeat.ts | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/src/agent.ts b/src/agent.ts index 821a4c269..bb5cd6af5 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -1289,7 +1289,7 @@ export class Agent { messages = [...messages, continueTurn()]; continuePrompt = { turn: randomUUID(), text: CONTINUE_ROW }; } else { - continuePrompt = toolLoopPrompt(messages.at(-1), index, this.name); + continuePrompt = toolLoopPrompt(messages.at(-1), index); } const request: NormalizedRequest = { diff --git a/src/silent-heartbeat.ts b/src/silent-heartbeat.ts index d819adb08..cfaf496f7 100644 --- a/src/silent-heartbeat.ts +++ b/src/silent-heartbeat.ts @@ -212,9 +212,8 @@ export function indexRequestOnlyRows(stored: readonly StoredRowLike[], agentName export function toolLoopPrompt( last: NormalizedMessage | undefined, index: RequestOnlyRowIndex, - agentName: string, ): RequestOnlyPrompt | undefined { - if (!last || last.participant === agentName) return undefined; + if (!last) return undefined; for (const block of last.content) { if (block.type !== 'tool_result') continue; const prompt = index.prompts.get(toolUseKey(block.toolUseId)); From a9a96469b8431403d1877466309fa39144b5dc36 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sun, 18 Oct 2026 00:48:35 -0400 Subject: [PATCH 08/11] test(requests): the retry and cooldown tests wait for what they assert on The retry test slept 50 ms for the error policy's retry, and the cooldown test 1.1 s for a 1 s hold's release; a loaded host can fire either timer later than that. The retry test now waits for the retry's call, which the policy's timer starts by itself. The cooldown test waits for the hold's release, which only queues the resumed attempt: the test's runUntilIdle then starts it, so a wait for its call would time out. --- test/silent-heartbeat-rows.test.ts | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/test/silent-heartbeat-rows.test.ts b/test/silent-heartbeat-rows.test.ts index 2c6a79907..72aa7bc9a 100644 --- a/test/silent-heartbeat-rows.test.ts +++ b/test/silent-heartbeat-rows.test.ts @@ -212,7 +212,16 @@ function rowWith(framework: AgentFramework, text: string) { return row; } -const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); +/** Wait for what a test's next assertions need, rather than for a length of + * time: a timer the framework set (a retry's backoff, a cooldown's release) + * can fire late on a loaded host. */ +async function waitFor(cond: () => boolean, what: string, timeoutMs = 15_000): Promise { + const deadline = Date.now() + timeoutMs; + while (!cond()) { + if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`); + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} describe('silent heartbeat stored rows', () => { it('keeps a tick reply out of the previous assistant message on the wire', async () => { @@ -735,7 +744,7 @@ describe('replies to a [Continue] turn', () => { await x.framework.runUntilIdle(); assert.equal(internal.providerAccelerationCooldowns.size, 1, 'the restart is held'); await wake(x, ordinaryEvent('arrived during the cooldown')); - await sleep(1_100); + await waitFor(() => internal.providerAccelerationCooldowns.size === 0, 'the cooldown\'s release'); await x.framework.runUntilIdle(); assert.equal(membrane.calls.length, 4, 'first, the continued reply, its held restart, the resumed attempt'); @@ -800,7 +809,7 @@ describe('replies to a [Continue] turn', () => { [createMockResponse(signed('sig-c2', 'all done'))]); (x.framework as any).pendingRequests.push({ agentName: 'assistant', reason: 'timer', source: 'framework', timestamp: Date.now() }); await x.framework.runUntilIdle(); - await sleep(50); + await waitFor(() => membrane.calls.length >= 4, 'the retry\'s call'); await x.framework.runUntilIdle(); assert.equal(membrane.calls.length, 4, 'first, the continued reply, its failed restart, the retry'); assert.ok(membrane.calls[3]!.messages.at(-1)!.content.some((b) => b.type === 'tool_result')); From 1895b2a25d76c9f7776cd667ac830ce5bb2d6b41 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sun, 18 Oct 2026 00:56:42 -0400 Subject: [PATCH 09/11] test(stop): a retry's failing compile fails the assembler the stream path calls The test of a write deferred behind the entry a failed retry left replaced the agent's buildActivationRequest to make the retry's compile throw. The stream path now compiles through the agent's own assembler, and buildActivationRequest is only its wrapper for previews, so the retry compiled, its stream started, and the test's check that the old frame left its entry failed. The test now replaces the assembler. --- test/stop-with-work-in-flight.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index ac513baf2..d5ce2bf80 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -677,7 +677,9 @@ describe('stop() with work still in flight', () => { framework.pushEvent({ type: 'external-message', source: 'test', content: 'go', metadata: {} } as unknown as ProcessEvent); await until(() => internals.activeStreams.has('assistant'), 'the first stream'); const handle = internals.activeStreams.get('assistant')!; - (framework.getAgent('assistant') as unknown as { buildActivationRequest: () => never }).buildActivationRequest = () => { + // The stream path compiles through the agent's own assembler + // (buildActivationRequest is only its wrapper for previews). + (framework.getAgent('assistant') as unknown as { assembleActivationRequest: () => never }).assembleActivationRequest = () => { throw new Error('the compile failed'); }; release(); From d3c45a7abdf2f773e57c80b477236eb2eb461ccc Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sun, 18 Oct 2026 10:07:35 -0400 Subject: [PATCH 10/11] test(guard): a dropped turn's cancel-record test gates the assembler the stream path calls #437 brought a test that holds a turn in its compile, by replacing the agent's buildActivationRequest, while stop() and start() run, and then checks that the next turn's own error stays a failure. The stream path compiles through the agent's own assembler, and buildActivationRequest is only its wrapper for previews, so the hold never took: the test waited for a compile that never entered it, and ran into its timeout. It now gates the assembler, as 1895b2a did for #481's test. Co-Authored-By: Claude Opus 5.5 --- test/tool-result-guard.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index 8b38a407e..b17a62274 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -3665,8 +3665,11 @@ test("a dropped turn's cancel record is not read by the next turn's stream: its const agent = framework.getAgent('assistant')!; const compile = gate(); const entered = gate(); - const build = agent.buildActivationRequest.bind(agent); - agent.buildActivationRequest = async (...args) => { entered.open(); await compile.opened; return build(...args); }; + // The stream path compiles through the agent's own assembler + // (buildActivationRequest is only its wrapper for previews). + const hooked = agent as unknown as { assembleActivationRequest: (...args: unknown[]) => Promise }; + const assemble = hooked.assembleActivationRequest.bind(agent); + hooked.assembleActivationRequest = async (...args) => { entered.open(); await compile.opened; return assemble(...args); }; const stream = membrane.streamYielding.bind(membrane); membrane.streamYielding = (...args) => { const result = stream(...args); From 7850864bdd1a5a3648745c11db91c03c3f6d9cc2 Mon Sep 17 00:00:00 2001 From: Duncan-2410 Date: Sun, 18 Oct 2026 10:16:11 -0400 Subject: [PATCH 11/11] docs(agent): requestSettings()'s doc names the assembler that builds a framework turn #514 wrote the doc on main, where buildActivationRequest built a framework turn's request. Here that function is only the wrapper for previews, and assembleActivationRequest builds the turn, so the doc names it. Co-Authored-By: Claude Opus 5.5 --- src/agent.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent.ts b/src/agent.ts index f538c9ea8..c3dc0931c 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -1351,7 +1351,7 @@ export class Agent { * The request fields that come from this agent's own settings: its model * config, prompt caching and cache TTL, prefill message, provider * parameters and participant name. Every request the agent sends takes - * them from here, a framework turn's (buildActivationRequest) and the + * them from here, a framework turn's (assembleActivationRequest) and the * direct API's (runInferenceWithInjections), so a setting reaches both or * neither. */