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..a57eaceb2 --- /dev/null +++ b/changelog.d/request-only-rows-stay.changed.md @@ -0,0 +1,40 @@ +- 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 (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'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/request-only-rows-stay.fixed.md b/changelog.d/request-only-rows-stay.fixed.md new file mode 100644 index 000000000..6d586111a --- /dev/null +++ b/changelog.d/request-only-rows-stay.fixed.md @@ -0,0 +1,3 @@ +- A silent heartbeat tick is now told apart by its server and its event id + together, so two servers' ticks whose events share an id each get their + own separator, before their own first row. diff --git a/src/agent.ts b/src/agent.ts index f1e1d56ad..c3dc0931c 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -5,13 +5,19 @@ import type { CacheWireReceipt, KvUnifiedRequestHooks } from './kv-unified-wire. import { ToolResultGuard } from './tool-result-guard.js'; import { assertBooleanSetting, holdsSettings } from './config-checks.js'; import { - indexSilentTickRows, - separateSilentHeartbeatTicks, + CONTINUE_ROW, + continueTurn, + indexRequestOnlyRows, + renderRequestOnlyRows, silentHeartbeatSeparatorTurn, + tickKey, + toolLoopPrompt, + type RequestOnlyPrompt, + type RequestOnlyRowIndex, type SilentHeartbeatTick, - type SilentTickIndex, type StoredRowLike, } from './silent-heartbeat.js'; +import { randomUUID } from 'node:crypto'; import { nonErrorResultText, truncateForHistory, @@ -29,6 +35,13 @@ export interface ActivationRequestOptions { export interface StartStreamResult { stream: YieldingStream; request: NormalizedRequest; + /** The request-only `[Continue]` prompt this request's reply follows, if + * 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[]; } @@ -1221,13 +1234,13 @@ export class Agent { * may show a withheld stub before its place is added; the activation waits * for them around its call. */ - async buildActivationRequest( + private async assembleActivationRequest( availableTools: ToolDefinition[], injections?: ContextInjection[], budget?: TokenBudget, compressionTools: ToolDefinition[] = availableTools, options: ActivationRequestOptions = {}, - ): Promise { + ): 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 @@ -1286,44 +1299,59 @@ 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 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 - && !ticks.started.has(options.silentHeartbeat.eventId); - + && !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) gets + // a reply that follows that message and carries no prompt. + let continuePrompt: RequestOnlyPrompt | undefined; 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()]; + continuePrompt = { turn: randomUUID(), text: CONTINUE_ROW }; + } else { + continuePrompt = toolLoopPrompt(messages.at(-1), index); } - return { + const request: NormalizedRequest = { messages: this.toolResultGuard.prepareRequest(messages), system: this.buildSystemPrompt(systemInjections), tools: availableTools.length > 0 ? availableTools : undefined, ...this.requestSettings(), ...(kvUnified ? { cacheMarkers: 'cm-owned' as const } : {}), }; + return { request, ...(continuePrompt ? { continuePrompt } : {}) }; } /** * 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. */ @@ -1344,22 +1372,35 @@ export class Agent { }; } - /** 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; + /** 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 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(), prompts: new Map() }; 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; } } @@ -1423,8 +1464,8 @@ export class Agent { // Compiled only once each withheld stub's annotation has finished, // including one that began during the compile, and after a stub edit that // failed has been retried (see afterAnnotations). - const request = await this.toolResultGuard.afterAnnotations( - () => this.buildActivationRequest(availableTools, injections, budget, compressionTools, options)); + const { request, continuePrompt } = await this.toolResultGuard.afterAnnotations( + () => this.assembleActivationRequest(availableTools, injections, budget, compressionTools, options)); request.messages = this.toolResultGuard.prepareRequest(request.messages, true); const receiptAware = (this.contextManager as unknown as { getStrategy?: () => unknown }) @@ -1473,6 +1514,7 @@ export class Agent { return { stream, request, + ...(continuePrompt ? { continuePrompt } : {}), ...(kvEnabled ? { takeKvSubmission: () => kvQueue.shift(), diff --git a/src/framework.ts b/src/framework.ts index b9ec18f1c..a447cd91f 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11,7 +11,7 @@ import type { Membrane, ContentBlock, NormalizedRequest, NormalizedResponse, Yie import { MembraneError } from '@animalabs/membrane'; import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError, stripUnpairedToolBlocks } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; -import 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 { @@ -2200,6 +2200,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 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(); + /** The trigger of `agentName`'s turn in progress, if a turn is running. */ getActiveTurnTrigger(agentName: string): InferenceRequest | undefined { return this.activeTurnTriggers.get(agentName); @@ -10487,6 +10497,20 @@ export class AgentFramework { this.midTurnInputSignals.add(agent.name); } + // 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); + } + // Check if any tool result requested endTurn const shouldEndTurn = currentState.toolResults.some(tc => tc.result.endTurn); @@ -10642,7 +10666,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) { @@ -10691,12 +10714,15 @@ export class AgentFramework { }); } else { // Non-streaming fallback: schedule re-inference, which continues - // this stream's logical turn (openLogical). + // this stream's logical turn (openLogical), 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, }); } } @@ -12419,9 +12445,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 } : {}), @@ -12767,7 +12795,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, @@ -12783,20 +12810,40 @@ 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. + * 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. 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 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.) */ - private silentTurnRowMetadata(agentName: string): MessageMetadata | undefined { - const tick = this.activeTurnTriggers.get(agentName)?.silentHeartbeat; - const stamp: SilentHeartbeatStamp | undefined = tick - ? { eventId: tick.eventId, serverId: tick.serverId, agentName } - : undefined; - return stamp ? { silentHeartbeat: stamp } : undefined; + private silentTurnRowMetadata(agentName: string): RequestOnlyStamps | undefined { + const trigger = this.activeTurnTriggers.get(agentName); + 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 plain-prose segments this turn suppressed without drafts (proseRouting=disabled). */ @@ -15100,15 +15147,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, // and stop() may close provider admission. Do not create a provider // stream after its generation lost ownership, or once admission closed. @@ -15139,6 +15177,7 @@ export class AgentFramework { request: compiledRequest, takeKvSubmission, drainKvSubmissionIds, + continuePrompt, } = await agent.startStreamWithInjections(tools, injections, undefined, compressionTools, { silentHeartbeat: trigger?.silentHeartbeat, }); @@ -15194,6 +15233,14 @@ export class AgentFramework { 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/src/index.ts b/src/index.ts index 4263295e0..ef57e6eff 100644 --- a/src/index.ts +++ b/src/index.ts @@ -168,6 +168,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 9d4601375..cfaf496f7 100644 --- a/src/silent-heartbeat.ts +++ b/src/silent-heartbeat.ts @@ -1,40 +1,70 @@ /** - * 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 + * 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, + * 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 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 - * (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 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]'; /** Identity of one authenticated silent heartbeat tick. */ export interface SilentHeartbeatTick { @@ -50,20 +80,63 @@ 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; +} + +/** 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; 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 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; + 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 { 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 }] }; } /** @@ -76,7 +149,7 @@ export function silentHeartbeatSeparatorTurn(): 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) { @@ -86,56 +159,108 @@ function identityKeys(content: ContentBlock[]): string[] { return keys; } +function toolUseKey(id: string): string { + return `tool_use:${id}`; +} + export interface StoredRowLike { participant: string; content: ContentBlock[]; metadata?: unknown; } -export interface SilentTickIndex { - /** Identity key of every assistant row this agent stored during a tick → tick event id. */ - rows: Map; - /** Ticks this agent has already stored rows for (its own, not another resident's). */ - started: Set; +export interface RequestOnlyRowIndex { + /** Identity key of every assistant row this agent stored during a silent + * 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; } -/** Index the rows this agent stored during silent ticks. */ -export function indexSilentTickRows(stored: readonly StoredRowLike[], agentName: string): SilentTickIndex { - const rows = new Map(); - const started = new Set(); +/** 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. 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 prompts = new Map(); 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); + const stamp = silentHeartbeatOf(row.metadata); + 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); + } + if (prompt) for (const key of keys) prompts.set(key, prompt); } - return { rows, started }; + return { ticks, prompts }; } /** - * 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 + * 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, +): RequestOnlyPrompt | 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)); + if (prompt) return prompt; + } + return undefined; +} + +export interface RenderedRequestOnlyRows { + messages: NormalizedMessage[]; + /** The ticks (`tickKey`) whose separator this window renders. A tick none + * of whose rows survive compilation renders none. */ + separatedTicks: Set; +} + +/** + * 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 separateSilentHeartbeatTicks( +export function renderRequestOnlyRows( messages: NormalizedMessage[], - rows: ReadonlyMap, + index: RequestOnlyRowIndex, agentName: string, -): NormalizedMessage[] { - if (rows.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 }; 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); - out.push(silentHeartbeatSeparatorTurn()); + const keys = identityKeys(message.content); + const tick = keys.map((key) => index.ticks.get(key)).find((t) => t !== undefined); + 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); + if (prompt !== undefined && !promptedTurns.has(prompt.turn)) { + promptedTurns.add(prompt.turn); + out.push(requestOnlyTurn(prompt.text)); } } out.push(message); } - return out; + return { messages: out, separatedTicks }; } diff --git a/src/types/agent.ts b/src/types/agent.ts index b1e82dd2f..2ecb0c6c5 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -411,11 +411,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; - /** 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. */ + /** 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 af98145e9..72aa7bc9a 100644 --- a/test/silent-heartbeat-rows.test.ts +++ b/test/silent-heartbeat-rows.test.ts @@ -15,24 +15,39 @@ 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, 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; @@ -64,22 +79,62 @@ 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 strategy that leaves every tool exchange out of the compiled window. */ +/** 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 { + 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')); } } @@ -91,8 +146,10 @@ 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; } = {}) { - 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({ @@ -149,6 +206,23 @@ 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; +} + +/** 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 () => { const x = await make(); @@ -189,10 +263,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 }); } }); @@ -212,14 +286,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('[heartbeat tick]'), '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 }); } }); @@ -234,16 +309,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,12 +354,40 @@ 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 }); } }); + 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 { @@ -299,8 +402,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 +418,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 +443,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 +465,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]); @@ -420,7 +523,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 }); } }); @@ -440,10 +543,380 @@ 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?.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'); + } 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', 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: prompt } }, + ]; + const index = indexRequestOnlyRows(stored, 'assistant'); + 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. + assert.deepEqual(renderRequestOnlyRows(bare([stored[0]!, stored[2]!]), index, 'assistant').messages.map(textOf), + ['earlier reply', SILENT_HEARTBEAT_SEPARATOR, '[Continue]', 'tick continued']); + }); + + 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(), [ + 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 () => { + 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 !== 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); + 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 }); } + }); + + 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; + 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 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'); + 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('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 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'); + + 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.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 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')); + + 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 continued = x.membrane.calls[1]!.messages; + const later = x.membrane.calls[2]!.messages; + 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 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, + 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, 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')); + 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 }); } + }); + + 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 }); } + }); }); 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 }); } }); diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index 45a5810e3..ef63b3184 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -679,7 +679,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(); diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index 9923d0fdb..b17a62274 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -1810,6 +1810,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}`); @@ -3599,8 +3608,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) => { @@ -3657,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); @@ -3720,8 +3731,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) => {