Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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、文件系统或进程。

Expand Down
Original file line number Diff line number Diff line change
@@ -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` 包装会话信封并复用持久事件名称:这会让观察者无法从名称判断事件是否已经持久提交。
79 changes: 56 additions & 23 deletions packages/core/src/agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
import {
AgentBusyError,
AgentClosedError,
type AgentExecutionEvent,
type AgentIdGenerator,
type AgentObserver,
type AgentOptions,
Expand All @@ -11,8 +10,10 @@ import {
type AssistantMessage,
assertNever,
type Clock,
type EventEnvelope,
type ExecutionId,
type ExecutionMetadata,
type ExecutionObservation,
isAbortError,
type ModelProvider,
type ModelStreamEvent,
Expand All @@ -21,8 +22,8 @@ import {
type SessionLog,
SettlementPersistenceError,
type SettlementSignalFactory,
type ToolSet,
type ToolResult,
type ToolSet,
type TurnId,
type UserMessage,
} from '@agent-harness/kernel';
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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 } : {}),
});
}
Expand Down Expand Up @@ -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,
Expand All @@ -490,22 +493,39 @@ export class Agent {
});
}
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.#notifyExecution({
type: toolSignal.aborted || isAbortError(error) ? 'tool.cancelled' : 'tool.failed',
this.#notifyObservation({
type: 'tool.invocation.finished',
sessionId: this.#session.sessionId,
turnId,
stepId,
Expand All @@ -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 } : {}),
});
}
Expand All @@ -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,
Expand All @@ -533,7 +554,8 @@ export class Agent {
toolCallId: call.id,
toolName: call.name,
...this.#withDuration(toolStartedAt),
status: result.status,
termination: 'returned',
resultStatus: result.status,
});
}
// 提交工具调用完成事件
Expand Down Expand Up @@ -624,7 +646,7 @@ export class Agent {
async #append(events: readonly SessionEvent[], signal: AbortSignal): Promise<void> {
const appended = await this.#session.append(events, { signal });
for (const envelope of appended) {
this.#notifyExecution({ type: 'session.event', envelope });
this.#notifySessionEvent(envelope);
}
}

Expand All @@ -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 生命周期。
}
}
}
Expand Down
8 changes: 2 additions & 6 deletions packages/hosts/web/src/session-registry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
);
Expand Down
92 changes: 4 additions & 88 deletions packages/kernel/src/agent.ts
Original file line number Diff line number Diff line change
@@ -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 状态。 */
Expand All @@ -24,84 +18,6 @@ export interface AgentIdGenerator {
message(): MessageId;
}

/** 观察 Agent 生命周期事件的回调。 */
export interface AgentObserver {
onStream?(event: ModelStreamEvent): void | Promise<void>;
onExecution?(event: AgentExecutionEvent): void | Promise<void>;
}

/** 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;
Expand Down
Loading
Loading