diff --git a/docs/architecture.md b/docs/architecture.md index 637f3aa..95b10a0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -47,7 +47,7 @@ Kernel 以 context、skills、memory、模型、会话和工具等能力域定 `AgentProfile` 带 ID、版本和静态指令。Runtime 只在使用默认 prompt 组装器时冻结并注入它;自定义 `PromptAssembler` 完全拥有组装顺序。Profile 当前不写入会话事件,审计需求必须使用独立契约。 -`AgentObserver` 接收持久 `session.event`、模型 usage 及模型/工具的开始、完成、失败观测事件。Runtime 注入的 Clock 使这些事件包含稳定关联 ID 与耗时;传入执行上下文时,所有模型和工具事件还直接携带冻结后的 `executionId`。投递是最佳努力,可靠 trace 与审计由 Host 或 Platform adapter 的 outbox 实现。 +`SessionEvent` 是 turn、step、assistant settlement 与工具结果的唯一持久生命周期事实;append 成功后,`AgentObserver.onSessionEvent` 直接接收对应的 `EventEnvelope`。模型流由 `onStream` 单独通知;模型与工具调用的开始、结束、用量和耗时属于 `ExecutionObservation`,通过 `onObservation` 通知。Runtime 注入的 Clock 使调用观测包含稳定关联 ID 与耗时;传入执行上下文时,观测还直接携带冻结后的 `executionId`。所有 observer 投递都是最佳努力,可靠 trace 与审计由 Host 或 Platform adapter 的 outbox 实现。 工具领域只位于 `kernel/tools/protocol.ts` 与 `kernel/tools/runtime.ts`。`ToolContract` 定义工具调用契约,`ExecutableTool` 附带直接执行器,`ToolSet` 提供一次运行固定的模型目录与调用路由;模型只接收 `ModelToolDefinition` 投影。`PolicyControlledToolSet` 在副作用前调用策略;Sandbox、Platform 许可与 MCP 通过 adapters 接入。Kernel 和 Core 不依赖外部网络协议、账户模型、SDK、文件系统或进程。 diff --git a/docs/decisions/0055-separate-session-facts-from-execution-observations.md b/docs/decisions/0055-separate-session-facts-from-execution-observations.md new file mode 100644 index 0000000..8be380a --- /dev/null +++ b/docs/decisions/0055-separate-session-facts-from-execution-observations.md @@ -0,0 +1,29 @@ +# ADR 0055:分离会话事实与执行观测 + +状态:已接受 + +## 背景 + +`AgentExecutionEvent` 同时转发已经提交的 `SessionEvent`,又定义模型与工具的瞬态执行事件。其中 `tool.completed` 既表示持久会话事实,也表示携带耗时的执行观测,但两者字段与语义不同。Kernel 中还出现了一套未接入运行时的 `AgentRuntimeEvent`、`EventEnvelope` 和流事件定义,形成第二套生命周期词汇。 + +## 决策 + +`SessionEvent` 是 turn、step、assistant settlement 与工具结果的唯一持久生命周期事实。`AgentObserver.onSessionEvent` 在 append 成功后直接接收其 `EventEnvelope`,不再通过 `AgentExecutionEvent` 包装。 + +模型和工具的开始、结束、耗时、底层异常与模型用量属于瞬态执行观测,由 `ExecutionObservation` 和 `AgentObserver.onObservation` 表达。观测事件使用 `model.invocation.*` 与 `tool.invocation.*` 名称,避免与持久会话事件同名。一次 invocation 只有一种结束事件,终态通过 `outcome` 或 `termination` 字段表达。 + +`ModelStreamEvent` 继续通过 `AgentObserver.onStream` 独立通知。删除没有调用方的聚合事件目录;各类事件在拥有其业务语义的 capability 目录定义,由 Kernel 公共入口统一导出。 + +## 后果 + +会话恢复、消息投影和持久化继续只依赖 `SessionEvent`。观察者能够明确区分已经提交的事实与可能丢失的瞬态信号,不再需要从 `session.event` 包装中取出信封,也不会遇到两个含义不同的 `tool.completed`。 + +调用方需要分别实现 `onSessionEvent` 与 `onObservation`。可靠 trace、外部投递和去重仍由 observer 下游负责;执行观测不会写入会话日志。 + +本决定取代 ADR 0016 和 ADR 0051 中由单一 `AgentExecutionEvent` 同时承载两类通知的接口形态,不改变其提交后通知、最佳努力投递和注入时钟等约束。 + +## 放弃的方案 + +- 不把全部执行观测写入 `SessionEvent`:调用耗时、底层异常名称和 usage 通知不是恢复模型可见状态所需的事实。 +- 不保留覆盖所有领域的 `AgentRuntimeEvent` 大联合:它会重复 session、model 和 observability 各自拥有的协议,并让不同变化原因集中到一个文件。 +- 不继续用 `AgentExecutionEvent` 包装会话信封并复用持久事件名称:这会让观察者无法从名称判断事件是否已经持久提交。 diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index 783dc3c..396eed9 100644 --- a/packages/core/src/agent.ts +++ b/packages/core/src/agent.ts @@ -2,7 +2,6 @@ import { AgentBusyError, AgentClosedError, - type AgentExecutionEvent, type AgentIdGenerator, type AgentObserver, type AgentOptions, @@ -11,8 +10,10 @@ import { type AssistantMessage, assertNever, type Clock, + type EventEnvelope, type ExecutionId, type ExecutionMetadata, + type ExecutionObservation, isAbortError, type ModelProvider, type ModelStreamEvent, @@ -21,8 +22,8 @@ import { type SessionLog, SettlementPersistenceError, type SettlementSignalFactory, - type ToolSet, type ToolResult, + type ToolSet, type TurnId, type UserMessage, } from '@agent-harness/kernel'; @@ -361,8 +362,8 @@ export class Agent { // 提交模型开始事件 if (modelStartedAt !== undefined) { - this.#notifyExecution({ - type: 'model.started', + this.#notifyObservation({ + type: 'model.invocation.started', sessionId: this.#session.sessionId, turnId, stepId, @@ -397,8 +398,8 @@ export class Agent { case 'usage': assertCumulativeUsage(event.inputTokens, event.outputTokens, inputTokens, outputTokens); usageSequence += 1; - this.#notifyExecution({ - type: 'model.usage', + this.#notifyObservation({ + type: 'model.invocation.usage', sessionId: this.#session.sessionId, turnId, stepId, @@ -416,26 +417,28 @@ export class Agent { } } if (modelStartedAt !== undefined) { - this.#notifyExecution({ - type: 'model.completed', + this.#notifyObservation({ + type: 'model.invocation.finished', sessionId: this.#session.sessionId, turnId, stepId, attemptId: attempt.attemptId, ...executionEvent, ...this.#withDuration(modelStartedAt), + outcome: 'completed', }); } } catch (error) { if (modelStartedAt !== undefined) { - this.#notifyExecution({ - type: modelSignal.aborted || isAbortError(error) ? 'model.cancelled' : 'model.failed', + this.#notifyObservation({ + type: 'model.invocation.finished', sessionId: this.#session.sessionId, turnId, stepId, attemptId: attempt.attemptId, ...executionEvent, ...this.#withDuration(modelStartedAt), + outcome: modelSignal.aborted || isAbortError(error) ? 'cancelled' : 'failed', ...(error instanceof Error ? { errorName: error.name } : {}), }); } @@ -477,8 +480,8 @@ export class Agent { const toolSignal = this.#deadlineSignal(signal, this.#toolTimeoutMs); const toolStartedAt = this.#now(); if (toolStartedAt !== undefined) { - this.#notifyExecution({ - type: 'tool.started', + this.#notifyObservation({ + type: 'tool.invocation.started', sessionId: this.#session.sessionId, turnId, stepId, @@ -490,6 +493,7 @@ export class Agent { }); } let result: ToolResult; + let toolInvocationFinished = false; try { result = await this.#tools.execute(call, { signal: toolSignal, @@ -497,6 +501,22 @@ export class Agent { }); } 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, @@ -504,8 +524,8 @@ export class Agent { }; } else { if (toolStartedAt !== undefined) { - this.#notifyExecution({ - type: toolSignal.aborted || isAbortError(error) ? 'tool.cancelled' : 'tool.failed', + this.#notifyObservation({ + type: 'tool.invocation.finished', sessionId: this.#session.sessionId, turnId, stepId, @@ -514,6 +534,7 @@ 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 } : {}), }); } @@ -522,9 +543,9 @@ export class Agent { } result = this.#boundToolResult(result); attempt.toolResults.set(call.id, result); - if (toolStartedAt !== undefined) { - this.#notifyExecution({ - type: 'tool.completed', + if (toolStartedAt !== undefined && !toolInvocationFinished) { + this.#notifyObservation({ + type: 'tool.invocation.finished', sessionId: this.#session.sessionId, turnId, stepId, @@ -533,7 +554,8 @@ export class Agent { toolCallId: call.id, toolName: call.name, ...this.#withDuration(toolStartedAt), - status: result.status, + termination: 'returned', + resultStatus: result.status, }); } // 提交工具调用完成事件 @@ -624,7 +646,7 @@ export class Agent { async #append(events: readonly SessionEvent[], signal: AbortSignal): Promise { const appended = await this.#session.append(events, { signal }); for (const envelope of appended) { - this.#notifyExecution({ type: 'session.event', envelope }); + this.#notifySessionEvent(envelope); } } @@ -639,13 +661,24 @@ export class Agent { } } - /** 向 observer 派发结构化执行事件(模型生成时的使用量、工具调用结果等),与持久化的会话日志一致。 */ - #notifyExecution(event: AgentExecutionEvent): void { + /** 向 observer 派发已经成功提交的会话事实。 */ + #notifySessionEvent(envelope: EventEnvelope): void { + for (const observer of this.#observers) { + try { + void Promise.resolve(observer.onSessionEvent?.(envelope)).catch(() => undefined); + } catch { + // 观察者只接收提交后的通知,其失败不能改变 Agent 生命周期。 + } + } + } + + /** 向 observer 派发不进入会话日志的模型与工具调用观测信号。 */ + #notifyObservation(observation: ExecutionObservation): void { for (const observer of this.#observers) { try { - void Promise.resolve(observer.onExecution?.(event)).catch(() => undefined); + void Promise.resolve(observer.onObservation?.(observation)).catch(() => undefined); } catch { - // 观察者只接收执行通知,其失败不能改变 Agent 生命周期。 + // 瞬态观测失败不能改变 Agent 生命周期。 } } } diff --git a/packages/hosts/web/src/session-registry.ts b/packages/hosts/web/src/session-registry.ts index 4042d7d..fdd34e8 100644 --- a/packages/hosts/web/src/session-registry.ts +++ b/packages/hosts/web/src/session-registry.ts @@ -295,12 +295,8 @@ export class SessionRegistry implements WebSessionController { const runtime = this.#factory.create( { onStream: (event) => events.publish('stream', event), - onExecution: (event) => - events.publish( - 'session', - event, - event.type === 'session.event' ? event.envelope.sequence : undefined, - ), + onSessionEvent: (envelope) => + events.publish('session', { type: 'session.event', envelope }, envelope.sequence), }, workspace, ); diff --git a/packages/kernel/src/agent.ts b/packages/kernel/src/agent.ts index 348f9c5..bd6ac82 100644 --- a/packages/kernel/src/agent.ts +++ b/packages/kernel/src/agent.ts @@ -1,14 +1,8 @@ import type { Clock, PromptAssembler } from './context/protocol.js'; -import type { - AttemptId, - ExecutionId, - MessageId, - SessionId, - StepId, - TurnId, -} from './identifiers.js'; -import type { ModelProvider, ModelStreamEvent } from './model.js'; -import type { EventEnvelope, SessionLog, TurnOutcome } from './session/protocol.js'; +import type { AttemptId, ExecutionId, MessageId, StepId, TurnId } from './identifiers.js'; +import type { ModelProvider } from './model.js'; +import type { AgentObserver } from './observability/protocol.js'; +import type { SessionLog, TurnOutcome } from './session/protocol.js'; import type { ToolCallId, ToolSet } from './tools/protocol.js'; /** 当前 Agent 状态。 */ @@ -24,84 +18,6 @@ export interface AgentIdGenerator { message(): MessageId; } -/** 观察 Agent 生命周期事件的回调。 */ -export interface AgentObserver { - onStream?(event: ModelStreamEvent): void | Promise; - onExecution?(event: AgentExecutionEvent): void | Promise; -} - -/** Agent 生命周期事件。 */ -export type AgentExecutionEvent = - | { readonly type: 'session.event'; readonly envelope: EventEnvelope } - | { - readonly type: 'model.started'; - readonly sessionId: SessionId; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly executionId?: ExecutionId; - readonly startedAt: Date; - } - | { - readonly type: 'model.completed' | 'model.failed' | 'model.cancelled'; - readonly sessionId: SessionId; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly executionId?: ExecutionId; - /** 时钟不可用时省略,不能影响已开始的观测生命周期。 */ - readonly durationMs?: number; - readonly errorName?: string; - } - | { - readonly type: 'tool.started'; - readonly sessionId: SessionId; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly executionId?: ExecutionId; - readonly toolCallId: ToolCallId; - readonly toolName: string; - readonly startedAt: Date; - } - | { - readonly type: 'tool.completed'; - readonly sessionId: SessionId; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly executionId?: ExecutionId; - readonly toolCallId: ToolCallId; - readonly toolName: string; - /** 时钟不可用时省略,不能影响已开始的观测生命周期。 */ - readonly durationMs?: number; - readonly status: 'success' | 'failure' | 'cancelled' | 'unknown'; - } - | { - readonly type: 'tool.failed' | 'tool.cancelled'; - readonly sessionId: SessionId; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly executionId?: ExecutionId; - readonly toolCallId: ToolCallId; - readonly toolName: string; - /** 时钟不可用时省略,不能影响已开始的观测生命周期。 */ - readonly durationMs?: number; - readonly errorName?: string; - } - | { - readonly type: 'model.usage'; - readonly sessionId: SessionId; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly executionId?: ExecutionId; - readonly usageSequence: number; - readonly inputTokensDelta: number; - readonly outputTokensDelta: number; - }; - /** 创建一个用于结算的 AbortSignal。 */ export interface SettlementSignalFactory { create(): AbortSignal; diff --git a/packages/kernel/src/events.ts b/packages/kernel/src/events.ts deleted file mode 100644 index 35c2e93..0000000 --- a/packages/kernel/src/events.ts +++ /dev/null @@ -1,299 +0,0 @@ -import type { ExecutionMetadata } from './agent.js'; -import type { - AttemptId, - ExecutionId, - MessageId, - SessionId, - StepId, - TurnId, -} from './identifiers.js'; -import type { - AssistantCompleted, - AssistantFailed, - AssistantInterrupted, - AssistantMessage, - UserMessage, -} from './model.js'; -import type { JsonValue, SerializedError } from './serialization.js'; -import type { - ToolCall, - ToolCallId, - ToolCancelled, - ToolFailure, - ToolSuccess, - ToolUnknown, -} from './tools/protocol.js'; - -/** - * Agent 一次运行中的事件。 - * 调度、attempt、step、turn、tool 都是 `AgentRuntimeEvent` 的子集。 - * 流式输出单独放在 `StreamEvent`,不进入这套事件。 - * - * 同一业务动作只保留一种事件。终态仍用不同的 `type` 区分。 - * 合并时保留原先会话事件和观测事件上的全部参数。 - */ -export type AgentRuntimeEvent = - | InboxEnqueuedEvent - | InboxRemovedEvent - | TurnStartedEvent - | UserMessageEvent - | StepStartedEvent - | AttemptStartedEvent - | AttemptCompletedEvent - | AttemptFailedEvent - | AttemptCancelledEvent - | AttemptInterruptedEvent - | AttemptUsageEvent - | ToolCallStartedEvent - | ToolCallCompletedEvent - | ToolCallFailedEvent - | ToolCallCancelledEvent - | ToolCallUnknownEvent - | StepCompletedEvent - | StepFailedEvent - | StepCancelledEvent - | StepInterruptedEvent - | TurnCompletedEvent - | TurnFailedEvent - | TurnCancelledEvent - | TurnInterruptedEvent; - -/** 用户输入已进入 inbox,等待 FIFO 领取。 */ -export interface InboxEnqueuedEvent { - readonly type: 'inbox.enqueued'; - readonly message: UserMessage; -} - -/** 用户输入在被领取前离开 inbox。 */ -export interface InboxRemovedEvent { - readonly type: 'inbox.removed'; - readonly messageId: MessageId; - readonly reason?: string; -} - -interface TurnScope { - readonly turnId: TurnId; -} - -/** turn 已开始。 */ -export interface TurnStartedEvent extends TurnScope { - readonly type: 'turn.started'; - readonly execution?: ExecutionMetadata; -} - -/** 用户消息已写入当前 turn。 */ -export interface UserMessageEvent { - readonly type: 'user.message'; - readonly message: UserMessage; -} - -interface TurnEndScope extends TurnScope { - readonly error?: SerializedError; -} - -/** turn 正常闭合。 */ -export interface TurnCompletedEvent extends TurnEndScope { - readonly type: 'turn.completed'; - readonly outcome: 'completed'; -} - -/** turn 因错误闭合。 */ -export interface TurnFailedEvent extends TurnEndScope { - readonly type: 'turn.failed'; - readonly outcome: 'failed'; -} - -/** turn 因调用方取消而闭合。 */ -export interface TurnCancelledEvent extends TurnEndScope { - readonly type: 'turn.cancelled'; - readonly outcome: 'cancelled'; -} - -/** turn 因进程中断而闭合。 */ -export interface TurnInterruptedEvent extends TurnEndScope { - readonly type: 'turn.interrupted'; - readonly outcome: 'interrupted'; -} - -interface StepScope { - readonly turnId: TurnId; - readonly stepId: StepId; -} - -/** step 已开始。 */ -export interface StepStartedEvent extends StepScope { - readonly type: 'step.started'; -} - -/** step 正常闭合。 */ -export interface StepCompletedEvent extends StepScope { - readonly type: 'step.completed'; - readonly outcome: 'completed'; -} - -/** step 因错误闭合。 */ -export interface StepFailedEvent extends StepScope { - readonly type: 'step.failed'; - readonly outcome: 'failed'; - readonly error: SerializedError; -} - -/** step 因调用方取消而闭合。 */ -export interface StepCancelledEvent extends StepScope { - readonly type: 'step.cancelled'; - readonly outcome: 'cancelled'; - readonly error: SerializedError; -} - -/** step 因进程中断而闭合。 */ -export interface StepInterruptedEvent extends StepScope { - readonly type: 'step.interrupted'; - readonly outcome: 'interrupted'; - readonly reason: string; -} - -/** 一次 attempt 或工具调用的关联身份。 */ -interface ExecutionScope { - readonly sessionId: SessionId; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly executionId?: ExecutionId; -} - -/** 时钟中途不可用时省略耗时。 */ -interface TimedExecution { - readonly durationMs?: number; -} - -/** attempt 开始,对应原先的 `model.started`。 */ -export interface AttemptStartedEvent extends ExecutionScope { - readonly type: 'attempt.started'; - readonly startedAt: Date; -} - -/** attempt 正常结束,对应原先的 `model.completed` 与成功的助手结算。 */ -export interface AttemptCompletedEvent extends ExecutionScope, TimedExecution { - readonly type: 'attempt.completed'; - readonly errorName?: string; - readonly message: AssistantMessage; - readonly settlement: AssistantCompleted; -} - -/** attempt 失败,对应原先的 `model.failed` 与失败的助手结算。 */ -export interface AttemptFailedEvent extends ExecutionScope, TimedExecution { - readonly type: 'attempt.failed'; - readonly errorName?: string; - readonly partialText: string; - readonly observedToolCalls: readonly ToolCall[]; - readonly error: SerializedError; - readonly settlement: AssistantFailed; -} - -/** attempt 被调用方取消,对应原先的 `model.cancelled`。 */ -export interface AttemptCancelledEvent extends ExecutionScope, TimedExecution { - readonly type: 'attempt.cancelled'; - readonly errorName?: string; - readonly partialText: string; - readonly observedToolCalls: readonly ToolCall[]; - readonly reason: string; - readonly settlement: AssistantInterrupted; -} - -/** attempt 因进程中断而无法按成功或失败闭合。 */ -export interface AttemptInterruptedEvent { - readonly type: 'attempt.interrupted'; - readonly turnId: TurnId; - readonly stepId: StepId; - readonly attemptId: AttemptId; - readonly partialText: string; - readonly observedToolCalls: readonly ToolCall[]; - readonly reason: string; - readonly settlement: AssistantInterrupted; -} - -/** 一次 attempt 内可累加的 token 增量。它不是流式 `usage`。 */ -export interface AttemptUsageEvent extends ExecutionScope { - readonly type: 'model.usage'; - readonly usageSequence: number; - readonly inputTokensDelta: number; - readonly outputTokensDelta: number; -} - -interface ToolScope extends ExecutionScope { - readonly toolCallId: ToolCallId; - readonly toolName: string; -} - -/** 开始执行一次工具调用,对应原先的 `tool.started`。 */ -export interface ToolCallStartedEvent extends ToolScope { - readonly type: 'tool.started'; - readonly startedAt: Date; -} - -/** 工具调用成功。保留会话中的 `result` 和观测中的 `status`。 */ -export interface ToolCallCompletedEvent extends ToolScope, TimedExecution { - readonly type: 'tool.completed'; - readonly output: JsonValue; - readonly result: ToolSuccess; - readonly status: 'success'; -} - -/** 工具调用失败。 */ -export interface ToolCallFailedEvent extends ToolScope, TimedExecution { - readonly type: 'tool.failed'; - readonly errorName?: string; - readonly error: SerializedError; - readonly result: ToolFailure; - readonly status: 'failure'; -} - -/** 工具调用被取消。 */ -export interface ToolCallCancelledEvent extends ToolScope, TimedExecution { - readonly type: 'tool.cancelled'; - readonly errorName?: string; - readonly reason: string; - readonly result: ToolCancelled; - readonly status: 'cancelled'; -} - -/** 无法确认工具是否执行及其外部副作用。 */ -export interface ToolCallUnknownEvent extends ToolScope, TimedExecution { - readonly type: 'tool.unknown'; - readonly reason: string; - readonly result: ToolUnknown; - readonly status: 'unknown'; -} - -/** - * 持久化时包裹一次运行事件的统一格式。 - * sequence 在会话内严格单调;timestamp 为 ISO-8601 字符串。 - */ -export interface EventEnvelope { - readonly version: 1; - readonly sessionId: SessionId; - readonly sequence: number; - readonly timestamp: string; - readonly event: AgentRuntimeEvent; -} - -/** 模型输出的文本增量。 */ -export interface TextDeltaEvent { - readonly type: 'text.delta'; - readonly text: string; -} - -/** 模型输出中观察到的一次工具调用。此时尚未执行工具。 */ -export interface ToolCallStreamEvent { - readonly type: 'tool.call'; - readonly call: ToolCall; -} - -/** 模型报告的累计 token 用量。 */ -export interface ModelUsageStreamEvent { - readonly type: 'usage'; - readonly inputTokens: number; - readonly outputTokens: number; -} - -export type StreamEvent = TextDeltaEvent | ToolCallStreamEvent | ModelUsageStreamEvent; diff --git a/packages/kernel/src/index.ts b/packages/kernel/src/index.ts index 20c85d9..f929c6e 100644 --- a/packages/kernel/src/index.ts +++ b/packages/kernel/src/index.ts @@ -6,6 +6,7 @@ export * from './exhaustiveness.js'; export * from './identifiers.js'; export * from './memory/protocol.js'; export * from './model.js'; +export * from './observability/protocol.js'; export * from './serialization.js'; export { analyzeSessionLifecycle } from './session/lifecycle.js'; export * from './session/protocol.js'; diff --git a/packages/kernel/src/observability/protocol.ts b/packages/kernel/src/observability/protocol.ts new file mode 100644 index 0000000..5b7e03b --- /dev/null +++ b/packages/kernel/src/observability/protocol.ts @@ -0,0 +1,74 @@ +import type { AttemptId, ExecutionId, SessionId, StepId, TurnId } from '../identifiers.js'; +import type { ModelStreamEvent } from '../model.js'; +import type { EventEnvelope } from '../session/protocol.js'; +import type { ToolCallId, ToolResult } from '../tools/protocol.js'; + +/** 观察 Agent 已提交的会话事实与瞬态执行信号。 */ +export interface AgentObserver { + onStream?(event: ModelStreamEvent): void | Promise; + onSessionEvent?(envelope: EventEnvelope): void | Promise; + onObservation?(observation: ExecutionObservation): void | Promise; +} + +/** 不进入会话日志的模型与工具调用观测信号。 */ +export type ExecutionObservation = + | ModelInvocationStarted + | ModelInvocationFinished + | ModelInvocationUsage + | ToolInvocationStarted + | ToolInvocationFinished; + +interface InvocationScope { + readonly sessionId: SessionId; + readonly turnId: TurnId; + readonly stepId: StepId; + readonly attemptId: AttemptId; + readonly executionId?: ExecutionId; +} + +interface FinishedInvocation extends InvocationScope { + /** 时钟中途不可用时省略,不能影响已开始的观测生命周期。 */ + readonly durationMs?: number; +} + +interface ModelInvocationStarted extends InvocationScope { + readonly type: 'model.invocation.started'; + readonly startedAt: Date; +} + +interface ModelInvocationFinished extends FinishedInvocation { + readonly type: 'model.invocation.finished'; + readonly outcome: 'completed' | 'failed' | 'cancelled'; + readonly errorName?: string; +} + +interface ModelInvocationUsage extends InvocationScope { + readonly type: 'model.invocation.usage'; + readonly usageSequence: number; + readonly inputTokensDelta: number; + readonly outputTokensDelta: number; +} + +interface ToolInvocationScope extends InvocationScope { + readonly toolCallId: ToolCallId; + readonly toolName: string; +} + +interface ToolInvocationStarted extends ToolInvocationScope { + readonly type: 'tool.invocation.started'; + readonly startedAt: Date; +} + +type ToolInvocationFinished = + | (FinishedInvocation & + ToolInvocationScope & { + readonly type: 'tool.invocation.finished'; + readonly termination: 'returned'; + readonly resultStatus: ToolResult['status']; + }) + | (FinishedInvocation & + ToolInvocationScope & { + readonly type: 'tool.invocation.finished'; + readonly termination: 'failed' | 'cancelled'; + readonly errorName?: string; + }); diff --git a/packages/test-support/src/demo.ts b/packages/test-support/src/demo.ts index aa43937..c341f31 100644 --- a/packages/test-support/src/demo.ts +++ b/packages/test-support/src/demo.ts @@ -54,10 +54,8 @@ export async function runDemo(): Promise { clock: { now: () => new Date('2026-10-13T00:00:00.000Z') }, observers: [ { - onExecution: (event) => { - if (event.type !== 'session.event') { - executionEvents.push(event.type); - } + onObservation: (observation) => { + executionEvents.push(observation.type); }, }, ], diff --git a/packages/test-support/test/agent.test.ts b/packages/test-support/test/agent.test.ts index 0ecce5e..8eb6f38 100644 --- a/packages/test-support/test/agent.test.ts +++ b/packages/test-support/test/agent.test.ts @@ -6,13 +6,13 @@ import { Agent } from '@agent-harness/core'; import { AgentBusyError, AgentClosedError, - type AgentExecutionEvent, AgentIdCollisionError, analyzeSessionLifecycle, brandId, DirectToolSet, type EventEnvelope, type ExecutionId, + type ExecutionObservation, type MessageId, type SessionId, type SessionLog, @@ -1150,6 +1150,7 @@ describe('Agent', () => { it('工具调用到达独立 deadline 后记录取消结果', async () => { const callId = brandId<'ToolCallId'>('call-timeout') as ToolCallId; const session = new InMemorySessionLog(sessionId); + const observed: ExecutionObservation[] = []; const model = new ScriptedModelProvider([ { events: [{ type: 'tool.call', call: { id: callId, name: 'wait', input: {} } }], @@ -1174,6 +1175,8 @@ describe('Agent', () => { toolTimeoutMs: 20, settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), + clock: { now: () => new Date(0) }, + observers: [{ onObservation: (observation) => observed.push(observation) }], }); await expect(agent.run(user())).resolves.toMatchObject({ content: '继续执行' }); @@ -1184,6 +1187,14 @@ describe('Agent', () => { 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: 'TimeoutError', + }), + ]); }); it('普通会话 I/O 响应 run signal,并用一个独立信号提交 terminal batch', async () => { @@ -1356,8 +1367,11 @@ describe('Agent', () => { onStream() { throw new Error('sync observer failed'); }, - onExecution() { - throw new Error('sync execution observer failed'); + onSessionEvent() { + throw new Error('sync session observer failed'); + }, + onObservation() { + throw new Error('sync observation observer failed'); }, }, { @@ -1365,9 +1379,13 @@ describe('Agent', () => { seen.push(event.type); throw new Error('async observer failed'); }, - async onExecution(event) { - seen.push(event.type); - throw new Error('async execution observer failed'); + async onSessionEvent(envelope) { + seen.push(envelope.event.type); + throw new Error('async session observer failed'); + }, + async onObservation(observation) { + seen.push(observation.type); + throw new Error('async observation observer failed'); }, }, ], @@ -1377,12 +1395,12 @@ describe('Agent', () => { expect(response.content).toBe('ok'); expect(seen).toContain('text.delta'); - expect(seen).toContain('session.event'); + expect(seen).toContain('turn.started'); expectClosed(await readEvents(session)); }); it('以注入时钟成对报告模型执行的开始与终态', async () => { - const observed: AgentExecutionEvent[] = []; + const observed: ExecutionObservation[] = []; const timestamps = [new Date('2026-10-13T00:00:00.000Z'), new Date('2026-10-13T00:00:00.025Z')]; const agent = new Agent({ model: new ScriptedModelProvider([{ events: [{ type: 'text.delta', text: 'ok' }] }]), @@ -1393,22 +1411,26 @@ describe('Agent', () => { settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), clock: { now: () => timestamps.shift() ?? new Date('invalid') }, - observers: [{ onExecution: (event) => observed.push(event) }], + observers: [{ onObservation: (observation) => observed.push(observation) }], }); await agent.run(user()); expect(observed.filter((event) => event.type.startsWith('model.'))).toEqual([ expect.objectContaining({ - type: 'model.started', + type: 'model.invocation.started', startedAt: new Date('2026-10-13T00:00:00.000Z'), }), - expect.objectContaining({ type: 'model.completed', durationMs: 25 }), + expect.objectContaining({ + type: 'model.invocation.finished', + outcome: 'completed', + durationMs: 25, + }), ]); }); it('模型失败或取消时为已开始的调用报告对应终态', async () => { - const failedObserved: AgentExecutionEvent[] = []; + const failedObserved: ExecutionObservation[] = []; const failedAgent = new Agent({ model: new ScriptedModelProvider([{ events: [], error: new Error('model failed') }]), session: new InMemorySessionLog(sessionId), @@ -1418,16 +1440,20 @@ describe('Agent', () => { settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), clock: { now: () => new Date() }, - observers: [{ onExecution: (event) => failedObserved.push(event) }], + observers: [{ onObservation: (observation) => failedObserved.push(observation) }], }); await expect(failedAgent.run(user())).rejects.toThrow('model failed'); expect(failedObserved.filter((event) => event.type.startsWith('model.'))).toEqual([ - expect.objectContaining({ type: 'model.started' }), - expect.objectContaining({ type: 'model.failed', errorName: 'Error' }), + expect.objectContaining({ type: 'model.invocation.started' }), + expect.objectContaining({ + type: 'model.invocation.finished', + outcome: 'failed', + errorName: 'Error', + }), ]); - const cancelledObserved: AgentExecutionEvent[] = []; + const cancelledObserved: ExecutionObservation[] = []; let prepareStarted!: () => void; const started = new Promise((resolve) => { prepareStarted = resolve; @@ -1453,7 +1479,7 @@ describe('Agent', () => { settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), clock: { now: () => new Date() }, - observers: [{ onExecution: (event) => cancelledObserved.push(event) }], + observers: [{ onObservation: (observation) => cancelledObserved.push(observation) }], }); const controller = new AbortController(); const run = cancelledAgent.run(user(), controller.signal); @@ -1462,14 +1488,18 @@ describe('Agent', () => { await expect(run).rejects.toThrow('cancel model'); expect(cancelledObserved.filter((event) => event.type.startsWith('model.'))).toEqual([ - expect.objectContaining({ type: 'model.started' }), - expect.objectContaining({ type: 'model.cancelled', errorName: 'Error' }), + expect.objectContaining({ type: 'model.invocation.started' }), + expect.objectContaining({ + type: 'model.invocation.finished', + outcome: 'cancelled', + errorName: 'Error', + }), ]); }); it('工具抛错或取消时报告对应终态,终态 ToolResult 仍报告完成', async () => { const callId = brandId<'ToolCallId'>('call-1') as ToolCallId; - const createAgent = (execute: ScriptedToolSet['execute'], observed: AgentExecutionEvent[]) => + const createAgent = (execute: ScriptedToolSet['execute'], observed: ExecutionObservation[]) => new Agent({ model: new ScriptedModelProvider([ { events: [{ type: 'tool.call', call: { id: callId, name: 'lookup', input: {} } }] }, @@ -1485,20 +1515,24 @@ describe('Agent', () => { settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), clock: { now: () => new Date() }, - observers: [{ onExecution: (event) => observed.push(event) }], + observers: [{ onObservation: (observation) => observed.push(observation) }], }); - const thrownObserved: AgentExecutionEvent[] = []; + const thrownObserved: ExecutionObservation[] = []; const thrownAgent = createAgent(async () => { throw new Error('tool failed'); }, thrownObserved); await expect(thrownAgent.run(user())).rejects.toThrow('tool failed'); expect(thrownObserved.filter((event) => event.type.startsWith('tool.'))).toEqual([ - expect.objectContaining({ type: 'tool.started' }), - expect.objectContaining({ type: 'tool.failed', errorName: 'Error' }), + expect.objectContaining({ type: 'tool.invocation.started' }), + expect.objectContaining({ + type: 'tool.invocation.finished', + termination: 'failed', + errorName: 'Error', + }), ]); - const cancelledThrownObserved: AgentExecutionEvent[] = []; + const cancelledThrownObserved: ExecutionObservation[] = []; let executionStarted!: () => void; const executionStartedPromise = new Promise((resolve) => { executionStarted = resolve; @@ -1517,19 +1551,27 @@ describe('Agent', () => { controller.abort(new Error('cancel tool')); await expect(cancelledRun).rejects.toThrow('cancel tool'); expect(cancelledThrownObserved.filter((event) => event.type.startsWith('tool.'))).toEqual([ - expect.objectContaining({ type: 'tool.started' }), - expect.objectContaining({ type: 'tool.cancelled', errorName: 'Error' }), + expect.objectContaining({ type: 'tool.invocation.started' }), + expect.objectContaining({ + type: 'tool.invocation.finished', + termination: 'cancelled', + errorName: 'Error', + }), ]); - const resultObserved: AgentExecutionEvent[] = []; + const resultObserved: ExecutionObservation[] = []; const resultAgent = createAgent( async () => ({ status: 'cancelled', callId, reason: 'tool cancelled' }), resultObserved, ); await resultAgent.run(user()); expect(resultObserved.filter((event) => event.type.startsWith('tool.'))).toEqual([ - expect.objectContaining({ type: 'tool.started' }), - expect.objectContaining({ type: 'tool.completed', status: 'cancelled' }), + expect.objectContaining({ type: 'tool.invocation.started' }), + expect.objectContaining({ + type: 'tool.invocation.finished', + termination: 'returned', + resultStatus: 'cancelled', + }), ]); }); @@ -1556,7 +1598,8 @@ describe('Agent', () => { it('在提交会话事件后发送生命周期通知,并关联模型用量', async () => { const session = new InMemorySessionLog(sessionId); - const observed: AgentExecutionEvent[] = []; + const sessionEvents: EventEnvelope[] = []; + const observed: ExecutionObservation[] = []; const agent = new Agent({ model: new ScriptedModelProvider([ { @@ -1573,29 +1616,21 @@ describe('Agent', () => { maxSteps: 4, settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), - observers: [{ onExecution: (event) => observed.push(event) }], + observers: [ + { + onSessionEvent: (envelope) => sessionEvents.push(envelope), + onObservation: (observation) => observed.push(observation), + }, + ], }); await agent.run(user()); - expect(observed.map((event) => event.type)).toEqual([ - 'session.event', - 'session.event', - 'session.event', - 'model.usage', - 'model.usage', - 'session.event', - 'session.event', - 'session.event', + expect(observed.map((observation) => observation.type)).toEqual([ + 'model.invocation.usage', + 'model.invocation.usage', ]); - expect( - observed - .filter( - (event): event is Extract => - event.type === 'session.event', - ) - .map((event) => event.envelope.event.type), - ).toEqual([ + expect(sessionEvents.map((envelope) => envelope.event.type)).toEqual([ 'turn.started', 'user.message', 'step.started', @@ -1603,8 +1638,8 @@ describe('Agent', () => { 'step.ended', 'turn.ended', ]); - expect(observed[3]).toMatchObject({ - type: 'model.usage', + expect(observed[0]).toMatchObject({ + type: 'model.invocation.usage', sessionId: 'session-1', turnId: 'turn-1', stepId: 'step-2', @@ -1613,8 +1648,8 @@ describe('Agent', () => { inputTokensDelta: 12, outputTokensDelta: 3, }); - expect(observed[4]).toMatchObject({ - type: 'model.usage', + expect(observed[1]).toMatchObject({ + type: 'model.invocation.usage', sessionId: 'session-1', turnId: 'turn-1', stepId: 'step-2', @@ -1623,20 +1658,13 @@ describe('Agent', () => { inputTokensDelta: 0, outputTokensDelta: 2, }); - expect( - observed - .filter( - (event): event is Extract => - event.type === 'session.event', - ) - .map((event) => event.envelope.sequence), - ).toEqual([1, 2, 3, 4, 5, 6]); + expect(sessionEvents.map((envelope) => envelope.sequence)).toEqual([1, 2, 3, 4, 5, 6]); }); it('模型与工具观测事件携带已冻结的 executionId', async () => { const callId = brandId<'ToolCallId'>('call-observed-execution') as ToolCallId; const executionId = brandId<'ExecutionId'>('execution-observed') as ExecutionId; - const observed: AgentExecutionEvent[] = []; + const observed: ExecutionObservation[] = []; const agent = new Agent({ model: new ScriptedModelProvider([ { @@ -1651,7 +1679,7 @@ describe('Agent', () => { settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), clock: { now: () => new Date(0) }, - observers: [{ onExecution: (event) => observed.push(event) }], + observers: [{ onObservation: (observation) => observed.push(observation) }], }); await agent.run(user(), { @@ -1662,16 +1690,15 @@ describe('Agent', () => { }, }); - const runtimeEvents = observed.filter((event) => event.type !== 'session.event'); - expect(runtimeEvents).not.toHaveLength(0); - expect(runtimeEvents).toEqual( - expect.arrayContaining([expect.objectContaining({ type: 'model.started' })]), + expect(observed).not.toHaveLength(0); + expect(observed).toEqual( + expect.arrayContaining([expect.objectContaining({ type: 'model.invocation.started' })]), ); - expect(runtimeEvents.every((event) => event.executionId === executionId)).toBe(true); + expect(observed.every((observation) => observation.executionId === executionId)).toBe(true); }); it('以 sessionId 区分相同 attempt 标识符的不同会话用量', async () => { - const observed: AgentExecutionEvent[] = []; + const observed: ExecutionObservation[] = []; const run = async (id: SessionId): Promise => { const agent = new Agent({ model: new ScriptedModelProvider([ @@ -1685,7 +1712,7 @@ describe('Agent', () => { maxSteps: 4, settlementSignals: new TestSettlementSignals(), prompts: defaultPrompts(), - observers: [{ onExecution: (event) => observed.push(event) }], + observers: [{ onObservation: (observation) => observed.push(observation) }], }); await agent.run(user()); }; @@ -1695,14 +1722,16 @@ describe('Agent', () => { await run(secondSessionId); const usage = observed.filter( - (event): event is Extract => - event.type === 'model.usage', + ( + observation, + ): observation is Extract => + observation.type === 'model.invocation.usage', ); expect(usage).toHaveLength(2); expect(usage.map((event) => event.sessionId)).toEqual([sessionId, secondSessionId]); expect(usage.map(({ sessionId: _sessionId, ...identity }) => identity)).toEqual([ { - type: 'model.usage', + type: 'model.invocation.usage', turnId: 'turn-1', stepId: 'step-2', attemptId: 'attempt-3', @@ -1711,7 +1740,7 @@ describe('Agent', () => { outputTokensDelta: 1, }, { - type: 'model.usage', + type: 'model.invocation.usage', turnId: 'turn-1', stepId: 'step-2', attemptId: 'attempt-3', diff --git a/packages/test-support/test/demo.test.ts b/packages/test-support/test/demo.test.ts index 09767b9..c7eb78d 100644 --- a/packages/test-support/test/demo.test.ts +++ b/packages/test-support/test/demo.test.ts @@ -7,13 +7,13 @@ describe('public demo', () => { response: 'Agent Harness Lab 0.1.0 演示完成:会话、工具和续步均已结算。', toolInput: { version: '0.1.0' }, executionEvents: [ - 'model.started', - 'model.completed', - 'tool.started', - 'tool.completed', - 'model.started', - 'model.usage', - 'model.completed', + 'model.invocation.started', + 'model.invocation.finished', + 'tool.invocation.started', + 'tool.invocation.finished', + 'model.invocation.started', + 'model.invocation.usage', + 'model.invocation.finished', ], eventTypes: [ 'turn.started',