From 4c0bdd44a6d5eaf40992c3b4d0c0cdff5cbcaf0d Mon Sep 17 00:00:00 2001 From: antra-tess Date: Thu, 22 Oct 2026 22:14:45 -0700 Subject: [PATCH 1/6] =?UTF-8?q?feat(prose):=20autoContinue=20=E2=80=94=20c?= =?UTF-8?q?ontinue=20a=20turn=20the=20model=20ends=20itself=20until=20skip?= =?UTF-8?q?=5Freply?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A turn the model ends itself (end_turn / stop_sequence), with prose or with nothing, is followed by another turn instead of a pause until the next event, unless it ended on a skip: the skip_reply tool (or any endTurn tool), or a last `>>skip_reply` (explicit) / `>>>skip_reply` hold (hybrid). Off by default; AgentConfig.autoContinue: true | { maxConsecutive: 1-20 } (3). The next turn reads a short [continue] system row, stored after the turn's [delivered] receipt, naming skip_reply as the way to stop; near the cap it says how many continuations are left. The wake (reason auto-continue, source framework) carries the turn's locus. The run count resets on any wake with a cause of its own and when a tool ends a turn. Refusals, max_tokens, the stall guard's ends, silent heartbeat ticks and ephemeral runs never continue; a ` !` continuation runs instead. Co-Authored-By: Claude Opus 5.5 --- .../feat-continue-after-prose.added.md | 14 + docs/auto-continue.md | 69 +++ src/agent.ts | 37 +- src/auto-continue.ts | 35 ++ src/framework.ts | 111 +++++ src/types/agent.ts | 29 ++ test/auto-continue.test.ts | 405 ++++++++++++++++++ 7 files changed, 698 insertions(+), 2 deletions(-) create mode 100644 changelog.d/feat-continue-after-prose.added.md create mode 100644 docs/auto-continue.md create mode 100644 src/auto-continue.ts create mode 100644 test/auto-continue.test.ts diff --git a/changelog.d/feat-continue-after-prose.added.md b/changelog.d/feat-continue-after-prose.added.md new file mode 100644 index 000000000..604c6f07b --- /dev/null +++ b/changelog.d/feat-continue-after-prose.added.md @@ -0,0 +1,14 @@ +- **Hosts:** `AgentConfig.autoContinue` (default off): a turn the model ends + itself, with prose or with nothing, is followed by another turn straight + away, until the agent calls `skip_reply` (or another tool that ends the + turn), ends on a `>>skip_reply` / `>>>skip_reply` envelope, or reaches + `maxConsecutive` continuations in a row (default 3, 1 to 20; the count + starts again on any wake with a cause of its own). The next turn reads a + short `[continue]` system row, stored after the turn's `[delivered]` + receipt, that names `skip_reply` as the way to stop and, near the cap, how + many continuations are left; it keeps the turn's locus. Its wake has + reason `auto-continue`. Refusals, `max_tokens` cut-offs, the stall guard's + ends, silent heartbeat ticks and ephemeral runs never continue, and a + ` !` continuation runs instead. A malformed setting is refused when the + agent is built. New exported type `AutoContinueConfig`. See + docs/auto-continue.md. diff --git a/docs/auto-continue.md b/docs/auto-continue.md new file mode 100644 index 000000000..d01d3a0fb --- /dev/null +++ b/docs/auto-continue.md @@ -0,0 +1,69 @@ +# Continuing until skip_reply (`autoContinue`) + +`autoContinue` is a default-off per-agent setting. With it on, a turn the +model ends itself is followed by another turn straight away, instead of the +usual pause until the next event (a message, a heartbeat, a timer). The agent +keeps going until it says it's done: by calling `skip_reply`, or another tool +that ends the turn, such as `sleep`. + +```json +"agent": { "autoContinue": true } // cap of 3 +"agent": { "autoContinue": { "maxConsecutive": 5 } } // 1 to 20 +``` + +## Which turns continue + +A turn continues when the model ended it (stop reason `end_turn` or +`stop_sequence`), whether its last response is prose or empty, unless: + +- a tool ended it: `skip_reply`, `sleep`, any endTurn tool; +- its last prose envelope is a skip: `>>skip_reply` in explicit mode, or a + `>>>skip_reply` in hybrid mode whose hold is still standing at the end; +- it asked for a ` !` continuation (`>>#ops !`), which runs instead, with no + notice, and counts toward nothing; +- it was a silent heartbeat tick; +- it ended with a refusal, a `max_tokens` cut-off, or the stall guard's + `no_progress` / `round_limit`; +- the agent is an ephemeral run (a subagent), or the framework is stopping. + +## What the agent reads + +Before the next turn a short system row joins the window, after the turn's +`[delivered]` receipt, so it is the newest thing the turn reads: + +``` +[continue] Your last turn ended without skip_reply, so another one starts now. To stop and wait for the next event, call skip_reply. +``` + +Near the cap the row also says how many continuations are left: + +``` +… Continuation 2 of 3 in a row: one more is left. +… This is the last continuation in a row (3 of 3): after this turn you wait for the next event either way. +``` + +An agent that can't call `skip_reply` is told to end with a `>>skip_reply` +(explicit) or `>>>skip_reply` (hybrid) envelope instead; in locus or disabled +mode without the tool, the row names no way to stop and the cap ends the run. + +The row is stored at the turn's end, after its response, and never retracted, +so the prompt-cache prefix stays stable. It is deliberately short and +event-shaped: longer instruction blocks about output, sent as user-role +messages, have tripped provider classifiers. + +## The next turn + +The continuation's wake has source `framework` and reason `auto-continue`. In +locus and hybrid modes it carries the turn's locus as its channel, so the next +turn's plain prose lands where the last one's did rather than in the +process-global last-inbound channel. A message that arrives meanwhile joins +the same turn, as any batched wake does. + +## The cap + +At most `maxConsecutive` continuations run in a row (default 3). The count +starts again when a turn starts with a cause of its own (anything but a +framework wake), and when a tool ends a turn. Each continued turn is a full +inference over the context (mostly cache reads), and in locus mode its prose +is published, so the cap bounds both the cost and the posts of an agent that +never skips. diff --git a/src/agent.ts b/src/agent.ts index dac053601..62eadfdb8 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -64,6 +64,7 @@ import type { } from '@animalabs/context-manager'; import type { AgentConfig, + AutoContinueConfig, AgentState, SameRoundThinkTextPolicy, SameRoundThinkTextPolicySource, @@ -102,7 +103,8 @@ const DEFAULT_SAME_ROUND_THINK_TEXT_POLICY: SameRoundThinkTextPolicy = 'public'; * `toolWrapperProseGuard`, `thinking.enabled` or `refusalHandling.autoRewind` * holds anything but true or false (assertBooleanSetting), or whose `thinking` * or `refusalHandling` is anything but an object, null or false - * (holdsSettings). The constructor runs it first, so the rule holds wherever + * (holdsSettings), or whose `autoContinue` resolveAutoContinue refuses. + * The constructor runs it first, so the rule holds wherever * an Agent is built. `create()` also runs it for each of its agents before * anything opens, and `createEphemeralAgent` before it reserves the name, so a * refusal there leaves nothing behind. Membrane reads `thinking.enabled`, and @@ -142,6 +144,33 @@ export function assertAgentSwitches(config: AgentConfig): void { 'true keeps a response whose whole prose is a tool-call-shaped wrapper from running or entering the agent\'s history, ' + 'and false or leaving it out doesn\'t', ); + resolveAutoContinue(config); +} + +/** `AutoContinueConfig.maxConsecutive` when left out. */ +export const DEFAULT_AUTO_CONTINUE_MAX = 3; +/** The most continuations in a row a config may ask for. */ +const MAX_AUTO_CONTINUE_MAX = 20; + +/** + * `AgentConfig.autoContinue` resolved: undefined when off (left out, null or + * false), else its cap. Anything else is refused: `true`, or an object whose + * `maxConsecutive` is left out or a whole number from 1 to 20. + */ +export function resolveAutoContinue(config: AgentConfig): { maxConsecutive: number } | undefined { + const setting = `Agent "${config.name}" autoContinue`; + const value = config.autoContinue; + if (value === true) return { maxConsecutive: DEFAULT_AUTO_CONTINUE_MAX }; + if (!holdsSettings(setting, value, '{ maxConsecutive: 3 }')) return undefined; + const max = (value as AutoContinueConfig).maxConsecutive; + if (max === undefined) return { maxConsecutive: DEFAULT_AUTO_CONTINUE_MAX }; + if (typeof max !== 'number' || !Number.isInteger(max) || max < 1 || max > MAX_AUTO_CONTINUE_MAX) { + throw new Error( + `${setting}.maxConsecutive must be a whole number from 1 to ${MAX_AUTO_CONTINUE_MAX}, not ${JSON.stringify(max) ?? String(max)}: ` + + 'the most turns in a row that continue without skip_reply before the agent waits for the next event', + ); + } + return { maxConsecutive: max }; } /** @@ -174,6 +203,9 @@ export class Agent { readonly proseRouting: 'locus' | 'explicit' | 'hybrid' | 'disabled'; /** Exact whole-response known-tool wrapper containment (default off). */ readonly toolWrapperProseGuard: boolean; + /** Continuation after a turn the model ends itself (see + * AgentConfig.autoContinue); undefined when off. */ + readonly autoContinue: { maxConsecutive: number } | undefined; /** Clock rows' interval and precision (see AgentConfig.clock); undefined * when the agent writes none of its own (a resident still reads those * another resident writes into the shared slot). */ @@ -307,7 +339,8 @@ export class Agent { this.toolResultGuard = new ToolResultGuard(config.name, contextManager, config.toolResultGuard === true); this.proseRouting = config.proseRouting ?? 'locus'; this.toolWrapperProseGuard = config.toolWrapperProseGuard === true; - this.clock = resolveClockConfig(config.name, config.clock); + this.autoContinue = resolveAutoContinue(config); + this.clock =resolveClockConfig(config.name, config.clock); this.cacheTtl = config.cacheTtl ?? '1h'; this.promptCaching = config.promptCaching !== false; this.prefillUserMessage = config.prefillUserMessage; diff --git a/src/auto-continue.ts b/src/auto-continue.ts new file mode 100644 index 000000000..d65cb29ad --- /dev/null +++ b/src/auto-continue.ts @@ -0,0 +1,35 @@ +/** + * autoContinue (AgentConfig.autoContinue, docs/auto-continue.md): a turn the + * model ends itself is followed by another, unless it ended on a skip. This + * module holds what doesn't need the framework: which ends continue, and the + * notice the next turn reads. + */ + +/** + * Stop reasons of a turn the model ended itself. The others don't continue: + * a refusal has its own handling (retries, rewind, the reaction), a + * `max_tokens` end was cut off rather than chosen, and `no_progress` and + * `round_limit` are the stall guard's verdict that the turn wasn't moving. + */ +export const AUTO_CONTINUE_STOP_REASONS: ReadonlySet = new Set(['end_turn', 'stop_sequence']); + +/** + * The notice a continuation's turn reads: that the last turn ended without a + * skip so another one starts, how to stop (`stop`, an imperative such as + * "call skip_reply"; left out when the agent has no way to skip), and, near + * the cap, how many continuations are left. `count` is this continuation's + * place in the run, 1 to `max`. Short and event-shaped on purpose: it rides + * as a system row in the window, where instruction blocks about output have + * tripped classifiers. + */ +export function autoContinueNotice(count: number, max: number, stop: string | undefined): string { + const opening = stop + ? `[continue] Your last turn ended without skip_reply, so another one starts now. To stop and wait for the next event, ${stop}.` + : '[continue] Your last turn ended on its own, so another one starts now.'; + const left = max - count; + if (left <= 0) { + return `${opening} This is the last continuation in a row (${count} of ${max}): after this turn you wait for the next event either way.`; + } + if (left === 1) return `${opening} Continuation ${count} of ${max} in a row: one more is left.`; + return opening; +} diff --git a/src/framework.ts b/src/framework.ts index 89228c2e0..4de099437 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -156,6 +156,7 @@ import { DEFAULT_OUTPUT_BOUNDS } from './code-execution/isolate/output.js'; import type { PreludeTool } from './code-execution/isolate/prelude.js'; import { bounded, renderList, renderReturn, VALUE_CHARS } from './code-execution/isolate/render.js'; import { splitProseRuns, withoutRoundsProse } from './prose-segments.js'; +import { AUTO_CONTINUE_STOP_REASONS, autoContinueNotice } from './auto-continue.js'; import { cumulativeDelta } from './usage-accounting.js'; import { stampThinkingTokenEstimates } from './thinking-token-stamp.js'; import { contentAsStored, nameAsStored } from './text-as-stored.js'; @@ -1094,6 +1095,13 @@ interface TurnRouting { hybridHold?: 'private' | 'failed'; /** The turn asked for a ` !` continuation; taken once. */ continuation: boolean; + /** + * Explicit mode: the turn's latest envelope was `>>skip_reply`, so it + * ended on a deliberate skip and autoContinue leaves it paused. (Hybrid + * reads `hybridHold === 'private'`, which a `>>>skip_reply` leaves until a + * destination is named.) + */ + lastEnvelopeSkipped?: boolean; /** Successful plain-prose destinations in delivery order; deduped by the receipt. */ deliveries: ChannelRef[]; /** Plain-prose segments the turn suppressed without drafts (proseRouting=disabled, or a round withheld before its scripts returned: ScriptRound.withhold), reported by its receipt. */ @@ -2734,6 +2742,10 @@ export class AgentFramework { /** Consecutive bounce-wakes per agent; capped so an agent that keeps * emitting unprefixed prose can't wake-loop (notices still append). */ private proseBounceStreaks: Map = new Map(); + /** Consecutive autoContinue continuations per agent, capped by its + * maxConsecutive. Cleared when a turn starts with a cause of its own + * (processInferenceRequests) or a tool ends a turn (skip_reply). */ + private autoContinueStreaks: Map = new Map(); private pendingAssistantBlocks: Map = new Map(); /** Streams the FRAMEWORK cancelled for non-terminal reasons, keyed * `${agentName}:${streamId}`: an endTurn tool result or a context-budget @@ -14787,6 +14799,9 @@ export class AgentFramework { // (Optional chaining: prototype-built harnesses leave the map unset.) if (requests.some((r) => !FRAMEWORK_WAKE_SOURCES.has(r.source))) { this.proseBounceStreaks?.delete(agentName); + // Likewise autoContinue's cap bounds only a run of continuations + // that nothing else feeds. + this.autoContinueStreaks?.delete(agentName); } // Route this turn's auto-published speech to the channel that triggered // it (item-3 redux). A batched wake may carry several triggering channels @@ -16553,6 +16568,94 @@ export class AgentFramework { }); } + /** + * autoContinue (AgentConfig.autoContinue): after a turn the model ended + * itself (AUTO_CONTINUE_STOP_REASONS), with prose or with nothing, start + * another, unless its last word was a skip (`>>skip_reply`, or a hybrid + * `>>>skip_reply` hold still standing). Runs at the logical turn end of a + * stream's `complete`, after the turn's deliveries and `[delivered]` + * receipt, and only when the turn asked for no ` !` continuation (that one + * runs instead). A tool that ends the turn (skip_reply, sleep) never + * reaches here, and clears the run's count. + * + * The next turn reads a `[continue]` notice, stored now so it is the newest + * row the turn compiles (after the receipt), and keeps this turn's locus + * through its wake's channelId. A run is capped at maxConsecutive; the + * last allowed notice says so, and the cap holds until a turn starts with + * a cause of its own. + */ + private autoContinueAfterTurn( + agent: Agent, + routing: TurnRouting, + trigger: InferenceRequest | undefined, + stopReason: string, + ): void { + const config = agent.autoContinue; + if (!config) return; + const name = agent.name; + if (!AUTO_CONTINUE_STOP_REASONS.has(stopReason)) return; + // A silent heartbeat tick is a private self-check, never the start of a run. + if (trigger?.suppressProse) return; + // A subagent's run ends with its result. + if (this.ephemeralRuns.has(name)) return; + if (this.providerAdmissionClosed || this.agents.get(name) !== agent) { + console.error(`[continue] ${name}: not continued — ${this.providerAdmissionClosed ? 'the framework is stopping' : 'the agent is no longer registered'}`); + return; + } + const skipped = agent.proseRouting === 'explicit' + ? routing.lastEnvelopeSkipped === true + : agent.proseRouting === 'hybrid' && routing.hybridHold === 'private'; + if (skipped) { + this.autoContinueStreaks.delete(name); + console.error(`[continue] ${name}: turn ended on a skip_reply envelope — waiting for the next event`); + return; + } + const count = (this.autoContinueStreaks.get(name) ?? 0) + 1; + if (count > config.maxConsecutive) { + console.error(`[continue] ${name}: ${config.maxConsecutive} continuation(s) in a row already — waiting for the next event`); + return; + } + this.autoContinueStreaks.set(name, count); + const stop = this.canAgentUseTool(name, 'skip_reply') + ? 'call skip_reply' + : agent.proseRouting === 'explicit' + ? 'end with a >>skip_reply envelope' + : agent.proseRouting === 'hybrid' + ? 'end with a >>>skip_reply envelope' + : undefined; + const text = autoContinueNotice(count, config.maxConsecutive, stop); + try { + // Stored straight into the window, as the receipt just was: the turn's + // stream is over, so the row lands after its response, and the next + // compile reads it as the newest row. + const id = agent.getContextManager().addMessage( + 'user', + [{ type: 'text', text }], + { system: true, kind: 'auto-continue', continuation: count, maxConsecutive: config.maxConsecutive } as MessageMetadata, + ); + this.emitTrace({ type: 'message:added', messageId: id, source: 'auto-continue' }); + } catch (err) { + // Without its notice the next turn would wake to nothing it can read + // as a cause: no notice, no continuation. + console.error(`[continue] ${name}: notice not stored, so not continued:`, err); + this.autoContinueStreaks.set(name, count - 1); + return; + } + console.error( + `[continue] ${name}: turn ended (${stopReason}) without skip_reply — continuation ${count} of ${config.maxConsecutive}` + + (routing.locus ? ` in ${this.channelRegistry?.nameShown(routing.locus) ?? routing.locus.channelId}` : ''), + ); + this.pendingRequests.push({ + agentName: name, + reason: 'auto-continue', + source: 'framework', + timestamp: Date.now(), + // The next turn speaks where this one did (locus and hybrid modes): + // the locus's server and id, as the framework's own wakes name a channel. + ...(routing.locus ? { channelId: routing.locus.channelId, channelServerId: routing.locus.serverId } : {}), + }); + } + /** The agent's routing state for its current turn, made when it has none. */ private routingOf(agentName: string): TurnRouting { let routing = this.turnRouting.get(agentName); @@ -16991,6 +17094,7 @@ export class AgentFramework { // A private round (one that called skip_reply) asks for no publication // or continuation, even when delivery runs after that turn has ended. if (parsed.continueTurn && !round.keepPrivate) routing.continuation = true; + routing.lastEnvelopeSkipped = parsed.kind === 'private'; if (parsed.kind === 'private') { // The text already lives in the assistant message (window/chronicle); @@ -18766,6 +18870,9 @@ export class AgentFramework { // deliveries have run. Explicit mode doesn't hold the turn // for them, so it checks at the end of its chain. if (cancelKind === 'turn_ended') { + // A tool ended the turn (skip_reply, sleep): the agent stopped on + // purpose, so a later autoContinue run counts from one again. + this.autoContinueStreaks.delete(agent.name); if (turnProseRouting === 'explicit') { void turnSpeechChain.then(() => this.requestProseContinuation(agent, streamRouting)); } else { @@ -19770,7 +19877,11 @@ export class AgentFramework { // A `!` continuation this turn asked for. This gives prose the // same continuation ability tool rounds have (wanted for // robotics-style loops where an end-of-turn pause is harmful). + // Otherwise autoContinue, for an agent that has it, continues a + // turn the model ended itself unless that end was a skip. + const askedToContinue = finalRouting.continuation; this.requestProseContinuation(agent, finalRouting); + if (!askedToContinue) this.autoContinueAfterTurn(agent, finalRouting, trigger, response.stopReason); break; } diff --git a/src/types/agent.ts b/src/types/agent.ts index 248717e6d..abb441e40 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -9,6 +9,13 @@ export type SameRoundThinkTextPolicySource = | 'recipe' | 'compatibility_default'; +/** `AgentConfig.autoContinue` as an object (`true` means `{}`). */ +export interface AutoContinueConfig { + /** Continuations in a row before the agent waits for the next event: + * a whole number, 1 to 20. Default 3. */ + maxConsecutive?: number; +} + /** * Configuration for an agent. */ @@ -219,6 +226,28 @@ export interface AgentConfig { */ toolWrapperProseGuard?: boolean; + /** + * Continue on the agent's own turn ending (off when left out): a turn the + * model ends itself (stop reason `end_turn` or `stop_sequence`), whether + * its last response is prose or empty, is followed by another turn, + * instead of a pause until the next event. A turn ended through + * `skip_reply` (or another tool that ends the turn, such as `sleep`), or + * whose last prose envelope is `>>skip_reply` (explicit) or `>>>skip_reply` + * (hybrid), pauses as usual. + * + * Before the next turn a short system notice joins the window, after the + * turn's `[delivered]` receipt, saying that the turn continues and that + * skip_reply stops it; near the cap it also says how many continuations + * are left. The next turn keeps the turn's locus (locus and hybrid modes). + * At most `maxConsecutive` (default 3) continuations run in a row: the + * count starts again when a turn starts with a cause of its own (a + * message, a heartbeat, a timer) or a tool ends one. Silent heartbeat + * ticks, refusals, `max_tokens` cut-offs, the stall guard's ends and + * ephemeral runs never continue. A ` !` continuation the turn asked for + * runs instead, and counts toward nothing. See docs/auto-continue.md. + */ + autoContinue?: boolean | AutoContinueConfig; + /** * Clock rows (off when left out): when a boundary of this interval on the * agent's local wall clock (counted from local midnight, in the diff --git a/test/auto-continue.test.ts b/test/auto-continue.test.ts new file mode 100644 index 000000000..c4b51c6a7 --- /dev/null +++ b/test/auto-continue.test.ts @@ -0,0 +1,405 @@ +/** + * autoContinue (docs/auto-continue.md): a turn the model ends itself, with + * prose or with nothing, is followed by another turn, unless it ended on a + * skip (the skip_reply tool, or a `>>skip_reply` / `>>>skip_reply` + * envelope). The next turn reads a `[continue]` notice; a run is capped. + */ + +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { tmpdir } from 'node:os'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import type { + AgentConfig, + Module, + ModuleContext, + ProcessState, + ProcessEvent, + EventResponse, + ToolDefinition, + ToolCall, + ToolResult, + InferenceRequest, +} from '../src/index.js'; +import { AgentFramework } from '../src/index.js'; +import { resolveAutoContinue } from '../src/agent.js'; +import { autoContinueNotice } from '../src/auto-continue.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +import type { ContentBlock, NormalizedResponse } from '@animalabs/membrane'; + +class TriggerModule implements Module { + readonly name = 'robot'; + framework: AgentFramework | null = null; + + async start(_ctx: ModuleContext): Promise {} + async stop(): Promise {} + getTools(): ToolDefinition[] { + return [{ name: 'move', description: 'Move the robot', inputSchema: { type: 'object', properties: {} } }]; + } + async handleToolCall(_call: ToolCall): Promise { + return { success: true, data: { ok: true } }; + } + async onProcess(event: ProcessEvent, _state: ProcessState): Promise { + if (event.type === 'external-message') { + const text = String((event as { content?: unknown }).content); + return { addMessages: [{ participant: 'Antra', content: [{ type: 'text', text }] }], requestInference: true }; + } + return {}; + } +} + +/** Registry stub, as main's prose-routing tests build it: a locus of + * world:commons on the stub server, two named channels, and a log of what + * was routed where (by channel id). */ +function stubRegistry(framework: AgentFramework) { + const routed: Array<{ text: string; locus: string | null }> = []; + const known: Record = { + '#alpha': { channelId: 'chan-alpha', label: '#alpha' }, + 'world:commons': { channelId: 'world:commons', label: 'world:commons' }, + }; + const deliverSpeech = async (_agent: string, text: string, locus?: { serverId?: string; channelId: string } | null) => { + routed.push({ text, locus: locus?.channelId ?? null }); + return { + status: 'delivered' as const, + destination: { serverId: locus?.serverId ?? 'stub', channelId: locus?.channelId ?? '' }, + at: Date.now(), + }; + }; + const registry: Record = { + resolveProseTarget: (spec: string) => + known[spec] ? { serverId: 'stub', ...known[spec] } : { error: 'no channel matches "' + spec + '"' }, + deliverSpeech, + routeSpeech: async (agent: string, text: string, locus?: { serverId?: string; channelId: string } | null) => { + const outcome = await deliverSpeech(agent, text, locus); + return { delivered: true, serverId: outcome.destination.serverId, channelId: outcome.destination.channelId }; + }, + resolveDestination: ({ channelId }: { channelId: string }) => ({ destination: { serverId: 'stub', channelId } }), + resolveLocus: () => ({ serverId: 'stub', channelId: 'world:commons' }), + getDefaultPublishChannel: () => null, + isChannelOpen: () => true, + getDescriptor: () => undefined, + getChannelTools: () => [], + channelsNamed: (name: string) => [{ serverId: 'stub', channelId: name }], + nameShown: (channel: { channelId: string }) => channel.channelId, + // skip_reply ends the turn, as the registry's own handler does. + handleChannelToolCall: async () => ({ success: true, endTurn: true, data: { skipped: true } }), + }; + (framework as unknown as { channelRegistry: unknown }).channelRegistry = new Proxy(registry, { + get: (target, prop: string) => (prop in target ? target[prop] : () => undefined), + }); + return routed; +} + +const text = (t: string): ContentBlock => ({ type: 'text', text: t } as ContentBlock); +const prose = (t: string): NormalizedResponse => createMockResponse([text(t)]); +const empty = (): NormalizedResponse => createMockResponse([]); +const skipReply = (id = 'sr'): NormalizedResponse => + createMockResponse([{ type: 'tool_use', id, name: 'skip_reply', input: {} } as ContentBlock], 'tool_use'); + +async function within(promise: Promise, label: string, ms = 8000): Promise { + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([promise, new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error('Timed out waiting for ' + label)), ms); + })]); + } finally { + clearTimeout(timer); + } +} + +describe('autoContinue', () => { + let tempDir: string; + let membrane: MockMembrane; + let module: TriggerModule; + + beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), 'auto-continue-test-')); + membrane = new MockMembrane(); + module = new TriggerModule(); + }); + + afterEach(() => { + rmSync(tempDir, { recursive: true, force: true }); + }); + + async function createFramework( + autoContinue: AgentConfig['autoContinue'], + proseRouting: AgentConfig['proseRouting'] = 'locus', + ): Promise { + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'You are a test agent.', proseRouting, autoContinue }], + modules: [module], + }); + module.framework = framework; + return framework; + } + + /** + * Answer the Nth turn with responses[N-1] the moment it starts (the mock + * hands every queued response to the first stream otherwise). A turn past + * the script calls skip_reply, so a stray continuation ends instead of + * hanging and the test fails on its count. Returns each turn's trigger. + */ + function script(framework: AgentFramework, responses: NormalizedResponse[]): InferenceRequest[] { + const triggers: InferenceRequest[] = []; + const internals = framework as unknown as { activeTurnTriggers: Map }; + framework.onTrace((e) => { + if ((e as { type: string }).type !== 'inference:started') return; + const turn = triggers.length; + const trigger = internals.activeTurnTriggers?.get('assistant'); + triggers.push(trigger ?? ({ reason: 'unknown' } as InferenceRequest)); + membrane.pushResponse(responses[turn] ?? skipReply(`stray-${turn}`)); + }); + return triggers; + } + + function trigger(framework: AgentFramework, content = 'hello'): void { + framework.pushEvent({ type: 'external-message', source: 'test', content, metadata: {} } as unknown as ProcessEvent); + } + + /** The window's rows: text and kind. */ + function rows(framework: AgentFramework): Array<{ text: string; kind?: string }> { + const messages = framework.getAgent('assistant')!.getContextManager().getAllMessages() as Array<{ + content: Array<{ type: string; text?: string }>; + metadata?: { kind?: string }; + }>; + return messages.map((m) => ({ + text: m.content.filter((b) => b.type === 'text').map((b) => b.text ?? '').join(''), + kind: m.metadata?.kind, + })); + } + + const notices = (framework: AgentFramework) => rows(framework).filter((r) => r.kind === 'auto-continue').map((r) => r.text); + + it('continues a turn that ends in prose, after its receipt, in the same channel, until skip_reply', async () => { + const framework = await createFramework(true); + const routed = stubRegistry(framework); + const triggers = script(framework, [prose('First thought.'), prose('Second thought.'), skipReply()]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + + assert.equal(triggers.length, 3, 'two continuations, then skip_reply ends the run'); + assert.deepEqual(triggers.slice(1).map((t) => t.reason), ['auto-continue', 'auto-continue']); + assert.ok(triggers.slice(1).every((t) => t.channelId === 'world:commons' && t.channelServerId === 'stub'), 'each continuation keeps the locus, server and id'); + assert.deepEqual(routed, [ + { text: 'First thought.', locus: 'world:commons' }, + { text: 'Second thought.', locus: 'world:commons' }, + ]); + const all = rows(framework); + const firstNotice = all.findIndex((r) => r.kind === 'auto-continue'); + assert.equal(all[firstNotice - 1]?.kind, 'delivery-receipt', 'the notice follows the turn\'s receipt'); + assert.deepEqual(notices(framework), [ + autoContinueNotice(1, 3, 'call skip_reply'), + autoContinueNotice(2, 3, 'call skip_reply'), + ]); + assert.match(notices(framework)[0]!, /To stop and wait for the next event, call skip_reply\.$/); + assert.match(notices(framework)[1]!, /Continuation 2 of 3 in a row: one more is left\.$/); + } finally { + await framework.stop(); + } + }); + + it('continues a turn that ends with no text', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const triggers = script(framework, [empty(), skipReply()]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2); + assert.equal(triggers[1]!.reason, 'auto-continue'); + assert.equal(notices(framework).length, 1); + } finally { + await framework.stop(); + } + }); + + it('stops at maxConsecutive, and says so on the last continuation', async () => { + const framework = await createFramework({ maxConsecutive: 2 }); + stubRegistry(framework); + // The model never skips: the cap ends the run. + const triggers = script(framework, [prose('one'), prose('two'), prose('three'), prose('four')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 3, 'the turn and two continuations'); + assert.deepEqual(notices(framework), [ + autoContinueNotice(1, 2, 'call skip_reply'), + autoContinueNotice(2, 2, 'call skip_reply'), + ]); + assert.match(notices(framework)[0]!, /Continuation 1 of 2 in a row: one more is left\.$/); + assert.match(notices(framework)[1]!, /This is the last continuation in a row \(2 of 2\): after this turn you wait for the next event either way\.$/); + } finally { + await framework.stop(); + } + }); + + it('starts a fresh run when a message wakes the agent after the cap', async () => { + const framework = await createFramework({ maxConsecutive: 1 }); + stubRegistry(framework); + const triggers = script(framework, [prose('a'), prose('b'), prose('c'), prose('d')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the first run'); + assert.equal(triggers.length, 2, 'one continuation, then the cap'); + trigger(framework, 'again'); + await within(framework.runUntilIdle(), 'the second run'); + assert.equal(triggers.length, 4, 'the message starts a turn, and its run continues once more'); + assert.equal(triggers[3]!.reason, 'auto-continue'); + } finally { + await framework.stop(); + } + }); + + it('does nothing when autoContinue is off', async () => { + const framework = await createFramework(undefined); + stubRegistry(framework); + const triggers = script(framework, [prose('Only this.')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the turn'); + assert.equal(triggers.length, 1); + assert.equal(notices(framework).length, 0); + } finally { + await framework.stop(); + } + }); + + for (const stop of ['max_tokens', 'refusal'] as const) { + it(`does not continue a turn that ended with ${stop}`, async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const triggers = script(framework, [createMockResponse([text('cut off')], stop)]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the turn'); + assert.equal(triggers.length, 1); + assert.equal(notices(framework).length, 0); + } finally { + await framework.stop(); + } + }); + } + + it('does not continue a silent heartbeat tick', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const triggers = script(framework, [prose('private check')]); + const internals = framework as unknown as { pendingRequests: InferenceRequest[] }; + try { + internals.pendingRequests.push({ agentName: 'assistant', reason: 'heartbeat', source: 'framework', timestamp: Date.now(), suppressProse: true } as InferenceRequest); + await within(framework.runUntilIdle(), 'the silent turn'); + assert.equal(triggers.length, 1); + assert.equal(notices(framework).length, 0); + } finally { + await framework.stop(); + } + }); + + it('a tool-ended turn clears the run, so the next run counts from one', async () => { + const framework = await createFramework({ maxConsecutive: 2 }); + stubRegistry(framework); + const internals = framework as unknown as { autoContinueStreaks: Map; pendingRequests: InferenceRequest[] }; + const triggers = script(framework, [skipReply(), prose('after'), skipReply('sr2')]); + try { + // A run already at its cap, then a framework wake (not a cause of its + // own) whose turn ends with skip_reply. + internals.autoContinueStreaks.set('assistant', 2); + internals.pendingRequests.push({ agentName: 'assistant', reason: 'prose-bounce', source: 'framework', timestamp: Date.now() }); + await within(framework.runUntilIdle(), 'the skip'); + assert.equal(triggers.length, 1); + assert.equal(internals.autoContinueStreaks.has('assistant'), false, 'skip_reply cleared the run'); + internals.pendingRequests.push({ agentName: 'assistant', reason: 'prose-bounce', source: 'framework', timestamp: Date.now() }); + await within(framework.runUntilIdle(), 'the next run'); + assert.equal(triggers.length, 3, 'a prose end continues again'); + assert.deepEqual(notices(framework), [autoContinueNotice(1, 2, 'call skip_reply')]); + } finally { + await framework.stop(); + } + }); + + it('explicit: a turn whose last envelope is >>skip_reply waits; one that ends on a delivery continues', async () => { + const framework = await createFramework(true, 'explicit'); + const routed = stubRegistry(framework); + const triggers = script(framework, [prose('>>#alpha posted'), prose('>>#alpha more\n>>skip_reply thinking it over')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run'); + assert.equal(triggers.length, 2, 'the delivery continues; the >>skip_reply ending waits'); + assert.deepEqual(routed.map((r) => r.text), ['posted', 'more']); + assert.deepEqual(notices(framework), [autoContinueNotice(1, 3, 'call skip_reply')]); + } finally { + await framework.stop(); + } + }); + + it('hybrid: a turn left private by >>>skip_reply waits', async () => { + const framework = await createFramework(true, 'hybrid'); + const routed = stubRegistry(framework); + const triggers = script(framework, [prose('Out loud.'), prose('>>>skip_reply\nkeeping this to myself')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run'); + assert.equal(triggers.length, 2); + assert.equal(triggers[1]!.channelId, 'world:commons', 'the continuation keeps the hybrid locus'); + assert.equal(triggers[1]!.channelServerId, 'stub'); + assert.deepEqual(routed.map((r) => r.text), ['Out loud.']); + assert.equal(notices(framework).length, 1); + } finally { + await framework.stop(); + } + }); + + it("a turn's ' !' continuation runs instead, with no notice and no count", async () => { + const framework = await createFramework(true, 'explicit'); + stubRegistry(framework); + const internals = framework as unknown as { autoContinueStreaks: Map }; + const triggers = script(framework, [prose('>>#alpha ! step one'), prose('>>skip_reply done')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run'); + assert.equal(triggers.length, 2); + assert.equal(triggers[1]!.reason, 'prose-continuation'); + assert.equal(notices(framework).length, 0); + assert.equal(internals.autoContinueStreaks.has('assistant'), false); + } finally { + await framework.stop(); + } + }); +}); + +describe('autoContinue config', () => { + const at = (autoContinue: unknown) => ({ name: 'a', model: 'm', systemPrompt: '', autoContinue } as AgentConfig); + + it('reads off, true, and a cap', () => { + assert.equal(resolveAutoContinue(at(undefined)), undefined); + assert.equal(resolveAutoContinue(at(false)), undefined); + assert.equal(resolveAutoContinue(at(null)), undefined); + assert.deepEqual(resolveAutoContinue(at(true)), { maxConsecutive: 3 }); + assert.deepEqual(resolveAutoContinue(at({})), { maxConsecutive: 3 }); + assert.deepEqual(resolveAutoContinue(at({ maxConsecutive: 7 })), { maxConsecutive: 7 }); + }); + + it('refuses anything else', () => { + for (const bad of ['yes', 1, [], { maxConsecutive: 0 }, { maxConsecutive: 21 }, { maxConsecutive: 2.5 }, { maxConsecutive: '3' }]) { + assert.throws(() => resolveAutoContinue(at(bad)), /Agent "a" autoContinue/, JSON.stringify(bad)); + } + }); +}); + +describe('autoContinueNotice', () => { + it('names the way to stop, and the count only near the cap', () => { + assert.equal( + autoContinueNotice(1, 3, 'call skip_reply'), + '[continue] Your last turn ended without skip_reply, so another one starts now. To stop and wait for the next event, call skip_reply.', + ); + assert.match(autoContinueNotice(2, 3, 'call skip_reply'), / Continuation 2 of 3 in a row: one more is left\.$/); + assert.match(autoContinueNotice(3, 3, 'call skip_reply'), / This is the last continuation in a row \(3 of 3\)/); + assert.match(autoContinueNotice(1, 1, 'call skip_reply'), /last continuation in a row \(1 of 1\)/); + assert.equal(autoContinueNotice(1, 5, undefined), '[continue] Your last turn ended on its own, so another one starts now.'); + }); +}); From 900fc6d360dd8389e8b3b32cfe2e0a0050f8450a Mon Sep 17 00:00:00 2001 From: antra-tess Date: Thu, 22 Oct 2026 23:05:32 -0700 Subject: [PATCH 2/6] =?UTF-8?q?fix(prose):=20review=20of=20#744=20?= =?UTF-8?q?=E2=80=94=20notice=20at=20the=20continuation's=20start,=20its?= =?UTF-8?q?=20channel=20outranked,=20guards=20tested?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The [continue] notice is written by the scheduler as the turn that takes the auto-continue wake starts (autoContinuePending → writeAutoContinueNotice), and only when the continuation is that turn's whole cause, so it never claims a turn that doesn't run. A wake the inference policy drops writes no notice and gives back its count (dropAutoContinue); a notice that can't be stored starts no continuation and gives back its count. - The continuation's channel is the batch's fallback: any other channel-bearing wake (ambient included) outranks it, though queued first. - resolveAutoContinue: docstring says what it accepts; a string or number is told it can be true, false or an object (strings: unquoted). - Docs/JSDoc/changelog: the receipt precedes the notice only outside explicit mode; a batched message takes the turn with no notice. - Tests (26): batched message (channel, no notice, fresh run), policy drop, notice store failure, stop instruction for explicit/hybrid/locus agents without skip_reply, stop_sequence continues, no_progress/round_limit don't, ephemeral/stopping/unregistered guards, the config message. Co-Authored-By: Claude Opus 5.5 --- .../feat-continue-after-prose.added.md | 19 ++- docs/auto-continue.md | 22 ++- src/agent.ts | 17 +- src/auto-continue.ts | 12 ++ src/framework.ts | 104 ++++++++--- src/types/agent.ts | 12 +- test/auto-continue.test.ts | 161 +++++++++++++++++- 7 files changed, 295 insertions(+), 52 deletions(-) diff --git a/changelog.d/feat-continue-after-prose.added.md b/changelog.d/feat-continue-after-prose.added.md index 604c6f07b..c92a4eaec 100644 --- a/changelog.d/feat-continue-after-prose.added.md +++ b/changelog.d/feat-continue-after-prose.added.md @@ -3,12 +3,13 @@ away, until the agent calls `skip_reply` (or another tool that ends the turn), ends on a `>>skip_reply` / `>>>skip_reply` envelope, or reaches `maxConsecutive` continuations in a row (default 3, 1 to 20; the count - starts again on any wake with a cause of its own). The next turn reads a - short `[continue]` system row, stored after the turn's `[delivered]` - receipt, that names `skip_reply` as the way to stop and, near the cap, how - many continuations are left; it keeps the turn's locus. Its wake has - reason `auto-continue`. Refusals, `max_tokens` cut-offs, the stall guard's - ends, silent heartbeat ticks and ephemeral runs never continue, and a - ` !` continuation runs instead. A malformed setting is refused when the - agent is built. New exported type `AutoContinueConfig`. See - docs/auto-continue.md. + starts again on any wake with a cause of its own). As the next turn starts + it reads a short `[continue]` system row (written only when the + continuation is that turn's whole cause, never for a wake that is dropped) + that names `skip_reply` as the way to stop and, near the cap, how many + continuations are left; it keeps the turn's locus unless a batched message + names another channel. Its wake has reason `auto-continue`. Refusals, + `max_tokens` cut-offs, the stall guard's ends, silent heartbeat ticks and + ephemeral runs never continue, and a ` !` continuation runs instead. A + malformed setting is refused when the agent is built. New exported type + `AutoContinueConfig`. See docs/auto-continue.md. diff --git a/docs/auto-continue.md b/docs/auto-continue.md index d01d3a0fb..361254d3e 100644 --- a/docs/auto-continue.md +++ b/docs/auto-continue.md @@ -28,8 +28,10 @@ A turn continues when the model ended it (stop reason `end_turn` or ## What the agent reads -Before the next turn a short system row joins the window, after the turn's -`[delivered]` receipt, so it is the newest thing the turn reads: +As the next turn starts, a short system row joins the window, so it is the +newest thing the turn reads (after the last turn's `[delivered]` receipt, +outside explicit mode, whose envelopes acknowledge themselves and write no +receipt): ``` [continue] Your last turn ended without skip_reply, so another one starts now. To stop and wait for the next event, call skip_reply. @@ -46,8 +48,12 @@ An agent that can't call `skip_reply` is told to end with a `>>skip_reply` (explicit) or `>>>skip_reply` (hybrid) envelope instead; in locus or disabled mode without the tool, the row names no way to stop and the cap ends the run. -The row is stored at the turn's end, after its response, and never retracted, -so the prompt-cache prefix stays stable. It is deliberately short and +The row is written by the scheduler as it starts the turn that takes the +continuation's wake, never earlier, so it never claims a turn that doesn't +run: a wake the inference policy drops writes no row and gives back its +count, and a row that can't be stored starts no continuation. It is appended +before the turn compiles and never retracted, so the prompt-cache prefix +stays stable. It is deliberately short and event-shaped: longer instruction blocks about output, sent as user-role messages, have tripped provider classifiers. @@ -56,8 +62,12 @@ messages, have tripped provider classifiers. The continuation's wake has source `framework` and reason `auto-continue`. In locus and hybrid modes it carries the turn's locus as its channel, so the next turn's plain prose lands where the last one's did rather than in the -process-global last-inbound channel. A message that arrives meanwhile joins -the same turn, as any batched wake does. +process-global last-inbound channel. + +A message that arrives meanwhile joins the same turn, as any batched wake +does, and then that turn is the message's: its channel outranks the +continuation's (ambient included, though the continuation's wake was queued +later), no `[continue]` row is written, and the run's count starts again. ## The cap diff --git a/src/agent.ts b/src/agent.ts index 62eadfdb8..06e7c0adb 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -5,7 +5,7 @@ import type { CacheWireReceipt, KvUnifiedRequestHooks } from './kv-unified-wire. import { ToolResultGuard } from './tool-result-guard.js'; import { withoutTokenEstimates } from './thinking-token-stamp.js'; import { injectionsAsStored } from './text-as-stored.js'; -import { assertBooleanSetting, holdsSettings } from './config-checks.js'; +import { assertBooleanSetting, holdsSettings, kindOf } from './config-checks.js'; import { CONTINUE_ROW, continueTurn, @@ -154,14 +154,21 @@ const MAX_AUTO_CONTINUE_MAX = 20; /** * `AgentConfig.autoContinue` resolved: undefined when off (left out, null or - * false), else its cap. Anything else is refused: `true`, or an object whose - * `maxConsecutive` is left out or a whole number from 1 to 20. + * false), else its cap. `true` takes the default cap; an object takes + * `maxConsecutive`, a whole number from 1 to 20 (the default when left out). + * Any other value is refused. */ export function resolveAutoContinue(config: AgentConfig): { maxConsecutive: number } | undefined { const setting = `Agent "${config.name}" autoContinue`; - const value = config.autoContinue; + const value: unknown = config.autoContinue; if (value === true) return { maxConsecutive: DEFAULT_AUTO_CONTINUE_MAX }; - if (!holdsSettings(setting, value, '{ maxConsecutive: 3 }')) return undefined; + if (value !== undefined && value !== null && value !== false && (typeof value !== 'object' || Array.isArray(value))) { + throw new Error( + `${setting} must be true, false or an object such as { maxConsecutive: 3 }, not ${kindOf(value)}` + + (typeof value === 'string' ? ' (true and false go unquoted)' : ''), + ); + } + if (!holdsSettings(setting, value as AgentConfig['autoContinue'], '{ maxConsecutive: 3 }')) return undefined; const max = (value as AutoContinueConfig).maxConsecutive; if (max === undefined) return { maxConsecutive: DEFAULT_AUTO_CONTINUE_MAX }; if (typeof max !== 'number' || !Number.isInteger(max) || max < 1 || max > MAX_AUTO_CONTINUE_MAX) { diff --git a/src/auto-continue.ts b/src/auto-continue.ts index d65cb29ad..d613522aa 100644 --- a/src/auto-continue.ts +++ b/src/auto-continue.ts @@ -11,6 +11,18 @@ * `max_tokens` end was cut off rather than chosen, and `no_progress` and * `round_limit` are the stall guard's verdict that the turn wasn't moving. */ +/** The reason on a continuation's wake (InferenceRequest.reason). */ +export const AUTO_CONTINUE_REASON = 'auto-continue'; + +/** What a queued continuation's notice says, held until its turn starts. */ +export interface PendingAutoContinue { + /** This continuation's place in the run, 1 to `max`. */ + count: number; + max: number; + /** How the agent stops (autoContinueNotice's `stop`), if it can. */ + stop: string | undefined; +} + export const AUTO_CONTINUE_STOP_REASONS: ReadonlySet = new Set(['end_turn', 'stop_sequence']); /** diff --git a/src/framework.ts b/src/framework.ts index 4de099437..818cf6536 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -156,7 +156,7 @@ import { DEFAULT_OUTPUT_BOUNDS } from './code-execution/isolate/output.js'; import type { PreludeTool } from './code-execution/isolate/prelude.js'; import { bounded, renderList, renderReturn, VALUE_CHARS } from './code-execution/isolate/render.js'; import { splitProseRuns, withoutRoundsProse } from './prose-segments.js'; -import { AUTO_CONTINUE_STOP_REASONS, autoContinueNotice } from './auto-continue.js'; +import { AUTO_CONTINUE_REASON, AUTO_CONTINUE_STOP_REASONS, autoContinueNotice, type PendingAutoContinue } from './auto-continue.js'; import { cumulativeDelta } from './usage-accounting.js'; import { stampThinkingTokenEstimates } from './thinking-token-stamp.js'; import { contentAsStored, nameAsStored } from './text-as-stored.js'; @@ -2746,6 +2746,9 @@ export class AgentFramework { * maxConsecutive. Cleared when a turn starts with a cause of its own * (processInferenceRequests) or a tool ends a turn (skip_reply). */ private autoContinueStreaks: Map = new Map(); + /** A queued continuation's notice, per agent, until the turn that takes its + * wake starts (writeAutoContinueNotice) or the wake is dropped. */ + private autoContinuePending: Map = new Map(); private pendingAssistantBlocks: Map = new Map(); /** Streams the FRAMEWORK cancelled for non-terminal reasons, keyed * `${agentName}:${streamId}`: an endTurn tool result or a context-budget @@ -14779,10 +14782,31 @@ export class AgentFramework { if (requests.some((r) => resumesLogicalTurn(r.reason))) { this.endContinuedTurn(agentName, { kind: 'dropped', why: 'the inference policy dropped its continuation' }); } + if (requests.some((r) => r.reason === AUTO_CONTINUE_REASON)) this.dropAutoContinue(agentName, 'the inference policy dropped its wake'); this.endPendingPrimary(agentName); continue; } + // An autoContinue wake's notice is written now, as the turn that takes + // it starts, so it never claims a turn that doesn't run. Only when the + // continuation is the turn's whole cause: a batch with a cause of its + // own (a message, a heartbeat) is that cause's turn, and its run of + // continuations starts again below. A notice that can't be stored + // starts no continuation: without it the turn would wake to nothing. + if (requests.some((r) => r.reason === AUTO_CONTINUE_REASON)) { + const pending = this.autoContinuePending.get(agentName); + this.autoContinuePending.delete(agentName); + if (pending && requests.every((r) => FRAMEWORK_WAKE_SOURCES.has(r.source)) + && !this.writeAutoContinueNotice(agent, pending)) { + this.autoContinueStreaks.set(agentName, pending.count - 1); + requests = requests.filter((r) => r.reason !== AUTO_CONTINUE_REASON); + if (requests.length === 0) { + this.endPendingPrimary(agentName); + continue; + } + } + } + // Start streaming inference (non-blocking — driveStream runs in background) // The request that grants the turn-alive exception must also define the // downstream turn semantics. An ordinary wake can already be pending @@ -14829,12 +14853,21 @@ export class AgentFramework { const eventAt = (r: InferenceRequest): number => r.wakeAt ?? r.timestamp; let ambientReq: InferenceRequest | undefined; let addressedReq: InferenceRequest | undefined; + // An autoContinue wake names the finished turn's locus. It speaks for + // the turn only when nothing else in the batch names a channel: a + // message that arrived meanwhile, ambient included, outranks it, though + // its wake was queued first. + let continuationReq: InferenceRequest | undefined; for (const r of requests) { if (!r.channelId) continue; + if (r.reason === AUTO_CONTINUE_REASON) { + continuationReq ??= r; + continue; + } if (!ambientReq || eventAt(r) >= eventAt(ambientReq)) ambientReq = r; if (r.addressed && (!addressedReq || eventAt(r) >= eventAt(addressedReq))) addressedReq = r; } - const channelReq = addressedReq ?? ambientReq; + const channelReq = addressedReq ?? ambientReq ?? continuationReq; // Telemetry provenance (who/where woke the agent). An ADDRESSED winner // is its own provenance, even when it names no author (a push event // can route without an author id): borrowing another request's author @@ -16578,11 +16611,14 @@ export class AgentFramework { * runs instead). A tool that ends the turn (skip_reply, sleep) never * reaches here, and clears the run's count. * - * The next turn reads a `[continue]` notice, stored now so it is the newest - * row the turn compiles (after the receipt), and keeps this turn's locus - * through its wake's channelId. A run is capped at maxConsecutive; the - * last allowed notice says so, and the cap holds until a turn starts with - * a cause of its own. + * This queues the wake and holds the notice it reads + * (autoContinuePending): the scheduler writes the `[continue]` row as the + * turn that takes the wake starts, and only when the continuation is that + * turn's whole cause (processInferenceRequests); a dropped wake writes + * none and gives back its count (dropAutoContinue). The wake names this + * turn's locus, which a batched message's channel outranks. A run is + * capped at maxConsecutive; the last allowed notice says so, and the cap + * holds until a turn starts with a cause of its own. */ private autoContinueAfterTurn( agent: Agent, @@ -16623,31 +16659,14 @@ export class AgentFramework { : agent.proseRouting === 'hybrid' ? 'end with a >>>skip_reply envelope' : undefined; - const text = autoContinueNotice(count, config.maxConsecutive, stop); - try { - // Stored straight into the window, as the receipt just was: the turn's - // stream is over, so the row lands after its response, and the next - // compile reads it as the newest row. - const id = agent.getContextManager().addMessage( - 'user', - [{ type: 'text', text }], - { system: true, kind: 'auto-continue', continuation: count, maxConsecutive: config.maxConsecutive } as MessageMetadata, - ); - this.emitTrace({ type: 'message:added', messageId: id, source: 'auto-continue' }); - } catch (err) { - // Without its notice the next turn would wake to nothing it can read - // as a cause: no notice, no continuation. - console.error(`[continue] ${name}: notice not stored, so not continued:`, err); - this.autoContinueStreaks.set(name, count - 1); - return; - } + this.autoContinuePending.set(name, { count, max: config.maxConsecutive, stop }); console.error( `[continue] ${name}: turn ended (${stopReason}) without skip_reply — continuation ${count} of ${config.maxConsecutive}` + (routing.locus ? ` in ${this.channelRegistry?.nameShown(routing.locus) ?? routing.locus.channelId}` : ''), ); this.pendingRequests.push({ agentName: name, - reason: 'auto-continue', + reason: AUTO_CONTINUE_REASON, source: 'framework', timestamp: Date.now(), // The next turn speaks where this one did (locus and hybrid modes): @@ -16656,6 +16675,39 @@ export class AgentFramework { }); } + /** + * Store a continuation's `[continue]` notice in the agent's window, as the + * turn that takes its wake starts: no turn is running, so the row lands + * after everything the last turn wrote (its `[delivered]` receipt + * included, outside explicit mode) and is the newest row the turn + * compiles. False when it couldn't be stored. + */ + private writeAutoContinueNotice(agent: Agent, pending: PendingAutoContinue): boolean { + const text = autoContinueNotice(pending.count, pending.max, pending.stop); + try { + const id = agent.getContextManager().addMessage( + 'user', + [{ type: 'text', text }], + { system: true, kind: 'auto-continue', continuation: pending.count, maxConsecutive: pending.max } as MessageMetadata, + ); + this.emitTrace({ type: 'message:added', messageId: id, source: 'auto-continue' }); + return true; + } catch (err) { + console.error(`[continue] ${agent.name}: notice not stored, so not continued:`, err); + return false; + } + } + + /** A queued continuation whose wake starts no turn: its notice is never + * written, and the run's count goes back to what it was before. */ + private dropAutoContinue(agentName: string, why: string): void { + const pending = this.autoContinuePending.get(agentName); + if (!pending) return; + this.autoContinuePending.delete(agentName); + this.autoContinueStreaks.set(agentName, pending.count - 1); + console.error(`[continue] ${agentName}: continuation ${pending.count} dropped — ${why}`); + } + /** The agent's routing state for its current turn, made when it has none. */ private routingOf(agentName: string): TurnRouting { let routing = this.turnRouting.get(agentName); diff --git a/src/types/agent.ts b/src/types/agent.ts index abb441e40..1a7bd9522 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -235,10 +235,14 @@ export interface AgentConfig { * whose last prose envelope is `>>skip_reply` (explicit) or `>>>skip_reply` * (hybrid), pauses as usual. * - * Before the next turn a short system notice joins the window, after the - * turn's `[delivered]` receipt, saying that the turn continues and that - * skip_reply stops it; near the cap it also says how many continuations - * are left. The next turn keeps the turn's locus (locus and hybrid modes). + * As the next turn starts, a short system notice joins the window (after + * the last turn's `[delivered]` receipt, outside explicit mode, which + * writes none), saying that the turn continues and that skip_reply stops + * it; near the cap it also says how many continuations are left. It's + * written only when the continuation is that turn's whole cause: a + * message batched with it makes it that message's turn, with no notice, + * in the message's channel. Otherwise the next turn keeps the turn's + * locus (locus and hybrid modes). * At most `maxConsecutive` (default 3) continuations run in a row: the * count starts again when a turn starts with a cause of its own (a * message, a heartbeat, a timer) or a tool ends one. Silent heartbeat diff --git a/test/auto-continue.test.ts b/test/auto-continue.test.ts index c4b51c6a7..39f7b2338 100644 --- a/test/auto-continue.test.ts +++ b/test/auto-continue.test.ts @@ -126,11 +126,15 @@ describe('autoContinue', () => { async function createFramework( autoContinue: AgentConfig['autoContinue'], proseRouting: AgentConfig['proseRouting'] = 'locus', + allowedTools?: AgentConfig['allowedTools'], ): Promise { const framework = await AgentFramework.create({ storePath: join(tempDir, 'test.chronicle'), membrane: membrane.asMembrane(), - agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'You are a test agent.', proseRouting, autoContinue }], + agents: [{ + name: 'assistant', model: 'test-model', systemPrompt: 'You are a test agent.', proseRouting, autoContinue, + ...(allowedTools ? { allowedTools } : {}), + }], modules: [module], }); module.framework = framework; @@ -269,7 +273,21 @@ describe('autoContinue', () => { } }); - for (const stop of ['max_tokens', 'refusal'] as const) { + it('continues a turn that ends on a stop sequence', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const triggers = script(framework, [createMockResponse([text('stopped at the sequence')], 'stop_sequence'), skipReply()]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2); + assert.equal(triggers[1]!.reason, 'auto-continue'); + } finally { + await framework.stop(); + } + }); + + for (const stop of ['max_tokens', 'refusal', 'no_progress', 'round_limit'] as const) { it(`does not continue a turn that ended with ${stop}`, async () => { const framework = await createFramework(true); stubRegistry(framework); @@ -322,6 +340,140 @@ describe('autoContinue', () => { } }); + it("a message that arrives during the turn takes the next turn: its channel, no notice, a fresh run", async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { pendingRequests: InferenceRequest[]; autoContinueStreaks: Map }; + const triggers = script(framework, [prose('Working in the commons.'), skipReply()]); + // An ambient message in another channel, queued while the first turn + // runs: it waits for that turn, then batches with the continuation. + let injected = false; + framework.onTrace((e) => { + if ((e as { type: string }).type !== 'inference:started' || injected) return; + injected = true; + internals.pendingRequests.push({ + agentName: 'assistant', reason: 'channel-message', source: 'stub', timestamp: Date.now(), + channelId: 'chan-b', channelServerId: 'stub', + }); + }); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2, 'one more turn, not two'); + assert.ok(triggers[1]!.reason === 'channel-message' || triggers[1]!.reason === 'auto-continue'); + assert.equal(triggers[1]!.channelId, 'chan-b', "the message's channel outranks the continuation's, though queued first"); + assert.deepEqual(notices(framework), [], 'the message is the turn\'s cause: no [continue] row'); + assert.equal(internals.autoContinueStreaks.has('assistant'), false, 'and the run starts again'); + } finally { + await framework.stop(); + } + }); + + it('a continuation the inference policy drops writes no notice and gives back its count', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { + inferencePolicy: { shouldInfer(agent: string, requests: InferenceRequest[]): boolean }; + autoContinueStreaks: Map; + autoContinuePending: Map; + }; + internals.inferencePolicy = { shouldInfer: (_agent, requests) => !requests.some((r) => r.reason === 'auto-continue') }; + const triggers = script(framework, [prose('Only this.')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 1); + assert.deepEqual(notices(framework), [], 'no notice claims a turn that never ran'); + assert.equal(internals.autoContinueStreaks.get('assistant') ?? 0, 0, 'the count is given back'); + assert.equal(internals.autoContinuePending.size, 0); + } finally { + await framework.stop(); + } + }); + + it('a notice that cannot be stored starts no continuation and gives back its count', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { autoContinueStreaks: Map }; + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage(role: string, content: unknown, metadata?: { kind?: string }): unknown; + }; + const original = cm.addMessage.bind(cm); + cm.addMessage = (role, content, metadata) => { + if (metadata?.kind === 'auto-continue') throw new Error('synthetic store failure'); + return original(role, content, metadata); + }; + const triggers = script(framework, [prose('Only this.')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 1, 'no turn wakes without its notice'); + assert.equal(internals.autoContinueStreaks.get('assistant') ?? 0, 0); + } finally { + await framework.stop(); + } + }); + + for (const [mode, first, stop] of [ + ['explicit', '>>#alpha hello', 'end with a >>skip_reply envelope'], + ['hybrid', 'hello', 'end with a >>>skip_reply envelope'], + ['locus', 'hello', undefined], + ] as const) { + it(`${mode}: an agent that can't call skip_reply is told ${stop ? 'its envelope' : 'no way to stop'}`, async () => { + const framework = await createFramework({ maxConsecutive: 1 }, mode, ['robot--move']); + stubRegistry(framework); + // Without skip_reply, the cap ends the run. + const triggers = script(framework, [prose(first), prose(first), prose(first)]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2); + assert.deepEqual(notices(framework), [autoContinueNotice(1, 1, stop)]); + } finally { + await framework.stop(); + } + }); + } + + it('queues nothing for an ephemeral run, a stopping framework or an agent no longer registered', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string): void; + pendingRequests: InferenceRequest[]; + ephemeralRuns: Map; + providerAdmissionClosed: boolean; + autoContinuePending: Map; + autoContinueStreaks: Map; + }; + const agent = framework.getAgent('assistant')!; + const queued = () => internals.pendingRequests.filter((r) => r.reason === 'auto-continue').length; + try { + internals.ephemeralRuns.set('assistant', {}); + internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn'); + internals.ephemeralRuns.delete('assistant'); + assert.equal(queued(), 0, 'an ephemeral run ends with its result'); + + internals.providerAdmissionClosed = true; + internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn'); + internals.providerAdmissionClosed = false; + assert.equal(queued(), 0, 'a stopping framework starts nothing'); + + internals.autoContinueAfterTurn(Object.create(agent), {}, undefined, 'end_turn'); + assert.equal(queued(), 0, 'an agent object no longer the registered one starts nothing'); + + // The same call for the registered agent does queue one. + internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn'); + assert.equal(queued(), 1); + assert.equal(internals.autoContinuePending.size, 1); + } finally { + internals.pendingRequests.length = 0; + internals.autoContinuePending.clear(); + internals.autoContinueStreaks.clear(); + await framework.stop(); + } + }); + it('explicit: a turn whose last envelope is >>skip_reply waits; one that ends on a delivery continues', async () => { const framework = await createFramework(true, 'explicit'); const routed = stubRegistry(framework); @@ -384,6 +536,11 @@ describe('autoContinue config', () => { assert.deepEqual(resolveAutoContinue(at({ maxConsecutive: 7 })), { maxConsecutive: 7 }); }); + it('tells a string or number it can be true, false or an object', () => { + assert.throws(() => resolveAutoContinue(at('true')), /must be true, false or an object such as \{ maxConsecutive: 3 \}, not a string \(true and false go unquoted\)/); + assert.throws(() => resolveAutoContinue(at(1)), /must be true, false or an object/); + }); + it('refuses anything else', () => { for (const bad of ['yes', 1, [], { maxConsecutive: 0 }, { maxConsecutive: 21 }, { maxConsecutive: 2.5 }, { maxConsecutive: '3' }]) { assert.throws(() => resolveAutoContinue(at(bad)), /Agent "a" autoContinue/, JSON.stringify(bad)); From 89a84132f275c9552ee5e0f79b2c7885aa5408a6 Mon Sep 17 00:00:00 2001 From: antra-tess Date: Thu, 22 Oct 2026 23:18:52 -0700 Subject: [PATCH 3/6] =?UTF-8?q?fix(prose):=20review=20of=20#744=20?= =?UTF-8?q?=E2=80=94=20refusal=20names=20the=20kind,=20never=20the=20value?= =?UTF-8?q?;=20harness-safe=20reset;=20doc=20placement?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - resolveAutoContinue's maxConsecutive refusal names the value's kind (a string, a number below 1, above 20, not whole), never the value: a ${VAR} landing there could be a credential (config-checks rule). Tested. - turn_ended's run reset uses optional chaining, like the scheduler's, for prototype-built harnesses. - AUTO_CONTINUE_STOP_REASONS carries its own doc comment. - Tests: the batched-message turn's trigger is pinned to the message; the silent tick arrives from a server, as real ticks do. Co-Authored-By: Claude Opus 5.5 --- src/agent.ts | 7 ++++++- src/auto-continue.ts | 12 ++++++------ src/framework.ts | 3 ++- test/auto-continue.test.ts | 19 +++++++++++++++++-- 4 files changed, 31 insertions(+), 10 deletions(-) diff --git a/src/agent.ts b/src/agent.ts index 06e7c0adb..3f6c55d5b 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -172,8 +172,13 @@ export function resolveAutoContinue(config: AgentConfig): { maxConsecutive: numb const max = (value as AutoContinueConfig).maxConsecutive; if (max === undefined) return { maxConsecutive: DEFAULT_AUTO_CONTINUE_MAX }; if (typeof max !== 'number' || !Number.isInteger(max) || max < 1 || max > MAX_AUTO_CONTINUE_MAX) { + // The value's kind, never the value (config-checks): a `${VAR}` that + // lands here could be a credential. + const kind = typeof max !== 'number' ? kindOf(max) + : !Number.isInteger(max) ? 'a number that isn\'t whole' + : max < 1 ? 'a number below 1' : `a number above ${MAX_AUTO_CONTINUE_MAX}`; throw new Error( - `${setting}.maxConsecutive must be a whole number from 1 to ${MAX_AUTO_CONTINUE_MAX}, not ${JSON.stringify(max) ?? String(max)}: ` + + `${setting}.maxConsecutive must be a whole number from 1 to ${MAX_AUTO_CONTINUE_MAX}, not ${kind}: ` + 'the most turns in a row that continue without skip_reply before the agent waits for the next event', ); } diff --git a/src/auto-continue.ts b/src/auto-continue.ts index d613522aa..e03f77658 100644 --- a/src/auto-continue.ts +++ b/src/auto-continue.ts @@ -5,12 +5,6 @@ * notice the next turn reads. */ -/** - * Stop reasons of a turn the model ended itself. The others don't continue: - * a refusal has its own handling (retries, rewind, the reaction), a - * `max_tokens` end was cut off rather than chosen, and `no_progress` and - * `round_limit` are the stall guard's verdict that the turn wasn't moving. - */ /** The reason on a continuation's wake (InferenceRequest.reason). */ export const AUTO_CONTINUE_REASON = 'auto-continue'; @@ -23,6 +17,12 @@ export interface PendingAutoContinue { stop: string | undefined; } +/** + * Stop reasons of a turn the model ended itself. The others don't continue: + * a refusal has its own handling (retries, rewind, the reaction), a + * `max_tokens` end was cut off rather than chosen, and `no_progress` and + * `round_limit` are the stall guard's verdict that the turn wasn't moving. + */ export const AUTO_CONTINUE_STOP_REASONS: ReadonlySet = new Set(['end_turn', 'stop_sequence']); /** diff --git a/src/framework.ts b/src/framework.ts index 818cf6536..a7cead518 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -18924,7 +18924,8 @@ export class AgentFramework { if (cancelKind === 'turn_ended') { // A tool ended the turn (skip_reply, sleep): the agent stopped on // purpose, so a later autoContinue run counts from one again. - this.autoContinueStreaks.delete(agent.name); + // (Optional chaining: prototype-built harnesses leave the map unset.) + this.autoContinueStreaks?.delete(agent.name); if (turnProseRouting === 'explicit') { void turnSpeechChain.then(() => this.requestProseContinuation(agent, streamRouting)); } else { diff --git a/test/auto-continue.test.ts b/test/auto-continue.test.ts index 39f7b2338..69212533b 100644 --- a/test/auto-continue.test.ts +++ b/test/auto-continue.test.ts @@ -309,7 +309,8 @@ describe('autoContinue', () => { const triggers = script(framework, [prose('private check')]); const internals = framework as unknown as { pendingRequests: InferenceRequest[] }; try { - internals.pendingRequests.push({ agentName: 'assistant', reason: 'heartbeat', source: 'framework', timestamp: Date.now(), suppressProse: true } as InferenceRequest); + // As a real tick arrives: from its MCPL server, not the framework. + internals.pendingRequests.push({ agentName: 'assistant', reason: 'heartbeat', source: 'heartbeat', timestamp: Date.now(), suppressProse: true } as InferenceRequest); await within(framework.runUntilIdle(), 'the silent turn'); assert.equal(triggers.length, 1); assert.equal(notices(framework).length, 0); @@ -360,7 +361,7 @@ describe('autoContinue', () => { trigger(framework); await within(framework.runUntilIdle(), 'the run to settle'); assert.equal(triggers.length, 2, 'one more turn, not two'); - assert.ok(triggers[1]!.reason === 'channel-message' || triggers[1]!.reason === 'auto-continue'); + assert.equal(triggers[1]!.reason, 'channel-message', 'the message, queued first, is the turn\'s trigger'); assert.equal(triggers[1]!.channelId, 'chan-b', "the message's channel outranks the continuation's, though queued first"); assert.deepEqual(notices(framework), [], 'the message is the turn\'s cause: no [continue] row'); assert.equal(internals.autoContinueStreaks.has('assistant'), false, 'and the run starts again'); @@ -546,6 +547,20 @@ describe('autoContinue config', () => { assert.throws(() => resolveAutoContinue(at(bad)), /Agent "a" autoContinue/, JSON.stringify(bad)); } }); + + it('names the value\'s kind, never the value, which could be a secret', () => { + const secret = 'sk-ant-not-a-real-key'; + const message = (bad: unknown): string => { + try { resolveAutoContinue(at(bad)); } catch (err) { return (err as Error).message; } + return ''; + }; + assert.match(message({ maxConsecutive: secret }), /not a string:/); + assert.ok(!message({ maxConsecutive: secret }).includes(secret)); + assert.ok(!message(secret).includes(secret)); + assert.match(message({ maxConsecutive: 0 }), /not a number below 1:/); + assert.match(message({ maxConsecutive: 21 }), /not a number above 20:/); + assert.match(message({ maxConsecutive: 2.5 }), /not a number that isn't whole:/); + }); }); describe('autoContinueNotice', () => { From a676cf8a039cdb5dc853c9e04fbceebf2cd008d0 Mon Sep 17 00:00:00 2001 From: antra-tess Date: Thu, 22 Oct 2026 23:27:46 -0700 Subject: [PATCH 4/6] =?UTF-8?q?fix(prose):=20review=20of=20#744=20?= =?UTF-8?q?=E2=80=94=20a=20silent=20tick=20doesn't=20restart=20a=20run;=20?= =?UTF-8?q?refusal=20tested=20where=20agents=20are=20built?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - hasOwnCauseForRun(): a silent heartbeat tick isn't a cause of its own for an autoContinue run. Batched with a continuation it no longer resets the count (a resident whose ticks come faster than its turns would never reach the cap) and no longer suppresses the continuation's notice. The bounce streak's reset is unchanged. - Tests: a silent tick, and another framework wake, batched with a continuation keep the cap and the notice; create() and createEphemeralAgent refuse a bad autoContinue (the latter before registering the name); the ' !' test reads the count as the continuation's turn starts. - Docs/JSDoc: silent ticks don't restart a run. Co-Authored-By: Claude Opus 5.5 --- docs/auto-continue.md | 5 ++++- src/framework.ts | 20 +++++++++++++++++--- src/types/agent.ts | 3 ++- test/auto-continue.test.ts | 36 +++++++++++++++++++++++++++++++++++- test/config-switches.test.ts | 9 +++++++++ 5 files changed, 67 insertions(+), 6 deletions(-) diff --git a/docs/auto-continue.md b/docs/auto-continue.md index 361254d3e..7e32a765d 100644 --- a/docs/auto-continue.md +++ b/docs/auto-continue.md @@ -73,7 +73,10 @@ later), no `[continue]` row is written, and the run's count starts again. At most `maxConsecutive` continuations run in a row (default 3). The count starts again when a turn starts with a cause of its own (anything but a -framework wake), and when a tool ends a turn. Each continued turn is a full +framework wake or a silent heartbeat tick), and when a tool ends a turn. A +silent tick batched with a continuation neither restarts the run nor takes +its notice, so a resident whose ticks come faster than its turns still +reaches the cap. Each continued turn is a full inference over the context (mostly cache reads), and in locus mode its prose is published, so the cap bounds both the cost and the posts of an agent that never skips. diff --git a/src/framework.ts b/src/framework.ts index a7cead518..6a331ea70 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1302,6 +1302,15 @@ function isTurnContinuation(reason: string): boolean { * other source — a channel message, a heartbeat or other server event, a * gate flush, a timer, an operator nudge — is a cause of its own. */ const FRAMEWORK_WAKE_SOURCES = new Set(['framework', 'scheduler']); + +/** Whether a wake is a cause of its own for an autoContinue run: one from + * outside the framework that isn't a silent heartbeat tick. A batch holding + * one is that cause's turn: the run starts again, and no `[continue]` row is + * written for the continuation it carries. A silent tick is neither: batched + * with a continuation it neither restarts the run nor takes its notice. */ +function hasOwnCauseForRun(r: InferenceRequest): boolean { + return !FRAMEWORK_WAKE_SOURCES.has(r.source) && r.suppressProse !== true; +} const CONVERSATION_ROUTER_STATE_ID = 'framework/conversation-router'; const INFERENCE_LOG_ID = 'framework/inference-log'; const PROCESS_LOG_ID = 'framework/process-log'; @@ -14796,7 +14805,7 @@ export class AgentFramework { if (requests.some((r) => r.reason === AUTO_CONTINUE_REASON)) { const pending = this.autoContinuePending.get(agentName); this.autoContinuePending.delete(agentName); - if (pending && requests.every((r) => FRAMEWORK_WAKE_SOURCES.has(r.source)) + if (pending && !requests.some(hasOwnCauseForRun) && !this.writeAutoContinueNotice(agent, pending)) { this.autoContinueStreaks.set(agentName, pending.count - 1); requests = requests.filter((r) => r.reason !== AUTO_CONTINUE_REASON); @@ -14823,8 +14832,13 @@ export class AgentFramework { // (Optional chaining: prototype-built harnesses leave the map unset.) if (requests.some((r) => !FRAMEWORK_WAKE_SOURCES.has(r.source))) { this.proseBounceStreaks?.delete(agentName); - // Likewise autoContinue's cap bounds only a run of continuations - // that nothing else feeds. + } + // Likewise autoContinue's cap bounds only a run of continuations that + // nothing else feeds. A silent heartbeat tick doesn't feed it: batched + // with a continuation it would otherwise restart the run at every tick, + // and a resident whose ticks come faster than its turns would never + // reach the cap (hasOwnCauseForRun). + if (requests.some(hasOwnCauseForRun)) { this.autoContinueStreaks?.delete(agentName); } // Route this turn's auto-published speech to the channel that triggered diff --git a/src/types/agent.ts b/src/types/agent.ts index 1a7bd9522..8ccf3784e 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -245,7 +245,8 @@ export interface AgentConfig { * locus (locus and hybrid modes). * At most `maxConsecutive` (default 3) continuations run in a row: the * count starts again when a turn starts with a cause of its own (a - * message, a heartbeat, a timer) or a tool ends one. Silent heartbeat + * message, a timer, a heartbeat that isn't a silent tick) or a tool ends + * one. Silent heartbeat * ticks, refusals, `max_tokens` cut-offs, the stall guard's ends and * ephemeral runs never continue. A ` !` continuation the turn asked for * runs instead, and counts toward nothing. See docs/auto-continue.md. diff --git a/test/auto-continue.test.ts b/test/auto-continue.test.ts index 69212533b..35b82369f 100644 --- a/test/auto-continue.test.ts +++ b/test/auto-continue.test.ts @@ -370,6 +370,34 @@ describe('autoContinue', () => { } }); + for (const [what, extra] of [ + ['a silent heartbeat tick', { reason: 'heartbeat', source: 'heartbeat', suppressProse: true }], + ['another framework wake', { reason: 'prose-bounce', source: 'framework' }], + ] as const) { + it(`${what} batched with a continuation neither restarts the run nor takes its notice`, async () => { + // Cap 1: if the batched wake restarted the run, the second turn's end + // would continue a third time. + const framework = await createFramework({ maxConsecutive: 1 }); + stubRegistry(framework); + const internals = framework as unknown as { pendingRequests: InferenceRequest[] }; + const triggers = script(framework, [prose('one'), prose('two'), prose('three')]); + let injected = false; + framework.onTrace((e) => { + if ((e as { type: string }).type !== 'inference:started' || injected) return; + injected = true; + internals.pendingRequests.push({ agentName: 'assistant', timestamp: Date.now(), ...extra } as InferenceRequest); + }); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2, 'the cap still holds'); + assert.deepEqual(notices(framework), [autoContinueNotice(1, 1, 'call skip_reply')], 'the continuation\'s notice is written'); + } finally { + await framework.stop(); + } + }); + } + it('a continuation the inference policy drops writes no notice and gives back its count', async () => { const framework = await createFramework(true); stubRegistry(framework); @@ -512,13 +540,19 @@ describe('autoContinue', () => { stubRegistry(framework); const internals = framework as unknown as { autoContinueStreaks: Map }; const triggers = script(framework, [prose('>>#alpha ! step one'), prose('>>skip_reply done')]); + // The count as the ' !' continuation's turn starts: before its own end + // (a skip, which clears it anyway) can hide anything. + const countAtStart: Array = []; + framework.onTrace((e) => { + if ((e as { type: string }).type === 'inference:started') countAtStart.push(internals.autoContinueStreaks.get('assistant')); + }); try { trigger(framework); await within(framework.runUntilIdle(), 'the run'); assert.equal(triggers.length, 2); assert.equal(triggers[1]!.reason, 'prose-continuation'); assert.equal(notices(framework).length, 0); - assert.equal(internals.autoContinueStreaks.has('assistant'), false); + assert.equal(countAtStart[1], undefined, "the ' !' continuation counted toward nothing"); } finally { await framework.stop(); } diff --git a/test/config-switches.test.ts b/test/config-switches.test.ts index 00c789761..1a1991440 100644 --- a/test/config-switches.test.ts +++ b/test/config-switches.test.ts @@ -82,6 +82,8 @@ const THINKING_ENABLED = 'true asks the provider for extended thinking, ' + does const AUTO_REWIND = 'true redacts the turn that tripped a refusal and runs again, up to maxRewinds times, ' + doesnt; const TOOL_RESULT_GUARD = 'true turns on tool-output admission, with one guarded retry after a provider refusal, ' + doesnt; const TOOL_WRAPPER_PROSE_GUARD = 'true keeps a response whose whole prose is a tool-call-shaped wrapper from running or entering the agent\'s history, ' + doesnt; +const AUTO_CONTINUE_MAX = + 'the most turns in a row that continue without skip_reply before the agent waits for the next event'; describe('config switches', () => { it('create() refuses a framework switch that is not true or false, and a settings object other than an object, null or false', async () => { @@ -107,6 +109,8 @@ describe('config switches', () => { refused({ refusalHandling: 'on' }, 'refusalHandling must be an object, such as { autoRewind: true }, not a string'), refused({ toolResultGuard: 0 }, `toolResultGuard must be true or false, not a number: ${TOOL_RESULT_GUARD}`), refused({ toolWrapperProseGuard: { on: true } }, `toolWrapperProseGuard must be true or false, not an object: ${TOOL_WRAPPER_PROSE_GUARD}`), + refused({ autoContinue: 'true' }, 'autoContinue must be true, false or an object such as { maxConsecutive: 3 }, not a string (true and false go unquoted)'), + refused({ autoContinue: { maxConsecutive: 0 } }, `autoContinue.maxConsecutive must be a whole number from 1 to 20, not a number below 1: ${AUTO_CONTINUE_MAX}`), ]); }); @@ -123,6 +127,11 @@ describe('config switches', () => { return true; }); assert.deepEqual(workerStates(), [], 'the refused config registered no state under its name'); + await assert.rejects(framework.createEphemeralAgent({ ...worker, autoContinue: { maxConsecutive: 21 } } as never), (error: Error) => { + assert.equal(error.message, `Agent "worker" autoContinue.maxConsecutive must be a whole number from 1 to 20, not a number above 20: ${AUTO_CONTINUE_MAX}`); + return true; + }); + assert.deepEqual(workerStates(), [], 'a refused autoContinue registered no state under its name either'); const created = await framework.createEphemeralAgent(worker); assert.notDeepEqual(workerStates(), [], 'a created agent registers its states under its name'); created.cleanup(); From be1eceecb468b331b737bff98e8f548ae919dd21 Mon Sep 17 00:00:00 2001 From: antra-tess Date: Thu, 22 Oct 2026 23:58:13 -0700 Subject: [PATCH 5/6] =?UTF-8?q?fix(prose):=20nissa-seru=20review=20of=20#7?= =?UTF-8?q?44=20=E2=80=94=20row=20at=20the=20stream's=20start,=20named=20a?= =?UTF-8?q?nd=20neutral;=20stop=20must=20be=20offered;=20quiesce=20drops?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The [continue] row is written in startAgentStream, past every gate that can still stop the turn (stopping framework, provider park/requeue, stated wait), as the newest of its start inputs; the trigger then goes on unmarked. The second per-agent map is gone: the wake's mark (InferenceRequest.autoContinueRun: stop, heldAfterSend; sole set by the scheduler) carries what the row says, the count is the run's. - Wording (antra's pick of nissa's): '[continue] , another turn starts now. If you're done, call skip_reply; anything you write without it is sent as usual.' The cap is mentioned only on the last continuation. The row names the agent: residents share one message slot. - A turn that sent and then closed in prose still continues (antra: held words can't be told from unfinished work), and its row says those words were held, not sent. - Continues only an agent offered a way to stop: skip_reply offered by the channel registry and allowed, or a skip envelope in explicit/hybrid mode. - A quiesce drops a queued continuation and gives back its count (antra). - No continuation on top of a queued bounce wake; none for conversation forks; a turn with no channel passes its unresolved channel on, so the continuation takes no fallback either. - Docs: cost per answer, hybrid pins vs locus, JSON examples without comments, the above. agent.ts: spacing. - Tests (36): each of the above, plus a continuation batched first with a message; every new guard mutation-checked. Co-Authored-By: Claude Opus 5.5 --- .../feat-continue-after-prose.added.md | 29 ++- docs/auto-continue.md | 107 +++++--- src/agent.ts | 2 +- src/auto-continue.ts | 49 ++-- src/framework.ts | 207 ++++++++++----- src/types/agent.ts | 52 ++-- test/auto-continue.test.ts | 238 ++++++++++++++---- 7 files changed, 481 insertions(+), 203 deletions(-) diff --git a/changelog.d/feat-continue-after-prose.added.md b/changelog.d/feat-continue-after-prose.added.md index c92a4eaec..82fb58507 100644 --- a/changelog.d/feat-continue-after-prose.added.md +++ b/changelog.d/feat-continue-after-prose.added.md @@ -2,14 +2,21 @@ itself, with prose or with nothing, is followed by another turn straight away, until the agent calls `skip_reply` (or another tool that ends the turn), ends on a `>>skip_reply` / `>>>skip_reply` envelope, or reaches - `maxConsecutive` continuations in a row (default 3, 1 to 20; the count - starts again on any wake with a cause of its own). As the next turn starts - it reads a short `[continue]` system row (written only when the - continuation is that turn's whole cause, never for a wake that is dropped) - that names `skip_reply` as the way to stop and, near the cap, how many - continuations are left; it keeps the turn's locus unless a batched message - names another channel. Its wake has reason `auto-continue`. Refusals, - `max_tokens` cut-offs, the stall guard's ends, silent heartbeat ticks and - ephemeral runs never continue, and a ` !` continuation runs instead. A - malformed setting is refused when the agent is built. New exported type - `AutoContinueConfig`. See docs/auto-continue.md. + `maxConsecutive` continuations in a row (default 3, 1 to 20). The count + starts again on any wake with a cause of its own (a silent heartbeat tick + isn't one). As the next turn starts it reads a short `[continue]` row + naming the agent, saying another turn starts, how to stop and that + unmarked words are sent as usual (and, on the last continuation, that it + waits for the next event after it); the row is written only when the + continuation is that turn's whole cause, never for a wake that doesn't run. + The next turn keeps the turn's locus unless a batched message names + another channel. Its wake has reason `auto-continue`. Not continued: + refusals, `max_tokens` cut-offs, the stall guard's ends, silent heartbeat + ticks, a turn whose prose bounced, ephemeral runs, conversation forks, an + agent offered no way to stop, and a quiesce (which drops a queued + continuation); a ` !` continuation runs instead. Each continuation is a + full inference: an agent that answers and ends its turn gets at least one + more per answer. A malformed setting is refused when the agent is built, + naming the value's kind, never the value. New exported types + `AutoContinueConfig` and `AutoContinueRun` (`InferenceRequest.autoContinueRun`). + See docs/auto-continue.md. diff --git a/docs/auto-continue.md b/docs/auto-continue.md index 7e32a765d..7a729710f 100644 --- a/docs/auto-continue.md +++ b/docs/auto-continue.md @@ -6,11 +6,27 @@ usual pause until the next event (a message, a heartbeat, a timer). The agent keeps going until it says it's done: by calling `skip_reply`, or another tool that ends the turn, such as `sleep`. +`true` turns it on with a cap of 3 continuations in a row: + ```json -"agent": { "autoContinue": true } // cap of 3 -"agent": { "autoContinue": { "maxConsecutive": 5 } } // 1 to 20 +{ "agent": { "autoContinue": true } } ``` +An object sets the cap, a whole number from 1 to 20: + +```json +{ "agent": { "autoContinue": { "maxConsecutive": 5 } } } +``` + +## What it costs + +An agent that answers and simply ends its turn gets at least one more full +inference per answer: the continuation it then ends with `skip_reply`. One +that never calls `skip_reply` gets up to `maxConsecutive` more, and in locus +and hybrid modes each of those turns' prose is published. Each continuation's +rows also stay in the window. Keep the cap low, and expect roughly one extra +inference per turn for an agent that does skip. + ## Which turns continue A turn continues when the model ended it (stop reason `end_turn` or @@ -20,54 +36,74 @@ A turn continues when the model ended it (stop reason `end_turn` or - its last prose envelope is a skip: `>>skip_reply` in explicit mode, or a `>>>skip_reply` in hybrid mode whose hold is still standing at the end; - it asked for a ` !` continuation (`>>#ops !`), which runs instead, with no - notice, and counts toward nothing; + row, and counts toward nothing; +- its prose bounced (explicit mode): the bounce already wakes the agent to + resend, under its own cap; - it was a silent heartbeat tick; - it ended with a refusal, a `max_tokens` cut-off, or the stall guard's `no_progress` / `round_limit`; -- the agent is an ephemeral run (a subagent), or the framework is stopping. +- the agent is an ephemeral run (a subagent) or a conversation fork, or the + framework is stopping; +- the agent has no way to stop that it's offered: in locus or disabled mode, + `skip_reply` must be offered (a channel registry provides it) and allowed + (`allowedTools`). Explicit and hybrid agents can always end with a skip + envelope. -## What the agent reads - -As the next turn starts, a short system row joins the window, so it is the -newest thing the turn reads (after the last turn's `[delivered]` receipt, -outside explicit mode, whose envelopes acknowledge themselves and write no -receipt): +A turn that sent with a send tool and then closed in prose (held, not sent, +because the turn had already sent) still continues: its closing words can't +be told from unfinished work. The row tells the agent those words were held. -``` -[continue] Your last turn ended without skip_reply, so another one starts now. To stop and wait for the next event, call skip_reply. -``` +## What the agent reads -Near the cap the row also says how many continuations are left: +As the continuation's turn starts, after every gate that could still stop it, +a short system row joins the window as the newest of the turn's inputs (after +the last turn's `[delivered]` receipt, outside explicit mode, whose envelopes +acknowledge themselves and write no receipt): ``` -… Continuation 2 of 3 in a row: one more is left. -… This is the last continuation in a row (3 of 3): after this turn you wait for the next event either way. +[continue] lena, another turn starts now. If you're done, call skip_reply; anything you write without it is sent as usual. ``` -An agent that can't call `skip_reply` is told to end with a `>>skip_reply` -(explicit) or `>>>skip_reply` (hybrid) envelope instead; in locus or disabled -mode without the tool, the row names no way to stop and the cap ends the run. - -The row is written by the scheduler as it starts the turn that takes the -continuation's wake, never earlier, so it never claims a turn that doesn't -run: a wake the inference policy drops writes no row and gives back its -count, and a row that can't be stored starts no continuation. It is appended -before the turn compiles and never retracted, so the prompt-cache prefix -stays stable. It is deliberately short and -event-shaped: longer instruction blocks about output, sent as user-role -messages, have tripped provider classifiers. +It names the agent, since residents read one shared message slot. When the +last turn's closing words were held because that turn had already sent, it +adds `Your closing words last turn were held, not sent: that turn had already +sent with a send tool.` The last continuation of a run adds `This is the last +continuation in a row: after this turn you wait for the next event either +way.` An agent that can't call `skip_reply` is told to end with a +`>>skip_reply` (explicit) or `>>>skip_reply` (hybrid) envelope instead. + +The wording is neutral on purpose: a row that reads as a lapse ("you ended +without…") or as a budget ("2 of 3, one more left") invites an apology or a +"done here" reply, which in locus and hybrid modes would itself be published. +It is short and event-shaped: longer instruction blocks about output, sent as +user-role messages, have tripped provider classifiers. + +The row is never written for a turn that doesn't run: a wake the inference +policy drops, or that a quiesce drops, writes none and gives back its count; +a row that can't be stored gives back its count, and withdraws a turn the +continuation alone caused. It is appended before the turn compiles and never +retracted, so the prompt-cache prefix stays stable. ## The next turn The continuation's wake has source `framework` and reason `auto-continue`. In -locus and hybrid modes it carries the turn's locus as its channel, so the next -turn's plain prose lands where the last one's did rather than in the -process-global last-inbound channel. +locus and hybrid modes it carries the turn's locus, so the next turn's plain +prose lands in the same locus rather than the process-global last-inbound +channel. (A hybrid `>>>` destination is a pin within a turn: the next turn +starts with none, as any turn does.) A turn whose wake named a channel that +resolves to none (it had no channel and took no fallback) passes that channel +on, so its continuation takes no fallback either. A message that arrives meanwhile joins the same turn, as any batched wake does, and then that turn is the message's: its channel outranks the continuation's (ambient included, though the continuation's wake was queued -later), no `[continue]` row is written, and the run's count starts again. +later), no `[continue]` row is written, and the run's count starts again. A +silent heartbeat tick batched with a continuation is not a cause of its own: +the continuation keeps its row and its count. + +A host quiesce drops a queued continuation: at resume, possibly hours later, +"another turn starts now" would no longer be true. The next real event wakes +the agent as usual. ## The cap @@ -75,8 +111,5 @@ At most `maxConsecutive` continuations run in a row (default 3). The count starts again when a turn starts with a cause of its own (anything but a framework wake or a silent heartbeat tick), and when a tool ends a turn. A silent tick batched with a continuation neither restarts the run nor takes -its notice, so a resident whose ticks come faster than its turns still -reaches the cap. Each continued turn is a full -inference over the context (mostly cache reads), and in locus mode its prose -is published, so the cap bounds both the cost and the posts of an agent that -never skips. +its row, so a resident whose ticks come faster than its turns still reaches +the cap. diff --git a/src/agent.ts b/src/agent.ts index 3f6c55d5b..da33f98dd 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -352,7 +352,7 @@ export class Agent { this.proseRouting = config.proseRouting ?? 'locus'; this.toolWrapperProseGuard = config.toolWrapperProseGuard === true; this.autoContinue = resolveAutoContinue(config); - this.clock =resolveClockConfig(config.name, config.clock); + this.clock = resolveClockConfig(config.name, config.clock); this.cacheTtl = config.cacheTtl ?? '1h'; this.promptCaching = config.promptCaching !== false; this.prefillUserMessage = config.prefillUserMessage; diff --git a/src/auto-continue.ts b/src/auto-continue.ts index e03f77658..2eef74dd3 100644 --- a/src/auto-continue.ts +++ b/src/auto-continue.ts @@ -2,21 +2,12 @@ * autoContinue (AgentConfig.autoContinue, docs/auto-continue.md): a turn the * model ends itself is followed by another, unless it ended on a skip. This * module holds what doesn't need the framework: which ends continue, and the - * notice the next turn reads. + * row the next turn reads. */ /** The reason on a continuation's wake (InferenceRequest.reason). */ export const AUTO_CONTINUE_REASON = 'auto-continue'; -/** What a queued continuation's notice says, held until its turn starts. */ -export interface PendingAutoContinue { - /** This continuation's place in the run, 1 to `max`. */ - count: number; - max: number; - /** How the agent stops (autoContinueNotice's `stop`), if it can. */ - stop: string | undefined; -} - /** * Stop reasons of a turn the model ended itself. The others don't continue: * a refusal has its own handling (retries, rewind, the reaction), a @@ -26,22 +17,26 @@ export interface PendingAutoContinue { export const AUTO_CONTINUE_STOP_REASONS: ReadonlySet = new Set(['end_turn', 'stop_sequence']); /** - * The notice a continuation's turn reads: that the last turn ended without a - * skip so another one starts, how to stop (`stop`, an imperative such as - * "call skip_reply"; left out when the agent has no way to skip), and, near - * the cap, how many continuations are left. `count` is this continuation's - * place in the run, 1 to `max`. Short and event-shaped on purpose: it rides - * as a system row in the window, where instruction blocks about output have - * tripped classifiers. + * The row a continuation's turn reads. It names the agent (residents read + * one shared message slot), says another turn starts, says how to stop + * (`stop`, an imperative such as "call skip_reply") and that anything written + * without it is sent as usual, and, on the last continuation of a run, that + * the agent waits for the next event after it. When the last turn's closing + * words were held because that turn had already sent, it says so. Short, + * neutral and event-shaped on purpose: it rides as a system row in the + * window, where instruction blocks about output have tripped classifiers, + * and wording that reads as a lapse or a budget invites a published reply. */ -export function autoContinueNotice(count: number, max: number, stop: string | undefined): string { - const opening = stop - ? `[continue] Your last turn ended without skip_reply, so another one starts now. To stop and wait for the next event, ${stop}.` - : '[continue] Your last turn ended on its own, so another one starts now.'; - const left = max - count; - if (left <= 0) { - return `${opening} This is the last continuation in a row (${count} of ${max}): after this turn you wait for the next event either way.`; - } - if (left === 1) return `${opening} Continuation ${count} of ${max} in a row: one more is left.`; - return opening; +export function autoContinueNotice(opts: { + agentName: string; + stop: string; + last: boolean; + heldAfterSend?: boolean; +}): string { + return `[continue] ${opts.agentName}, another turn starts now.` + + (opts.heldAfterSend + ? ' Your closing words last turn were held, not sent: that turn had already sent with a send tool.' + : '') + + ` If you're done, ${opts.stop}; anything you write without it is sent as usual.` + + (opts.last ? ' This is the last continuation in a row: after this turn you wait for the next event either way.' : ''); } diff --git a/src/framework.ts b/src/framework.ts index 6a331ea70..f36bc7bb4 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -71,8 +71,7 @@ import type { AgentRuntimeSettingsSnapshot, AgentSettingsExtension, SameRoundThinkTextPolicy, - ModuleMessage, -} from './types/index.js'; + ModuleMessage, AutoContinueRun } from './types/index.js'; import { asModuleRow, heldHandle, heldWriteId, matchesMessageQuery } from './held-writes.js'; import { ProcessQueueImpl } from './queue.js'; import { DEFAULT_SHUTDOWN_TIMEOUT_MS, validateShutdownTimeout, shutdownErrorMessage, waitForShutdown, ShutdownTimeoutError } from './shutdown.js'; @@ -156,7 +155,7 @@ import { DEFAULT_OUTPUT_BOUNDS } from './code-execution/isolate/output.js'; import type { PreludeTool } from './code-execution/isolate/prelude.js'; import { bounded, renderList, renderReturn, VALUE_CHARS } from './code-execution/isolate/render.js'; import { splitProseRuns, withoutRoundsProse } from './prose-segments.js'; -import { AUTO_CONTINUE_REASON, AUTO_CONTINUE_STOP_REASONS, autoContinueNotice, type PendingAutoContinue } from './auto-continue.js'; +import { AUTO_CONTINUE_REASON, AUTO_CONTINUE_STOP_REASONS, autoContinueNotice } from './auto-continue.js'; import { cumulativeDelta } from './usage-accounting.js'; import { stampThinkingTokenEstimates } from './thinking-token-stamp.js'; import { contentAsStored, nameAsStored } from './text-as-stored.js'; @@ -1311,6 +1310,12 @@ const FRAMEWORK_WAKE_SOURCES = new Set(['framework', 'scheduler']); function hasOwnCauseForRun(r: InferenceRequest): boolean { return !FRAMEWORK_WAKE_SOURCES.has(r.source) && r.suppressProse !== true; } + +/** Whether a wake carries an autoContinue continuation: its own wake, or a + * requeued turn trigger that kept the mark (InferenceRequest.autoContinueRun). */ +function carriesContinuation(r: InferenceRequest): boolean { + return r.autoContinueRun !== undefined; +} const CONVERSATION_ROUTER_STATE_ID = 'framework/conversation-router'; const INFERENCE_LOG_ID = 'framework/inference-log'; const PROCESS_LOG_ID = 'framework/process-log'; @@ -2755,9 +2760,6 @@ export class AgentFramework { * maxConsecutive. Cleared when a turn starts with a cause of its own * (processInferenceRequests) or a tool ends a turn (skip_reply). */ private autoContinueStreaks: Map = new Map(); - /** A queued continuation's notice, per agent, until the turn that takes its - * wake starts (writeAutoContinueNotice) or the wake is dropped. */ - private autoContinuePending: Map = new Map(); private pendingAssistantBlocks: Map = new Map(); /** Streams the FRAMEWORK cancelled for non-terminal reasons, keyed * `${agentName}:${streamId}`: an endTurn tool result or a context-budget @@ -14675,6 +14677,21 @@ export class AgentFramework { // kept): heartbeat/gate wakes arrive once per tick and nothing else // bounds a multi-day maintenance window's accumulation. if (this.quiesced) { + // An autoContinue continuation doesn't outlast a quiesce: at resume, + // possibly hours later, "another turn starts now" would be untrue. + // Its own wake is dropped and the run's count given back; a requeued + // trigger that carried one parks without the mark. + if (requests.some(carriesContinuation)) { + this.giveBackAutoContinue(agentName, 'the host is quiesced'); + requests = requests + .filter((r) => r.reason !== AUTO_CONTINUE_REASON) + .map((r) => { + if (!carriesContinuation(r)) return r; + const { autoContinueRun: _mark, ...rest } = r; + return rest; + }); + if (requests.length === 0) continue; + } // Continuations of a held turn (budget restart, non-streaming tool // results) pass; everything else parks. const passThrough = requests.filter((r) => isTurnContinuation(r.reason)); @@ -14791,30 +14808,21 @@ export class AgentFramework { if (requests.some((r) => resumesLogicalTurn(r.reason))) { this.endContinuedTurn(agentName, { kind: 'dropped', why: 'the inference policy dropped its continuation' }); } - if (requests.some((r) => r.reason === AUTO_CONTINUE_REASON)) this.dropAutoContinue(agentName, 'the inference policy dropped its wake'); + if (requests.some(carriesContinuation)) this.giveBackAutoContinue(agentName, 'the inference policy dropped its wake'); this.endPendingPrimary(agentName); continue; } - // An autoContinue wake's notice is written now, as the turn that takes - // it starts, so it never claims a turn that doesn't run. Only when the - // continuation is the turn's whole cause: a batch with a cause of its - // own (a message, a heartbeat) is that cause's turn, and its run of - // continuations starts again below. A notice that can't be stored - // starts no continuation: without it the turn would wake to nothing. - if (requests.some((r) => r.reason === AUTO_CONTINUE_REASON)) { - const pending = this.autoContinuePending.get(agentName); - this.autoContinuePending.delete(agentName); - if (pending && !requests.some(hasOwnCauseForRun) - && !this.writeAutoContinueNotice(agent, pending)) { - this.autoContinueStreaks.set(agentName, pending.count - 1); - requests = requests.filter((r) => r.reason !== AUTO_CONTINUE_REASON); - if (requests.length === 0) { - this.endPendingPrimary(agentName); - continue; - } - } - } + // An autoContinue continuation is the turn's when nothing in the batch + // is a cause of its own (hasOwnCauseForRun): then the turn's trigger + // carries its mark, and the turn writes its `[continue]` row as it + // starts (startAgentStream), past every gate that could still stop it. + // A batch with a cause of its own is that cause's turn: no row, and the + // run starts again below. + const continuation = requests.find(carriesContinuation); + const continuationMark = continuation?.autoContinueRun && !requests.some(hasOwnCauseForRun) + ? { ...continuation.autoContinueRun, sole: requests.every(carriesContinuation) } + : undefined; // Start streaming inference (non-blocking — driveStream runs in background) // The request that grants the turn-alive exception must also define the @@ -14927,6 +14935,8 @@ export class AgentFramework { // channel is that message's, never one server's id with another's. channelId: channelReq?.channelId, channelServerId: channelReq?.channelServerId, + // Only the mark this batch decided, never one requests[0] carried. + autoContinueRun: continuationMark, addressed: addressedReq !== undefined, // A context-budget restart continues the same logical turn: it keeps // the channel for routing but names no author — the restart is its @@ -16617,7 +16627,7 @@ export class AgentFramework { /** * autoContinue (AgentConfig.autoContinue): after a turn the model ended - * itself (AUTO_CONTINUE_STOP_REASONS), with prose or with nothing, start + * itself (AUTO_CONTINUE_STOP_REASONS), with prose or with nothing, queue * another, unless its last word was a skip (`>>skip_reply`, or a hybrid * `>>>skip_reply` hold still standing). Runs at the logical turn end of a * stream's `complete`, after the turn's deliveries and `[delivered]` @@ -16625,20 +16635,23 @@ export class AgentFramework { * runs instead). A tool that ends the turn (skip_reply, sleep) never * reaches here, and clears the run's count. * - * This queues the wake and holds the notice it reads - * (autoContinuePending): the scheduler writes the `[continue]` row as the - * turn that takes the wake starts, and only when the continuation is that - * turn's whole cause (processInferenceRequests); a dropped wake writes - * none and gives back its count (dropAutoContinue). The wake names this - * turn's locus, which a batched message's channel outranks. A run is - * capped at maxConsecutive; the last allowed notice says so, and the cap - * holds until a turn starts with a cause of its own. + * It continues only an agent that has a way to stop it is offered: the + * skip_reply tool, or in explicit/hybrid mode a skip envelope. It doesn't + * add a continuation to a bounce wake the turn's prose already queued, and + * never continues an ephemeral run or a conversation fork. The wake names + * the turn's locus, or the unresolved channel of a turn that had none, so + * the next turn speaks, or doesn't, where this one did; a batched message's + * channel outranks it. Its mark (InferenceRequest.autoContinueRun) holds + * what the `[continue]` row says, written as the turn starts + * (startAgentStream). A run is capped at maxConsecutive; the last allowed + * row says so, and the cap holds until a turn starts with a cause of its own. */ private autoContinueAfterTurn( agent: Agent, routing: TurnRouting, trigger: InferenceRequest | undefined, stopReason: string, + opts: { heldAfterSend?: boolean } = {}, ): void { const config = agent.autoContinue; if (!config) return; @@ -16646,8 +16659,9 @@ export class AgentFramework { if (!AUTO_CONTINUE_STOP_REASONS.has(stopReason)) return; // A silent heartbeat tick is a private self-check, never the start of a run. if (trigger?.suppressProse) return; - // A subagent's run ends with its result. - if (this.ephemeralRuns.has(name)) return; + // A subagent's run ends with its result; a conversation fork answers + // its home channel's message. + if (this.ephemeralRuns.has(name) || this.conversationAgentHomes.has(name)) return; if (this.providerAdmissionClosed || this.agents.get(name) !== agent) { console.error(`[continue] ${name}: not continued — ${this.providerAdmissionClosed ? 'the framework is stopping' : 'the agent is no longer registered'}`); return; @@ -16660,66 +16674,92 @@ export class AgentFramework { console.error(`[continue] ${name}: turn ended on a skip_reply envelope — waiting for the next event`); return; } + const stop = this.autoContinueStop(agent); + if (!stop) { + console.error(`[continue] ${name}: not continued — no way to stop is offered (no skip_reply tool, no skip envelope in ${agent.proseRouting} mode)`); + return; + } + // The turn's prose bounced, and the bounce already wakes the agent to + // resend it (bounceProse, with its own cap). + if (this.pendingRequests.some((r) => r.agentName === name && r.reason === 'prose-bounce')) { + console.error(`[continue] ${name}: not continued — a bounce wake is already queued`); + return; + } const count = (this.autoContinueStreaks.get(name) ?? 0) + 1; if (count > config.maxConsecutive) { console.error(`[continue] ${name}: ${config.maxConsecutive} continuation(s) in a row already — waiting for the next event`); return; } this.autoContinueStreaks.set(name, count); - const stop = this.canAgentUseTool(name, 'skip_reply') - ? 'call skip_reply' - : agent.proseRouting === 'explicit' - ? 'end with a >>skip_reply envelope' - : agent.proseRouting === 'hybrid' - ? 'end with a >>>skip_reply envelope' - : undefined; - this.autoContinuePending.set(name, { count, max: config.maxConsecutive, stop }); console.error( `[continue] ${name}: turn ended (${stopReason}) without skip_reply — continuation ${count} of ${config.maxConsecutive}` + (routing.locus ? ` in ${this.channelRegistry?.nameShown(routing.locus) ?? routing.locus.channelId}` : ''), ); + // The next turn speaks where this one did (locus and hybrid modes): the + // locus's server and id, as the framework's own wakes name a channel. A + // turn that had no channel (noChannel: its wake named one that resolves + // to none) passes its wake's channel on, so the next has none either, + // rather than taking the global fallback this one refused. + const channel = routing.locus + ? { channelId: routing.locus.channelId, channelServerId: routing.locus.serverId } + : routing.noChannel && trigger?.channelId + ? { channelId: trigger.channelId, ...(trigger.channelServerId ? { channelServerId: trigger.channelServerId } : {}) } + : {}; this.pendingRequests.push({ agentName: name, reason: AUTO_CONTINUE_REASON, source: 'framework', timestamp: Date.now(), - // The next turn speaks where this one did (locus and hybrid modes): - // the locus's server and id, as the framework's own wakes name a channel. - ...(routing.locus ? { channelId: routing.locus.channelId, channelServerId: routing.locus.serverId } : {}), + ...channel, + autoContinueRun: { stop, ...(opts.heldAfterSend ? { heldAfterSend: true } : {}) }, }); } + /** How an autoContinue agent stops a run, as its `[continue]` row says it, + * or undefined when no way is offered: the skip_reply tool when the + * channel registry offers it and the agent may call it, else a skip + * envelope in explicit or hybrid mode. */ + private autoContinueStop(agent: Agent): string | undefined { + if (this.channelRegistry && this.canAgentUseTool(agent.name, 'skip_reply')) return 'call skip_reply'; + if (agent.proseRouting === 'explicit') return 'end with a >>skip_reply envelope'; + if (agent.proseRouting === 'hybrid') return 'end with a >>>skip_reply envelope'; + return undefined; + } + /** - * Store a continuation's `[continue]` notice in the agent's window, as the - * turn that takes its wake starts: no turn is running, so the row lands - * after everything the last turn wrote (its `[delivered]` receipt - * included, outside explicit mode) and is the newest row the turn - * compiles. False when it couldn't be stored. + * Store a continuation's `[continue]` row in the agent's window as its turn + * starts (startAgentStream), with the run's count as it stands and the + * agent's cap. False when it couldn't be stored. */ - private writeAutoContinueNotice(agent: Agent, pending: PendingAutoContinue): boolean { - const text = autoContinueNotice(pending.count, pending.max, pending.stop); + private writeAutoContinueNotice(agent: Agent, mark: AutoContinueRun): boolean { + const count = this.autoContinueStreaks.get(agent.name) ?? 1; + const max = agent.autoContinue?.maxConsecutive ?? count; + const text = autoContinueNotice({ + agentName: agent.name, stop: mark.stop, last: count >= max, heldAfterSend: mark.heldAfterSend, + }); try { const id = agent.getContextManager().addMessage( 'user', [{ type: 'text', text }], - { system: true, kind: 'auto-continue', continuation: pending.count, maxConsecutive: pending.max } as MessageMetadata, + { system: true, kind: 'auto-continue', agentName: agent.name, continuation: count, maxConsecutive: max } as MessageMetadata, ); this.emitTrace({ type: 'message:added', messageId: id, source: 'auto-continue' }); + console.error(`[continue] ${agent.name}: continuation ${count} of ${max} starts`); return true; } catch (err) { - console.error(`[continue] ${agent.name}: notice not stored, so not continued:`, err); + console.error(`[continue] ${agent.name}: row not stored, so not continued:`, err); return false; } } - /** A queued continuation whose wake starts no turn: its notice is never - * written, and the run's count goes back to what it was before. */ - private dropAutoContinue(agentName: string, why: string): void { - const pending = this.autoContinuePending.get(agentName); - if (!pending) return; - this.autoContinuePending.delete(agentName); - this.autoContinueStreaks.set(agentName, pending.count - 1); - console.error(`[continue] ${agentName}: continuation ${pending.count} dropped — ${why}`); + /** A continuation whose turn doesn't run: the run's count goes back to + * what it was before it was queued. */ + private giveBackAutoContinue(agentName: string, why: string): void { + const count = this.autoContinueStreaks?.get(agentName); + if (!count) return; + if (count <= 1) this.autoContinueStreaks.delete(agentName); + else this.autoContinueStreaks.set(agentName, count - 1); + console.error(`[continue] ${agentName}: continuation ${count} dropped — ${why}`); } /** The agent's routing state for its current turn, made when it has none. */ @@ -17979,6 +18019,35 @@ export class AgentFramework { // vector 25). Not once the host is quiesced, when it waits for resume(). if (assembled && !this.quiesced && this.deferredMessages.length > 0) this.flushForStart(agent, 'assembly'); + // An autoContinue continuation's `[continue]` row, written here: the + // turn has passed every gate that could still stop it (a stopping + // framework, a provider park's requeue, a stated wait), and the row is + // the newest of its inputs, before its checkpoint. The trigger goes on + // without the mark, so a retry doesn't write it again. A row that can't + // be stored gives back the run's count, and withdraws a turn the + // continuation alone caused: it would wake to nothing it can read. + if (attempt === 0 && !resumesTurn && trigger?.autoContinueRun) { + const mark = trigger.autoContinueRun; + const { autoContinueRun: _mark, ...unmarked } = trigger; + trigger = unmarked; + if (!this.writeAutoContinueNotice(agent, mark)) { + this.giveBackAutoContinue(agent.name, 'its row could not be stored'); + if (mark.sole) { + console.error(`[continue] ${agent.name}: turn not started (the continuation was its only cause)`); + if (this.activeTurnTokens.get(agent.name) === turnToken) { + this.activeTurnTokens.delete(agent.name); + this.activeTurnTriggers.delete(agent.name); + } + if (!this.activeTurnTokens.has(agent.name) + && this.deferredMessages.length > 0 && !this.pendingAssistantBlocks.has(agent.name)) { + this.readdDrained(this.drainDeferredFor(agent.name), 'withdrawn-turn flush'); + this.ackDeferredWrites(); + } + return false; + } + } + } + // The turn has passed its gate, so from here it goes on with a copy of // its trigger without a batch wake's marks or the mark of a wake to read // a render. A retry of it then comes back as an ordinary wake by whatever @@ -19948,7 +20017,13 @@ export class AgentFramework { // turn the model ended itself unless that end was a skip. const askedToContinue = finalRouting.continuation; this.requestProseContinuation(agent, finalRouting); - if (!askedToContinue) this.autoContinueAfterTurn(agent, finalRouting, trigger, response.stopReason); + if (!askedToContinue) { + // Closing words held, not sent, because the turn had already + // sent with a send tool: the continuation's row says so. + const heldAfterSend = turnProseRouting !== 'explicit' && turnSilenced && silenceCause === 'send' + && terminalContent.some((b) => b.type === 'text' && (b as { text: string }).text.trim().length > 0); + this.autoContinueAfterTurn(agent, finalRouting, trigger, response.stopReason, { heldAfterSend }); + } break; } diff --git a/src/types/agent.ts b/src/types/agent.ts index 8ccf3784e..e11e82b2a 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -9,6 +9,17 @@ export type SameRoundThinkTextPolicySource = | 'recipe' | 'compatibility_default'; +/** What a continuation's `[continue]` row says (InferenceRequest.autoContinueRun). */ +export interface AutoContinueRun { + /** How the agent stops, as the row puts it ("call skip_reply"). */ + stop: string; + /** The turn before ended in words that were held, not sent, because it + * had already sent with a send tool. */ + heldAfterSend?: boolean; + /** Set by the scheduler: the continuation is the turn's only cause. */ + sole?: boolean; +} + /** `AgentConfig.autoContinue` as an object (`true` means `{}`). */ export interface AutoContinueConfig { /** Continuations in a row before the agent waits for the next event: @@ -233,23 +244,24 @@ export interface AgentConfig { * instead of a pause until the next event. A turn ended through * `skip_reply` (or another tool that ends the turn, such as `sleep`), or * whose last prose envelope is `>>skip_reply` (explicit) or `>>>skip_reply` - * (hybrid), pauses as usual. + * (hybrid), pauses as usual. So does one whose prose bounced (the bounce + * wakes the agent), and an agent offered no way to stop (locus or disabled + * mode without an offered `skip_reply`). * - * As the next turn starts, a short system notice joins the window (after - * the last turn's `[delivered]` receipt, outside explicit mode, which - * writes none), saying that the turn continues and that skip_reply stops - * it; near the cap it also says how many continuations are left. It's - * written only when the continuation is that turn's whole cause: a - * message batched with it makes it that message's turn, with no notice, - * in the message's channel. Otherwise the next turn keeps the turn's - * locus (locus and hybrid modes). - * At most `maxConsecutive` (default 3) continuations run in a row: the - * count starts again when a turn starts with a cause of its own (a - * message, a timer, a heartbeat that isn't a silent tick) or a tool ends - * one. Silent heartbeat - * ticks, refusals, `max_tokens` cut-offs, the stall guard's ends and - * ephemeral runs never continue. A ` !` continuation the turn asked for - * runs instead, and counts toward nothing. See docs/auto-continue.md. + * As the next turn starts, a short `[continue]` system row naming the agent + * joins the window: another turn starts, how to stop, and that unmarked + * words are sent as usual; on the last continuation of a run, that the + * agent waits for the next event after it. It's written only when the + * continuation is that turn's whole cause: a message batched with it makes + * it that message's turn, with no row, in the message's channel. Otherwise + * the next turn keeps the turn's locus (locus and hybrid modes). At most + * `maxConsecutive` (default 3) continuations run in a row: the count + * starts again when a turn starts with a cause of its own (a message, a + * timer, a heartbeat that isn't a silent tick) or a tool ends one. Silent + * heartbeat ticks, refusals, `max_tokens` cut-offs, the stall guard's + * ends, ephemeral runs, conversation forks and a quiesce never continue. A + * ` !` continuation the turn asked for runs instead, and counts toward + * nothing. See docs/auto-continue.md. */ autoContinue?: boolean | AutoContinueConfig; @@ -463,6 +475,14 @@ export interface InferenceRequest { * channel of its own. */ channelServerId?: string; + /** + * Framework-set: this wake carries an autoContinue continuation + * (AgentConfig.autoContinue), with what its `[continue]` row says. The + * scheduler keeps the mark on the turn's trigger only when the + * continuation is that turn's whole cause, and the turn writes the row as + * it starts. Hosts don't set it. + */ + autoContinueRun?: AutoContinueRun; /** * True when the triggering message explicitly addressed the agent (mention, * reply-to-bot, DM — `chat:addressed` in MCPL RFC-001 terms). When a wake diff --git a/test/auto-continue.test.ts b/test/auto-continue.test.ts index 35b82369f..947193cc8 100644 --- a/test/auto-continue.test.ts +++ b/test/auto-continue.test.ts @@ -35,7 +35,11 @@ class TriggerModule implements Module { async start(_ctx: ModuleContext): Promise {} async stop(): Promise {} getTools(): ToolDefinition[] { - return [{ name: 'move', description: 'Move the robot', inputSchema: { type: 'object', properties: {} } }]; + return [ + { name: 'move', description: 'Move the robot', inputSchema: { type: 'object', properties: {} } }, + // A send tool by its bare name: a turn that calls it holds its later prose. + { name: 'send_message', description: 'Explicitly send a message', inputSchema: { type: 'object', properties: { text: { type: 'string' } } } }, + ]; } async handleToolCall(_call: ToolCall): Promise { return { success: true, data: { ok: true } }; @@ -75,12 +79,19 @@ function stubRegistry(framework: AgentFramework) { return { delivered: true, serverId: outcome.destination.serverId, channelId: outcome.destination.channelId }; }, resolveDestination: ({ channelId }: { channelId: string }) => ({ destination: { serverId: 'stub', channelId } }), - resolveLocus: () => ({ serverId: 'stub', channelId: 'world:commons' }), + // As the registry's resolver does through the framework's turn state: a + // turn whose wake named no resolvable channel (noChannel) has no locus + // and takes no fallback. + resolveLocus: (agentName: string) => + (framework as unknown as { turnRouting: Map }).turnRouting.get(agentName)?.noChannel + ? null + : { serverId: 'stub', channelId: 'world:commons' }, getDefaultPublishChannel: () => null, isChannelOpen: () => true, getDescriptor: () => undefined, getChannelTools: () => [], - channelsNamed: (name: string) => [{ serverId: 'stub', channelId: name }], + // 'ghost' names no registered channel, as a host's unresolvable id would. + channelsNamed: (name: string) => (name === 'ghost' ? [] : [{ serverId: 'stub', channelId: name }]), nameShown: (channel: { channelId: string }) => channel.channelId, // skip_reply ends the turn, as the registry's own handler does. handleChannelToolCall: async () => ({ success: true, endTurn: true, data: { skipped: true } }), @@ -92,6 +103,9 @@ function stubRegistry(framework: AgentFramework) { } const text = (t: string): ContentBlock => ({ type: 'text', text: t } as ContentBlock); +/** The row the test agent reads for a continuation. */ +const row = (last: boolean, stop = 'call skip_reply', heldAfterSend = false): string => + autoContinueNotice({ agentName: 'assistant', stop, last, heldAfterSend }); const prose = (t: string): NormalizedResponse => createMockResponse([text(t)]); const empty = (): NormalizedResponse => createMockResponse([]); const skipReply = (id = 'sr'): NormalizedResponse => @@ -147,7 +161,7 @@ describe('autoContinue', () => { * the script calls skip_reply, so a stray continuation ends instead of * hanging and the test fails on its count. Returns each turn's trigger. */ - function script(framework: AgentFramework, responses: NormalizedResponse[]): InferenceRequest[] { + function script(framework: AgentFramework, responses: Array): InferenceRequest[] { const triggers: InferenceRequest[] = []; const internals = framework as unknown as { activeTurnTriggers: Map }; framework.onTrace((e) => { @@ -155,7 +169,8 @@ describe('autoContinue', () => { const turn = triggers.length; const trigger = internals.activeTurnTriggers?.get('assistant'); triggers.push(trigger ?? ({ reason: 'unknown' } as InferenceRequest)); - membrane.pushResponse(responses[turn] ?? skipReply(`stray-${turn}`)); + // A turn with tool rounds takes all its responses at its start. + for (const response of [responses[turn] ?? skipReply(`stray-${turn}`)].flat()) membrane.pushResponse(response); }); return triggers; } @@ -196,12 +211,8 @@ describe('autoContinue', () => { const all = rows(framework); const firstNotice = all.findIndex((r) => r.kind === 'auto-continue'); assert.equal(all[firstNotice - 1]?.kind, 'delivery-receipt', 'the notice follows the turn\'s receipt'); - assert.deepEqual(notices(framework), [ - autoContinueNotice(1, 3, 'call skip_reply'), - autoContinueNotice(2, 3, 'call skip_reply'), - ]); - assert.match(notices(framework)[0]!, /To stop and wait for the next event, call skip_reply\.$/); - assert.match(notices(framework)[1]!, /Continuation 2 of 3 in a row: one more is left\.$/); + assert.deepEqual(notices(framework), [row(false), row(false)]); + assert.equal(notices(framework)[0], "[continue] assistant, another turn starts now. If you're done, call skip_reply; anything you write without it is sent as usual.", 'it names the agent'); } finally { await framework.stop(); } @@ -231,12 +242,8 @@ describe('autoContinue', () => { trigger(framework); await within(framework.runUntilIdle(), 'the run to settle'); assert.equal(triggers.length, 3, 'the turn and two continuations'); - assert.deepEqual(notices(framework), [ - autoContinueNotice(1, 2, 'call skip_reply'), - autoContinueNotice(2, 2, 'call skip_reply'), - ]); - assert.match(notices(framework)[0]!, /Continuation 1 of 2 in a row: one more is left\.$/); - assert.match(notices(framework)[1]!, /This is the last continuation in a row \(2 of 2\): after this turn you wait for the next event either way\.$/); + assert.deepEqual(notices(framework), [row(false), row(true)], 'only the last says it is the last'); + assert.match(notices(framework)[1]!, / This is the last continuation in a row: after this turn you wait for the next event either way\.$/); } finally { await framework.stop(); } @@ -335,7 +342,7 @@ describe('autoContinue', () => { internals.pendingRequests.push({ agentName: 'assistant', reason: 'prose-bounce', source: 'framework', timestamp: Date.now() }); await within(framework.runUntilIdle(), 'the next run'); assert.equal(triggers.length, 3, 'a prose end continues again'); - assert.deepEqual(notices(framework), [autoContinueNotice(1, 2, 'call skip_reply')]); + assert.deepEqual(notices(framework), [row(false)]); } finally { await framework.stop(); } @@ -372,7 +379,8 @@ describe('autoContinue', () => { for (const [what, extra] of [ ['a silent heartbeat tick', { reason: 'heartbeat', source: 'heartbeat', suppressProse: true }], - ['another framework wake', { reason: 'prose-bounce', source: 'framework' }], + // Not a prose-bounce: a queued bounce wake takes the place of a continuation. + ['another framework wake', { reason: 'framework-test-wake', source: 'framework' }], ] as const) { it(`${what} batched with a continuation neither restarts the run nor takes its notice`, async () => { // Cap 1: if the batched wake restarted the run, the second turn's end @@ -391,7 +399,7 @@ describe('autoContinue', () => { trigger(framework); await within(framework.runUntilIdle(), 'the run to settle'); assert.equal(triggers.length, 2, 'the cap still holds'); - assert.deepEqual(notices(framework), [autoContinueNotice(1, 1, 'call skip_reply')], 'the continuation\'s notice is written'); + assert.deepEqual(notices(framework), [row(true)], 'the continuation\'s notice is written'); } finally { await framework.stop(); } @@ -404,7 +412,6 @@ describe('autoContinue', () => { const internals = framework as unknown as { inferencePolicy: { shouldInfer(agent: string, requests: InferenceRequest[]): boolean }; autoContinueStreaks: Map; - autoContinuePending: Map; }; internals.inferencePolicy = { shouldInfer: (_agent, requests) => !requests.some((r) => r.reason === 'auto-continue') }; const triggers = script(framework, [prose('Only this.')]); @@ -414,7 +421,6 @@ describe('autoContinue', () => { assert.equal(triggers.length, 1); assert.deepEqual(notices(framework), [], 'no notice claims a turn that never ran'); assert.equal(internals.autoContinueStreaks.get('assistant') ?? 0, 0, 'the count is given back'); - assert.equal(internals.autoContinuePending.size, 0); } finally { await framework.stop(); } @@ -446,9 +452,8 @@ describe('autoContinue', () => { for (const [mode, first, stop] of [ ['explicit', '>>#alpha hello', 'end with a >>skip_reply envelope'], ['hybrid', 'hello', 'end with a >>>skip_reply envelope'], - ['locus', 'hello', undefined], ] as const) { - it(`${mode}: an agent that can't call skip_reply is told ${stop ? 'its envelope' : 'no way to stop'}`, async () => { + it(`${mode}: an agent that can't call skip_reply is told its envelope`, async () => { const framework = await createFramework({ maxConsecutive: 1 }, mode, ['robot--move']); stubRegistry(framework); // Without skip_reply, the cap ends the run. @@ -457,52 +462,189 @@ describe('autoContinue', () => { trigger(framework); await within(framework.runUntilIdle(), 'the run to settle'); assert.equal(triggers.length, 2); - assert.deepEqual(notices(framework), [autoContinueNotice(1, 1, stop)]); + assert.deepEqual(notices(framework), [row(true, stop)]); } finally { await framework.stop(); } }); } - it('queues nothing for an ephemeral run, a stopping framework or an agent no longer registered', async () => { + for (const [what, allowed, registry] of [ + ['may not call skip_reply', ['robot--move'], true], + ['has no channel registry offering skip_reply', undefined, false], + ] as const) { + it(`locus: an agent that ${what} is not continued: it would have no way to stop`, async () => { + const framework = await createFramework(true, 'locus', allowed ? [...allowed] : undefined); + if (registry) stubRegistry(framework); + const triggers = script(framework, [prose('hello'), prose('again')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 1); + assert.deepEqual(notices(framework), []); + } finally { + await framework.stop(); + } + }); + } + + it('queues nothing for an ephemeral run, a conversation fork, a stopping framework or an agent no longer registered', async () => { const framework = await createFramework(true); stubRegistry(framework); const internals = framework as unknown as { - autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string): void; + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string, opts?: unknown): void; pendingRequests: InferenceRequest[]; ephemeralRuns: Map; + conversationAgentHomes: Map; providerAdmissionClosed: boolean; - autoContinuePending: Map; autoContinueStreaks: Map; }; const agent = framework.getAgent('assistant')!; - const queued = () => internals.pendingRequests.filter((r) => r.reason === 'auto-continue').length; + const queued = () => internals.pendingRequests.filter((r) => r.reason === 'auto-continue'); try { internals.ephemeralRuns.set('assistant', {}); internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn'); internals.ephemeralRuns.delete('assistant'); - assert.equal(queued(), 0, 'an ephemeral run ends with its result'); + assert.equal(queued().length, 0, 'an ephemeral run ends with its result'); + + internals.conversationAgentHomes.set('assistant', 'chan-alpha'); + internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn'); + internals.conversationAgentHomes.delete('assistant'); + assert.equal(queued().length, 0, 'a conversation fork answers its home'); internals.providerAdmissionClosed = true; internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn'); internals.providerAdmissionClosed = false; - assert.equal(queued(), 0, 'a stopping framework starts nothing'); + assert.equal(queued().length, 0, 'a stopping framework starts nothing'); internals.autoContinueAfterTurn(Object.create(agent), {}, undefined, 'end_turn'); - assert.equal(queued(), 0, 'an agent object no longer the registered one starts nothing'); + assert.equal(queued().length, 0, 'an agent object no longer the registered one starts nothing'); - // The same call for the registered agent does queue one. - internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn'); - assert.equal(queued(), 1); - assert.equal(internals.autoContinuePending.size, 1); + // The same call for the registered agent does queue one, marked. + internals.autoContinueAfterTurn(agent, {}, undefined, 'end_turn', { heldAfterSend: true }); + assert.equal(queued().length, 1); + assert.deepEqual(queued()[0]!.autoContinueRun, { stop: 'call skip_reply', heldAfterSend: true }); } finally { internals.pendingRequests.length = 0; - internals.autoContinuePending.clear(); internals.autoContinueStreaks.clear(); await framework.stop(); } }); + it("the row says the last turn's closing words were held when it had already sent", async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string, opts?: unknown): void; + }; + const triggers = script(framework, [skipReply()]); + try { + internals.autoContinueAfterTurn(framework.getAgent('assistant')!, {}, undefined, 'end_turn', { heldAfterSend: true }); + await within(framework.runUntilIdle(), 'the continuation'); + assert.equal(triggers.length, 1); + assert.deepEqual(notices(framework), [row(false, 'call skip_reply', true)]); + } finally { + await framework.stop(); + } + }); + + it("a turn that sent and then closed in prose continues, and its row says those words were held", async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const sendRound = createMockResponse([ + text('Posting the summary.'), + { type: 'tool_use', id: 's1', name: 'robot--send_message', input: { text: 'the summary' } } as ContentBlock, + ], 'tool_use'); + const triggers = script(framework, [[sendRound, prose('Done.')], skipReply()]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2, 'it still continues: a held "Done." can\'t be told from unfinished work'); + assert.deepEqual(notices(framework), [row(false, 'call skip_reply', true)]); + } finally { + await framework.stop(); + } + }); + + it("a message batched after a queued continuation still takes the turn: no row, a fresh run", async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string): void; + pendingRequests: InferenceRequest[]; + autoContinueStreaks: Map; + }; + const triggers = script(framework, [skipReply()]); + try { + // The continuation is queued first, so it is the batch's first request. + internals.autoContinueAfterTurn(framework.getAgent('assistant')!, {}, undefined, 'end_turn'); + internals.pendingRequests.push({ agentName: 'assistant', reason: 'channel-message', source: 'stub', timestamp: Date.now() }); + await within(framework.runUntilIdle(), 'the turn'); + assert.equal(triggers.length, 1); + assert.equal(triggers[0]!.reason, 'auto-continue', 'the first request names the trigger'); + assert.equal(triggers[0]!.autoContinueRun, undefined, 'but the turn carries no continuation mark'); + assert.deepEqual(notices(framework), []); + assert.equal(internals.autoContinueStreaks.has('assistant'), false); + } finally { + await framework.stop(); + } + }); + + it('a quiesce drops a queued continuation and gives back its count', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string): void; + processInferenceRequests(): Promise; + pendingRequests: InferenceRequest[]; + quiesced: boolean; + autoContinueStreaks: Map; + }; + try { + internals.autoContinueAfterTurn(framework.getAgent('assistant')!, {}, undefined, 'end_turn'); + assert.equal(internals.autoContinueStreaks.get('assistant'), 1); + internals.quiesced = true; + await internals.processInferenceRequests(); + assert.equal(internals.pendingRequests.filter((r) => r.reason === 'auto-continue').length, 0, 'not parked for resume'); + assert.equal(internals.autoContinueStreaks.get('assistant') ?? 0, 0, 'its count is given back'); + assert.deepEqual(notices(framework), []); + } finally { + internals.quiesced = false; + await framework.stop(); + } + }); + + it('explicit: a turn whose prose bounced is woken by the bounce, not also continued', async () => { + const framework = await createFramework(true, 'explicit'); + stubRegistry(framework); + const triggers = script(framework, [prose('no prefix here'), prose('>>skip_reply fine')]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2, 'one wake, not a bounce and a continuation'); + assert.equal(triggers[1]!.reason, 'prose-bounce'); + assert.deepEqual(notices(framework), []); + } finally { + await framework.stop(); + } + }); + + it('a turn that had no channel passes that on: its continuation takes no fallback either', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { pendingRequests: InferenceRequest[] }; + const triggers = script(framework, [prose('nowhere to say this'), skipReply()]); + try { + internals.pendingRequests.push({ agentName: 'assistant', reason: 'channel-message', source: 'stub', timestamp: Date.now(), channelId: 'ghost' }); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2); + assert.equal(triggers[1]!.reason, 'auto-continue'); + assert.equal(triggers[1]!.channelId, 'ghost', 'the continuation names the same unresolved channel'); + } finally { + await framework.stop(); + } + }); + it('explicit: a turn whose last envelope is >>skip_reply waits; one that ends on a delivery continues', async () => { const framework = await createFramework(true, 'explicit'); const routed = stubRegistry(framework); @@ -512,7 +654,7 @@ describe('autoContinue', () => { await within(framework.runUntilIdle(), 'the run'); assert.equal(triggers.length, 2, 'the delivery continues; the >>skip_reply ending waits'); assert.deepEqual(routed.map((r) => r.text), ['posted', 'more']); - assert.deepEqual(notices(framework), [autoContinueNotice(1, 3, 'call skip_reply')]); + assert.deepEqual(notices(framework), [row(false)]); } finally { await framework.stop(); } @@ -598,14 +740,20 @@ describe('autoContinue config', () => { }); describe('autoContinueNotice', () => { - it('names the way to stop, and the count only near the cap', () => { + it('names the agent and the way to stop, says what unmarked words do, and the cap only on the last', () => { + assert.equal( + autoContinueNotice({ agentName: 'lena', stop: 'call skip_reply', last: false }), + "[continue] lena, another turn starts now. If you're done, call skip_reply; anything you write without it is sent as usual.", + ); + assert.equal( + autoContinueNotice({ agentName: 'lena', stop: 'end with a >>skip_reply envelope', last: true }), + "[continue] lena, another turn starts now. If you're done, end with a >>skip_reply envelope; anything you write without it is sent as usual. " + + 'This is the last continuation in a row: after this turn you wait for the next event either way.', + ); assert.equal( - autoContinueNotice(1, 3, 'call skip_reply'), - '[continue] Your last turn ended without skip_reply, so another one starts now. To stop and wait for the next event, call skip_reply.', + autoContinueNotice({ agentName: 'lena', stop: 'call skip_reply', last: false, heldAfterSend: true }), + '[continue] lena, another turn starts now. Your closing words last turn were held, not sent: that turn had already sent with a send tool. ' + + "If you're done, call skip_reply; anything you write without it is sent as usual.", ); - assert.match(autoContinueNotice(2, 3, 'call skip_reply'), / Continuation 2 of 3 in a row: one more is left\.$/); - assert.match(autoContinueNotice(3, 3, 'call skip_reply'), / This is the last continuation in a row \(3 of 3\)/); - assert.match(autoContinueNotice(1, 1, 'call skip_reply'), /last continuation in a row \(1 of 1\)/); - assert.equal(autoContinueNotice(1, 5, undefined), '[continue] Your last turn ended on its own, so another one starts now.'); }); }); From e05a31153623c4845bbec9268e1a55ba0ed5e4ec Mon Sep 17 00:00:00 2001 From: antra-tess Date: Fri, 23 Oct 2026 00:09:46 -0700 Subject: [PATCH 6/6] =?UTF-8?q?fix(prose):=20review=20of=20#744=20?= =?UTF-8?q?=E2=80=94=20a=20bounced=20close=20is=20the=20bounce's;=20the=20?= =?UTF-8?q?shared=20slot=20waits;=20a=20landed=20row=20counts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A turn whose closing prose bounced (explicit or hybrid) isn't continued: the bounce and its cap answer it, and continuing once the cap is spent restarted the loop the cap breaks (nissa-seru's run: 6 turns where off takes 3). TurnRouting.lastEnvelopeBounced, set by bounceProse, reset per envelope; replaces the queued-bounce check. - The row isn't written into the residents' shared slot while another reader's turn is alive: the scheduler requeues the continuation until it ends (as a clock row waits); a race at the write point counts as not stored. - A row whose write threw after it landed counts as written (read back with storedSince, as the clock row does), so no turn is withdrawn under a row. - A continuation a resumed turn takes in gives back its count. - Tests (40): bounce loop past the cap, shared-slot wait, landed-but-thrown, non-sole store failure (turn runs, count back), quiesce strip-and-park of a requeued marked trigger; each mutation-checked. Co-Authored-By: Claude Opus 5.5 --- .../feat-continue-after-prose.added.md | 2 +- docs/auto-continue.md | 16 ++- src/framework.ts | 84 +++++++++--- src/types/agent.ts | 6 +- test/auto-continue.test.ts | 122 +++++++++++++++++- 5 files changed, 199 insertions(+), 31 deletions(-) diff --git a/changelog.d/feat-continue-after-prose.added.md b/changelog.d/feat-continue-after-prose.added.md index 82fb58507..914c80c7c 100644 --- a/changelog.d/feat-continue-after-prose.added.md +++ b/changelog.d/feat-continue-after-prose.added.md @@ -12,7 +12,7 @@ The next turn keeps the turn's locus unless a batched message names another channel. Its wake has reason `auto-continue`. Not continued: refusals, `max_tokens` cut-offs, the stall guard's ends, silent heartbeat - ticks, a turn whose prose bounced, ephemeral runs, conversation forks, an + ticks, a turn whose closing prose bounced, ephemeral runs, conversation forks, an agent offered no way to stop, and a quiesce (which drops a queued continuation); a ` !` continuation runs instead. Each continuation is a full inference: an agent that answers and ends its turn gets at least one diff --git a/docs/auto-continue.md b/docs/auto-continue.md index 7a729710f..44c65c459 100644 --- a/docs/auto-continue.md +++ b/docs/auto-continue.md @@ -37,8 +37,9 @@ A turn continues when the model ended it (stop reason `end_turn` or `>>>skip_reply` in hybrid mode whose hold is still standing at the end; - it asked for a ` !` continuation (`>>#ops !`), which runs instead, with no row, and counts toward nothing; -- its prose bounced (explicit mode): the bounce already wakes the agent to - resend, under its own cap; +- its closing prose bounced (explicit or hybrid mode): the bounce, with the + bounce-wake cap, is that turn's answer, and continuing once the cap is + spent would restart the loop the cap breaks; - it was a silent heartbeat tick; - it ended with a refusal, a `max_tokens` cut-off, or the stall guard's `no_progress` / `round_limit`; @@ -64,7 +65,9 @@ acknowledge themselves and write no receipt): [continue] lena, another turn starts now. If you're done, call skip_reply; anything you write without it is sent as usual. ``` -It names the agent, since residents read one shared message slot. When the +It names the agent, since residents read one shared message slot. For the +same reason it is never written while another resident's turn is alive: the +continuation waits for that turn to end, as a clock row does. When the last turn's closing words were held because that turn had already sent, it adds `Your closing words last turn were held, not sent: that turn had already sent with a send tool.` The last continuation of a run adds `This is the last @@ -79,9 +82,10 @@ It is short and event-shaped: longer instruction blocks about output, sent as user-role messages, have tripped provider classifiers. The row is never written for a turn that doesn't run: a wake the inference -policy drops, or that a quiesce drops, writes none and gives back its count; -a row that can't be stored gives back its count, and withdraws a turn the -continuation alone caused. It is appended before the turn compiles and never +policy drops, or that a quiesce drops, writes none and gives back its count, +as does one a resumed turn (a budget restart) takes in; a row that can't be +stored gives back its count, and withdraws a turn the continuation alone +caused. A write that throws after its row landed counts as written. It is appended before the turn compiles and never retracted, so the prompt-cache prefix stays stable. ## The next turn diff --git a/src/framework.ts b/src/framework.ts index f36bc7bb4..2054d3727 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1101,6 +1101,13 @@ interface TurnRouting { * destination is named.) */ lastEnvelopeSkipped?: boolean; + /** + * Explicit or hybrid mode: the turn's latest envelope bounced (no + * destination, an unresolved or refused one, a failed send). Its bounce, + * with the bounce-wake cap, is that turn's answer, so autoContinue leaves + * it to the bounce. + */ + lastEnvelopeBounced?: boolean; /** Successful plain-prose destinations in delivery order; deduped by the receipt. */ deliveries: ChannelRef[]; /** Plain-prose segments the turn suppressed without drafts (proseRouting=disabled, or a round withheld before its scripts returned: ScriptRound.withhold), reported by its receipt. */ @@ -14820,9 +14827,23 @@ export class AgentFramework { // A batch with a cause of its own is that cause's turn: no row, and the // run starts again below. const continuation = requests.find(carriesContinuation); - const continuationMark = continuation?.autoContinueRun && !requests.some(hasOwnCauseForRun) + let continuationMark = continuation?.autoContinueRun && !requests.some(hasOwnCauseForRun) ? { ...continuation.autoContinueRun, sole: requests.every(carriesContinuation) } : undefined; + // A batch that resumes a logical turn (a budget restart, a guard retry) + // takes the continuation's wake into that turn, which writes no start + // inputs: its count goes back, and it writes no row. + if (continuationMark && requests.some((r) => resumesLogicalTurn(r.reason))) { + this.giveBackAutoContinue(agentName, 'a resumed turn took its wake'); + continuationMark = undefined; + } + // Its row goes into the residents' shared slot, which no write may + // enter while another reader's turn is alive (as for a clock row): the + // continuation waits until that turn has ended. + if (continuationMark && this.sharedSlotBusy(agent)) { + this.pendingRequests.push(...requests); + continue; + } // Start streaming inference (non-blocking — driveStream runs in background) // The request that grants the turn-alive exception must also define the @@ -16636,9 +16657,9 @@ export class AgentFramework { * reaches here, and clears the run's count. * * It continues only an agent that has a way to stop it is offered: the - * skip_reply tool, or in explicit/hybrid mode a skip envelope. It doesn't - * add a continuation to a bounce wake the turn's prose already queued, and - * never continues an ephemeral run or a conversation fork. The wake names + * skip_reply tool, or in explicit/hybrid mode a skip envelope. It leaves + * a turn whose closing prose bounced to the bounce, and never continues an + * ephemeral run or a conversation fork. The wake names * the turn's locus, or the unresolved channel of a turn that had none, so * the next turn speaks, or doesn't, where this one did; a batched message's * channel outranks it. Its mark (InferenceRequest.autoContinueRun) holds @@ -16679,10 +16700,11 @@ export class AgentFramework { console.error(`[continue] ${name}: not continued — no way to stop is offered (no skip_reply tool, no skip envelope in ${agent.proseRouting} mode)`); return; } - // The turn's prose bounced, and the bounce already wakes the agent to - // resend it (bounceProse, with its own cap). - if (this.pendingRequests.some((r) => r.agentName === name && r.reason === 'prose-bounce')) { - console.error(`[continue] ${name}: not continued — a bounce wake is already queued`); + // The turn's closing prose bounced: the bounce, with its own cap, is + // explicit and hybrid mode's answer to that turn, and continuing past + // the cap would restart the loop the cap breaks. + if (routing.lastEnvelopeBounced) { + console.error(`[continue] ${name}: not continued — the turn's closing prose bounced`); return; } const count = (this.autoContinueStreaks.get(name) ?? 0) + 1; @@ -16737,19 +16759,44 @@ export class AgentFramework { const text = autoContinueNotice({ agentName: agent.name, stop: mark.stop, last: count >= max, heldAfterSend: mark.heldAfterSend, }); + const before = this.storedCountOf(agent); + let id: MessageId; try { - const id = agent.getContextManager().addMessage( + id = agent.getContextManager().addMessage( 'user', [{ type: 'text', text }], { system: true, kind: 'auto-continue', agentName: agent.name, continuation: count, maxConsecutive: max } as MessageMetadata, ); - this.emitTrace({ type: 'message:added', messageId: id, source: 'auto-continue' }); - console.error(`[continue] ${agent.name}: continuation ${count} of ${max} starts`); - return true; } catch (err) { - console.error(`[continue] ${agent.name}: row not stored, so not continued:`, err); - return false; + // A write can throw after its row landed (context-manager's own + // propagation failing): read back from where the store stood, as the + // clock row does, so a row that is there isn't called missing. + const landed = before === undefined + ? undefined + : this.storedSince(agent, before, (m) => { + const meta = m.metadata as { kind?: unknown; agentName?: unknown } | undefined; + return meta?.kind === 'auto-continue' && meta.agentName === agent.name; + }, 'continue', 'the continuation row'); + if (!landed || landed.length === 0) { + console.error(`[continue] ${agent.name}: row not stored${landed ? '' : ' (whether it landed is unknown)'}, so not continued:`, err); + return false; + } + console.error(`[continue] ${agent.name}: the continuation row landed though its write threw:`, err); + id = landed[0]!.id; } + this.emitTrace({ type: 'message:added', messageId: id, source: 'auto-continue' }); + console.error(`[continue] ${agent.name}: continuation ${count} of ${max} starts`); + return true; + } + + /** Whether another reader of the residents' shared slot has a turn alive, + * so no row may be written there now (as storeDueClock checks). */ + private sharedSlotBusy(agent: Agent): boolean { + if (!this.sharedSlotAgents.has(agent)) return false; + for (const reader of this.sharedSlotAgents) { + if (reader !== agent && this.turnAlive(reader)) return true; + } + return false; } /** A continuation whose turn doesn't run: the run's count goes back to @@ -17056,6 +17103,7 @@ export class AgentFramework { ): Promise { const routing = round.routing; for (const envelope of splitHybridEnvelopes(rawText)) { + routing.lastEnvelopeBounced = false; const normalized = envelope; const attempted = normalized.startsWith('>>>'); const parsed = parseHybridProsePrefix(normalized); @@ -17201,6 +17249,7 @@ export class AgentFramework { // or continuation, even when delivery runs after that turn has ended. if (parsed.continueTurn && !round.keepPrivate) routing.continuation = true; routing.lastEnvelopeSkipped = parsed.kind === 'private'; + routing.lastEnvelopeBounced = false; if (parsed.kind === 'private') { // The text already lives in the assistant message (window/chronicle); @@ -17373,6 +17422,7 @@ export class AgentFramework { after: { substituted?: Draft; outcome?: 'unknown' } = {}, ): void { const name = agent.name; + round.routing.lastEnvelopeBounced = true; if (after.outcome === 'unknown' && !after.substituted) { // Words with no draft behind them, whose send threw, may have gone // out. Like an unknown outcome, that is marker-only: holding them as a @@ -18030,8 +18080,10 @@ export class AgentFramework { const mark = trigger.autoContinueRun; const { autoContinueRun: _mark, ...unmarked } = trigger; trigger = unmarked; - if (!this.writeAutoContinueNotice(agent, mark)) { - this.giveBackAutoContinue(agent.name, 'its row could not be stored'); + // Another reader's turn may have started since the scheduler checked. + const busy = this.sharedSlotBusy(agent); + if (busy || !this.writeAutoContinueNotice(agent, mark)) { + this.giveBackAutoContinue(agent.name, busy ? "another resident's turn is alive in the shared slot" : 'its row could not be stored'); if (mark.sole) { console.error(`[continue] ${agent.name}: turn not started (the continuation was its only cause)`); if (this.activeTurnTokens.get(agent.name) === turnToken) { diff --git a/src/types/agent.ts b/src/types/agent.ts index e11e82b2a..dd49f2cb5 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -244,9 +244,9 @@ export interface AgentConfig { * instead of a pause until the next event. A turn ended through * `skip_reply` (or another tool that ends the turn, such as `sleep`), or * whose last prose envelope is `>>skip_reply` (explicit) or `>>>skip_reply` - * (hybrid), pauses as usual. So does one whose prose bounced (the bounce - * wakes the agent), and an agent offered no way to stop (locus or disabled - * mode without an offered `skip_reply`). + * (hybrid), pauses as usual. So does one whose closing prose bounced (the + * bounce and its cap answer it), and an agent offered no way to stop + * (locus or disabled mode without an offered `skip_reply`). * * As the next turn starts, a short `[continue]` system row naming the agent * joins the window: another turn starts, how to stop, and that unmarked diff --git a/test/auto-continue.test.ts b/test/auto-continue.test.ts index 947193cc8..917b14263 100644 --- a/test/auto-continue.test.ts +++ b/test/auto-continue.test.ts @@ -141,6 +141,7 @@ describe('autoContinue', () => { autoContinue: AgentConfig['autoContinue'], proseRouting: AgentConfig['proseRouting'] = 'locus', allowedTools?: AgentConfig['allowedTools'], + others: AgentConfig[] = [], ): Promise { const framework = await AgentFramework.create({ storePath: join(tempDir, 'test.chronicle'), @@ -148,7 +149,7 @@ describe('autoContinue', () => { agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'You are a test agent.', proseRouting, autoContinue, ...(allowedTools ? { allowedTools } : {}), - }], + }, ...others], modules: [module], }); module.framework = framework; @@ -590,6 +591,116 @@ describe('autoContinue', () => { } }); + it("waits for another resident's turn before writing its row into the shared slot", async () => { + const framework = await createFramework(true, 'locus', undefined, [{ name: 'other', model: 'test-model', systemPrompt: 'Another resident.' }]); + stubRegistry(framework); + const internals = framework as unknown as { + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string): void; + processInferenceRequests(): Promise; + pendingRequests: InferenceRequest[]; + activeTurnTokens: Map; + }; + const triggers = script(framework, [skipReply()]); + try { + internals.autoContinueAfterTurn(framework.getAgent('assistant')!, {}, undefined, 'end_turn'); + internals.activeTurnTokens.set('other', 424242); + await internals.processInferenceRequests(); + assert.equal(triggers.length, 0, 'no turn while the other resident\'s turn is alive'); + assert.deepEqual(notices(framework), [], 'and no row in the middle of it'); + assert.equal(internals.pendingRequests.filter((r) => r.reason === 'auto-continue').length, 1, 'the continuation waits'); + internals.activeTurnTokens.delete('other'); + await within(framework.runUntilIdle(), 'the continuation'); + assert.equal(triggers.length, 1); + assert.deepEqual(notices(framework), [row(false)]); + } finally { + internals.activeTurnTokens.delete('other'); + await framework.stop(); + } + }); + + it('a row whose write threw after it landed counts as written: the turn runs, with one row', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage(role: string, content: unknown, metadata?: { kind?: string }): unknown; + }; + const original = cm.addMessage.bind(cm); + cm.addMessage = (role, content, metadata) => { + const id = original(role, content, metadata); + if (metadata?.kind === 'auto-continue') throw new Error('synthetic propagation failure after the write'); + return id; + }; + const triggers = script(framework, [prose('one'), skipReply()]); + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the run to settle'); + assert.equal(triggers.length, 2, 'the continuation runs'); + assert.deepEqual(notices(framework), [row(false)], 'its one row is there'); + } finally { + await framework.stop(); + } + }); + + it('a row that fails for a continuation batched with another framework wake gives back its count; the turn runs', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string): void; + pendingRequests: InferenceRequest[]; + autoContinueStreaks: Map; + }; + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + addMessage(role: string, content: unknown, metadata?: { kind?: string }): unknown; + }; + const original = cm.addMessage.bind(cm); + cm.addMessage = (role, content, metadata) => { + if (metadata?.kind === 'auto-continue') throw new Error('synthetic store failure'); + return original(role, content, metadata); + }; + const triggers = script(framework, [skipReply()]); + try { + internals.autoContinueAfterTurn(framework.getAgent('assistant')!, {}, undefined, 'end_turn'); + internals.pendingRequests.push({ agentName: 'assistant', reason: 'framework-test-wake', source: 'framework', timestamp: Date.now() }); + await within(framework.runUntilIdle(), 'the turn'); + assert.equal(triggers.length, 1, 'the other wake still has its turn'); + assert.deepEqual(notices(framework), []); + assert.equal(internals.autoContinueStreaks.get('assistant') ?? 0, 0, 'the count is given back'); + } finally { + await framework.stop(); + } + }); + + it('a quiesce parks a requeued trigger without its continuation mark, and gives back the count', async () => { + const framework = await createFramework(true); + stubRegistry(framework); + const internals = framework as unknown as { + autoContinueAfterTurn(agent: unknown, routing: unknown, trigger: unknown, stopReason: string): void; + processInferenceRequests(): Promise; + pendingRequests: InferenceRequest[]; + quiesced: boolean; + autoContinueStreaks: Map; + }; + try { + internals.autoContinueAfterTurn(framework.getAgent('assistant')!, {}, undefined, 'end_turn'); + // Swap the bare wake for a requeued turn trigger that carried the mark + // (as a provider park's requeue hands back), with a cause of its own. + const wake = internals.pendingRequests.find((r) => r.reason === 'auto-continue')!; + internals.pendingRequests.length = 0; + internals.pendingRequests.push({ ...wake, reason: 'framework-test-wake' }); + internals.quiesced = true; + await internals.processInferenceRequests(); + const parked = internals.pendingRequests.filter((r) => r.agentName === 'assistant'); + assert.equal(parked.length, 1, 'the trigger parks for resume'); + assert.equal(parked[0]!.reason, 'framework-test-wake'); + assert.equal(parked[0]!.autoContinueRun, undefined, 'without the mark, so resume writes no stale row'); + assert.equal(internals.autoContinueStreaks.get('assistant') ?? 0, 0); + } finally { + internals.quiesced = false; + internals.pendingRequests.length = 0; + await framework.stop(); + } + }); + it('a quiesce drops a queued continuation and gives back its count', async () => { const framework = await createFramework(true); stubRegistry(framework); @@ -614,15 +725,16 @@ describe('autoContinue', () => { } }); - it('explicit: a turn whose prose bounced is woken by the bounce, not also continued', async () => { + it("explicit: a turn whose closing prose bounced is left to the bounce, past its cap too", async () => { const framework = await createFramework(true, 'explicit'); stubRegistry(framework); - const triggers = script(framework, [prose('no prefix here'), prose('>>skip_reply fine')]); + // Every turn ends in unprefixed prose: the message's turn, then the two + // bounce wakes the cap allows. With the setting off it is the same 3. + const triggers = script(framework, [prose('no prefix'), prose('still none'), prose('none again'), prose('and again')]); try { trigger(framework); await within(framework.runUntilIdle(), 'the run to settle'); - assert.equal(triggers.length, 2, 'one wake, not a bounce and a continuation'); - assert.equal(triggers[1]!.reason, 'prose-bounce'); + assert.deepEqual(triggers.slice(1).map((t) => t.reason), ['prose-bounce', 'prose-bounce'], 'no continuation, before or after the bounce cap is spent'); assert.deepEqual(notices(framework), []); } finally { await framework.stop();