diff --git a/.changeset/controller-review-recovery.md b/.changeset/controller-review-recovery.md new file mode 100644 index 000000000..41eb11301 --- /dev/null +++ b/.changeset/controller-review-recovery.md @@ -0,0 +1,5 @@ +--- +"@agentskit/core": patch +--- + +Keep older pending or superseded tool calls from blocking later controller turns, recover decisions after a pending-registration save failure, and resume interrupted streaming messages loaded from memory. Use a fresh run ID for durable decision executions outside a live stream. diff --git a/.changeset/fair-tools-resume.md b/.changeset/fair-tools-resume.md new file mode 100644 index 000000000..96919f30a --- /dev/null +++ b/.changeset/fair-tools-resume.md @@ -0,0 +1,7 @@ +--- +"@agentskit/core": patch +--- + +Register streamed confirmations on arrival, treat identical in-flight decisions as no-ops, and preserve active streams when a decision is rejected. Resume only the latest completed assistant turn after every confirmation is decided, with results immediately following their originating assistant in call order. + +Restarting streamed decisions requires current ChatMemory because immutable arrival snapshots may omit subsequent calls. Without current conversation memory, an incomplete streamed snapshot raises AK_CONFIG_INVALID before claiming or executing. diff --git a/.doc-bridge/capabilities.json b/.doc-bridge/capabilities.json index 9c549ddb0..45547a6e6 100644 --- a/.doc-bridge/capabilities.json +++ b/.doc-bridge/capabilities.json @@ -7,7 +7,7 @@ "name": "AgentsKit", "root": "." }, - "contentHash": "85ceaef8fdb67f5b5644c4b297c3b9bade5667bb1458af891e07718f69633d7f", + "contentHash": "8efbc15b9798e169144aed1ad9efab22b1418a9053a014ee8806baea29b7f4e0", "artifacts": { "index": ".doc-bridge/index.json", "llmsTxt": ".doc-bridge/llms.txt" diff --git a/.doc-bridge/index.json b/.doc-bridge/index.json index fee47dd87..f1028b304 100644 --- a/.doc-bridge/index.json +++ b/.doc-bridge/index.json @@ -1,8 +1,8 @@ { "schemaVersion": 1, - "contentHash": "85ceaef8fdb67f5b5644c4b297c3b9bade5667bb1458af891e07718f69633d7f", + "contentHash": "8efbc15b9798e169144aed1ad9efab22b1418a9053a014ee8806baea29b7f4e0", "contentHashAlgo": "sha256-normalized-v1", - "generatedAt": "2026-10-08T03:57:45.715Z", + "generatedAt": "2026-10-08T05:36:46.849Z", "project": { "name": "AgentsKit", "root": "." @@ -3723,7 +3723,7 @@ "document", "human" ], - "contentHash": "a45c2e7c4960a50d190312d9eabc737f8973aa7d2b1720270a630578d0880423" + "contentHash": "c0805db69400cae3a7a0abc8f1dbacfc9bae0078798f4a2157df89c22c5bdf32" }, { "id": "document:apps/docs-next/content/docs/reference/packages/cross-platform.mdx", @@ -6050,7 +6050,7 @@ "document", "human" ], - "contentHash": "6e1205db68cf23787d54c5b5e30edc45263ebebb31bf12381caf66080304a13a" + "contentHash": "542f2d33984ae4c31a163e5749abdfc65dd8c9153b89d63db1184f29d5a711c3" }, { "id": "document:docs/architecture/adrs/0043-partition-bound-rag.md", @@ -7022,7 +7022,7 @@ "document", "project" ], - "contentHash": "41c9985b7138e1f2880ca675c7dbecdcf8a154c2b8d836af116d65d98e14ba22" + "contentHash": "5f3bb79e91401ee25cb9dcda785dea22b99d3605c5f62c528fc3623f8f3965b4" }, { "id": "document:packages/cross-platform/CONVENTIONS.md", @@ -18869,6 +18869,7 @@ "title": "controller-decision-internal.ts", "path": "packages/core/src/controller-decision-internal.ts", "symbols": [ + "buildToolContinuation", "decide" ], "tags": [ @@ -18876,7 +18877,7 @@ "ts", "src" ], - "contentHash": "69d0754cd15289a22b914a9f518de47c2dd0987a21dada4c5575b0aa2fed8d6b" + "contentHash": "79928bb33e5ed6a967353e3914694dbc928d90c8ed1f86c75500c373427a9518" }, { "id": "module:packages/core/src/controller-helpers.ts", @@ -18889,6 +18890,7 @@ "buildRetrievalMessage", "buildToolContinuation", "createControllerToolLoop", + "insertToolResults", "mapMessageById", "mapToolCallById", "mergeSystemMessages", @@ -18900,7 +18902,7 @@ "ts", "src" ], - "contentHash": "41bd7d1aa578b0ebc774eb7038ede6ef6d74662e0f627690f4e4fb27143dd90d" + "contentHash": "b8e745c83c267b54438be20d3243e7a2c463017e7a62926a994b1a695a354d48" }, { "id": "module:packages/core/src/controller-persistence.ts", @@ -18931,7 +18933,7 @@ "ts", "src" ], - "contentHash": "fa49e4c60cfaa2f824d943fca30c4d98f8bdb490a3accd7f5234ffc0e1478335" + "contentHash": "7cc1474de6d6faea4c5b17bd28e883ad0785c1e525988168f0cba60c04a2026c" }, { "id": "module:packages/core/src/controller.ts", @@ -18946,7 +18948,7 @@ "ts", "src" ], - "contentHash": "bfaa4fdc1be8b3536dcc1df4d1b5a8d4259f88b8ef7a3a15efa342e30cdaa6b8" + "contentHash": "c64ffe19da240db2d517cc5782e37e87a8defe6364fff4ae0c4c340d81a1ac60" }, { "id": "module:packages/core/src/errors.ts", @@ -20262,7 +20264,7 @@ "test", "tests" ], - "contentHash": "5682144bdbb19e39ccde70a760bf83f57899833a8e91a811d8da41d33cb51991" + "contentHash": "fab8069b6983f1d8da3218ab536ad9f7ecdd6c87f93d398357819d70218da4b0" }, { "id": "module:packages/core/tests/controller-edit-regenerate.test.ts", @@ -20313,7 +20315,7 @@ "mjs", "tests" ], - "contentHash": "376489948ab4d4b2d5941f97d5110ff3b72949467f8f3f41423984f0b750443a" + "contentHash": "7bc64ff6c8b16285878dad1847c17d258396f5509dc32feee41e240ccc473b2b" }, { "id": "module:packages/core/tests/controller-tool-call.test.ts", @@ -38294,7 +38296,7 @@ "changes": {} }, "inputs": { - "hash": "ee634f0a707bb1a44a72879697761ccd8d8979c3fe4438c5120cbbf977dd3829", + "hash": "743a3f6ba78828b3446b96d6e1bd1d91559448c65fdcaf092a9c27daec5bb91a", "fileCount": 2546, "projectionVersion": 2 }, @@ -38317,9 +38319,9 @@ "projection": { "type": "retrieval-index", "schemaVersion": 1, - "contentHash": "c351080629b1827cfb7fd526e0ea3c45664c22b69012a955e77c58c293e03d76", + "contentHash": "7eaaccf8e662f3f69d83cd6b10d747f392e82339268008f475c09b99a80d4f67", "contentHashAlgo": "sha256-normalized-v1", - "snapshotHash": "b8340635b73aa608508f3a93e6461a7343c158545cf34b3e25b6eaf99ee3bc5b", + "snapshotHash": "e2f485f514c517eb77ebaade41b6c6342a13e6b36c6a6b0d91a8cd677a4f3b9c", "overlayHash": "f99666ff9119084dd25c2974ae64eb6935ec2aa785ff5ad478fb9d541f7a5e94", "configurationHash": "9662cc07ea4ab203ff2ddf457d2db48b5375a7ab4a1f8fbc79dfebcaa8f9a00c", "lexiconVersion": 1, @@ -38351,11 +38353,11 @@ ], "averageFieldLength": { "aliases": 3.0139944022391045, - "body": 61.660135945621754, + "body": 61.61895241903239, "headings": 4.034386245501799, "path": 7.611755297880848, "summary": 4.562574970011995, - "symbols": 7.442622950819672, + "symbols": 7.445821671331467, "title": 3.359456217512995 } }, @@ -51096,7 +51098,7 @@ "path": "apps/docs-next/content/docs/for-agents/core.mdx", "symbols": "", "summary": "Stable TypeScript contracts (Adapter, Tool, Memory, Retriever, Skill, Runtime) + `createChatController` + primitives. Target: <10 KB gzipped, zero runtime deps.", - "body": "Purpose Stable TypeScript contracts (Adapter, Tool, Memory, Retriever, Skill, Runtime) + createChatController + primitives. Target: <10 KB gzipped, zero runtime deps. Install Primary exports (main entry) - createChatController(config) — state machine behind every UI - proposeToolCall(proposal) — controller method that places a trusted, validated action into the same confirmation lifecycle; it never executes directly. The standalone helper is available from @agentskit/core/tool-proposal . Returns { getState, subscribe, send, stop, retry, edit, regenerate, setInput, clear, decide, approve, deny, updateConfig } . - defineTool(config) — typed tool factory with JSON-Schema input inference. - controller.send(string ContentPart[]) preserves multimodal parts and text projection. - controller.decide(id, 'approve' 'deny', reason?) uses ChatConfig.decisionStore to validate the current conversation before claiming, prioritizing live controller state over memory and preserving unsaved later turns and using the pending snapshot only as a fallback ( ToolDecisionStore , ToolDecisionRecord ). The store provides insert-if-absent putPending , linearizable get , atomic pending-to-claimed claim , and matching-claim terminal settle . Scope it to an application-authorized conversation. Same-decision terminal replay returns the recorded tool outcome; unknown IDs raise AK ACTION NOT FOUND , competing or in-flight decisions raise AK ACTION ALREADY DECIDED . Failed and claimed records are never automatically retried. Serialize different decisions per conversation to protect transcript writes. - approve / deny are deprecated. With a decision store they delegate to decide , including atomic claims and typed errors; without a store they preserve generation-local behavior. - ChatConfig.authorizeToolCall — trusted host authorization before tool proposal and again before execution. Callback failures default to deny. - createInMemoryMemory / createLocalStorageMemory — default ChatMemory implementations. - createStaticRetriever / formatRetrievedDocuments — RAG-less retrieval using token substring overlap; use @agentskit/rag for semantic production retrieval. - executeToolCall , consumeStream , createEventEmitter , parseToolArgs , safeParseArgs , isRecord , createId , generateId , buildMessage — low-level helpers. parseToolArgs returns { args, valid } so adapters can fail closed; safeParseArgs remains the compatibility value-only parser. IDs use crypto.randomUUID() and retain the caller-provided prefix. - isRecord(value) accepts non-null objects other than arrays, including Date , Map , and class instances; use a schema validator when plain JSON objects are required. - AgentsKitError , AdapterError , ToolError , MemoryError , ConfigError , RuntimeError , SandboxError , SkillError , ErrorCodes — error taxonomy. - Token budget: compileBudget , approximateCounter . - Progressive tool args: createProgressiveArgParser , executeToolProgressively . - createVirtualizedMemory — hot-window + cold-retriever wrapper (also re-exported from @agentskit/memory ). - Untrusted-input fencing lives in the @agentskit/core/security entry: fenceUntrustedContent(content, opts?) + UNTRUSTED CONTENT DIRECTIVE wrap attacker-influenced text so the model treats it as data, not instructions (the mitigation companion to createInjectionDetector ). - Multi-modal content parts: textPart , imagePart , audioPart , videoPart , filePart , partsToText , normalizeContent , filterParts . - Agent-loop internals (advanced): buildToolMap , activateSkills , executeSafeTool , createToolLifecycle . - Memory serialization: serializeMessages , deserializeMessages ; validate untrusted records with validateMemoryRecord from @agentskit/core/memory-validation . Subpath exports (zero main-bundle weight) Subpath Purpose Recipe --- --- --- @agentskit/core/agent-schema Declarative agent YAML/JSON + validator @agentskit/core/prompt-experiments A/B prompts with feature flags @agentskit/core/auto-summarize Auto-summar", + "body": "Purpose Stable TypeScript contracts (Adapter, Tool, Memory, Retriever, Skill, Runtime) + createChatController + primitives. Target: <10 KB gzipped, zero runtime deps. Install Primary exports (main entry) - createChatController(config) — state machine behind every UI - proposeToolCall(proposal) — controller method that places a trusted, validated action into the same confirmation lifecycle; it never executes directly. The standalone helper is available from @agentskit/core/tool-proposal . Returns { getState, subscribe, send, stop, retry, edit, regenerate, setInput, clear, decide, approve, deny, updateConfig } . - defineTool(config) — typed tool factory with JSON-Schema input inference. - controller.send(string ContentPart[]) preserves multimodal parts and text projection. - controller.decide(id, 'approve' 'deny', reason?) uses ChatConfig.decisionStore to validate the current conversation before claiming, prioritizing live controller state over memory and preserving unsaved later turns and using the pending snapshot only as a fallback ( ToolDecisionStore , ToolDecisionRecord ). The store provides insert-if-absent putPending , linearizable get , atomic pending-to-claimed claim , and matching-claim terminal settle . Scope it to an application-authorized conversation. Idempotency compares only the decision, retaining the original denial reason. Same-decision terminal replay returns the recorded tool outcome; unknown IDs raise AK ACTION NOT FOUND , conflicting decisions or targets that are no longer pending raise AK ACTION ALREADY DECIDED . Failed and claimed records are never automatically retried. Serialize different decisions per conversation to protect transcript writes. - Identical decisions are idempotent even while claimed: they return the recorded outcome, or the original call while execution is in flight, without invoking the tool or model again. Streamed confirmations are registered before onToolCall . After restart, streamed decisions require current ChatMemory : arrival snapshots are immutable and may omit later calls; without a current conversation, AK CONFIG INVALID rejects an incomplete streaming snapshot before claiming. A live controller does not require memory. - approve / deny are deprecated. With a decision store they delegate to decide , including atomic claims and typed errors; without a store they preserve generation-local behavior. - ChatConfig.authorizeToolCall — trusted host authorization before tool proposal and again before execution. Callback failures default to deny. - createInMemoryMemory / createLocalStorageMemory — default ChatMemory implementations. - createStaticRetriever / formatRetrievedDocuments — RAG-less retrieval using token substring overlap; use @agentskit/rag for semantic production retrieval. - executeToolCall , consumeStream , createEventEmitter , parseToolArgs , safeParseArgs , isRecord , createId , generateId , buildMessage — low-level helpers. parseToolArgs returns { args, valid } so adapters can fail closed; safeParseArgs remains the compatibility value-only parser. IDs use crypto.randomUUID() and retain the caller-provided prefix. - isRecord(value) accepts non-null objects other than arrays, including Date , Map , and class instances; use a schema validator when plain JSON objects are required. - AgentsKitError , AdapterError , ToolError , MemoryError , ConfigError , RuntimeError , SandboxError , SkillError , ErrorCodes — error taxonomy. - Token budget: compileBudget , approximateCounter . - Progressive tool args: createProgressiveArgParser , executeToolProgressively . - createVirtualizedMemory — hot-window + cold-retriever wrapper (also re-exported from @agentskit/memory ). - Untrusted-input fencing lives in the @agentskit/core/security entry: fenceUntrustedContent(content, opts?) + UNTRUSTED CONTENT DIRECTIVE wrap attacker-influenced text so the model treats it as data, not instructions (the mitigation companion to createInjectionDetector ). - Multi-modal content parts: textPar", "aliases": "core document agent" }, "graph": { @@ -51243,7 +51245,7 @@ } ] }, - "contentHash": "573989e1de71fdaeefe6492d54e3ca8d59283fc01028008ed784b8184fce7e65", + "contentHash": "5a6c5ed5dcee2a4368afb53deb724ff0f189fe8008b742912f46fd3d409799a9", "provenance": "observed", "confidence": "observed" }, @@ -59983,7 +59985,7 @@ "path": "apps/docs-next/content/docs/reference/packages/core.mdx", "symbols": "", "summary": "import { ContributeCallout } from '@/components/contribute/contribute-callout'", - "body": "import { ContributeCallout } from '@/components/contribute/contribute-callout' The shared contract layer for AgentsKit: TypeScript types, the headless chat controller, stream helpers, and building blocks used by @agentskit/react , @agentskit/ink , @agentskit/runtime , and adapters. No third-party runtime dependencies — keep this package small and stable. When to use - You implement a custom adapter , tool, memory, or UI on top of official types. - You need createChatController without React (advanced integrations). - You want to understand messages, tool calls, and stream chunks across the ecosystem. You usually do not import core directly in a typical React app except for types — prefer and . Install Most feature packages already depend on core ; you add it explicitly when authoring libraries or sharing types. Public exports (overview) Chat controller and config Export Role -------- ------ createChatController Headless state machine: send, stream, tools, memory, skills, retriever Types: ChatConfig , ChatController , ChatState , ChatReturn Configuration and controller shape The controller merges system prompts, runs retrieval, dispatches tool calls, saves via ChatMemory after successful turns, skips persistence for failed or aborted turns, surfaces background memory and skill activation failures through onError , and exposes subscribe/update patterns consumed by UI packages. Durable decisions and multimodal input controller.send accepts string ContentPart[] . User messages retain their parts and a partsToText projection in content ; provider support remains adapter-specific. ChatReturn.send has the same widened signature. Implementers of ChatController must supply the required decide member and accept parts in send . Configure ChatConfig.decisionStore with a conversation-scoped ToolDecisionStore to use await controller.decide(toolCallId, 'approve') or await controller.decide(toolCallId, 'deny', reason) . Pending message snapshots survive controller replacement and process restart through that store. Configured chat memory also receives pending snapshots before proposal completion. ToolDecisionRecord stores toolCallId , messages , status ( pending , claimed , complete , failed , denied ), and optional decision , reason , and outcome . Implement putPending as insert-if-absent, get as a linearizable read, claim as an atomic pending-to-claimed transition returning the snapshot only to its winner, and settle as a durable terminal write for the matching claim. A database implementation can use UPDATE ... WHERE status = 'pending' RETURNING ... . The application must scope access to the authorized conversation. Before claiming, the controller loads and validates the current conversation, prioritizing live state (including unsaved turns), then configured chat memory, then the stored snapshot when neither exists. It preserves later messages and patches the target call, revalidates and reauthorizes the tool, persists its outcome, and resumes the model. Same-decision terminal replay returns the stored tool outcome without another tool or model invocation. Unknown IDs raise AK ACTION NOT FOUND ; in-flight or conflicting decisions raise AK ACTION ALREADY DECIDED . Tool failures become terminal failed records and reach the model as tool errors. A new attempt needs a new proposal ID. For a stuck claimed record, inspect get(id) and domain execution evidence, stop the original worker, and terminally settle the matching claim with a proven outcome or explicit indeterminate failure. Never reset or delete the claim. Reconcile the target call in current chat memory and continue with a new user turn; a retry requires a separately authorized new proposal ID. See . A crash after claim leaves an indeterminate claimed record requiring application reconciliation; core never automatically retries side effects. Serialize different decisions in one conversation to avoid lost transcript writes. Legacy approve / deny are deprecated: with a decision store they", + "body": "import { ContributeCallout } from '@/components/contribute/contribute-callout' The shared contract layer for AgentsKit: TypeScript types, the headless chat controller, stream helpers, and building blocks used by @agentskit/react , @agentskit/ink , @agentskit/runtime , and adapters. No third-party runtime dependencies — keep this package small and stable. When to use - You implement a custom adapter , tool, memory, or UI on top of official types. - You need createChatController without React (advanced integrations). - You want to understand messages, tool calls, and stream chunks across the ecosystem. You usually do not import core directly in a typical React app except for types — prefer and . Install Most feature packages already depend on core ; you add it explicitly when authoring libraries or sharing types. Public exports (overview) Chat controller and config Export Role -------- ------ createChatController Headless state machine: send, stream, tools, memory, skills, retriever Types: ChatConfig , ChatController , ChatState , ChatReturn Configuration and controller shape The controller merges system prompts, runs retrieval, dispatches tool calls, saves via ChatMemory after successful turns, skips persistence for failed or aborted turns, surfaces background memory and skill activation failures through onError , and exposes subscribe/update patterns consumed by UI packages. Durable decisions and multimodal input controller.send accepts string ContentPart[] . User messages retain their parts and a partsToText projection in content ; provider support remains adapter-specific. ChatReturn.send has the same widened signature. Implementers of ChatController must supply the required decide member and accept parts in send . Configure ChatConfig.decisionStore with a conversation-scoped ToolDecisionStore to use await controller.decide(toolCallId, 'approve') or await controller.decide(toolCallId, 'deny', reason) . Pending message snapshots survive controller replacement and process restart through that store. Configured chat memory also receives pending snapshots before proposal completion. ToolDecisionRecord stores toolCallId , messages , status ( pending , claimed , complete , failed , denied ), and optional decision , reason , and outcome . Implement putPending as insert-if-absent, get as a linearizable read, claim as an atomic pending-to-claimed transition returning the snapshot only to its winner, and settle as a durable terminal write for the matching claim. A database implementation can use UPDATE ... WHERE status = 'pending' RETURNING ... . The application must scope access to the authorized conversation. Before claiming, the controller loads and validates the current conversation, prioritizing live state (including unsaved turns), then configured chat memory, then the stored snapshot when neither exists. It preserves later messages and patches the target call, revalidates and reauthorizes the tool, persists its outcome, and resumes the model only for the latest completed assistant message when all confirmations in that target assistant message are decided. Tool results follow their originating assistant message in call order. Idempotency compares the decision only; a repeated denial with a different reason retains the original reason and outcome. Same-decision terminal replay returns the stored tool outcome without another tool or model invocation. Unknown IDs raise AK ACTION NOT FOUND ; conflicting decisions or targets that are no longer pending raise AK ACTION ALREADY DECIDED . Tool failures become terminal failed records and reach the model as tool errors. A new attempt needs a new proposal ID. For a stuck claimed record, inspect get(id) and domain execution evidence, stop the original worker, and terminally settle the matching claim with a proven outcome or explicit indeterminate failure. Never reset or delete the claim. Reconcile the target call in current chat memory and continue with a new user turn; a retry requires a s", "aliases": "document human" }, "graph": { @@ -60125,7 +60127,7 @@ } ] }, - "contentHash": "a45c2e7c4960a50d190312d9eabc737f8973aa7d2b1720270a630578d0880423", + "contentHash": "c0805db69400cae3a7a0abc8f1dbacfc9bae0078798f4a2157df89c22c5bdf32", "provenance": "observed", "confidence": "observed" }, @@ -72036,7 +72038,7 @@ "path": "docs/architecture/adrs/0042-durable-tool-decisions.md", "symbols": "", "summary": "Status: Proposed (implementation authorized; evidence review remains external)", - "body": "ADR 0042 — Durable tool decisions in core 1.x - Status : Proposed (implementation authorized; evidence review remains external) - Date : 2026-10-06 Context Legacy controller approvals use generation-local identity and cannot resume a confirmation in another controller. Distributed callers need an atomic storage boundary rather than a process-local mutex. Decision Add ChatController.decide(toolCallId, decision, reason?) and ChatConfig.decisionStore . Deprecate approve and deny ; with a decision store they delegate to decide , sharing atomic claims and typed errors, and without a store they retain generation-local behavior. Existing callers remain compatible, but structural implementations of ChatController must implement the required decide member. Add ContentPart[] to controller and shared ChatReturn send , retaining string input and its empty-input behavior. The conversation-scoped store creates immutable pending records containing a serializable message snapshot, atomically transitions pending to claimed with its decision, and stores a terminal tool outcome. Claim returns the authoritative snapshot only to its winner. A database implementation can use UPDATE ... WHERE status = 'pending' RETURNING ... ; creation must be insert-if-absent. Reads and writes must be durable and linearizable per tool call. Store access is an application-authorized conversation boundary, not an authorization token supplied by an untrusted client. Before claiming, the controller loads and validates the current conversation: live state takes precedence over configured chat memory, and the pending snapshot is used only when no current conversation is available. Loading and sibling reconciliation operate on cloned messages so existing subscriber snapshots remain immutable. After claim, it stops the stream and merges only reconciled terminal sibling outcomes by message and call ID into the latest live state, preserving chunks, new confirmations and turns received during storage awaits. It rechecks that the target remains pending before execution; a removed or changed target terminally settles the claim as failed without executing or resuming the model. It patches the target call while preserving later turns, revalidates and reauthorizes execution through the existing tool pipeline, saves the terminal outcome before model continuation, and never automatically reclaims a claimed or failed record. Same-decision terminal replay returns the recorded tool outcome without invoking tool or model. Unknown IDs and competing decisions raise typed errors; an in-flight duplicate raises a conflict. Pending snapshots are persisted before proposal completion when the store is configured. Consequences Core adds no database implementation or runtime dependency. HTTP schema, identity mapping, stream transport, and provider-specific persistence are integration responsibilities. A crash after claim can leave an indeterminate claimed record; exactly-once admission does not promise transactional domain writes or automatic crash recovery. Applications must reconcile such records without automatically rerunning side effects. Concurrent different calls in one conversation require application-level conversation serialization to avoid lost transcript updates. Store-backed legacy approvals and denials use the same claim boundary as durable decisions. Implementation is lazy-loaded to preserve the root import budget. Tests use independently specified synthetic stores and snapshots, without private consumer details. Operational recovery of a claimed decision The operator reads store.get(toolCallId) in the authorized conversation scope and verifies status , decision , arguments in messages , and any domain transaction or tool audit evidence. Stop the original worker and serialize conversation writes before reconciliation; elapsed time alone does not prove that execution never occurred. Never release a claim by deleting it or resetting it to pending . If the domain evidence proves compl", + "body": "ADR 0042 — Durable tool decisions in core 1.x - Status : Proposed (implementation authorized; evidence review remains external) - Date : 2026-10-06 Context Legacy controller approvals use generation-local identity and cannot resume a confirmation in another controller. Distributed callers need an atomic storage boundary rather than a process-local mutex. Decision Add ChatController.decide(toolCallId, decision, reason?) and ChatConfig.decisionStore . Deprecate approve and deny ; with a decision store they delegate to decide , sharing atomic claims and typed errors, and without a store they retain generation-local behavior. Existing callers remain compatible, but structural implementations of ChatController must implement the required decide member. Add ContentPart[] to controller and shared ChatReturn send , retaining string input and its empty-input behavior. The conversation-scoped store creates immutable pending records containing a serializable message snapshot, atomically transitions pending to claimed with its decision, and stores a terminal tool outcome. Claim returns the authoritative snapshot only to its winner. A database implementation can use UPDATE ... WHERE status = 'pending' RETURNING ... ; creation must be insert-if-absent. Reads and writes must be durable and linearizable per tool call. Store access is an application-authorized conversation boundary, not an authorization token supplied by an untrusted client. Before claiming, the controller loads and validates the current conversation: live state takes precedence over configured chat memory, and the pending snapshot is used only when no current conversation is available. Loading and sibling reconciliation operate on cloned messages so existing subscriber snapshots remain immutable. After claim, it rechecks the live target before merging only reconciled terminal sibling outcomes by message and call ID into the latest live state, preserving chunks, new confirmations and turns received during storage awaits. It rechecks that the target remains pending before execution; a removed or changed target terminally settles the claim as failed without executing or resuming the model. A rejected decision leaves any active stream untouched. It patches the target call while preserving later turns, revalidates and reauthorizes execution through the existing tool pipeline, saves the terminal outcome before model continuation (only from the latest completed assistant when all calls are decided, with tool results inserted immediately after their originating message in call order), and never automatically reclaims a claimed or failed record. Same-decision terminal replay returns the recorded tool outcome without invoking tool or model. Unknown IDs and competing decisions raise typed errors; an identical in-flight duplicate returns the original call without execution. Pending snapshots are persisted on streamed call arrival before callbacks, and before proposal completion when the store is configured. Their immutable arrival snapshots may omit later streamed calls, so reconstruction after restart requires current ChatMemory; an incomplete streaming fallback snapshot fails before claim with AK CONFIG INVALID. Consequences Core adds no database implementation or runtime dependency. HTTP schema, identity mapping, stream transport, and provider-specific persistence are integration responsibilities. A crash after claim can leave an indeterminate claimed record; exactly-once admission does not promise transactional domain writes or automatic crash recovery. Applications must reconcile such records without automatically rerunning side effects. Concurrent different calls in one conversation require application-level conversation serialization to avoid lost transcript updates. Store-backed legacy approvals and denials use the same claim boundary as durable decisions. Implementation is lazy-loaded to preserve the root import budget. Tests use independently specified synthetic stores and snap", "aliases": "document human" }, "graph": { @@ -72069,7 +72071,7 @@ } ] }, - "contentHash": "6e1205db68cf23787d54c5b5e30edc45263ebebb31bf12381caf66080304a13a", + "contentHash": "542f2d33984ae4c31a163e5749abdfc65dd8c9153b89d63db1184f29d5a711c3", "provenance": "observed", "confidence": "observed" }, @@ -78079,7 +78081,7 @@ "path": "packages/core/README.md", "symbols": "", "summary": "Profile: major-package", - "body": "@agentskit/core Profile: Promise - prepare: (messages: Message[], reconciled: Map) => Promise<{ generation: number; correlation: AgentEventContext }> + load: (snapshot: Message[]) => Promise<{ messages: Message[]; fallback: boolean }> + prepare: (messages: Message[], reconcile: (current: Message[]) => Message[] | undefined) => Promise<{ generation: number; correlation: AgentEventContext } | undefined> tool: (name: string) => ToolDefinition | undefined runTool: (tool: ToolDefinition | undefined, call: ToolCall, onPartial: (result: string) => void, generation: number, correlation: AgentEventContext) => Promise patch: (assistantId: string, toolCallId: string, patch: Partial) => void finish: (toolCallId: string) => void isCurrent: (generation: number) => boolean resume: (assistantId: string, generation: number, correlation: AgentEventContext) => Promise + setMessages: (messages: Message[]) => void messages: () => Message[] persist: (correlation: AgentEventContext) => Promise } @@ -30,10 +34,17 @@ export async function decide( const pending = await store.get(id) if (!pending) throw new ToolError({ code: ErrorCodes.AK_ACTION_NOT_FOUND, message: 'Tool decision not found' }) if (pending.status !== 'pending') { - if (pending.decision === decision && pending.outcome && ['complete', 'failed', 'denied'].includes(pending.status)) return pending.outcome + if (pending.decision === decision) { + const replay = pending.outcome ?? pending.messages.flatMap(message => message.toolCalls ?? []).find(call => call.id === id) + if (replay) return replay + } throw new ToolError({ code: ErrorCodes.AK_ACTION_ALREADY_DECIDED, message: 'Tool decision already claimed or decided' }) } - const messages = await context.load(pending.messages) + const loaded = await context.load(pending.messages) + if (loaded.fallback && pending.messages.some(message => message.status === 'streaming')) { + throw new ConfigError({ code: ErrorCodes.AK_CONFIG_INVALID, message: 'Resuming a streamed tool decision requires current ChatMemory' }) + } + const messages = loaded.messages.map(message => ({ ...message, toolCalls: message.toolCalls?.map(call => ({ ...call })) })) const reconciled = new Map() for (const message of messages) { if (!message.toolCalls) continue @@ -56,19 +67,29 @@ export async function decide( if (!record) { const existing = await store.get(id) if (!existing) throw new ToolError({ code: ErrorCodes.AK_ACTION_NOT_FOUND, message: 'Tool decision not found' }) - if (existing.decision === decision && existing.outcome && ['complete', 'failed', 'denied'].includes(existing.status)) return existing.outcome + if (existing.decision === decision) { + const replay = existing.outcome ?? existing.messages.flatMap(message => message.toolCalls ?? []).find(call => call.id === id) + if (replay) return replay + } throw new ToolError({ code: ErrorCodes.AK_ACTION_ALREADY_DECIDED, message: 'Tool decision already claimed or decided' }) } if (record.toolCallId !== id || record.status !== 'claimed' || record.decision !== decision || !Array.isArray(record.messages)) { throw new ConfigError({ code: ErrorCodes.AK_CONFIG_INVALID, message: 'Decision store returned an invalid claim' }) } - const { generation, correlation } = await context.prepare(messages, reconciled) + const prepared = await context.prepare(messages, latest => { + if (!latest.some(message => message.toolCalls?.some(call => call.id === id && call.status === 'requires_confirmation'))) return undefined + return latest.map(message => ({ ...message, toolCalls: message.toolCalls?.map(call => { + const outcome = reconciled.get(message.id)?.find(loaded => loaded.id === call.id) + return call.id !== id && call.status === 'requires_confirmation' && outcome ? outcome : call + }) })) + }) const current = context.messages().find(m => m.id === message.id)?.toolCalls?.find(c => c.id === id) - if (current?.status !== 'requires_confirmation') { + if (!prepared || current?.status !== 'requires_confirmation') { const error = new ToolError({ code: ErrorCodes.AK_ACTION_ALREADY_DECIDED, message: 'Tool decision is no longer pending' }) await store.settle({ ...record, status: 'failed', outcome: { ...call, status: 'error', error: error.message }, messages: context.messages() }) throw error } + const { generation, correlation } = prepared call = current let outcome: ToolCall if (decision === 'deny') { @@ -85,6 +106,11 @@ export async function decide( } } context.patch(message.id, id, outcome) + const transcript = context.messages() + const assistant = transcript.find(current => current.id === message.id) + if (assistant && assistant.status !== 'streaming' && transcript[transcript.length - 1]?.id !== assistant.id) { + context.setMessages(insertToolResults(transcript, assistant.id, assistant.toolCalls ?? [], buildMessage)) + } context.finish(id) const status = outcome.status === 'complete' ? 'complete' : 'failed' await store.settle({ @@ -94,6 +120,11 @@ export async function decide( messages: context.messages(), }) await context.persist(correlation) - if (context.isCurrent(generation)) await context.resume(message.id, generation, correlation) + const latest = context.messages() + const last = latest[latest.length - 1] + if (context.isCurrent(generation) && last?.id === message.id && last?.status !== 'streaming' + && !last.toolCalls?.some(call => call.status !== 'complete' && call.status !== 'error')) { + await context.resume(message.id, generation, correlation) + } return outcome } diff --git a/packages/core/src/controller-helpers.ts b/packages/core/src/controller-helpers.ts index 7e557efb0..1f87e54b7 100644 --- a/packages/core/src/controller-helpers.ts +++ b/packages/core/src/controller-helpers.ts @@ -75,6 +75,22 @@ export function sameToolLifecycle( return true } +export function insertToolResults( + messages: Message[], + assistantId: string, + calls: ToolCall[], + buildMsg: (init: { role: Message['role']; content: string; toolCallId?: string; status?: Message['status'] }) => Message, +): Message[] { + return messages.flatMap(message => { + if (message.id !== assistantId && message.status === 'streaming') return [message] + const settled = message.id === assistantId ? calls : message.toolCalls ?? [] + if (!settled.length || settled.some(call => call.status !== 'complete' && call.status !== 'error')) return [message] + const results = settled.filter(call => !messages.some(result => result.role === 'tool' && result.toolCallId === call.id)) + .map(call => buildMsg({ role: 'tool', content: call.result ?? call.error ?? '', toolCallId: call.id })) + return [{ ...message, status: 'complete' as const }, ...results] + }) +} + /** Build tool-result messages + a fresh streaming assistant for multi-turn tool loops. */ export function buildToolContinuation( messages: Message[], @@ -82,20 +98,10 @@ export function buildToolContinuation( calls: ToolCall[], buildMsg: (init: { role: Message['role']; content: string; toolCallId?: string; status?: Message['status'] }) => Message, ): { messages: Message[]; nextAssistantId: string } { - const results = calls.map(call => - buildMsg({ - role: 'tool', - content: call.result ?? call.error ?? '', - toolCallId: call.id, - }), - ) const nextA = buildMsg({ role: 'assistant', content: '', status: 'streaming' }) return { messages: [ - ...messages.map(message => - message.id === assistantId ? { ...message, status: 'complete' as const } : message - ), - ...results, + ...insertToolResults(messages, assistantId, calls, buildMsg), nextA, ], nextAssistantId: nextA.id, @@ -195,9 +201,11 @@ export function createControllerToolLoop({ getConfig, getState, getCorrelation, if (done && shouldPersist) await persist(getState().messages, getCorrelation()) } - const continueTools = (aid: string, calls: ToolCall[]): string => { + const continueTools = async (aid: string, calls: ToolCall[]): Promise => { + const { buildToolContinuation } = await import('./controller-decision-internal.js') let nextId = '' set(current => { + if (current.messages[current.messages.length - 1]?.id !== aid) return current const { messages: next, nextAssistantId } = buildToolContinuation( current.messages, aid, @@ -231,13 +239,14 @@ export function createControllerToolLoop({ getConfig, getState, getCorrelation, // Nothing to feed back, or something still awaiting confirmation — // stop here; the caller drives the next step. - if (!calls.length || waits) { - if (waits) await persistPending() + if (!calls.length || waits || getState().messages[getState().messages.length - 1]?.id !== id) { await finalize(id, !waits) + if (waits) await persistPending() return } - id = continueTools(id, calls) + id = await continueTools(id, calls) + if (!id) return const ok = await run(id, '', g, correlation) if (!ok) return } diff --git a/packages/core/src/controller-tool-call.ts b/packages/core/src/controller-tool-call.ts index 256f02ea9..4096fb49d 100644 --- a/packages/core/src/controller-tool-call.ts +++ b/packages/core/src/controller-tool-call.ts @@ -28,7 +28,7 @@ interface ToolCallContext { patchCall: PatchCall runTool: RunTool correlation: AgentEventContext - registerToolCall: (id: string) => void + registerToolCall: (id: string) => void | Promise } export async function handleControllerToolCall({ @@ -76,9 +76,8 @@ export async function handleControllerToolCall({ await auth(authorize, toolCall, { messages, tool, phase: 'propose' }) if (!isCurrentGeneration()) return - if (tool?.requiresConfirmation) registerToolCall(toolCall.id) - setMessage(assistantId, current => ({ ...current, toolCalls: [...(current.toolCalls ?? []), toolCall] })) + if (tool?.requiresConfirmation) await registerToolCall(toolCall.id) await onToolCall?.(toolCall, { messages, tool }) if (!isCurrentGeneration()) return diff --git a/packages/core/src/controller.ts b/packages/core/src/controller.ts index 19737507c..9db2d6bf3 100644 --- a/packages/core/src/controller.ts +++ b/packages/core/src/controller.ts @@ -49,6 +49,7 @@ import type { let lifecycle = createToolLifecycle(toolMap) let skillTools: ToolDefinition[] = [] const approvalGenerations = new Map() + let pendingRegistration = Promise.resolve() let hydrated = false let active = false const authorize: NonNullable = async (call, context) => { @@ -206,7 +207,11 @@ import type { patchCall, runTool, correlation, - registerToolCall: id => approvalGenerations.set(id, g), + registerToolCall: id => { + approvalGenerations.set(id, g) + pendingRegistration = persistPending() + return pendingRegistration + }, }) }, onToolResult(content) { @@ -432,32 +437,35 @@ import type { }) }, async decide(tid, decision, reason) { + const { decide } = await import('./controller-decision-internal.js') + await pendingRegistration.catch(() => {}) let loadedLive = false let loadedGeneration = gen - return import('./controller-decision-internal.js').then(m => m.decide(tid, decision, reason, { + return decide(tid, decision, reason, { store: config.decisionStore, messages: () => state.messages, load: async snapshot => { await activate() loadedGeneration = gen loadedLive = state.messages.length > 0 const messages = loadedLive ? state.messages : await persistence.load() - return (messages.length ? messages : snapshot).map(message => ({ ...message, toolCalls: message.toolCalls?.map(call => ({ ...call })) })) + return { fallback: !messages.length, messages: messages.length ? messages : snapshot } }, tool: name => toolMap.get(name), runTool, patch: patchCall, finish: id => { approvalGenerations.delete(id) }, isCurrent: generation => generation === gen, + setMessages: controller.setMessages, persist: correlation => persist(state.messages, correlation, true), resume, - prepare: async (messages, reconciled) => { + prepare: async (messages, reconcile) => { const useLive = loadedLive || state.messages.length > 0 || gen !== loadedGeneration - controller.stop() - set(current => ({ ...current, messages: (useLive ? current.messages : messages).map(message => { - return { ...message, status: message.status === 'streaming' ? 'complete' as const : message.status, toolCalls: message.toolCalls?.map(call => { - const outcome = reconciled.get(message.id)?.find(loaded => loaded.id === call.id) - return call.id !== tid && call.status === 'requires_confirmation' && outcome ? outcome : call - }) } - }), error: null })) - return { generation: gen, correlation: beginRun() } + let current = useLive ? state.messages : messages + if (state.status !== 'streaming') current = current.map(message => + message.status === 'streaming' ? { ...message, status: 'complete' as const } : message, + ) + const next = reconcile(current) + if (!next) return undefined + set(current => ({ ...current, messages: next, error: null })) + return { generation: gen, correlation: state.status === 'streaming' ? activeCorrelation : beginRun() } }, - })) + }) }, async approve(tid) { if (config.decisionStore) { diff --git a/packages/core/tests/controller-decision-internal.test.ts b/packages/core/tests/controller-decision-internal.test.ts index 09024f552..04690edf5 100644 --- a/packages/core/tests/controller-decision-internal.test.ts +++ b/packages/core/tests/controller-decision-internal.test.ts @@ -1,6 +1,6 @@ import { describe, it, expect, vi } from 'vitest' import { createChatController, createInMemoryMemory, imagePart, filePart, textPart } from '../src/index' -import type { AdapterFactory, AdapterRequest, ToolDecisionRecord, ToolDecisionStore, ChatConfig } from '../src/index' +import type { AdapterFactory, AdapterRequest, ToolDecisionRecord, ToolDecisionStore, ChatConfig, StreamChunk } from '../src/index' import { createMockAdapter } from './helpers' const copy = (value: T): T => JSON.parse(JSON.stringify(value)) as T @@ -44,6 +44,110 @@ function fixture(overrides: Partial = {}) { const proposal = { id: 'call-1', name: 'write', args: { value: 7 } } describe('durable controller decisions', () => { + it('continues an auto-run turn despite an older undecided confirmation', async () => { + const requests: AdapterRequest[] = [] + const weather = vi.fn(() => 'sunny') + const { config } = fixture({ tools: [ + { name: 'write', requiresConfirmation: true, execute: () => 'stored' }, + { name: 'weather', execute: weather }, + ] }) + const chat = createChatController({ ...config, adapter: { + createSource(request) { + requests.push(request) + let chunks: StreamChunk[] = [{ type: 'text', content: 'Sunny today' }] + if (requests.length === 1) chunks = [{ type: 'tool_call', toolCall: { ...proposal, args: '{}' } }] + if (requests.length === 2) chunks = [{ type: 'tool_call', toolCall: { id: 'weather', name: 'weather', args: '{}' } }] + return createMockAdapter([...chunks, { type: 'done' }]).createSource(request) + }, + } }) + await chat.send('write') + await chat.send('weather?') + expect(requests).toHaveLength(3) + expect(weather).toHaveBeenCalledTimes(1) + expect(chat.getState().messages.flatMap(message => message.toolCalls ?? []).find(call => call.id === proposal.id)?.status).toBe('requires_confirmation') + expect(requests[2].messages.find(message => message.toolCallId === 'weather')?.content).toBe('sunny') + expect(chat.getState().messages.at(-1)?.content).toBe('Sunny today') + }) + + it('resumes the latest decision despite a running call from a superseded generation', async () => { + let release = () => {} + const gate = new Promise(resolve => { release = resolve }) + const { config, requests } = fixture({ tools: [ + { name: 'write', requiresConfirmation: true, execute: () => 'stored' }, + { name: 'slow', execute: async () => { await gate; return 'old result' } }, + ] }) + const chat = createChatController({ ...config, adapter: createMockAdapter([ + { type: 'tool_call', toolCall: { id: 'old-running', name: 'slow', args: '{}' } }, { type: 'done' }, + ]) }) + const sending = chat.send('slow') + await vi.waitUntil(() => chat.getState().messages.at(-1)?.toolCalls?.[0].status === 'running') + chat.updateConfig({ adapter: config.adapter }) + try { + await chat.send('later') + await chat.proposeToolCall(proposal) + await chat.decide(proposal.id, 'approve') + expect(requests).toHaveLength(2) + expect(requests[1].messages.find(message => message.toolCallId === proposal.id)?.content).toBe('stored') + expect(requests[1].correlation?.runId).not.toBe(requests[0].correlation?.runId) + expect(chat.getState().messages.at(-1)?.content).toBe('Finished') + } finally { release(); await sending } + }) + + it.each(['decide', 'approve', 'deny'] as const)('allows %s after a single pending-registration save failure', async action => { + const memory = createInMemoryMemory() + const onError = vi.fn() + const { chat, config, execute } = fixture({ memory, onError }) + await chat.proposeToolCall(proposal) + const save = memory.save.bind(memory) + memory.save = vi.fn().mockRejectedValueOnce(new Error('offline once')).mockImplementation(save) + chat.updateConfig({ adapter: createMockAdapter([ + { type: 'tool_call', toolCall: { ...proposal, id: 'failed-arrival', args: '{}' } }, { type: 'done' }, + ]) }) + await chat.send('another write') + expect(chat.getState().error).toMatchObject({ code: 'AK_MEMORY_SAVE_FAILED' }) + expect(onError).toHaveBeenCalledWith(expect.objectContaining({ code: 'AK_MEMORY_SAVE_FAILED' })) + expect(await config.decisionStore!.get('failed-arrival')).toBeUndefined() + if (action === 'decide') await chat.decide(proposal.id, 'approve') + else await chat[action](proposal.id) + expect(execute).toHaveBeenCalledTimes(action === 'deny' ? 0 : 1) + expect(await config.decisionStore!.get(proposal.id)).toMatchObject({ status: action === 'deny' ? 'denied' : 'complete' }) + expect(await memory.load()).toEqual(chat.getState().messages) + }) + + it('completes a persisted streaming assistant after restart and resumes in call order', async () => { + const memory = createInMemoryMemory() + const { config, requests, execute } = fixture({ memory }) + const messages = [{ id: 'user', role: 'user' as const, content: 'write twice', createdAt: 1 }, { + id: 'interrupted', role: 'assistant' as const, content: 'partial', createdAt: 2, status: 'streaming' as const, + toolCalls: [ + { ...proposal, status: 'requires_confirmation' as const }, + { ...proposal, id: 'call-2', status: 'requires_confirmation' as const }, + ], + }] + await memory.save(messages) + for (const call of messages[1].toolCalls!) await config.decisionStore!.putPending({ toolCallId: call.id, status: 'pending', messages }) + const restarted = createChatController(config) + await restarted.decide('call-2', 'approve') + expect(requests).toHaveLength(0) + await restarted.decide(proposal.id, 'approve') + expect(execute).toHaveBeenCalledTimes(2) + expect(requests).toHaveLength(1) + expect(requests[0].messages.map(message => message.role)).toEqual(['user', 'assistant', 'tool', 'tool', 'assistant']) + expect(requests[0].messages.filter(message => message.role === 'tool').map(message => [message.toolCallId, message.content])).toEqual([[proposal.id, 'stored'], ['call-2', 'stored']]) + expect(restarted.getState().messages.at(-1)?.content).toBe('Finished') + expect(await memory.load()).toEqual(restarted.getState().messages) + }) + + it('replays a denial with a different reason without replacing the recorded reason', async () => { + const { chat, config, execute, requests } = fixture() + await chat.proposeToolCall(proposal) + const first = await chat.decide(proposal.id, 'deny', 'original') + expect(await chat.decide(proposal.id, 'deny', 'different')).toEqual(first) + expect(await config.decisionStore!.get(proposal.id)).toMatchObject({ reason: 'original' }) + expect(execute).not.toHaveBeenCalled() + expect(requests).toHaveLength(1) + }) + it.each([[false, true], [true, true], [false, false]])('preserves later turns when restarted=%s, memory=%s', async (restarted, persisted) => { const memory = persisted ? createInMemoryMemory() : undefined const { chat, config } = fixture({ memory }) @@ -53,12 +157,150 @@ describe('durable controller decisions', () => { const current = restarted ? createChatController(config) : chat await current.decide(proposal.id, 'approve') const messages = current.getState().messages - expect(messages.slice(0, before.length).map(m => [m.id, m.content])).toEqual(before.map(m => [m.id, m.content])) + expect(messages.filter(message => message.role !== 'tool').slice(0, before.length).map(m => [m.id, m.content])).toEqual(before.map(m => [m.id, m.content])) expect(messages[0].toolCalls?.[0].status).toBe('complete') expect(messages.at(-1)?.content).toBe('Finished') if (memory) expect(await memory.load()).toEqual(messages) }) + it('does not resume an older call while a later confirmation waits', async () => { + const { chat, requests } = fixture() + await chat.proposeToolCall(proposal) + await chat.send('later') + await chat.proposeToolCall({ ...proposal, id: 'later-call' }) + const before = requests.length + await chat.decide(proposal.id, 'approve') + expect(requests).toHaveLength(before) + expect(chat.getState().messages[1]).toMatchObject({ role: 'tool', toolCallId: proposal.id, content: 'stored' }) + expect(chat.getState().messages.at(-1)?.toolCalls?.[0].status).toBe('requires_confirmation') + await chat.decide('later-call', 'deny') + expect(requests).toHaveLength(before + 1) + const messages = requests.at(-1)!.messages + for (let index = 0; index < messages.length; index++) { + const calls = messages[index].toolCalls ?? [] + expect(messages.slice(index + 1, index + 1 + calls.length).map(message => message.toolCallId)).toEqual(calls.map(call => call.id)) + } + }) + + it.each(['approve', 'deny'] as const)('registers streamed calls before %s and accepts identical double clicks', async action => { + const { config, execute, requests } = fixture() + let release = () => {} + const gate = new Promise(resolve => { release = resolve }) + let arrived = false + const chat = createChatController({ ...config, adapter: { + createSource(request) { + if (request.messages.some(message => message.role === 'tool')) return config.adapter.createSource(request) + return { + async *stream() { + yield { type: 'tool_call' as const, toolCall: { ...proposal, args: '{}' } } + arrived = true + await gate + yield { type: 'tool_call' as const, toolCall: { ...proposal, id: 'call-2', args: '{}' } } + yield { type: 'done' as const } + }, + abort: vi.fn(release), + } + }, + }, onToolCall: async call => { await Promise.all([chat[action](call.id), chat[action](call.id)]) } }) + const sending = chat.send('write twice') + try { + await vi.waitUntil(() => arrived) + expect(requests).toHaveLength(0) + expect((await config.decisionStore!.get(proposal.id))?.decision).toBe(action) + } finally { release() } + await sending + expect(requests).toHaveLength(1) + expect(requests[0].messages.filter(message => message.role === 'tool').map(message => message.toolCallId)).toEqual([proposal.id, 'call-2']) + expect(execute).toHaveBeenCalledTimes(action === 'approve' ? 2 : 0) + }) + + it('does not abort a live stream when the claimed target is no longer pending', async () => { + const { chat, config, execute } = fixture() + await chat.proposeToolCall(proposal) + let releaseClaim = () => {} + let claimed = false + const claimGate = new Promise(resolve => { releaseClaim = resolve }) + const claim = config.decisionStore!.claim.bind(config.decisionStore) + config.decisionStore!.claim = async (...args) => { const record = await claim(...args); claimed = true; await claimGate; return record } + const deciding = chat.decide(proposal.id, 'approve') + const rejected = expect(deciding).rejects.toMatchObject({ code: 'AK_ACTION_ALREADY_DECIDED' }) + await vi.waitUntil(() => claimed) + let releaseStream = () => {} + const gate = new Promise(resolve => { releaseStream = resolve }) + const abort = vi.fn(releaseStream) + chat.updateConfig({ adapter: { createSource: () => ({ + async *stream() { yield { type: 'text' as const, content: 'early' }; await gate; yield { type: 'text' as const, content: ' late' }; yield { type: 'done' as const } }, + abort, + }) } }) + const sending = chat.send('later') + await vi.waitUntil(() => chat.getState().messages.at(-1)?.content === 'early') + chat.setMessages(chat.getState().messages.slice(1)) + releaseClaim() + await rejected + expect(abort).not.toHaveBeenCalled() + expect(chat.getState().status).toBe('streaming') + releaseStream() + await sending + expect(chat.getState().messages.at(-1)?.content).toBe('early late') + expect(execute).not.toHaveBeenCalled() + }) + + it('orders mixed decisions made in reverse order and avoids duplicate results', async () => { + const { config, requests, execute } = fixture() + const chat = createChatController({ ...config, adapter: { + createSource(request) { + if (request.messages.some(message => message.role === 'tool')) return config.adapter.createSource(request) + return createMockAdapter([ + { type: 'tool_call', toolCall: { ...proposal, args: '{}' } }, + { type: 'tool_call', toolCall: { ...proposal, id: 'call-2', args: '{}' } }, { type: 'done' }, + ]).createSource(request) + }, + } }) + await chat.send('write twice') + await chat.decide('call-2', 'approve') + expect(requests).toHaveLength(0) + await chat.decide(proposal.id, 'deny', 'cancelled') + expect(requests).toHaveLength(1) + const messages = requests[0].messages + const index = messages.findIndex(message => message.toolCalls?.length === 2) + expect(messages.slice(index + 1, index + 3).map(message => [message.toolCallId, message.content])).toEqual([ + [proposal.id, 'Permission denied: cancelled'], ['call-2', 'stored'], + ]) + await chat.send('next') + const tools = requests.at(-1)!.messages.filter(message => message.role === 'tool') + expect(tools.map(message => message.toolCallId)).toEqual([proposal.id, 'call-2']) + expect(execute).toHaveBeenCalledTimes(1) + }) + + it('allows subscriber approval while arrival persistence is still saving', async () => { + const memory = createInMemoryMemory() + const save = memory.save.bind(memory) + let release = () => {} + const gate = new Promise(resolve => { release = resolve }) + memory.save = async messages => { await gate; await save(messages) } + const { config, requests, execute } = fixture({ memory }) + const chat = createChatController({ ...config, adapter: { + createSource(request) { + return request.messages.some(message => message.role === 'tool') + ? config.adapter.createSource(request) + : createMockAdapter([{ type: 'tool_call', toolCall: { ...proposal, args: '{}' } }, { type: 'done' }]).createSource(request) + }, + } }) + let approving: Promise | undefined + const unsubscribe = chat.subscribe(() => { + if (!approving && chat.getState().messages.some(message => message.toolCalls?.some(call => call.id === proposal.id))) { + approving = chat.approve(proposal.id) + } + }) + const sending = chat.send('write') + await vi.waitUntil(() => approving !== undefined) + release() + await Promise.all([sending, approving]) + unsubscribe() + expect(execute).toHaveBeenCalledTimes(1) + expect(requests).toHaveLength(1) + }) + it('detects a non-atomic read/await/write claim', async () => { const port = store(new Map(), false) await port.putPending({ toolCallId: 'id', messages: [], status: 'pending' }) @@ -108,11 +350,13 @@ describe('durable controller decisions', () => { await vi.waitUntil(() => started) const before = copy(chat.getState().messages) await chat.decide(proposal.id, 'approve') + expect(chat.getState().status).toBe('streaming') + release() await sending const messages = chat.getState().messages - expect(messages.slice(0, before.length).map(message => [message.id, message.content])).toEqual(before.map(message => [message.id, message.content])) + expect(messages.filter(message => message.role !== 'tool').slice(0, before.length).map(message => [message.id, message.content])).toEqual(before.map(message => [message.id, message.content])) expect(messages[0].toolCalls?.[0]).toMatchObject({ status: 'complete', result: 'stored' }) - expect(messages.at(-1)?.content).toBe('Finished') + expect(messages.at(-1)?.content).toBe('partial response') expect(await memory.load()).toEqual(messages) expect(execute).toHaveBeenCalledTimes(1) }) @@ -171,11 +415,11 @@ describe('durable controller decisions', () => { releaseStore() await deciding const messages = chat.getState().messages - expect(messages.slice(0, before.length).map(message => [message.id, message.content])).toEqual(before.map(message => [message.id, message.content])) + expect(messages.filter(message => message.role !== 'tool').slice(0, before.length).map(message => [message.id, message.content])).toEqual(before.map(message => [message.id, message.content])) expect(messages.flatMap(message => message.toolCalls ?? []).find(call => call.id === 'new-call')).toMatchObject({ status: 'requires_confirmation' }) expect(messages[0].toolCalls?.[0]).toMatchObject({ status: 'complete', result: 'stored' }) expect(await memory.load()).toEqual(messages) - expect((await port.get(proposal.id))?.messages).toEqual(copy(messages.slice(0, before.length))) + expect((await port.get(proposal.id))?.messages).toEqual(copy(messages)) expect(execute).toHaveBeenCalledTimes(1) }) @@ -219,7 +463,8 @@ describe('durable controller decisions', () => { } const deciding = chat.decide(proposal.id, 'approve') await vi.waitUntil(() => held) - await expect(chat[action](proposal.id)).rejects.toMatchObject({ code: 'AK_ACTION_ALREADY_DECIDED' }) + if (action === 'approve') await expect(chat[action](proposal.id)).resolves.toBeUndefined() + else await expect(chat[action](proposal.id)).rejects.toMatchObject({ code: 'AK_ACTION_ALREADY_DECIDED' }) expect(execute).not.toHaveBeenCalled() release() await deciding @@ -238,7 +483,7 @@ describe('durable controller decisions', () => { const before = copy(chat.getState().messages) memory.save = save await chat.decide(proposal.id, 'approve') - expect(chat.getState().messages.slice(0, before.length).map(message => message.content)).toEqual(before.map(message => message.content)) + expect(chat.getState().messages.filter(message => message.role !== 'tool').slice(0, before.length).map(message => message.content)).toEqual(before.map(message => message.content)) expect(await memory.load()).toEqual(chat.getState().messages) }) @@ -287,14 +532,13 @@ describe('durable controller decisions', () => { expect(chat.getState().messages.flatMap(message => message.toolCalls ?? []).find(call => call.id === 'sibling')).toMatchObject({ status: 'complete', result: 'external' }) }) - it('admits one of 100 parallel decisions across independent controllers, repeated 100 times', async () => { + it('executes once for 100 identical parallel decisions across independent controllers, repeated 100 times', async () => { for (let round = 0; round < 100; round++) { const { config, chat, execute } = fixture() const id = `parallel-${round}` await chat.proposeToolCall({ ...proposal, id }) const results = await Promise.allSettled(Array.from({ length: 100 }, () => createChatController(config).decide(id, 'approve'))) - expect(results.filter(r => r.status === 'fulfilled')).toHaveLength(1) - for (const result of results) if (result.status === 'rejected') expect(result.reason.code).toBe('AK_ACTION_ALREADY_DECIDED') + expect(results.filter(r => r.status === 'fulfilled')).toHaveLength(100) expect(execute).toHaveBeenCalledTimes(1) } }) @@ -328,7 +572,7 @@ describe('durable controller decisions', () => { const { config, chat, execute } = fixture() await chat.proposeToolCall(proposal) await config.decisionStore!.claim(proposal.id, 'approve') - await expect(createChatController(config).decide(proposal.id, 'approve')).rejects.toMatchObject({ code: 'AK_ACTION_ALREADY_DECIDED' }) + await expect(createChatController(config).decide(proposal.id, 'approve')).resolves.toMatchObject({ id: proposal.id }) expect(execute).not.toHaveBeenCalled() }) @@ -351,7 +595,7 @@ describe('durable controller decisions', () => { }) it('persists model-produced pending calls and resumes them in a new controller', async () => { - const { config, execute } = fixture() + const { config, execute } = fixture({ memory: createInMemoryMemory() }) const chat = createChatController({ ...config, adapter: createMockAdapter([ { type: 'tool_call', toolCall: { ...proposal, args: JSON.stringify(proposal.args) } }, { type: 'done' }, ]) }) @@ -362,7 +606,7 @@ describe('durable controller decisions', () => { }) it('resumes a multi-call assistant only after all persisted confirmations settle', async () => { - const { config, requests, execute } = fixture() + const { config, requests, execute } = fixture({ memory: createInMemoryMemory() }) const chat = createChatController({ ...config, adapter: createMockAdapter([ { type: 'tool_call', toolCall: { ...proposal, args: '{}' } }, { type: 'tool_call', toolCall: { ...proposal, id: 'call-2', args: '{}' } }, { type: 'done' }, @@ -376,6 +620,19 @@ describe('durable controller decisions', () => { expect(requests[0].messages.filter(m => m.role === 'tool')).toHaveLength(2) }) + it('rejects a streamed snapshot after restart without memory before claiming', async () => { + const { config, execute, requests } = fixture() + const chat = createChatController({ ...config, adapter: createMockAdapter([ + { type: 'tool_call', toolCall: { ...proposal, args: '{}' } }, + { type: 'tool_call', toolCall: { ...proposal, id: 'call-2', args: '{}' } }, { type: 'done' }, + ]) }) + await chat.send('write twice') + await expect(createChatController(config).decide(proposal.id, 'approve')).rejects.toMatchObject({ code: 'AK_CONFIG_INVALID' }) + expect((await config.decisionStore!.get(proposal.id))?.status).toBe('pending') + expect(execute).not.toHaveBeenCalled() + expect(requests).toHaveLength(0) + }) + it('fails closed on an invalid external claim', async () => { const decisionStore = store() decisionStore.claim = async () => ({ toolCallId: 'wrong', status: 'claimed', decision: 'approve', messages: [] }) @@ -398,7 +655,7 @@ describe('durable controller decisions', () => { const { chat, config, execute, requests } = fixture({ decisionStore }) await chat.proposeToolCall(proposal) await expect(chat.decide(proposal.id, 'approve')).rejects.toThrow('terminal write failed') - await expect(createChatController(config).decide(proposal.id, 'approve')).rejects.toMatchObject({ code: 'AK_ACTION_ALREADY_DECIDED' }) + await expect(createChatController(config).decide(proposal.id, 'approve')).resolves.toMatchObject({ id: proposal.id }) expect(execute).toHaveBeenCalledTimes(1) expect(requests).toHaveLength(0) }) diff --git a/packages/core/tests/controller-restart.mjs b/packages/core/tests/controller-restart.mjs index f25ac4c35..5aad4916a 100644 --- a/packages/core/tests/controller-restart.mjs +++ b/packages/core/tests/controller-restart.mjs @@ -11,8 +11,8 @@ if (!mode) { const directory = mkdtempSync(join(tmpdir(), 'core-restart-')) try { for (const format of ['esm', 'cjs']) { - for (const mode of ['propose', 'decide']) { - const child = await runCommand(process.execPath, [fileURLToPath(import.meta.url), mode, format, join(directory, `${format}.json`)]) + for (const mode of ['propose', 'decide', 'stream-propose', 'stream-decide']) { + const child = await runCommand(process.execPath, [fileURLToPath(import.meta.url), mode, format, join(directory, `${format}-${mode.startsWith('stream-') ? 'stream' : 'proposal'}.json`)]) assert.equal(child.code, 0, child.stderr) process.stdout.write(child.stdout) } @@ -44,18 +44,37 @@ if (!mode) { }, } let executions = 0, requests = 0 + const memory = { + async load() { return existsSync(`${recordPath}.memory`) ? JSON.parse(readFileSync(`${recordPath}.memory`, 'utf8')) : [] }, + async save(messages) { writeFileSync(`${recordPath}.memory`, JSON.stringify(messages)) }, + } const chat = createChatController({ decisionStore, + ...(mode.startsWith('stream-') ? { memory } : {}), + onToolCall() { + if (mode === 'stream-propose') { + assert.equal(read().messages.at(-1).status, 'streaming') + console.log(JSON.stringify({ mode, format, status: read().status })) + process.exit(0) + } + }, tools: [{ name: 'write', requiresConfirmation: true, execute() { executions++; return 'stored' } }], adapter: { createSource(request) { + if (mode === 'stream-propose') return { + abort() {}, + async *stream() { yield { type: 'tool_call', toolCall: { id: 'restart', name: 'write', args: '{}' } }; yield { type: 'done' } }, + } requests++ assert.equal(request.messages.find(m => m.role === 'tool')?.content, 'stored') return { abort() {}, async *stream() { yield { type: 'text', content: 'Finished' }; yield { type: 'done' } } } }, }, }) - if (mode === 'propose') { + if (mode === 'stream-propose') { + await chat.send('write') + assert.fail('Expected process termination during streaming') + } else if (mode === 'propose') { await chat.proposeToolCall({ id: 'restart', name: 'write', args: {} }) assert.equal(read().status, 'pending') assert.equal(executions, 0) diff --git a/readme-standard-v1.json b/readme-standard-v1.json index 20853aadf..07c37c9cb 100644 --- a/readme-standard-v1.json +++ b/readme-standard-v1.json @@ -519,7 +519,7 @@ "docs/STABILITY.md", "scripts/readme-fixtures.test.mjs" ], - "sourceHash": "sha256:98eb02aa354070ff757a74d59010f0d24900da5dec3eea89f920dedbe468b984" + "sourceHash": "sha256:e6346be6222d7760d6682774b3c775cb14dda438182be499796e8f7609d37c8b" }, "exceptions": [] },