From baab36938536616d208517c5c1c7eaf89a0d2518 Mon Sep 17 00:00:00 2001 From: hanpj <2568545091@qq.com> Date: Sun, 18 Oct 2026 16:32:23 +0800 Subject: [PATCH 1/2] refactor: tighten agent runtime boundaries --- packages/core/src/agent.ts | 401 +++++++++++++---------- packages/core/src/inbox.ts | 9 +- packages/kernel/src/serialization.ts | 20 ++ packages/kernel/src/tools/runtime.ts | 32 +- packages/kernel/test/domain.test.ts | 21 +- packages/test-support/test/agent.test.ts | 215 ++++++++++++ 6 files changed, 494 insertions(+), 204 deletions(-) diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index 396eed9..661929e 100644 --- a/packages/core/src/agent.ts +++ b/packages/core/src/agent.ts @@ -8,20 +8,26 @@ import { type AgentPhase, type AgentRunOptions, type AssistantMessage, + type AttemptId, + assertJsonValue, assertNever, type Clock, type EventEnvelope, type ExecutionId, type ExecutionMetadata, type ExecutionObservation, + freezeJsonValue, isAbortError, type ModelProvider, + type ModelRequest, type ModelStreamEvent, type PromptAssembler, type SessionEvent, type SessionLog, SettlementPersistenceError, type SettlementSignalFactory, + type StepId, + type ToolCall, type ToolResult, type ToolSet, type TurnId, @@ -74,10 +80,7 @@ export class Agent { #inboxDrain: Promise | undefined; constructor(options: AgentOptions) { - // 参数校验 - if (!Number.isInteger(options.maxSteps) || options.maxSteps < 1) { - throw new Error('maxSteps must be a positive integer'); - } + assertPositiveInteger('maxSteps', options.maxSteps); for (const [name, value] of [ ['runTimeoutMs', options.runTimeoutMs], ['modelAttemptTimeoutMs', options.modelAttemptTimeoutMs], @@ -86,12 +89,11 @@ export class Agent { ['maxToolResultBytes', options.maxToolResultBytes], ['maxQueuedRuns', options.maxQueuedRuns], ] as const) { - if (value !== undefined && (!Number.isSafeInteger(value) || value < 1)) { - throw new TypeError(`${name} 必须是正整数`); + if (value !== undefined) { + assertPositiveInteger(name, value); } } - // 赋值参数 this.#model = options.model; this.#session = options.session; this.#tools = options.tools; @@ -106,7 +108,7 @@ export class Agent { this.#maxModelOutputBytes = options.maxModelOutputBytes; this.#maxToolResultBytes = options.maxToolResultBytes ?? DEFAULT_MAX_TOOL_RESULT_BYTES; this.#maxQueuedRuns = options.maxQueuedRuns ?? DEFAULT_MAX_QUEUED_RUNS; - this.#observers = options.observers ?? []; + this.#observers = Object.freeze([...(options.observers ?? [])]); this.#idReservations = new AgentIdReservations(options.reservedIds); } @@ -152,34 +154,39 @@ export class Agent { return Promise.reject(new AgentClosedError('Agent 已停止接纳新的 run')); } + let accepted: AcceptedRun; + try { + accepted = acceptRun(message, options); + } catch (error) { + return Promise.reject(error); + } + return new Promise((resolve, reject) => { - const externalSignal = isAbortSignal(options) - ? options - : (options.signal ?? new AbortController().signal); let queued: QueuedRun; const onAbort = (): void => { if (!this.#inbox.remove(queued)) { return; } queued.clearAbortListener(); - reject(externalSignal.reason); + reject(accepted.signal.reason); }; queued = { - message, - options, + message: accepted.message, + signal: accepted.signal, + ...(accepted.execution === undefined ? {} : { execution: accepted.execution }), resolve, reject, - clearAbortListener: () => externalSignal.removeEventListener('abort', onAbort), + clearAbortListener: () => accepted.signal.removeEventListener('abort', onAbort), }; - if (externalSignal.aborted) { - reject(externalSignal.reason); + if (accepted.signal.aborted) { + reject(accepted.signal.reason); return; } if (this.#inbox.size >= this.#maxQueuedRuns) { reject(new AgentBusyError(`Agent 输入队列已达到上限 ${this.#maxQueuedRuns}`)); return; } - externalSignal.addEventListener('abort', onAbort, { once: true }); + accepted.signal.addEventListener('abort', onAbort, { once: true }); this.#inbox.enqueue(queued); this.#ensureInboxDraining(); }); @@ -201,13 +208,13 @@ export class Agent { /** 串行领取已接纳输入，并仅结算各自调用，不让单项失败中断后续领取。 */ async #drainInbox(): Promise { for (;;) { - const queued = this.#inbox.claim(); // 领取最早尚未执行的输入 + const queued = this.#inbox.claim(); if (queued === undefined) { return; } queued.clearAbortListener(); try { - queued.resolve(await this.#runTurn(queued.message, queued.options)); + queued.resolve(await this.#runTurn(queued.message, queued.signal, queued.execution)); } catch (error) { queued.reject(error); } @@ -217,14 +224,9 @@ export class Agent { /** 执行已从 inbox 领取的一条输入。 */ async #runTurn( message: UserMessage, - options: AbortSignal | AgentRunOptions, + externalSignal: AbortSignal, + execution: ExecutionMetadata | undefined, ): Promise { - const externalSignal = isAbortSignal(options) - ? options - : (options.signal ?? new AbortController().signal); - const execution = freezeExecutionMetadata( - isAbortSignal(options) ? undefined : options.execution, - ); const turn: ActiveTurn = { turnId: this.#idReservations.turn(this.#idGenerator.turn()), started: false, @@ -242,7 +244,6 @@ export class Agent { this.#phase = { kind: 'running', turnId: turn.turnId }; try { - // 提交 turn 开始事件 await this.#append( [ { @@ -255,24 +256,19 @@ export class Agent { ); turn.started = true; - // 提交用户消息事件 this.#idReservations.message(message.id); await this.#append([{ type: 'user.message', message }], signal); - // 循环执行 step，直至无工具调用或触及步数上限 - for (let stepNumber = 1; stepNumber <= this.#maxSteps; stepNumber += 1) { + for (let stepNumber = 1; ; stepNumber += 1) { throwIfAborted(signal); - // 创建 step 生命周期 const stepId = this.#idReservations.step(this.#idGenerator.step()); const step = createStep(stepId); turn.activeStep = step; this.#phase = { kind: 'running', turnId: turn.turnId, stepId }; - // 执行一步 step，获取返回的 assistant 消息 const messageFromStep = await this.#runStep(turn.turnId, step, signal, execution); - // 如果 step 没有工具调用，则提交 turn 结束事件 if (messageFromStep.toolCalls.length === 0) { await this.#append( [{ type: 'turn.ended', turnId: turn.turnId, outcome: 'completed' }], @@ -282,15 +278,11 @@ export class Agent { return messageFromStep; } - // 如果 step 数量达到上限，则抛出错误 if (stepNumber === this.#maxSteps) { throw new Error(`agent exceeded maxSteps (${this.#maxSteps})`); } } - - throw new Error('agent loop ended without a final assistant message'); } catch (error) { - // 构建 terminal settlement 事件 const outcome = signal.aborted || isAbortError(error) ? 'cancelled' : 'failed'; const terminalEvents = buildTerminalEvents({ error, @@ -298,7 +290,6 @@ export class Agent { turn, }); - // 如果 terminal settlement 事件不为空，则提交结算事件 if (terminalEvents.length > 0) { try { const settlementSignal = this.#settlementSignalFactory.create(); @@ -324,24 +315,64 @@ export class Agent { signal: AbortSignal, execution: ExecutionMetadata | undefined, ): Promise { - const executionEvent = this.#executionEvent(execution); - // 提交 step 开始事件 const stepId = step.stepId; await this.#append([{ type: 'step.started', turnId, stepId }], signal); step.started = true; throwIfAborted(signal); - // 组装请求 + const sessionEvents = await this.#session.read({ signal }); const request = await waitForSignal( - this.#promptAssembler.assemble( - await this.#session.read({ signal }), - this.#tools.list(), - signal, - ), + this.#promptAssembler.assemble(sessionEvents, this.#tools.list(), signal), + signal, + ); + const attempt = await this.#runModelAttempt(request, turnId, step, signal, execution); + throwIfAborted(signal); + + const assistantMessage: AssistantMessage = { + id: this.#idReservations.message(this.#idGenerator.message()), + role: 'assistant', + content: attempt.partialText, + toolCalls: Object.freeze([...attempt.observedToolCalls]), + }; + attempt.settlement = { + status: 'completed', + attemptId: attempt.attemptId, + message: assistantMessage, + }; + await this.#append( + [{ type: 'assistant.settled', turnId, stepId, settlement: attempt.settlement }], signal, ); + attempt.settlementPersisted = true; - // 准备模型；准备失败尚未形成 attempt，保持其不会消耗 attempt 标识符的恢复语义。 + for (const call of attempt.observedToolCalls) { + const result = await this.#executeToolCall( + call, + turnId, + stepId, + attempt.attemptId, + signal, + execution, + ); + attempt.toolResults.set(call.id, result); + await this.#append([{ type: 'tool.completed', turnId, stepId, result }], signal); + attempt.persistedToolResults.add(call.id); + } + + await this.#append([{ type: 'step.ended', turnId, stepId, outcome: 'completed' }], signal); + step.ended = true; + return assistantMessage; + } + + /** 准备并消费一次模型调用；prepare 失败时不会创建或消耗 attempt 身份。 */ + async #runModelAttempt( + request: ModelRequest, + turnId: TurnId, + step: Step, + signal: AbortSignal, + execution: ExecutionMetadata | undefined, + ): Promise { + const stepId = step.stepId; const modelSignal = this.#deadlineSignal(signal, this.#modelAttemptTimeoutMs); const prepared = await this.#model.prepare(modelSignal); throwIfAborted(modelSignal); @@ -354,13 +385,13 @@ export class Agent { persistedToolResults: new Set(), }; step.attempt = attempt; + const executionScope = this.#executionObservationScope(execution); let outputBytes = 0; let inputTokens = 0; let outputTokens = 0; let usageSequence = 0; const modelStartedAt = this.#now(); - // 提交模型开始事件 if (modelStartedAt !== undefined) { this.#notifyObservation({ type: 'model.invocation.started', @@ -368,25 +399,23 @@ export class Agent { turnId, stepId, attemptId: attempt.attemptId, - ...executionEvent, + ...executionScope, startedAt: modelStartedAt, }); } try { - // 循环获取模型流式生成时的中间输出 - for await (const event of prepared.stream(request, modelSignal)) { + for await (const emittedEvent of prepared.stream(request, modelSignal)) { throwIfAborted(modelSignal); + const event = snapshotModelStreamEvent(emittedEvent); this.#notifyStream(event); switch (event.type) { - // 文本增量 case 'text.delta': outputBytes += utf8ByteLength(event.text); this.#assertOutputBudget(outputBytes); attempt.partialText += event.text; break; - // 工具调用 case 'tool.call': outputBytes += utf8ByteLength(JSON.stringify(event.call)); this.#assertOutputBudget(outputBytes); @@ -394,7 +423,6 @@ export class Agent { attempt.observedToolCalls.push(event.call); break; - // 使用量 case 'usage': assertCumulativeUsage(event.inputTokens, event.outputTokens, inputTokens, outputTokens); usageSequence += 1; @@ -404,7 +432,7 @@ export class Agent { turnId, stepId, attemptId: attempt.attemptId, - ...executionEvent, + ...executionScope, usageSequence, inputTokensDelta: event.inputTokens - inputTokens, outputTokensDelta: event.outputTokens - outputTokens, @@ -416,6 +444,7 @@ export class Agent { assertNever(event); } } + throwIfAborted(modelSignal); if (modelStartedAt !== undefined) { this.#notifyObservation({ type: 'model.invocation.finished', @@ -423,7 +452,7 @@ export class Agent { turnId, stepId, attemptId: attempt.attemptId, - ...executionEvent, + ...executionScope, ...this.#withDuration(modelStartedAt), outcome: 'completed', }); @@ -436,7 +465,7 @@ export class Agent { turnId, stepId, attemptId: attempt.attemptId, - ...executionEvent, + ...executionScope, ...this.#withDuration(modelStartedAt), outcome: modelSignal.aborted || isAbortError(error) ? 'cancelled' : 'failed', ...(error instanceof Error ? { errorName: error.name } : {}), @@ -444,129 +473,82 @@ export class Agent { } throw error; } - throwIfAborted(signal); - - // 构建 assistant 消息 - const assistantMessage: AssistantMessage = { - id: this.#idReservations.message(this.#idGenerator.message()), - role: 'assistant', - content: attempt.partialText, - toolCalls: attempt.observedToolCalls, - }; - - // 构建结算结果 - attempt.settlement = { - status: 'completed', - attemptId: attempt.attemptId, - message: assistantMessage, - }; + return attempt; + } - // 提交结算事件 - await this.#append( - [ - { - type: 'assistant.settled', - turnId, - stepId, - settlement: attempt.settlement, - }, - ], - signal, - ); - attempt.settlementPersisted = true; + /** 执行并规范化一次工具调用，保证已开始的调用恰好发出一个终态观测。 */ + async #executeToolCall( + call: ToolCall, + turnId: TurnId, + stepId: StepId, + attemptId: AttemptId, + signal: AbortSignal, + execution: ExecutionMetadata | undefined, + ): Promise { + const toolSignal = this.#deadlineSignal(signal, this.#toolTimeoutMs); + const executionScope = this.#executionObservationScope(execution); + const toolStartedAt = this.#now(); + if (toolStartedAt !== undefined) { + this.#notifyObservation({ + type: 'tool.invocation.started', + sessionId: this.#session.sessionId, + turnId, + stepId, + attemptId, + ...executionScope, + toolCallId: call.id, + toolName: call.name, + startedAt: toolStartedAt, + }); + } - // 执行工具调用 - for (const call of attempt.observedToolCalls) { - const toolSignal = this.#deadlineSignal(signal, this.#toolTimeoutMs); - const toolStartedAt = this.#now(); + try { + const result = await this.#tools.execute(call, { + signal: toolSignal, + ...(execution === undefined ? {} : { execution }), + }); + throwIfAborted(toolSignal); if (toolStartedAt !== undefined) { this.#notifyObservation({ - type: 'tool.invocation.started', + type: 'tool.invocation.finished', sessionId: this.#session.sessionId, turnId, stepId, - attemptId: attempt.attemptId, - ...executionEvent, + attemptId, + ...executionScope, toolCallId: call.id, toolName: call.name, - startedAt: toolStartedAt, + ...this.#withDuration(toolStartedAt), + termination: 'returned', + resultStatus: result.status, }); } - let result: ToolResult; - let toolInvocationFinished = false; - try { - result = await this.#tools.execute(call, { - signal: toolSignal, - ...(execution === undefined ? {} : { execution }), - }); - } catch (error) { - if (toolSignal.aborted && !signal.aborted) { - if (toolStartedAt !== undefined) { - this.#notifyObservation({ - type: 'tool.invocation.finished', - sessionId: this.#session.sessionId, - turnId, - stepId, - attemptId: attempt.attemptId, - ...executionEvent, - toolCallId: call.id, - toolName: call.name, - ...this.#withDuration(toolStartedAt), - termination: 'cancelled', - ...(error instanceof Error ? { errorName: error.name } : {}), - }); - toolInvocationFinished = true; - } - result = { - status: 'cancelled', - callId: call.id, - reason: `工具调用超过 toolTimeoutMs (${this.#toolTimeoutMs})`, - }; - } else { - if (toolStartedAt !== undefined) { - this.#notifyObservation({ - type: 'tool.invocation.finished', - sessionId: this.#session.sessionId, - turnId, - stepId, - attemptId: attempt.attemptId, - ...executionEvent, - toolCallId: call.id, - toolName: call.name, - ...this.#withDuration(toolStartedAt), - termination: toolSignal.aborted || isAbortError(error) ? 'cancelled' : 'failed', - ...(error instanceof Error ? { errorName: error.name } : {}), - }); - } - throw error; - } - } - result = this.#boundToolResult(result); - attempt.toolResults.set(call.id, result); - if (toolStartedAt !== undefined && !toolInvocationFinished) { + return this.#boundToolResult(result); + } catch (error) { + if (toolStartedAt !== undefined) { this.#notifyObservation({ type: 'tool.invocation.finished', sessionId: this.#session.sessionId, turnId, stepId, - attemptId: attempt.attemptId, - ...executionEvent, + attemptId, + ...executionScope, toolCallId: call.id, toolName: call.name, ...this.#withDuration(toolStartedAt), - termination: 'returned', - resultStatus: result.status, + termination: toolSignal.aborted || isAbortError(error) ? 'cancelled' : 'failed', + ...(error instanceof Error ? { errorName: error.name } : {}), }); } - // 提交工具调用完成事件 - await this.#append([{ type: 'tool.completed', turnId, stepId, result }], signal); - attempt.persistedToolResults.add(call.id); + if (toolSignal.aborted && !signal.aborted) { + return { + status: 'cancelled', + callId: call.id, + reason: `工具调用超过 toolTimeoutMs (${this.#toolTimeoutMs})`, + }; + } + throw error; } - - // 提交 step 结束事件 - await this.#append([{ type: 'step.ended', turnId, stepId, outcome: 'completed' }], signal); - step.ended = true; - return assistantMessage; } /** 控制单次 run 的模型输出预算，避免模型无限输出。 */ @@ -612,7 +594,7 @@ export class Agent { } } - #executionEvent(execution: ExecutionMetadata | undefined): { + #executionObservationScope(execution: ExecutionMetadata | undefined): { readonly executionId?: ExecutionId; } { return execution === undefined ? {} : { executionId: execution.id }; @@ -652,38 +634,105 @@ export class Agent { /** 向 observer 派发LLM流事件（模型生成时的中间输出）；用于在 Agent 外部展示模型生成过程，不做持久化。 */ #notifyStream(event: ModelStreamEvent): void { - for (const observer of this.#observers) { - try { - void Promise.resolve(observer.onStream?.(event)).catch(() => undefined); - } catch { - // 观察者只接收瞬态通知，其失败不能改变 Agent 生命周期。 - } - } + this.#notifyObservers((observer) => observer.onStream?.(event)); } /** 向 observer 派发已经成功提交的会话事实。 */ #notifySessionEvent(envelope: EventEnvelope): void { - for (const observer of this.#observers) { - try { - void Promise.resolve(observer.onSessionEvent?.(envelope)).catch(() => undefined); - } catch { - // 观察者只接收提交后的通知，其失败不能改变 Agent 生命周期。 - } - } + this.#notifyObservers((observer) => observer.onSessionEvent?.(envelope)); } /** 向 observer 派发不进入会话日志的模型与工具调用观测信号。 */ #notifyObservation(observation: ExecutionObservation): void { + this.#notifyObservers((observer) => observer.onObservation?.(observation)); + } + + #notifyObservers(deliver: (observer: AgentObserver) => void | Promise): void { for (const observer of this.#observers) { try { - void Promise.resolve(observer.onObservation?.(observation)).catch(() => undefined); + void Promise.resolve(deliver(observer)).catch(() => undefined); } catch { - // 瞬态观测失败不能改变 Agent 生命周期。 + // 观察者仅接收通知，其失败不能改变 Agent 生命周期。 } } } } +interface AcceptedRun { + readonly message: UserMessage; + readonly signal: AbortSignal; + readonly execution?: ExecutionMetadata; +} + +/** 在跨越排队边界前校验并复制调用方输入。 */ +function acceptRun(message: UserMessage, options: AbortSignal | AgentRunOptions): AcceptedRun { + const shorthandSignal = isAbortSignal(options); + const signal = shorthandSignal ? options : (options.signal ?? new AbortController().signal); + const execution = freezeExecutionMetadata(shorthandSignal ? undefined : options.execution); + return Object.freeze({ + message: snapshotUserMessage(message), + signal, + ...(execution === undefined ? {} : { execution }), + }); +} + +function snapshotUserMessage(message: UserMessage): UserMessage { + if (typeof message !== 'object' || message === null) { + throw new TypeError('message 必须是对象'); + } + if (typeof message.id !== 'string' || message.id.length === 0) { + throw new TypeError('message.id 必须是非空字符串'); + } + if (message.role !== 'user') { + throw new TypeError('message.role 必须是 user'); + } + if (typeof message.content !== 'string') { + throw new TypeError('message.content 必须是字符串'); + } + return Object.freeze({ id: message.id, role: message.role, content: message.content }); +} + +function assertPositiveInteger(name: string, value: number): void { + if (!Number.isSafeInteger(value) || value < 1) { + throw new TypeError(`${name} 必须是正整数`); + } +} + +/** 在通知 observer 或保存工具调用前取得模型事件的不可变快照。 */ +function snapshotModelStreamEvent(event: ModelStreamEvent): ModelStreamEvent { + switch (event.type) { + case 'text.delta': + if (typeof event.text !== 'string') { + throw new TypeError('模型 text.delta.text 必须是字符串'); + } + return Object.freeze({ type: event.type, text: event.text }); + case 'tool.call': + return Object.freeze({ type: event.type, call: snapshotToolCall(event.call) }); + case 'usage': + return Object.freeze({ + type: event.type, + inputTokens: event.inputTokens, + outputTokens: event.outputTokens, + }); + default: + return assertNever(event); + } +} + +function snapshotToolCall(call: ToolCall): ToolCall { + if (typeof call !== 'object' || call === null) { + throw new TypeError('模型 tool.call.call 必须是对象'); + } + if (typeof call.id !== 'string' || call.id.length === 0) { + throw new TypeError('模型 tool.call.call.id 必须是非空字符串'); + } + if (typeof call.name !== 'string' || call.name.length === 0) { + throw new TypeError('模型 tool.call.call.name 必须是非空字符串'); + } + assertJsonValue(call.input); + return Object.freeze({ id: call.id, name: call.name, input: freezeJsonValue(call.input) }); +} + /** 计算模型事件文本的 UTF-8 字节数，供单次输出预算使用。 */ function utf8ByteLength(value: string): number { return textEncoder.encode(value).byteLength; diff --git a/packages/core/src/inbox.ts b/packages/core/src/inbox.ts index 2cd1dd1..d7a8771 100644 --- a/packages/core/src/inbox.ts +++ b/packages/core/src/inbox.ts @@ -1,11 +1,12 @@ -import type { AgentRunOptions, AssistantMessage, UserMessage } from '@agent-harness/kernel'; +import type { AssistantMessage, ExecutionMetadata, UserMessage } from '@agent-harness/kernel'; /** 尚未开始的 turn；每个调用者只会收到自己消息对应的结果。 */ export interface QueuedRun { readonly message: UserMessage; - readonly options: AbortSignal | AgentRunOptions; - readonly resolve: (message: AssistantMessage) => void; // 执行成功时，调用 resolve 通知调用者 - readonly reject: (error: unknown) => void; // 执行失败时，调用 reject 通知调用者 + readonly signal: AbortSignal; + readonly execution?: ExecutionMetadata; + readonly resolve: (message: AssistantMessage) => void; + readonly reject: (error: unknown) => void; readonly clearAbortListener: () => void; } diff --git a/packages/kernel/src/serialization.ts b/packages/kernel/src/serialization.ts index b4718b2..637a68f 100644 --- a/packages/kernel/src/serialization.ts +++ b/packages/kernel/src/serialization.ts @@ -14,6 +14,26 @@ export function assertJsonValue(value: unknown): asserts value is JsonValue { assertJsonValueAt(value, '$', new Set()); } +/** 复制并冻结已校验的 JSON 值，避免跨异步边界共享调用方可变对象。 */ +export function freezeJsonValue(value: JsonValue): JsonValue { + if (value === null || typeof value !== 'object') { + return value; + } + if (Array.isArray(value)) { + return Object.freeze(value.map(freezeJsonValue)); + } + const copy: Record = {}; + for (const [key, child] of Object.entries(value)) { + Object.defineProperty(copy, key, { + value: freezeJsonValue(child), + enumerable: true, + writable: false, + configurable: false, + }); + } + return Object.freeze(copy); +} + function assertJsonValueAt(value: unknown, path: string, ancestors: Set): void { if (value === null || typeof value === 'string' || typeof value === 'boolean') { return; diff --git a/packages/kernel/src/tools/runtime.ts b/packages/kernel/src/tools/runtime.ts index b2e0ffd..b413af6 100644 --- a/packages/kernel/src/tools/runtime.ts +++ b/packages/kernel/src/tools/runtime.ts @@ -1,6 +1,11 @@ import type { ExecutionMetadata } from '../agent.js'; import type { JsonValue } from '../serialization.js'; -import { assertJsonValue, isAbortError, serializeError } from '../serialization.js'; +import { + assertJsonValue, + freezeJsonValue, + isAbortError, + serializeError, +} from '../serialization.js'; import type { ExecutableTool, ModelToolDefinition, @@ -68,7 +73,7 @@ export class SandboxToolSet implements ToolSet { return settleRegisteredCall(call, context, this.#definitions, (_tool, input) => { assertJsonValue(input); const task = Object.freeze({ - call: Object.freeze({ ...call, input: freezeJson(input) }), + call: Object.freeze({ ...call, input: freezeJsonValue(input) }), ...(this.#workspace === undefined ? {} : { workspace: this.#workspace }), ...(this.#limits === undefined ? {} : { limits: this.#limits }), }); @@ -139,7 +144,7 @@ export class CompositeToolSet implements ToolSet { Object.freeze({ name: definition.name, description: definition.description, - inputSchema: freezeJson(definition.inputSchema as JsonValue) as Readonly< + inputSchema: freezeJsonValue(definition.inputSchema as JsonValue) as Readonly< Record >, }), @@ -318,7 +323,7 @@ function denialReason(decision: unknown): string { function freezeToolCall(call: ToolCall): ToolCall { assertJsonValue(call.input); - return Object.freeze({ id: call.id, name: call.name, input: freezeJson(call.input) }); + return Object.freeze({ id: call.id, name: call.name, input: freezeJsonValue(call.input) }); } function freezeExecutionContext( @@ -333,22 +338,3 @@ function freezeExecutionContext( }), }); } - -function freezeJson(value: JsonValue): JsonValue { - if (value === null || typeof value !== 'object') { - return value; - } - if (Array.isArray(value)) { - return Object.freeze(value.map(freezeJson)); - } - const copy: Record = {}; - for (const [key, child] of Object.entries(value)) { - Object.defineProperty(copy, key, { - value: freezeJson(child), - enumerable: true, - writable: false, - configurable: false, - }); - } - return Object.freeze(copy); -} diff --git a/packages/kernel/test/domain.test.ts b/packages/kernel/test/domain.test.ts index a43cf9a..7c1f60a 100644 --- a/packages/kernel/test/domain.test.ts +++ b/packages/kernel/test/domain.test.ts @@ -1,5 +1,11 @@ import { describe, expect, it } from 'vitest'; -import { assertJsonValue, brandId, isAbortError, serializeError } from '../src/index.ts'; +import { + assertJsonValue, + brandId, + freezeJsonValue, + isAbortError, + serializeError, +} from '../src/index.ts'; describe('domain', () => { describe('brandId', () => { @@ -39,6 +45,19 @@ describe('domain', () => { }); }); + describe('freezeJsonValue', () => { + it('递归复制并冻结 JSON 值', () => { + const source = { nested: { values: [1, 2] } }; + const snapshot = freezeJsonValue(source); + + source.nested.values.push(3); + + expect(snapshot).toEqual({ nested: { values: [1, 2] } }); + expect(Object.isFrozen(snapshot)).toBe(true); + expect(Object.isFrozen((snapshot as { readonly nested: object }).nested)).toBe(true); + }); + }); + describe('serializeError / isAbortError', () => { it('序列化 Error 与非 Error 值', () => { expect(serializeError(new TypeError('bad'))).toEqual({ diff --git a/packages/test-support/test/agent.test.ts b/packages/test-support/test/agent.test.ts index 8eb6f38..76a3f24 100644 --- a/packages/test-support/test/agent.test.ts +++ b/packages/test-support/test/agent.test.ts @@ -7,6 +7,7 @@ import { AgentBusyError, AgentClosedError, AgentIdCollisionError, + type AgentObserver, analyzeSessionLifecycle, brandId, DirectToolSet, @@ -112,6 +113,72 @@ describe('Agent', () => { expect(model.requests).toHaveLength(2); }); + it('在输入进入队列时冻结消息、执行上下文与取消信号', async () => { + const session = new InMemorySessionLog(sessionId); + let firstPrepared!: () => void; + let releaseFirst!: () => void; + const prepared = new Promise((resolve) => { + firstPrepared = resolve; + }); + const release = new Promise((resolve) => { + releaseFirst = resolve; + }); + const model = new ScriptedModelProvider([ + { + events: [{ type: 'text.delta', text: 'first reply' }], + prepare: async () => { + firstPrepared(); + await release; + }, + }, + { events: [{ type: 'text.delta', text: 'second reply' }] }, + ]); + const agent = new Agent({ + model, + session, + tools: new DirectToolSet([]), + ids: new SequentialAgentIds(), + maxSteps: 4, + settlementSignals: new TestSettlementSignals(), + prompts: defaultPrompts(), + }); + const first = agent.run(user('first')); + await prepared; + + const queuedMessage = { + id: brandId<'MessageId'>('user-2') as MessageId, + role: 'user' as const, + content: 'second', + }; + const acceptedSignal = new AbortController(); + const replacementSignal = new AbortController(); + const execution = { + id: brandId<'ExecutionId'>('execution-2') as ExecutionId, + subject: 'user-2', + authorizationVersion: 'v1', + }; + const options = { signal: acceptedSignal.signal, execution }; + const second = agent.run(queuedMessage, options); + + queuedMessage.content = 'mutated'; + execution.subject = 'attacker'; + options.signal = replacementSignal.signal; + replacementSignal.abort(new Error('不应影响已接纳输入')); + releaseFirst(); + + await expect(first).resolves.toMatchObject({ content: 'first reply' }); + await expect(second).resolves.toMatchObject({ content: 'second reply' }); + const events = await readEvents(session); + expect( + events + .filter((envelope) => envelope.event.type === 'user.message') + .map((envelope) => (envelope.event as { readonly message: UserMessage }).message.content), + ).toEqual(['first', 'second']); + expect( + events.filter((envelope) => envelope.event.type === 'turn.started').at(1)?.event, + ).toMatchObject({ execution: { subject: 'user-2' } }); + }); + it('输入队列达到上限时拒绝新输入，但不影响已接纳输入', async () => { const session = new InMemorySessionLog(sessionId); let signalFirstPrepared!: () => void; @@ -1197,6 +1264,52 @@ describe('Agent', () => { ]); }); + it('工具在 deadline 后迟返时仍记录取消结果', async () => { + const callId = brandId<'ToolCallId'>('call-late-return') as ToolCallId; + const session = new InMemorySessionLog(sessionId); + const observed: ExecutionObservation[] = []; + const tools = new ScriptedToolSet( + [{ name: 'wait', description: '等待', inputSchema: {} }], + (call, context) => + new Promise((resolve) => { + context.signal.addEventListener( + 'abort', + () => resolve({ status: 'success', callId: call.id, output: { late: true } }), + { once: true }, + ); + }), + ); + const agent = new Agent({ + model: new ScriptedModelProvider([ + { events: [{ type: 'tool.call', call: { id: callId, name: 'wait', input: {} } }] }, + { events: [{ type: 'text.delta', text: '继续执行' }] }, + ]), + session, + tools, + ids: new SequentialAgentIds(), + maxSteps: 4, + toolTimeoutMs: 10, + settlementSignals: new TestSettlementSignals(), + prompts: defaultPrompts(), + clock: { now: () => new Date(0) }, + observers: [{ onObservation: (observation) => observed.push(observation) }], + }); + + await expect(agent.run(user())).resolves.toMatchObject({ content: '继续执行' }); + expect( + (await readEvents(session)).find((envelope) => envelope.event.type === 'tool.completed') + ?.event, + ).toMatchObject({ type: 'tool.completed', result: { status: 'cancelled' } }); + expect(observed.filter((observation) => observation.type.startsWith('tool.'))).toEqual([ + expect.objectContaining({ type: 'tool.invocation.started' }), + expect.objectContaining({ + type: 'tool.invocation.finished', + termination: 'cancelled', + errorName: 'AbortError', + }), + ]); + }); + it('普通会话 I/O 响应 run signal，并用一个独立信号提交 terminal batch', async () => { let signalReadStarted: (() => void) | undefined; const readStarted = new Promise((resolve) => { @@ -1399,6 +1512,108 @@ describe('Agent', () => { expectClosed(await readEvents(session)); }); + it('observer 不能修改 Core 随后消费的模型事件或构造后追加 observer', async () => { + const callId = brandId<'ToolCallId'>('call-observer-isolation') as ToolCallId; + const calls: Array<{ readonly name: string; readonly input: unknown }> = []; + const mutationResults: boolean[] = []; + let appendedObserverCalled = false; + const observers: AgentObserver[] = [ + { + onStream(event) { + if (event.type === 'text.delta') { + mutationResults.push(Reflect.set(event, 'text', 'mutated')); + } + if (event.type === 'tool.call') { + mutationResults.push(Reflect.set(event.call, 'name', 'mutated')); + mutationResults.push(Reflect.set(event.call.input as object, 'query', 'mutated')); + } + }, + }, + ]; + const session = new InMemorySessionLog(sessionId); + const agent = new Agent({ + model: new ScriptedModelProvider([ + { + events: [ + { type: 'text.delta', text: 'before tool' }, + { + type: 'tool.call', + call: { id: callId, name: 'lookup', input: { query: 'original' } }, + }, + ], + }, + { events: [{ type: 'text.delta', text: 'done' }] }, + ]), + session, + tools: new ScriptedToolSet( + [{ name: 'lookup', description: '查询', inputSchema: {} }], + async (call) => { + calls.push({ name: call.name, input: call.input }); + return { status: 'success', callId: call.id, output: { ok: true } }; + }, + ), + ids: new SequentialAgentIds(), + maxSteps: 4, + settlementSignals: new TestSettlementSignals(), + prompts: defaultPrompts(), + observers, + }); + observers.push({ onStream: () => (appendedObserverCalled = true) }); + + await expect(agent.run(user())).resolves.toMatchObject({ content: 'done' }); + expect(mutationResults).toEqual([false, false, false, false]); + expect(calls).toEqual([{ name: 'lookup', input: { query: 'original' } }]); + expect(appendedObserverCalled).toBe(false); + const assistantSettlements = (await readEvents(session)).filter( + (envelope) => envelope.event.type === 'assistant.settled', + ); + expect(assistantSettlements[0]?.event).toMatchObject({ + settlement: { message: { content: 'before tool' } }, + }); + }); + + it('模型流在 deadline 后迟返时按取消结算', async () => { + const observed: ExecutionObservation[] = []; + const session = new InMemorySessionLog(sessionId); + const agent = new Agent({ + model: { + id: 'late-return-model', + async prepare() { + return { + async *stream(_request, signal) { + await new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { once: true }); + }); + }, + }; + }, + }, + session, + tools: new DirectToolSet([]), + ids: new SequentialAgentIds(), + maxSteps: 4, + modelAttemptTimeoutMs: 10, + settlementSignals: new TestSettlementSignals(), + prompts: defaultPrompts(), + clock: { now: () => new Date(0) }, + observers: [{ onObservation: (observation) => observed.push(observation) }], + }); + + await expect(agent.run(user())).rejects.toMatchObject({ name: 'AbortError' }); + expect(observed.filter((observation) => observation.type.startsWith('model.'))).toEqual([ + expect.objectContaining({ type: 'model.invocation.started' }), + expect.objectContaining({ + type: 'model.invocation.finished', + outcome: 'cancelled', + errorName: 'AbortError', + }), + ]); + expect((await readEvents(session)).at(-1)?.event).toMatchObject({ + type: 'turn.ended', + outcome: 'cancelled', + }); + }); + it('以注入时钟成对报告模型执行的开始与终态', async () => { const observed: ExecutionObservation[] = []; const timestamps = [new Date('2026-10-13T00:00:00.000Z'), new Date('2026-10-13T00:00:00.025Z')]; From 7fac6fa333c1aaa1a2b5716c8f191359bcfb785c Mon Sep 17 00:00:00 2001 From: hanpj <2568545091@qq.com> Date: Sun, 18 Oct 2026 16:39:18 +0800 Subject: [PATCH 2/2] fix: normalize runtime completion boundaries --- packages/core/src/agent.ts | 87 ++++++++++++++++++++---- packages/test-support/test/agent.test.ts | 84 +++++++++++++++++++++-- 2 files changed, 152 insertions(+), 19 deletions(-) diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index 661929e..111548c 100644 --- a/packages/core/src/agent.ts +++ b/packages/core/src/agent.ts @@ -375,7 +375,7 @@ export class Agent { const stepId = step.stepId; const modelSignal = this.#deadlineSignal(signal, this.#modelAttemptTimeoutMs); const prepared = await this.#model.prepare(modelSignal); - throwIfAborted(modelSignal); + this.#throwIfDeadlineAborted(signal, modelSignal); const attempt: Attempt = { attemptId: this.#idReservations.attempt(this.#idGenerator.attempt()), partialText: '', @@ -406,7 +406,7 @@ export class Agent { try { for await (const emittedEvent of prepared.stream(request, modelSignal)) { - throwIfAborted(modelSignal); + this.#throwIfDeadlineAborted(signal, modelSignal); const event = snapshotModelStreamEvent(emittedEvent); this.#notifyStream(event); switch (event.type) { @@ -444,7 +444,7 @@ export class Agent { assertNever(event); } } - throwIfAborted(modelSignal); + this.#throwIfDeadlineAborted(signal, modelSignal); if (modelStartedAt !== undefined) { this.#notifyObservation({ type: 'model.invocation.finished', @@ -458,6 +458,7 @@ export class Agent { }); } } catch (error) { + const terminalError = this.#deadlineError(error, signal, modelSignal); if (modelStartedAt !== undefined) { this.#notifyObservation({ type: 'model.invocation.finished', @@ -467,11 +468,11 @@ export class Agent { attemptId: attempt.attemptId, ...executionScope, ...this.#withDuration(modelStartedAt), - outcome: modelSignal.aborted || isAbortError(error) ? 'cancelled' : 'failed', - ...(error instanceof Error ? { errorName: error.name } : {}), + outcome: modelSignal.aborted || isAbortError(terminalError) ? 'cancelled' : 'failed', + ...(terminalError instanceof Error ? { errorName: terminalError.name } : {}), }); } - throw error; + throw terminalError; } return attempt; } @@ -503,11 +504,12 @@ export class Agent { } try { - const result = await this.#tools.execute(call, { + const returned = await this.#tools.execute(call, { signal: toolSignal, ...(execution === undefined ? {} : { execution }), }); - throwIfAborted(toolSignal); + this.#throwIfDeadlineAborted(signal, toolSignal); + const result = this.#boundToolResult(this.#snapshotToolResult(call, returned)); if (toolStartedAt !== undefined) { this.#notifyObservation({ type: 'tool.invocation.finished', @@ -523,8 +525,9 @@ export class Agent { resultStatus: result.status, }); } - return this.#boundToolResult(result); + return result; } catch (error) { + const terminalError = this.#deadlineError(error, signal, toolSignal); if (toolStartedAt !== undefined) { this.#notifyObservation({ type: 'tool.invocation.finished', @@ -536,8 +539,8 @@ export class Agent { toolCallId: call.id, toolName: call.name, ...this.#withDuration(toolStartedAt), - termination: toolSignal.aborted || isAbortError(error) ? 'cancelled' : 'failed', - ...(error instanceof Error ? { errorName: error.name } : {}), + termination: toolSignal.aborted || isAbortError(terminalError) ? 'cancelled' : 'failed', + ...(terminalError instanceof Error ? { errorName: terminalError.name } : {}), }); } if (toolSignal.aborted && !signal.aborted) { @@ -547,7 +550,7 @@ export class Agent { reason: `工具调用超过 toolTimeoutMs (${this.#toolTimeoutMs})`, }; } - throw error; + throw terminalError; } } @@ -563,6 +566,66 @@ export class Agent { return AbortSignal.any([parent, AbortSignal.timeout(timeoutMs)]); } + /** 父取消统一为 AbortError；局部 deadline 保留原始 TimeoutError。 */ + #throwIfDeadlineAborted(parent: AbortSignal, deadline: AbortSignal): void { + if (parent.aborted) { + throwIfAborted(parent); + } + if (deadline.aborted) { + throw deadline.reason; + } + } + + #deadlineError(error: unknown, parent: AbortSignal, deadline: AbortSignal): unknown { + try { + this.#throwIfDeadlineAborted(parent, deadline); + return error; + } catch (deadlineError) { + return deadlineError; + } + } + + /** 校验 ToolSet 返回值与当前调用关联，并隔离实现方仍可修改的载荷。 */ + #snapshotToolResult(call: ToolCall, result: ToolResult): ToolResult { + if (typeof result !== 'object' || result === null) { + throw new TypeError('工具结果必须是对象'); + } + if (result.callId !== call.id) { + throw new TypeError(`工具结果 callId 必须匹配当前调用 (${call.id})`); + } + switch (result.status) { + case 'success': + assertJsonValue(result.output); + return Object.freeze({ + status: result.status, + callId: call.id, + output: freezeJsonValue(result.output), + }); + case 'failure': + if ( + typeof result.error !== 'object' || + result.error === null || + typeof result.error.name !== 'string' || + typeof result.error.message !== 'string' + ) { + throw new TypeError('工具失败结果必须包含字符串 name 与 message'); + } + return Object.freeze({ + status: result.status, + callId: call.id, + error: Object.freeze({ name: result.error.name, message: result.error.message }), + }); + case 'cancelled': + case 'unknown': + if (typeof result.reason !== 'string') { + throw new TypeError(`工具 ${result.status} 结果必须包含字符串 reason`); + } + return Object.freeze({ status: result.status, callId: call.id, reason: result.reason }); + default: + return assertNever(result); + } + } + /** 限制工具返回的可变载荷，避免其无限写入会话与后续 prompt。 */ #boundToolResult(result: ToolResult): ToolResult { const bytes = utf8ByteLength(JSON.stringify(this.#toolResultPayload(result))); diff --git a/packages/test-support/test/agent.test.ts b/packages/test-support/test/agent.test.ts index 76a3f24..400d9d7 100644 --- a/packages/test-support/test/agent.test.ts +++ b/packages/test-support/test/agent.test.ts @@ -19,6 +19,7 @@ import { type SessionLog, SettlementPersistenceError, type ToolCallId, + type ToolResult, type UserMessage, } from '@agent-harness/kernel'; import { @@ -1305,11 +1306,80 @@ describe('Agent', () => { expect.objectContaining({ type: 'tool.invocation.finished', termination: 'cancelled', - errorName: 'AbortError', + errorName: 'TimeoutError', }), ]); }); + it('工具返回错误 callId 时在写入会话前拒绝结果', async () => { + const callId = brandId<'ToolCallId'>('call-expected') as ToolCallId; + const wrongCallId = brandId<'ToolCallId'>('call-wrong') as ToolCallId; + const session = new InMemorySessionLog(sessionId); + const observed: ExecutionObservation[] = []; + const agent = new Agent({ + model: new ScriptedModelProvider([ + { events: [{ type: 'tool.call', call: { id: callId, name: 'lookup', input: {} } }] }, + ]), + session, + tools: new ScriptedToolSet( + [{ name: 'lookup', description: '查询', inputSchema: {} }], + async () => ({ status: 'success', callId: wrongCallId, output: { ok: true } }), + ), + ids: new SequentialAgentIds(), + maxSteps: 4, + settlementSignals: new TestSettlementSignals(), + prompts: defaultPrompts(), + clock: { now: () => new Date(0) }, + observers: [{ onObservation: (observation) => observed.push(observation) }], + }); + + await expect(agent.run(user())).rejects.toThrow('工具结果 callId 必须匹配当前调用'); + const toolCompleted = (await readEvents(session)).find( + (envelope) => envelope.event.type === 'tool.completed', + )?.event; + expect(toolCompleted).toMatchObject({ + type: 'tool.completed', + result: { status: 'failure', callId }, + }); + expect(toolCompleted).not.toMatchObject({ result: { callId: wrongCallId } }); + expect( + observed.filter((observation) => observation.type === 'tool.invocation.finished'), + ).toEqual([expect.objectContaining({ termination: 'failed', errorName: 'TypeError' })]); + }); + + it('工具返回非法载荷时只发送一个调用终态观测', async () => { + const callId = brandId<'ToolCallId'>('call-invalid-result') as ToolCallId; + const cyclic: Record = {}; + cyclic.self = cyclic; + const invalidResult = { + status: 'success', + callId, + output: cyclic, + } as unknown as ToolResult; + const observed: ExecutionObservation[] = []; + const agent = new Agent({ + model: new ScriptedModelProvider([ + { events: [{ type: 'tool.call', call: { id: callId, name: 'lookup', input: {} } }] }, + ]), + session: new InMemorySessionLog(sessionId), + tools: new ScriptedToolSet( + [{ name: 'lookup', description: '查询', inputSchema: {} }], + async () => invalidResult, + ), + ids: new SequentialAgentIds(), + maxSteps: 4, + settlementSignals: new TestSettlementSignals(), + prompts: defaultPrompts(), + clock: { now: () => new Date(0) }, + observers: [{ onObservation: (observation) => observed.push(observation) }], + }); + + await expect(agent.run(user())).rejects.toThrow('包含循环引用'); + expect( + observed.filter((observation) => observation.type === 'tool.invocation.finished'), + ).toEqual([expect.objectContaining({ termination: 'failed', errorName: 'TypeError' })]); + }); + it('普通会话 I/O 响应 run signal，并用一个独立信号提交 terminal batch', async () => { let signalReadStarted: (() => void) | undefined; const readStarted = new Promise((resolve) => { @@ -1572,7 +1642,7 @@ describe('Agent', () => { }); }); - it('模型流在 deadline 后迟返时按取消结算', async () => { + it('模型流在 deadline 后迟返时按局部超时失败结算', async () => { const observed: ExecutionObservation[] = []; const session = new InMemorySessionLog(sessionId); const agent = new Agent({ @@ -1599,18 +1669,18 @@ describe('Agent', () => { observers: [{ onObservation: (observation) => observed.push(observation) }], }); - await expect(agent.run(user())).rejects.toMatchObject({ name: 'AbortError' }); + await expect(agent.run(user())).rejects.toMatchObject({ name: 'TimeoutError' }); expect(observed.filter((observation) => observation.type.startsWith('model.'))).toEqual([ expect.objectContaining({ type: 'model.invocation.started' }), expect.objectContaining({ type: 'model.invocation.finished', outcome: 'cancelled', - errorName: 'AbortError', + errorName: 'TimeoutError', }), ]); expect((await readEvents(session)).at(-1)?.event).toMatchObject({ type: 'turn.ended', - outcome: 'cancelled', + outcome: 'failed', }); }); @@ -1707,7 +1777,7 @@ describe('Agent', () => { expect.objectContaining({ type: 'model.invocation.finished', outcome: 'cancelled', - errorName: 'Error', + errorName: 'AbortError', }), ]); }); @@ -1770,7 +1840,7 @@ describe('Agent', () => { expect.objectContaining({ type: 'tool.invocation.finished', termination: 'cancelled', - errorName: 'Error', + errorName: 'AbortError', }), ]);