From e4df73937333661896eb354734edc584edd6f926 Mon Sep 17 00:00:00 2001 From: hanpj <2568545091@qq.com> Date: Fri, 16 Oct 2026 18:40:20 +0800 Subject: [PATCH 1/2] refactor: clarify agent loop names and define runtime events. Co-authored-by: Cursor --- packages/core/src/agent.ts | 64 +++---- packages/core/src/inbox.ts | 4 +- packages/core/src/lifecycle.ts | 2 +- packages/kernel/src/events.ts | 300 +++++++++++++++++++++++++++++++++ 4 files changed, 338 insertions(+), 32 deletions(-) create mode 100644 packages/kernel/src/events.ts diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index eeae44a..783dc3c 100644 --- a/packages/core/src/agent.ts +++ b/packages/core/src/agent.ts @@ -53,11 +53,11 @@ export class Agent { readonly #model: ModelProvider; readonly #session: SessionLog; readonly #tools: ToolSet; - readonly #ids: AgentIdGenerator; + readonly #idGenerator: AgentIdGenerator; readonly #maxSteps: number; - readonly #settlementSignals: SettlementSignalFactory; + readonly #settlementSignalFactory: SettlementSignalFactory; readonly #clock: Clock | undefined; - readonly #prompts: PromptAssembler; + readonly #promptAssembler: PromptAssembler; readonly #runTimeoutMs: number | undefined; readonly #modelAttemptTimeoutMs: number; readonly #toolTimeoutMs: number; @@ -65,12 +65,12 @@ export class Agent { readonly #maxToolResultBytes: number; readonly #maxQueuedRuns: number; readonly #observers: readonly AgentObserver[]; - readonly #reservations: AgentIdReservations; // 标识符预留池，保证同一个 Agent 里，turn / step / attempt / message / toolCall 的 ID 不重复 + readonly #idReservations: AgentIdReservations; readonly #inbox = new AgentInbox(); #phase: AgentPhase = { kind: 'idle' }; - #controller: AbortController | undefined; + #activeRunAbortController: AbortController | undefined; #accepting = true; - #draining: Promise | undefined; // 正在 draining 的 promise，用于保证只有一个 drain 在执行 + #inboxDrain: Promise | undefined; constructor(options: AgentOptions) { // 参数校验 @@ -94,11 +94,11 @@ export class Agent { this.#model = options.model; this.#session = options.session; this.#tools = options.tools; - this.#ids = options.ids; + this.#idGenerator = options.ids; this.#maxSteps = options.maxSteps; - this.#settlementSignals = options.settlementSignals; + this.#settlementSignalFactory = options.settlementSignals; this.#clock = options.clock; - this.#prompts = options.prompts; + this.#promptAssembler = options.prompts; this.#runTimeoutMs = options.runTimeoutMs; this.#modelAttemptTimeoutMs = options.modelAttemptTimeoutMs ?? DEFAULT_MODEL_ATTEMPT_TIMEOUT_MS; this.#toolTimeoutMs = options.toolTimeoutMs ?? DEFAULT_TOOL_TIMEOUT_MS; @@ -106,7 +106,7 @@ export class Agent { this.#maxToolResultBytes = options.maxToolResultBytes ?? DEFAULT_MAX_TOOL_RESULT_BYTES; this.#maxQueuedRuns = options.maxQueuedRuns ?? DEFAULT_MAX_QUEUED_RUNS; this.#observers = options.observers ?? []; - this.#reservations = new AgentIdReservations(options.reservedIds); + this.#idReservations = new AgentIdReservations(options.reservedIds); } get phase(): AgentPhase { @@ -115,7 +115,7 @@ export class Agent { /** 取消当前活动 run；不影响是否继续接纳新的 run。 */ cancel(reason: unknown = new Error('agent run cancelled')): void { - this.#controller?.abort(reason); + this.#activeRunAbortController?.abort(reason); } /** 永久停止接纳新 run，并取消当前活动。重复调用安全。 */ @@ -128,12 +128,12 @@ export class Agent { /** 等待当前 drain 完成，其中包含活动 turn 与已接纳的排队输入。 */ async whenIdle(): Promise { for (;;) { - const draining = this.#draining; - if (draining === undefined) { + const inboxDrain = this.#inboxDrain; + if (inboxDrain === undefined) { return; } - await draining; - if (this.#draining === undefined) { + await inboxDrain; + if (this.#inboxDrain === undefined) { return; } } @@ -186,11 +186,11 @@ export class Agent { /** 确保只有一个 FIFO drain；单个输入失败不会阻断后续输入。 */ #ensureInboxDraining(): void { - if (this.#draining !== undefined) { + if (this.#inboxDrain !== undefined) { return; } - this.#draining = this.#drainInbox().finally(() => { - this.#draining = undefined; + this.#inboxDrain = this.#drainInbox().finally(() => { + this.#inboxDrain = undefined; if (this.#inbox.hasPending) { this.#ensureInboxDraining(); } @@ -200,7 +200,7 @@ export class Agent { /** 串行领取已接纳输入，并仅结算各自调用，不让单项失败中断后续领取。 */ async #drainInbox(): Promise { for (;;) { - const queued = this.#inbox.claim(); + const queued = this.#inbox.claim(); // 领取最早尚未执行的输入 if (queued === undefined) { return; } @@ -225,12 +225,12 @@ export class Agent { isAbortSignal(options) ? undefined : options.execution, ); const turn: ActiveTurn = { - turnId: this.#reservations.turn(this.#ids.turn()), + turnId: this.#idReservations.turn(this.#idGenerator.turn()), started: false, ended: false, }; const controller = new AbortController(); - this.#controller = controller; + this.#activeRunAbortController = controller; const timeoutSignal = this.#runTimeoutMs === undefined ? undefined : AbortSignal.timeout(this.#runTimeoutMs); const signal = AbortSignal.any([ @@ -255,7 +255,7 @@ export class Agent { turn.started = true; // 提交用户消息事件 - this.#reservations.message(message.id); + this.#idReservations.message(message.id); await this.#append([{ type: 'user.message', message }], signal); // 循环执行 step，直至无工具调用或触及步数上限 @@ -263,7 +263,7 @@ export class Agent { throwIfAborted(signal); // 创建 step 生命周期 - const stepId = this.#reservations.step(this.#ids.step()); + const stepId = this.#idReservations.step(this.#idGenerator.step()); const step = createStep(stepId); turn.activeStep = step; this.#phase = { kind: 'running', turnId: turn.turnId, stepId }; @@ -300,7 +300,7 @@ export class Agent { // 如果 terminal settlement 事件不为空，则提交结算事件 if (terminalEvents.length > 0) { try { - const settlementSignal = this.#settlementSignals.create(); + const settlementSignal = this.#settlementSignalFactory.create(); await this.#append(terminalEvents, settlementSignal); } catch (persistenceError) { throw new SettlementPersistenceError(error, persistenceError, outcome); @@ -308,7 +308,7 @@ export class Agent { } throw error; } finally { - this.#controller = undefined; + this.#activeRunAbortController = undefined; this.#phase = { kind: 'idle' }; } } @@ -332,7 +332,11 @@ export class Agent { // 组装请求 const request = await waitForSignal( - this.#prompts.assemble(await this.#session.read({ signal }), this.#tools.list(), signal), + this.#promptAssembler.assemble( + await this.#session.read({ signal }), + this.#tools.list(), + signal, + ), signal, ); @@ -341,7 +345,7 @@ export class Agent { const prepared = await this.#model.prepare(modelSignal); throwIfAborted(modelSignal); const attempt: Attempt = { - attemptId: this.#reservations.attempt(this.#ids.attempt()), + attemptId: this.#idReservations.attempt(this.#idGenerator.attempt()), partialText: '', observedToolCalls: [], settlementPersisted: false, @@ -354,6 +358,8 @@ export class Agent { let outputTokens = 0; let usageSequence = 0; const modelStartedAt = this.#now(); + + // 提交模型开始事件 if (modelStartedAt !== undefined) { this.#notifyExecution({ type: 'model.started', @@ -383,7 +389,7 @@ export class Agent { case 'tool.call': outputBytes += utf8ByteLength(JSON.stringify(event.call)); this.#assertOutputBudget(outputBytes); - this.#reservations.toolCall(event.call.id); + this.#idReservations.toolCall(event.call.id); attempt.observedToolCalls.push(event.call); break; @@ -439,7 +445,7 @@ export class Agent { // 构建 assistant 消息 const assistantMessage: AssistantMessage = { - id: this.#reservations.message(this.#ids.message()), + id: this.#idReservations.message(this.#idGenerator.message()), role: 'assistant', content: attempt.partialText, toolCalls: attempt.observedToolCalls, diff --git a/packages/core/src/inbox.ts b/packages/core/src/inbox.ts index 4c24d4f..2cd1dd1 100644 --- a/packages/core/src/inbox.ts +++ b/packages/core/src/inbox.ts @@ -4,8 +4,8 @@ import type { AgentRunOptions, AssistantMessage, UserMessage } from '@agent-harn export interface QueuedRun { readonly message: UserMessage; readonly options: AbortSignal | AgentRunOptions; - readonly resolve: (message: AssistantMessage) => void; - readonly reject: (error: unknown) => void; + readonly resolve: (message: AssistantMessage) => void; // 执行成功时，调用 resolve 通知调用者 + readonly reject: (error: unknown) => void; // 执行失败时，调用 reject 通知调用者 readonly clearAbortListener: () => void; } diff --git a/packages/core/src/lifecycle.ts b/packages/core/src/lifecycle.ts index 7f6a7d5..b74dfab 100644 --- a/packages/core/src/lifecycle.ts +++ b/packages/core/src/lifecycle.ts @@ -13,7 +13,7 @@ import { type RuntimeFailureOutcome = Extract; -/** 跟踪单次模型 attempt 的流式累积、结算与工具结果，供失败路径补写事件。 */ +/** 一次 LLM 调用 */ export interface Attempt { readonly attemptId: AttemptId; partialText: string; // 模型生成时的中间输出 diff --git a/packages/kernel/src/events.ts b/packages/kernel/src/events.ts new file mode 100644 index 0000000..879ca93 --- /dev/null +++ b/packages/kernel/src/events.ts @@ -0,0 +1,300 @@ +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; From a671038eb352a00b2341fbcc9aab9d2cb2632b17 Mon Sep 17 00:00:00 2001 From: hanpj <2568545091@qq.com> Date: Fri, 16 Oct 2026 18:43:39 +0800 Subject: [PATCH 2/2] style: remove extra blank line in runtime event catalog. Co-authored-by: Cursor --- packages/kernel/src/events.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/kernel/src/events.ts b/packages/kernel/src/events.ts index 879ca93..35c2e93 100644 --- a/packages/kernel/src/events.ts +++ b/packages/kernel/src/events.ts @@ -58,7 +58,6 @@ export type AgentRuntimeEvent = | TurnCancelledEvent | TurnInterruptedEvent; - /** 用户输入已进入 inbox，等待 FIFO 领取。 */ export interface InboxEnqueuedEvent { readonly type: 'inbox.enqueued';