diff --git a/changelog.d/guard-withheld-names-original.fixed.md b/changelog.d/guard-withheld-names-original.fixed.md new file mode 100644 index 000000000..ee0e2cd84 --- /dev/null +++ b/changelog.d/guard-withheld-names-original.fixed.md @@ -0,0 +1,60 @@ +- A tool result the guard withholds now says where its original is. The + guard's notice ("Tool result withheld by the guard. The tool has already + executed.") is followed by the original's place: a workspace file under the + first writable mount's `tool-results/` (named + `-withheld---`, one per result, with any + character outside `A-Za-z0-9._-` replaced and the name cut to 80 + characters, so a long tool use id is cut), or, with no writable workspace + or a failed write, the guard's audit record (`framework/tool-result-guard`), + for an operator. The place holds the result as stored, which is its text: + for a result over the inline cap, its capped text, whose own note names the + file with the whole output. A failed write's own error stays in the audit + record and the logs; the audit record also says when no file was written or + a write threw. When saving the audit record itself + had failed as the result arrived, the stub says so rather than that the + original is kept there. This holds for every withheld settlement: a refusal, + an aborted stream, a batch that was never submitted, and one whose audit sync + failed. The notice still names no refusal and no category. Before this, a + withheld result left only the notice, so the agent couldn't tell what the + tool had done (agent-framework #277). The file is an ordinary workspace + file: the agent can read it back deliberately, and so can whoever else reads + the mount, such as another agent sharing it or a synced copy. +- How the stub is written. Where each original went is queued first, as an + audit record of its own, and the stub's edit is then made at once. If it + fails, it is retried before the next activation, compile of a request that + is sent, maintenance pass or stop, including for an ephemeral run or a + closed fork that has left the agent map, and the batch stays held out of + summaries until it lands. Until then, or if the store keeps failing or the + process stops first, the stub is the bare notice, and the `annotated` audit + record describes the spill's outcome when the annotation finishes, with `unfinished` for a write still running. That record is not revised when the write later settles. A file whose copy on disk fails + is still in the workspace's tree, where its tools read it, so the stub names + it; for this, `WorkspaceModule.writeBinary`'s result for that case carries + `data: { inChronicle: true }`, including when a post-commit operation throws. Inline truncation notes also name that readable tree file, and the disk-copy failure is logged and traced. Each batch has one 30-second deadline for all its withheld writes. A write still under way then is named as unfinished, with the audit record as the fallback. The guard stops waiting for it there; the write goes on, and may finish or fail later. `stop()` still waits for the workspace's own turns, which such a write holds, bounded only by a teardown timeout passed to `stop()` (indefinitely without one, as since #248). Until it + settles, no other withheld original is written, for this batch or a later + one of any agent: their stubs name the audit record. Writes can start before the first deadline fires; those batches each use their own deadline. A write that never settles keeps new withheld spills disabled for the process lifetime. A deadline that fires names its file on stderr; a stall's skips are logged once, naming the first file, and counted when it clears; an exhausted batch's skips are logged once per batch. The inline cap's spills are outside this deadline. Requests sent by the framework and the direct Agent API are compiled only once each annotation started so far has finished, + including one that began while the compile waited or ran, and the refusal + retry; a conversation fork seeded from its template copies only stubs whose + annotation has finished. A request preview doesn't wait, so rollback, + undo and redo never wait on a workspace write, and a preview can show a stub + before its place is added, as can public `compileContext()`. Stop lets running annotations finish while the + workspace is up; one that starts after that writes no file and is saved + before the store closes. A crash before a stub is written leaves the notice + alone, and the staged audit record still holds the original. If the active + branch changes while the originals are being written, no more are written: + a write that has not set its tree entry is refused where it commits, and it and the rest are recorded as skipped for the branch change. A write whose tree entry was already set remains on that branch. While the store is away the + stub stays as it was; if the branch is back by the edit, those results' + stubs name the audit record. So no original's tree entry lands on another + branch. For this, `WorkspaceModule.writeBinary` takes an optional `branch`: + the file's tree entry lands on that branch or isn't written, and a refused + write's result carries `data: { branchLeft: true }` (it isn't traced as a + failed spill). A mount has one directory on disk for every branch, so with + `autoMaterialize` a file's copy there, made after its tree entry, is there + whichever branch is active, like any other write's. The audit records `annotated` + (where each original went) and `stub_updated` (the stub's edit) carry the + batch's `sourceBranch`. The direct Agent API's refusal retry is now + compiled afresh, as the framework's is, so its longer stubs fit the budget. + +- With `maintenanceIntervalMs: 0`, periodic retries of retained guards are disabled; stop still retries them. A failing audit record blocks later queued edits and their compression holds until its write succeeds. Shutdown reports an error if that retry still fails. +- A framework-driven turn that compiled across stop is discarded before driving its stream, with a line saying why, even if the framework has restarted. A `start()` after a finished `stop()` starts a new lifecycle, with the spill gate open again; a `start()` made while a `stop()` is still draining isn't a supported order. Direct Agent calls remain governed by their caller's abort signal; a refusal retry checks that signal again after waiting for annotations. The direct API reads its unguarded tool results once, after the first compile, and sends the same ones with the refusal retry. +- Concurrent starts on one Agent, a direct inference call beside a framework-driven turn, are not made safe by this change: an Agent compiles while its status still reads idle, as before, and the annotation wait lengthens that window by up to the batch's deadline. `abortInference()` returns false during a direct call's initial assembly. +- `save_recent_image` reports a tree-only write as saved, including its path, size, MIME type, `inChronicle` and `diskCopyError` (the failure after the tree entry, whichever it was). A later failure in a multi-image save, returned or thrown, includes that file among the already-saved paths. An RFC-005 reference fetched to such a file names it. diff --git a/docs/tool-result-guard.md b/docs/tool-result-guard.md index efd50b5ad..4bddaa4e9 100644 --- a/docs/tool-result-guard.md +++ b/docs/tool-result-guard.md @@ -24,10 +24,14 @@ structured provider `stopReason: 'refusal'`, not keywords in output, ordinary provider errors, or natural-language refusal text. On that signal, every result in the batch is withheld. Tool calls, result IDs, -and error flags remain intact; each entire payload becomes: +and error flags remain intact; each entire payload becomes the guard's notice, > Tool result withheld by the guard. The tool has already executed. +followed by where the original is (see "Where a withheld original goes" +below). The same holds for every withheld settlement: a refusal, an aborted +stream, a batch that was never submitted, and one whose audit sync failed. + The refused attempt's partial assistant output is discarded. Inference is retried once on the same model, within the same logical turn; executed tools are never automatically run again. Refusal details stay in operational logs, @@ -86,11 +90,15 @@ placeholder is not summarized while awaiting a decision: it is staged with a context-manager compression hold (`addMessage(..., { holdCompression: true })`, context-manager >= 0.12), placed before the strategy's ingress callback runs. Every settlement path (accept, withhold, turn-ended, unsubmitted, aborted) -edits in the final content first and then releases the hold. Audit-only -failures cannot keep a settled message held. If an acceptance edit fails, -the guard keeps the accepted payload and its hold in a separate retry queue; -it does not re-arm provider submission or refusal handling. Compression sees -only the settled message: accepted output, or the notice for a withheld batch. +edits in the final content first and then releases the hold: an accepted +batch's payload, or a withheld batch's stub, which is edited once the host +has reported where each of its originals went, including a failed, skipped +or unfinished write, so a withheld batch stays held meanwhile. +Once a history edit succeeds, its own audit-only failure does not retain that message's hold. If an acceptance +edit or a stub's edit fails, the guard keeps the content and its hold in a +separate retry queue; it does not re-arm provider submission or refusal +handling. Compression sees only the settled message: accepted output, or a +withheld batch's stub. Holds are in-memory; a reopened process has no pending batch and therefore no holds. Raw pending output goes directly to the provider. A clean following response @@ -106,18 +114,25 @@ the provider receives placeholders instead of the originals, and the batch settl provider. Failed edits and audit writes are retried in order before the next real -activation or guard audit write, during maintenance, and after stream teardown -at shutdown. Preview/compile calls do not flush the queue. A history edit is -retried only on its original branch and only if the message still contains the -guard's placeholder or the accepted payload; later operator edits, removals, -and branch changes take precedence. Such an acceptance is audited with +activation, compile of a request that is sent, or guard audit write, during +maintenance, and after stream teardown at shutdown, including those of an +ephemeral run or a closed conversation fork that has left the agent map. + +With `maintenanceIntervalMs: 0`, periodic retries are disabled; stop still retries retained guards. The queue is ordered: a failing audit write blocks later queued history edits and their compression holds until the write succeeds. Shutdown retries the same queue and reports an error if it still fails. + +Previews do not flush the queue. A history edit is retried only on its +original branch and only if the message still contains the guard's +placeholder or the content the edit writes; later operator edits, removals, +and branch changes take precedence. Such an edit is audited with `historyUpdate: "superseded"` and its `sourceBranch` rather than overwriting the operator's view. Originals in the audit are never edited or removed. Storage retries are in-memory. If the process dies before storage recovers, unsynced records or unfinished edits cannot be promised to survive. Originals -already made durable remain in Chronicle; reopened context conservatively -retains any withholding notice whose acceptance edit did not persist. +already made durable remain in Chronicle. A reopened context keeps the +withholding notice of any batch whose acceptance edit or stub edit did not +persist, with no compression hold, and nothing edits it later; an +`annotated` record that was saved still names a written file. A stream that ends without a clean round and without a successor (abort, exhausted error retries; also an aborted direct `Agent.runInference`) @@ -130,6 +145,81 @@ deliberately conservative: an interrupted submission does not establish that the output was accepted. Similarly, if context compilation omits a pending exchange, its unsubmitted payload is not admitted to memory. +## Where a withheld original goes + +Each withheld result's original is written, as the result was stored (its +text; for a result over the inline cap, its capped text, whose own note names +the file with the whole output), to a file under the first writable mount's +`tool-results/` directory, named +`-withheld---.txt` with any character outside +`A-Za-z0-9._-` replaced and the name cut to 80 characters. Its stub names it: + +> Tool result withheld by the guard. The tool has already executed. Its result +> is in workspace file work/tool-results/2026-10-10-withheld-1a2b3c4d5e6f-0-toolu_01.txt. + +Otherwise the stub names the guard's audit record, for an operator: + +- with no writable workspace, or once the framework is stopping: "Its result is + kept in the guard's audit record (framework/tool-result-guard), for an + operator."; +- when the write failed: "Writing its result to workspace file … failed, so it + is kept only in the guard's audit record …". The error itself stays in the + audit record and the logs; +- when the write is still under way at its batch's deadline: "Its result was still being written to workspace file … when this note was made; if that file is missing, it is kept in the guard's audit record …". + +Each batch has one 30-second deadline, starting when the batch is queued, so a request waiting on a chain of batches waits at most one deadline past the latest one's arrival. Each successive write uses the remaining time. Once the deadline passes, the request can carry the unfinished-write notice; later results in the batch name the audit record. The underlying write continues, and may finish or fail later. The guard stops waiting for it at the deadline, but `stop()` still waits for the workspace's own turns, which such a write holds, bounded only by a teardown timeout passed to `stop()`, and indefinitely without one. A deadline that fires logs the file's path to stderr. While any write remains overdue, new withheld spills are skipped as `workspace_unanswered`, logged once for the stall and counted when it clears. A write that never settles keeps that gate closed for the process lifetime. Other batches can start writes before the first deadline fires, each within its own deadline. This bound covers withheld spills; the inline cap's original spill has no deadline. + +A file whose tree entry was set is readable through the workspace's tools, on the branch that was current when the entry was set, even if the rest of its write fails: its disk copy refused, or a throw after the entry. `WorkspaceModule.writeBinary` reports `data: { inChronicle: true }` for that outcome. Both withheld stubs and inline truncation notes name the tree file, while stderr and `tool:spill_failed` report the failure (the workspace's own `workspace:materialize-failed` event is emitted only for a refused push). `save_recent_image` also counts that file as saved, with its path, size, MIME type, `inChronicle` and `diskCopyError` (the failure after the tree entry, whichever it was) in the saved receipt, and an RFC-005 reference fetched to such a file names it. A write that throws before its tree entry is set fails the call, naming the images already saved. + +When saving the staged audit record itself had failed as the result arrived, +the stub says the result went to the audit record "though saving that record +had failed when the result arrived", rather than that it is kept there. + +The file is an ordinary workspace file. The agent can read it back +deliberately, and so can whoever else reads the mount: another agent sharing +it, a process, or a synced copy. Withholding keeps the original out of the +agent's context, not out of its workspace. + +If the active branch changes while the originals are being written (a +rollback, an undo, a host's switch), no more are written, and the rest are +recorded as skipped for the branch change. While the store is away from the +batch's branch, the stub stays as it was; if the branch is back by the edit, +as after an undo and a redo, each skipped result's stub names the audit +record. A write already under way names the batch's branch, and the workspace +refuses it where it commits the file (`WorkspaceModule.writeBinary`'s `branch` +option, whose refusal carries `data: { branchLeft: true }`), so no original's +tree entry lands on another branch. With `autoMaterialize`, a mount has one +directory on disk for every branch, so a written file's copy there, made after +its tree entry, is there whichever branch is active, like any other write's. + +Two audit records describe it, each with `batchId`, `messageId` and +`sourceBranch`: + +- `annotated`, queued first, on its own: one entry in `results` per result. + `{ toolUseId, path }` is a written file; with `error`, a failed write; with + `unfinished: true`, a write still under way at the deadline. `{ toolUseId, + error }` is a write that threw. `{ toolUseId, skipped }` is no write: + `no_workspace_file` (no writable workspace, or the framework was stopping), + `workspace_unanswered` (an earlier write was still under way past its + deadline), `batch_deadline` (the batch had no time left to start another write), or `branch_changed` (the workspace declined the write before its tree entry was set). + +The `annotated` record describes what each spill reported when the batch's annotation finished. An `unfinished` entry is not updated when the underlying write later settles. +- `stub_updated`: the stub's edit, with `historyUpdate: "superseded"` when an + operator's edit, removal, rollback or branch change came first. + +The stub's edit is made at once. If it fails, it waits in the retry queue +above, and the batch stays held until it lands, so no summary is made of a +stub that doesn't yet say where its original went. Meanwhile a request +compiled while it is failing, as the refusal retry is, carries the bare +notice, and the `annotated` record, once saved, names the file. An activation +or a direct API call retries the queue first, and fails while the edit keeps +failing, as it does for a failing acceptance edit. +Requests sent by the framework or the direct Agent inference API are compiled after the guard's annotations finish, including the refusal retry. A conversation fork also waits before compiling its template's seed. A preview, including public `compileContext()`, does not wait and can show a stub before its place is added. + +Concurrent starts on one Agent, a direct inference call beside a framework-driven turn, are not made safe here: an Agent compiles its request while its status still reads idle (or ready), as before this change, and the wait for annotations lengthens that window, by up to the batch's deadline. `abortInference()` returns false during a direct call's initial assembly; a caller's own abort signal is checked before the request is sent, and again after the refusal retry's annotation wait. The direct API reads its unguarded tool results once, after the first compile, and sends the same ones with the refusal retry. + +At stop, the framework lets running annotations finish while the workspace is up. An annotation that starts after that drain writes no file and is saved before the store closes. A framework-driven turn that compiled across stop is discarded before its stream is driven, even if the same framework has since restarted. A `start()` after a finished `stop()` starts a new lifecycle, with the spill gate open again; a `start()` made while a `stop()` is still draining isn't a supported order. These lifecycle gates cover framework-driven turns; direct Agent calls use their caller's abort signal. + For example, an operator can inspect records without changing the agent's view: @@ -160,3 +250,6 @@ retroactive and does not rewrite results already accepted into memory. - **Held stream text on abort.** Text held while a submitted batch is pending is dropped (not previewed) if the stream aborts or errors before the round settles. +- **Withheld originals in the workspace.** Every reader of the mount can read + a withheld original's file, and nothing removes it; delete it as any + workspace file. diff --git a/src/agent.ts b/src/agent.ts index 412707a33..5dd7c3d45 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -2,7 +2,7 @@ import type { Membrane, NormalizedMessage, NormalizedRequest, ContentBlock, Yiel import { isAbortedResponse } from '@animalabs/membrane'; import { createHash } from 'node:crypto'; import type { CacheWireReceipt, KvUnifiedRequestHooks } from './kv-unified-wire.js'; -import { ToolResultGuard, TOOL_RESULT_GUARD_NOTICE } from './tool-result-guard.js'; +import { ToolResultGuard } from './tool-result-guard.js'; import { indexSilentTickRows, separateSilentHeartbeatTicks, @@ -670,28 +670,37 @@ export class Agent { this.toolResultGuard.storeResults(content, wireResults, this._state.toolResults); } - // Compile context (with optional injections) - const { messages, systemInjections } = await this.compileWithInjections(budget, injections); - - // If we have pending tool results, add them - if (this._state.status === 'ready' && !guardedResults) { - const toolResultMessages = this.buildToolResultMessages(this._state.toolResults); - messages.push(...toolResultMessages); - } - - const request: NormalizedRequest = { - messages: this.toolResultGuard.prepareRequest(messages, true), - system: this.buildSystemPrompt(systemInjections), - config: { - model: this.model, - maxTokens: this.maxTokens, - ...(this.temperature !== undefined && { temperature: this.temperature }), - ...(this.thinking !== undefined && { thinking: this.thinking }), - }, - tools: tools.length > 0 ? tools : undefined, - ...(this.providerParams && { providerParams: this.providerParams }), - assistantParticipant: this.name, + // Compile the context (with optional injections) into a request. The + // guard's refusal retry is assembled the same way, afresh. It is compiled + // only once each withheld stub's annotation has finished, including one + // that began during the compile, and after a stub edit that failed has + // been retried; one whose store still fails is compiled as the bare + // notice. + let unguardedResults: ReturnType | undefined; + const assemble = async (): Promise => { + const { messages, systemInjections } = await this.toolResultGuard.afterAnnotations( + () => this.compileWithInjections(budget, injections)); + // Read once, after the first compile, since results may have become + // ready while it ran; the refusal retry, assembled while the agent is + // inferring, sends the same ones. + unguardedResults ??= this._state.status === 'ready' && !guardedResults + ? this.buildToolResultMessages(this._state.toolResults) : []; + messages.push(...unguardedResults); + return { + messages: this.toolResultGuard.prepareRequest(messages, true), + system: this.buildSystemPrompt(systemInjections), + config: { + model: this.model, + maxTokens: this.maxTokens, + ...(this.temperature !== undefined && { temperature: this.temperature }), + ...(this.thinking !== undefined && { thinking: this.thinking }), + }, + tools: tools.length > 0 ? tools : undefined, + ...(this.providerParams && { providerParams: this.providerParams }), + assistantParticipant: this.name, + }; }; + const request = await assemble(); const abortController = new AbortController(); if (options?.signal) { @@ -704,7 +713,7 @@ export class Agent { // Set state to inferring this._inferenceStartedAt = Date.now(); - const inferencePromise = this.doInference(request, abortController.signal); + const inferencePromise = this.doInference(request, abortController.signal, assemble); this._state = { status: 'inferring', promise: inferencePromise, abortController }; try { @@ -793,6 +802,10 @@ export class Agent { * it does not touch agent state and does not call the membrane, and * `ContextManager.compile` is itself side-effect-free (compression runs in * the background). Safe to call regardless of the agent's current status. + * It doesn't wait for the tool-result guard's annotations, so a preview + * never waits on a workspace write (rollback, undo and redo report one), and + * may show a withheld stub before its place is added; the activation waits + * for them around its call. */ async buildActivationRequest( availableTools: ToolDefinition[], @@ -961,7 +974,11 @@ export class Agent { this.lastStreamRealInputTokens = 0; this.lastStreamOutputTokens = 0; - const request = await this.buildActivationRequest(availableTools, injections, budget, compressionTools, options); + // Compiled only once each withheld stub's annotation has finished, + // including one that began during the compile, and after a stub edit that + // failed has been retried (see afterAnnotations). + const request = await this.toolResultGuard.afterAnnotations( + () => this.buildActivationRequest(availableTools, injections, budget, compressionTools, options)); request.messages = this.toolResultGuard.prepareRequest(request.messages, true); const receiptAware = (this.contextManager as unknown as { getStrategy?: () => unknown }) @@ -1179,9 +1196,13 @@ export class Agent { private async doInference( request: NormalizedRequest, - signal?: AbortSignal + signal: AbortSignal | undefined, + /** Compiles the context into a request again, for the guard's retry. */ + reassemble: () => Promise, ): Promise { - let response = await this.membrane.stream(request, { signal }); + let response = signal?.aborted + ? { aborted: true as const, reason: 'user' as const } + : await this.membrane.stream(request, { signal }); // Usage of a refused attempt abandoned by the guard; it was billed and // is added to the retry's usage below rather than overwritten. let abandonedUsage: { inputTokens: number; outputTokens: number } | undefined; @@ -1191,12 +1212,14 @@ export class Agent { const ids = this.toolResultGuard.withhold('unknown'); if (ids) { abandonedUsage = response.usage; - const withheld = new Set(ids); - request = { ...request, messages: request.messages.map((message) => ({ - ...message, content: message.content.map((block) => block.type === 'tool_result' && withheld.has(block.toolUseId) - ? { type: 'tool_result', toolUseId: block.toolUseId, content: TOOL_RESULT_GUARD_NOTICE, isError: block.isError } : block), - })) }; - response = await this.membrane.stream(request, { signal }); + // The retry is compiled afresh, like every request: once each stub's + // annotation has finished (agent-framework #277), and against the + // budget without the batch's reservation, so a stub longer than its + // original can't carry the request past the budget. + const retry = await reassemble(); + response = signal?.aborted + ? { aborted: true, reason: 'user', partialUsage: response.usage } + : await this.membrane.stream(retry, { signal }); } } diff --git a/src/framework.ts b/src/framework.ts index 446231423..34c2ef445 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -89,6 +89,7 @@ import { ChannelRegistry, type ChannelToolOrigin } from './mcpl/channel-registry import { ConversationRouter, CONVERSATION_ROUTING_DEPRECATION_NOTICE } from './mcpl/conversation-router.js'; import { safeSlice } from './safe-slice.js'; import type { WorkspaceModule } from './modules/workspace/index.js'; +import type { ToolResultGuard, WithheldSpill } from './tool-result-guard.js'; import { toolResultDataToHistoryString, truncateForHistory, @@ -814,6 +815,10 @@ class DefaultErrorPolicy implements ErrorPolicy { /** Default sync interval in milliseconds */ const DEFAULT_SYNC_INTERVAL_MS = 1000; +/** Total time one withheld batch may spend awaiting workspace writes. + * At the deadline its unfinished write is named as such, and later + * originals stay in the audit. This bound applies to withheld spills. */ +const WITHHELD_SPILL_DEADLINE_MS = 30_000; /** Poll pending context-strategy maintenance independently of user activity. */ const DEFAULT_MAINTENANCE_INTERVAL_MS = 5000; /** Bound one pass so a large backlog yields to inference and other agents. */ @@ -1056,6 +1061,8 @@ export class AgentFramework { private providerAccelerationJitterMs = PROVIDER_ACCELERATION_JITTER_MS; private providerHoldHook: ProviderHoldHook | undefined; private providerAdmissionClosed = false; + /** Invalidates framework turns still assembling when stop begins. */ + private lifecycleGeneration = 0; /** Last time we reported stale (busy-requeued) inference requests, per agent. */ private staleWarnAt = new Map(); /** Per-agent last inference activity (epoch ms), for /healthz + doctor tooling. */ @@ -1981,6 +1988,10 @@ export class AgentFramework { this.running = true; this.providerAdmissionClosed = false; + // A new lifecycle: a turn compiled before an earlier stop() never drives, + // and withheld spills, closed for the generation a stop ended, are open + // for this one. + this.lifecycleGeneration++; this.loopPromise = this.runLoop(); // Start periodic sync timer (if enabled) @@ -2043,6 +2054,10 @@ export class AgentFramework { // store is still open, rather than leaving them to a reboot replay. this.ackDeferredWrites(); this.providerAdmissionClosed = true; + this.lifecycleGeneration++; + // The generation this stop ends; withheld spills close for it alone, so a + // start() made before this stop finishes reopens them for its own. + const stoppingGeneration = this.lifecycleGeneration; this.queue.close(); this.tuneOutCoordinator?.stop(); @@ -2109,6 +2124,18 @@ export class AgentFramework { this.eventGate?.dispose(); this.livenessWatchdog?.stop(); + // A stopping stream's batch settles as withheld, and its stub names where + // the original went: let those writes finish while the workspace is up. + // Tracked here rather than per agent, since an ephemeral run or a closed + // fork leaves the agent map while its annotation can still be running. + while (this.guardAnnotations.size > 0) await Promise.all([...this.guardAnnotations]); + // An annotation that starts after this point, such as one from a direct + // Agent API call stop doesn't cancel, starts no write to the stopping + // workspace: its stub names the audit record. A write that was past its + // deadline isn't waited for here, but the workspace's own stop waits for + // the mount turn it holds, bounded only by teardownTimeoutMs. + this.guardSpillsClosedFor = stoppingGeneration; + // Stop modules and MCPL servers in parallel const shutdownPromises: Promise[] = [this.moduleRegistry.stopAll(teardownTimeoutMs)]; if (this.mcplServerRegistry) { @@ -2123,13 +2150,19 @@ export class AgentFramework { const teardownSucceeded = shutdownErrors.length === 0; + // Those later annotations write nothing, so they finish at once; let them + // record and edit their stubs before the store closes. + while (this.guardAnnotations.size > 0) await Promise.all([...this.guardAnnotations]); + // Streams may queue storage repairs while being cancelled above. Retry - // after their teardown, while the store is still open, before final sync. - for (const agent of this.agents.values()) { + // after their teardown, while the store is still open, before final sync: + // every live agent's guard, and every other guard that isn't idle, whose + // ephemeral run or closed fork has left the agent map. + for (const [guard, name] of this.guardsWithStorageWork()) { try { - agent.toolResultGuard.flushUnrecorded(); + guard.flushUnrecorded(); } catch (error) { - console.error(`[tool-result-guard] agent=${agent.name} could not finish queued storage work at stop:`, error); + console.error(`[tool-result-guard] agent=${name} could not finish queued storage work at stop:`, error); shutdownErrors.push(error); } } @@ -2376,6 +2409,17 @@ export class AgentFramework { } private async runQueuedMaintenance(): Promise { + // A guard whose agent has left the map gets no activation to retry its + // queue; retry it here. It leaves busyGuards once nothing of it is left. + const live = new Set([...this.agents.values()].map((agent) => agent.toolResultGuard)); + for (const [guard, name] of this.busyGuards) { + if (live.has(guard)) continue; + try { + guard.flushUnrecorded(); + } catch (error) { + console.error(`[tool-result-guard] agent=${name} storage retry failed during maintenance:`, error); + } + } const queued = [...this.agents.values()].flatMap((agent) => { try { agent.toolResultGuard.flushUnrecorded(); @@ -2633,7 +2677,14 @@ export class AgentFramework { if (!mount) return null; const path = `${mount}/refs/${fileName}`; const res = await workspace.writeBinary(path, data, mimeType); - return res.success ? path : null; + if (res.success) return path; + // Stored, with its disk copy failed: the workspace's tools read it + // from the tree, so the reference is fetched to that path. + if ((res.data as { inChronicle?: boolean } | undefined)?.inChronicle === true) { + console.error(`[refs] ${path} is in the workspace's Chronicle tree, but the rest of its write failed: ${res.error ?? 'unknown error'}`); + return path; + } + return null; }, ); } @@ -2806,7 +2857,38 @@ export class AgentFramework { }; } - private restoreToolResultGuardSetting(agent: Agent): void { + /** Every guard's annotation still running, for stop() (agent-framework #277). */ + private readonly guardAnnotations = new Set>(); + /** Each guard that isn't idle, by agent name, as the guard reports it. An + * ephemeral run or a closed fork leaves the agent map with an annotation + * maybe still running, or a stub edit, acceptance edit or audit write + * queued, which maintenance and stop then retry (agent-framework #277). A + * guard is here from the moment it has work left until the moment it has + * none, however that comes about. */ + private readonly busyGuards = new Map(); + /** Withheld originals' writes still under way past their deadline: while + * any is, the workspace isn't answering and no other withheld original's + * write is started. Several can be, when writes of different agents' + * batches stalled together. */ + private readonly stalledSpills = new Set>(); + /** Withheld spills skipped since the current stall began. */ + private stallSkips = 0; + /** See WITHHELD_SPILL_DEADLINE_MS; an instance field so tests can shorten it. */ + private withheldSpillDeadlineMs = WITHHELD_SPILL_DEADLINE_MS; + /** The lifecycle generation whose stop() has let the running annotations + * finish: a withheld original is then no longer written to a workspace + * that is stopping, and its stub names the audit record. A start() after + * that stop() has finished has a generation of its own, so the spill gate + * is open again for an instance kept with a store it doesn't own and that + * is still open (start() doesn't restore stopped modules or MCPL + * connections, and refuses an owned store that stop() closed). A start() + * made while a stop() is still draining isn't a supported order. */ + private guardSpillsClosedFor: number | null = null; + + /** A new agent's guard: its persisted setting, the tool-results spill its + * withheld originals are written to, and the set stop() waits on + * (agent-framework #277). */ + private adoptToolResultGuard(agent: Agent): void { const enabled = (this.store.getStateJson(FRAMEWORK_STATE_ID) as { toolResultGuards?: Record; } | null)?.toolResultGuards?.[agent.name]; @@ -2814,6 +2896,28 @@ export class AgentFramework { throw new Error(`Invalid persisted tool_result_guard for ${agent.name}: expected a boolean`); } agent.toolResultGuard.setOverride(enabled); + const guard = agent.toolResultGuard; + guard.setHost({ + spill: (label, text, branch, deadline) => this.spillWithheld(label, text, branch, deadline), + deadline: () => Date.now() + this.withheldSpillDeadlineMs, + track: (annotation) => { + this.guardAnnotations.add(annotation); + void annotation.finally(() => this.guardAnnotations.delete(annotation)); + }, + idle: (idle) => { + if (idle) this.busyGuards.delete(guard); + else this.busyGuards.set(guard, agent.name); + }, + }); + } + + /** Each guard that may have queued storage work, once, by agent name: the + * live agents' and every other that isn't idle. */ + private guardsWithStorageWork(): Map { + const guards = new Map(); + for (const agent of this.agents.values()) guards.set(agent.toolResultGuard, agent.name); + for (const [guard, name] of this.busyGuards) if (!guards.has(guard)) guards.set(guard, name); + return guards; } private buildThinkTool( @@ -4263,7 +4367,7 @@ export class AgentFramework { }); const agent = new Agent(config, contextManager, this.membrane); - this.restoreToolResultGuardSetting(agent); + this.adoptToolResultGuard(agent); agent.markContextConsumed(); // RFC-006: nothing stored before now is replaceable this.ephemeralCandidates.set(agent, contextManager); @@ -6569,7 +6673,7 @@ export class AgentFramework { }); const agent = new Agent(config, contextManager, this.membrane); - this.restoreToolResultGuardSetting(agent); + this.adoptToolResultGuard(agent); agent.markContextConsumed(); // RFC-006: nothing stored before now is replaceable this.sharedSlotAgents.add(agent); const restoredSettings = this.readAgentRuntimeSettings(config.name); @@ -6646,7 +6750,7 @@ export class AgentFramework { allowedTools: [...SUBCONSCIOUS_TOOL_NAMES, 'think', 'skip_reply', 'end_turn'], }; const agent = new Agent(agentConfig, contextManager, this.membrane); - this.restoreToolResultGuardSetting(agent); + this.adoptToolResultGuard(agent); agent.markContextConsumed(); // RFC-006: nothing stored before now is replaceable this.sharedSlotAgents.add(agent); // reads the shared slot through its merged view this.agents.set(name, agent); @@ -7527,7 +7631,11 @@ export class AgentFramework { // seeding again would stack another template copy on top of it. const { messages: existing } = await contextManager.compile(); if (existing.length === 0) { - const { messages: compiled } = await template.getContextManager().compile(); + // The copy is the fork's own, and the template's annotation edits only + // the template's message: seed only once each of the template's + // annotations has finished. + const { messages: compiled } = await template.toolResultGuard.afterAnnotations( + () => template.getContextManager().compile()); for (const msg of compiled) { const participant = msg.participant === template.name ? name : msg.participant; contextManager.addMessage(participant, msg.content); @@ -7536,7 +7644,7 @@ export class AgentFramework { const config: AgentConfig = { ...templateConfig, name, strategy: undefined }; const agent = new Agent(config, contextManager, this.membrane); - this.restoreToolResultGuardSetting(agent); + this.adoptToolResultGuard(agent); agent.markContextConsumed(); // RFC-006: nothing stored before now is replaceable this.agents.set(name, agent); this.agentConfigs.set(name, config); @@ -9277,6 +9385,7 @@ export class AgentFramework { turnToken: number, ownsProviderGate: boolean, ): Promise { + const lifecycle = this.lifecycleGeneration; const continuingTurn = resumesLogicalTurn(trigger?.reason); const turnProseRouting = trigger?.suppressProse ? 'disabled' : agent.proseRouting; // Flush messages deferred during the PREVIOUS turn — before the @@ -9544,9 +9653,10 @@ export class AgentFramework { injections = injections ? [...injections, silentInjection] : [silentInjection]; } - // An ephemeral watchdog may dispose this Agent while hooks/compile await. - // Do not create a provider stream after its generation lost ownership. - if (this.agents.get(agent.name) !== agent) { + // An ephemeral watchdog may dispose this Agent while hooks/compile await, + // and stop() may close provider admission. Do not create a provider + // stream after its generation lost ownership, or once admission closed. + if (this.agents.get(agent.name) !== agent || this.providerAdmissionClosed) { this.channelRegistry?.stopTyping(); this.eventGate?.onInferenceEnded(agent.name); return false; @@ -9571,9 +9681,25 @@ export class AgentFramework { } = await agent.startStreamWithInjections(tools, injections, undefined, compressionTools, { silentHeartbeat: trigger?.silentHeartbeat, }); - if (this.agents.get(agent.name) !== agent) { + // The compile can wait for annotations (each batch has one spill + // deadline), so check ownership, admission, and the lifecycle again. + // A stop followed by start must not revive this earlier turn. + // Membrane's yielding stream starts its request when it is first + // iterated, so cancelling it here, before driveStream, sends nothing. + if (this.agents.get(agent.name) !== agent || this.providerAdmissionClosed || lifecycle !== this.lifecycleGeneration) { + console.error(`[turn] ${agent.name}: a turn compiled while ${ + this.agents.get(agent.name) !== agent ? 'the agent was replaced or removed' + : this.providerAdmissionClosed ? 'the framework was stopping' + : 'the framework was stopped and started again'} is dropped before it sends anything`); stream.cancel(); agent.cancelStream(); + if (this.agents.get(agent.name) === agent) { + // The compile took the agent's stream id, so a frame that started + // this one (a guard or budget restart) no longer owns the entries + // it left, and its teardown leaves them to this frame. + this.activeStreams.delete(agent.name); + this.pendingAssistantBlocks.delete(agent.name); + } this.channelRegistry?.stopTyping(); this.eventGate?.onInferenceEnded(agent.name); return false; @@ -11979,6 +12105,133 @@ export class AgentFramework { return { blocks, spilled }; } + /** Where a tool result labelled `label` is written: `/tool-results/