diff --git a/packages/agent/README.md b/packages/agent/README.md index a2c6efbe..05f9a622 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -8,9 +8,9 @@ Pi Agent 接入层:能力注册中心、`AgentRuntimePort` 及其 Pi 实现、 - 模型只能调用本次 `capabilities` 白名单里的能力。白名单有三层保护: 1. 注册层:只把白名单能力注册成 Pi 工具,其它工具名(例如 `bash`)由 harness 合成 unavailable 错误; 2. `before_tool` 钩子再校验一次,白名单外的调用一律 block,并产生 `tool_call_rejected` 事件; - 3. 每次执行都走 `CapabilityRegistry.invoke`,依次做作用域、schema、授权、幂等键、输出 schema 和审计。 + 3. 每次执行都走 `CapabilityRegistry.invoke`,依次做能力类别上限(allowedKinds)、schema、授权、幂等键、输出 schema 和审计。 - **不提供** bash/进程执行、任意 SQL、任意文件读写、任意网络/HTTP 代理、读取 Cookie/Token、安装依赖或修改代码的工具。`NodeExecutionEnv` 只作为会话目录的 FileSystem 使用,它的 exec 不会暴露给模型。 -- 授权只在服务端判定:`authorize` 回调查 automation_grant/approval。模型参数里写的"用户已同意"不起作用。没有注入 `authorize` 时,propose/execute 一律拒绝。 +- 授权只在服务端判定:`authorize` 回调查 automation_grant/approval。模型参数里写的"用户已同意"不起作用。没有注入 `authorize` 时,propose/approve/execute 一律拒绝。 - 幂等键由服务端按操作语义生成(默认是 `id:v版本:sha256(规范化输入)`,可以注入自定义生成器),通过 `ctx.idempotencyKey` 交给 handler,不采用模型给的值。 - 工具的输入输出在进入模型上下文、事件和审计之前,都要经过 `redactForModel`(cookie/passkey/token/apiKey/authorization/私有下载 URL/URL 内嵌账号密码)处理。`apiKey` 只在请求时经 provider `auth.resolve` 交给 SDK。 - 子代理:`subagentsEnabled` 在 v0.1 中只做记录。SDK 本身不提供子代理,以后由 `agent.delegate` 能力实现,并继承父任务的授权与预算。 @@ -18,30 +18,24 @@ Pi Agent 接入层:能力注册中心、`AgentRuntimePort` 及其 Pi 实现、 ## 新增一个能力 +能力描述只在 `@umt/contracts` 定义(B3):`CapabilitySpec`(描述符字段 + 真实 zod schema)、`CapabilityKind`(read / preview / propose / approve / execute)、`RiskLevel`(low / medium / high)、`CapabilityReplay`(safe / never)。本包不再有自己的描述类型。 + +首批 14 个能力一律用 `capabilitySpec(id)` 取规格(元数据来自 `CAPABILITY_CATALOG`,schema 来自 `CAPABILITY_IO`),server 模块只写处理器: + ```ts -registry.register( - { - id: "download.submit", // 域.动作;对 Pi 暴露为 download__submit - version: 1, - title: "提交下载", - description: "提交下载计划到下载器。成功只代表已提交,不代表已入库。", // 给模型看的中文说明 - inputSchema: z.object({ planId: z.string() }), - outputSchema: z.object({ taskId: z.string(), status: z.literal("submitted") }), - scope: "execute", // read | preview | propose | execute - riskLevel: "medium", // low | medium | high | critical - idempotent: true, - supportsPreview: true, - replay: "never", // 有副作用的能力必须是 never;只读能力用 safe - }, - async (input, ctx) => { - // ctx: actorId/taskId/scopeIds/signal + idempotencyKey + authorization - // 失败时抛 CapabilityError("precondition" | "not_found" | "conflict" | "unauthorized" | "budget", message) - return { taskId: await downloads.submit(input.planId, ctx.idempotencyKey!, ctx.signal), status: "submitted" }; - }, -); +import { capabilitySpec } from "@umt/contracts"; + +registry.register(capabilitySpec("download.submit"), async (input, ctx) => { + // input 已按 DownloadSubmitInput 校验;ctx: actorId/taskId/scopeIds/signal + idempotencyKey + authorization + // 失败时抛 CapabilityError("precondition" | "not_found" | "conflict" | "unauthorized" | "budget", message) + const task = await downloads.submit(input.planId, ctx.idempotencyKey!, ctx.signal); + return { taskId: task.id, planId: input.planId, downloadTask: null }; +}); ``` -要点:输出只写业务需要的字段,因为 zod 会剥掉多余字段;不要把凭证放进输出;有副作用的 handler 要按 `idempotencyKey` 到操作账本去重。崩溃恢复时,`replay:"never"` 的工具会得到"结果未知",模型需要用查询类能力核对状态。 +目录外的新能力:先在 contracts 的 `CAPABILITY_IDS` / `CAPABILITY_CATALOG` / `CAPABILITY_IO` 各加一项,再按上面注册。测试里可以手写 `CapabilitySpec` 对象(`kind` 是能力类别,`scope` 是 HTTP 调用所需的 ApiScope)。 + +要点:输出只写业务需要的字段,因为 zod 会剥掉多余字段;不要把凭证放进输出;有副作用的 handler 要按 `idempotencyKey` 到操作账本去重。propose / approve / execute 三类都必须经过 `authorize`,并由注册中心生成幂等键。崩溃恢复时,`replay:"never"` 的工具会得到"结果未知",模型需要用查询类能力核对状态。 ## 在 server 中装配(W2) @@ -53,11 +47,16 @@ const registry = new CapabilityRegistry({ audit: (rec) => auditRepo.insert({ type: "capability.invoke", ...rec }), // 写 audit_event idempotencyKey: (d, input, ctx) => opLedger.keyFor(d.id, input, ctx), // 可选 }); -registerAllCapabilities(registry); // catalog.lookup / release.search / … / inventory.query +for (const m of modules) m.registerCapabilities?.(registry, ctx); // 各 server 模块用 capabilitySpec(id) 注册处理器 const runtime = new PiAgentRuntime({ registry, - onEvent: (e) => agentRunEvents.append(e).then(() => sse.publish(`agent_run:${e.runId}`, e)), // 先落库(带 seq)再推 SSE + // e 是 contracts 的 AgentRunEvent;同一 run 内按 seq 串行回调。先落库再推 SSE: + onEvent: async (e) => { + await db.insertInto("agent_run_event").values(toDbRow(agentRunEventToRow(e))).execute(); // (run_id, seq) 唯一 + const bus = agentRunEventToBusEvent(e); // agent.run_status / agent.text_delta / agent.tool_call,其余返回 null + if (bus) events.publish(bus.type, bus.payload); + }, }); // 在持久任务 agentRun 的处理器里执行(不要在聊天 HTTP 请求里驱动): @@ -68,7 +67,8 @@ const run = await runtime.createRun({ capabilities: ["catalog.lookup", "inventory.query", "download.plan", "download.submit"], settings: { ...settingsRow.agent, apiKey: await secrets.decrypt(settingsRow.agent.apiKeyRef) }, sessionDir: join(dataDir, "agent-sessions"), - actor: { actorId, taskId: task.id, scopeIds, allowedScopes }, // "先给方案"模式传 ['read','preview','propose'] + actor: { actorId, taskId: task.id, scopeIds, allowedKinds }, // "先给方案"模式传 ['read','preview','propose'] + firstSeq, // 续跑时 = 该 run 在 agent_run_event 中的 max(seq)+1 }); const outcome = run.pendingOperations > 0 ? await run.resume() : await run.prompt(); await run.close(); @@ -79,4 +79,8 @@ await run.close(); ## 事件 -`AgentEvent` = `{ runId, seq, at } &` 下列之一:`run_started`、`text_delta`、`thinking_delta`、`tool_call_start`/`tool_call_end`(带脱敏截断后的输入/输出摘要和 `recovery` 标记)、`tool_call_rejected`、`compaction`(start/end)、`run_completed`(usage + 最终文本)、`run_failed`(reason: budget | aborted | error)。`run.events(fromSeq)` 回放已有事件后继续跟随新事件,`close()` 后结束。 +事件类型只在 contracts 定义(`AgentRunEvent` / `AgentRunEventBody` / `AgentRunEventType`,B9):`{ runId, seq, at } &` 下列之一:`run_started`、`text_delta`、`thinking_delta`、`tool_call_start`/`tool_call_end`(带脱敏截断后的输入/输出摘要和 `recovery` 标记)、`tool_call_rejected`、`compaction`(start/end)、`run_completed`(`AgentUsageTotals` + 最终文本)、`run_failed`(reason: budget | aborted | error)。 + +- 落库:`agent_run_event(run_id, seq, type, at, payload)`,主键 (run_id, seq);`agentRunEventToRow` / `agentRunEventFromRow` 负责转换。 +- 推送:`agentRunEventToBusEvent` 映射到全局 SSE 的 `agent.run_status` / `agent.text_delta` / `agent.tool_call`(phase:start→started,end→succeeded/failed,rejected→denied)。 +- `run.events(fromSeq)` 回放本进程已有事件后继续跟随新事件,`close()` 后结束;跨进程回放读 `agent_run_event`。 diff --git a/packages/agent/src/capability/registry.ts b/packages/agent/src/capability/registry.ts index 00b82c86..5ee9f5eb 100644 --- a/packages/agent/src/capability/registry.ts +++ b/packages/agent/src/capability/registry.ts @@ -1,24 +1,29 @@ import { createHash } from "node:crypto"; +import { + ApiScope, + AUTHORIZED_KINDS, + CapabilityKind, + CapabilityReplay, + type CapabilitySpec, + RiskLevel, +} from "@umt/contracts"; import { z } from "zod"; import { summarizeForLog } from "../redact.ts"; import { CapabilityError, isCapabilityError } from "./errors.ts"; import { capabilityIdToToolName, isCapabilityId } from "./tool-name.ts"; -import { - type AuditSink, - type AuthorizationChecker, - type AuthorizationDecision, - CAPABILITY_SCOPES, - type CapabilityAuditRecord, - type CapabilityDescriptor, - type CapabilityHandler, - type CapabilityScope, - type HandlerContext, - type IdempotencyKeyGenerator, - type InvokeContext, +import type { + AuditSink, + AuthorizationChecker, + AuthorizationDecision, + CapabilityAuditRecord, + CapabilityHandler, + HandlerContext, + IdempotencyKeyGenerator, + InvokeContext, } from "./types.ts"; export interface RegisteredCapability { - descriptor: CapabilityDescriptor; + descriptor: CapabilitySpec; handler: CapabilityHandler; toolName: string; } @@ -26,11 +31,11 @@ export interface RegisteredCapability { export interface CapabilityRegistryOptions { /** * 授权回调(服务端注入):检查 automation_grant / approval。 - * 未注入时:read/preview 放行,propose/execute 一律拒绝(安全默认)。 + * 未注入时:read/preview 放行,propose/approve/execute 一律拒绝(安全默认)。 */ authorize?: AuthorizationChecker; - /** 需要走授权回调的作用域,默认 propose + execute */ - authorizeScopes?: readonly CapabilityScope[]; + /** 需要走授权回调的类别,默认 contracts 的 AUTHORIZED_KINDS(propose/approve/execute) */ + authorizeKinds?: readonly CapabilityKind[]; /** 审计回调;每次调用(含拒绝/校验失败)都会触发一次 */ audit?: AuditSink; /** 幂等键生成(服务端基于操作语义生成,不信任模型);默认对 id+version+规范化输入做 sha256 */ @@ -64,7 +69,7 @@ export class CapabilityRegistry { } register( - descriptor: CapabilityDescriptor, + descriptor: CapabilitySpec, handler: CapabilityHandler, ): this { if (!isCapabilityId(descriptor.id)) throw new Error(`能力 ID 必须是 域.动作 形式: ${descriptor.id}`); @@ -72,14 +77,17 @@ export class CapabilityRegistry { if (!Number.isInteger(descriptor.version) || descriptor.version < 1) { throw new Error(`能力版本必须是正整数: ${descriptor.id}`); } - if (!CAPABILITY_SCOPES.includes(descriptor.scope)) throw new Error(`未知作用域: ${descriptor.scope}`); + if (!CapabilityKind.safeParse(descriptor.kind).success) throw new Error(`未知能力类别: ${descriptor.kind}`); + if (!ApiScope.safeParse(descriptor.scope).success) throw new Error(`未知 API scope: ${descriptor.scope}`); + if (!RiskLevel.safeParse(descriptor.riskLevel).success) throw new Error(`未知风险级别: ${descriptor.riskLevel}`); + if (!CapabilityReplay.safeParse(descriptor.replay).success) throw new Error(`未知 replay: ${descriptor.replay}`); if (!descriptor.description.trim()) throw new Error(`能力缺少给模型看的说明: ${descriptor.id}`); - if (descriptor.scope === "execute" && descriptor.replay === "safe" && !descriptor.idempotent) { + if (sideEffecting(descriptor.kind) && descriptor.replay === "safe" && !descriptor.idempotent) { throw new Error(`有副作用且非幂等的能力不能声明 replay:'safe': ${descriptor.id}`); } const toolName = capabilityIdToToolName(descriptor.id); this.entries.set(descriptor.id, { - descriptor: descriptor as unknown as CapabilityDescriptor, + descriptor: descriptor as unknown as CapabilitySpec, handler: handler as unknown as CapabilityHandler, toolName, }); @@ -131,7 +139,7 @@ export class CapabilityRegistry { const record: CapabilityAuditRecord = { capabilityId: descriptor.id, version: descriptor.version, - scope: descriptor.scope, + kind: descriptor.kind, riskLevel: descriptor.riskLevel, actorId: ctx.actorId, taskId: ctx.taskId, @@ -156,10 +164,10 @@ export class CapabilityRegistry { if (ctx.signal.aborted) throw await fail(new CapabilityError("precondition", "调用已取消"), "denied"); - // 1. 作用域:调用方给出的上限 - if (ctx.allowedScopes && !ctx.allowedScopes.includes(descriptor.scope)) { + // 1. 类别:调用方给出的上限 + if (ctx.allowedKinds && !ctx.allowedKinds.includes(descriptor.kind)) { throw await fail( - new CapabilityError("unauthorized", `当前任务不允许 ${descriptor.scope} 级能力: ${descriptor.id}`), + new CapabilityError("unauthorized", `当前任务不允许 ${descriptor.kind} 类能力: ${descriptor.id}`), "denied", ); } @@ -177,9 +185,9 @@ export class CapabilityRegistry { } // 3. 授权(服务端判定,模型声明无效) - const authorizeScopes = this.options.authorizeScopes ?? ["propose", "execute"]; + const authorizeKinds = this.options.authorizeKinds ?? AUTHORIZED_KINDS; let decision: AuthorizationDecision | undefined; - if (authorizeScopes.includes(descriptor.scope)) { + if (authorizeKinds.includes(descriptor.kind)) { if (!this.options.authorize) { throw await fail(new CapabilityError("unauthorized", `未配置授权校验,拒绝 ${descriptor.id}`), "denied"); } @@ -193,9 +201,9 @@ export class CapabilityRegistry { } } - // 4. 幂等键(只对有副作用的作用域生成) + // 4. 幂等键(只对有副作用的类别生成) let idempotencyKey: string | undefined; - if (descriptor.scope === "propose" || descriptor.scope === "execute") { + if (sideEffecting(descriptor.kind)) { idempotencyKey = (this.options.idempotencyKey ?? defaultIdempotencyKey)(descriptor, parsed.data, ctx); } @@ -233,6 +241,11 @@ export class CapabilityRegistry { } } +/** propose/approve/execute 会落变更;read/preview 不会 */ +function sideEffecting(kind: CapabilityKind): boolean { + return kind === "propose" || kind === "approve" || kind === "execute"; +} + function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } diff --git a/packages/agent/src/capability/types.ts b/packages/agent/src/capability/types.ts index 62da32dd..1fc4056e 100644 --- a/packages/agent/src/capability/types.ts +++ b/packages/agent/src/capability/types.ts @@ -1,43 +1,18 @@ -import type { CapabilityContext } from "@umt/contracts"; +import type { CapabilityContext, CapabilityKind, CapabilitySpec, RiskLevel } from "@umt/contracts"; import type { z } from "zod"; /** - * 能力作用域(由弱到强): - * - read:只读查询 - * - preview:生成预览/方案,不落任何变更 - * - propose:生成并保存变更计划(MutationPlan),等待授权 - * - execute:产生外部副作用(下载提交、Copy/Move、策略发布…) + * 能力描述统一取自 @umt/contracts(B3): + * - 运行时规格 `CapabilitySpec`(描述符字段 + 真实 zod schema),首批能力用 `capabilitySpec(id)` 组装; + * - 类别 `CapabilityKind`(read/preview/propose/approve/execute),风险 `RiskLevel`,重放 `CapabilityReplay`。 + * 本包不再定义自己的一套。 */ -export const CAPABILITY_SCOPES = ["read", "preview", "propose", "execute"] as const; -export type CapabilityScope = (typeof CAPABILITY_SCOPES)[number]; - -export const CAPABILITY_RISK_LEVELS = ["low", "medium", "high", "critical"] as const; -export type CapabilityRiskLevel = (typeof CAPABILITY_RISK_LEVELS)[number]; - -/** - * 能力描述。description 会原样给模型看,用中文写清用途、边界与返回含义。 - * replay:崩溃恢复时 harness 是否可以用持久化参数重跑;有副作用的一律 'never'。 - */ -export interface CapabilityDescriptor { - id: string; - version: number; - title: string; - description: string; - inputSchema: I; - outputSchema: O; - scope: CapabilityScope; - riskLevel: CapabilityRiskLevel; - idempotent: boolean; - supportsPreview: boolean; - replay: "safe" | "never"; -} - /** * 调用上下文:contracts 的 CapabilityContext(actorId/taskId/scopeIds/signal)加上本包需要的字段。 - * allowedScopes:本次调用方允许触达的最高作用域集合(例如"先给方案"模式只给 read/preview/propose)。 + * allowedKinds:本次调用方允许触达的能力类别(例如"先给方案"模式只给 read/preview/propose)。 */ export interface InvokeContext extends CapabilityContext { - allowedScopes?: readonly CapabilityScope[]; + allowedKinds?: readonly CapabilityKind[]; /** 调用来源,仅用于审计 */ source?: "agent" | "web" | "cli" | "mcp" | "task"; /** agent 调用时的 tool call id / invocation id,用于审计对账 */ @@ -57,7 +32,7 @@ export type CapabilityHandler Promise> | z.input; export interface AuthorizationRequest { - descriptor: CapabilityDescriptor; + descriptor: CapabilitySpec; input: unknown; ctx: InvokeContext; } @@ -74,15 +49,15 @@ export type AuthorizationChecker = ( request: AuthorizationRequest, ) => Promise | AuthorizationDecision; -export type IdempotencyKeyGenerator = (descriptor: CapabilityDescriptor, input: unknown, ctx: InvokeContext) => string; +export type IdempotencyKeyGenerator = (descriptor: CapabilitySpec, input: unknown, ctx: InvokeContext) => string; export type AuditOutcome = "ok" | "denied" | "invalid" | "error"; export interface CapabilityAuditRecord { capabilityId: string; version: number; - scope: CapabilityScope; - riskLevel: CapabilityRiskLevel; + kind: CapabilityKind; + riskLevel: RiskLevel; actorId: string; taskId: string; source?: InvokeContext["source"]; diff --git a/packages/agent/src/pi/runtime.ts b/packages/agent/src/pi/runtime.ts index 7959cc13..58589115 100644 --- a/packages/agent/src/pi/runtime.ts +++ b/packages/agent/src/pi/runtime.ts @@ -10,6 +10,7 @@ import { type RunResult, } from "@earendil-works/pi-agent-core"; import type { Usage } from "@earendil-works/pi-ai"; +import type { AgentBudgetKind, AgentRunEvent, AgentRunEventBody, AgentUsageTotals } from "@umt/contracts"; import type { TSchema } from "typebox"; import { isCapabilityError } from "../capability/errors.ts"; import type { CapabilityRegistry, RegisteredCapability } from "../capability/registry.ts"; @@ -17,12 +18,8 @@ import { toolNameToCapabilityId } from "../capability/tool-name.ts"; import { redactForModel, redactString, summarizeForLog } from "../redact.ts"; import { EventChannel } from "../runtime/event-channel.ts"; import { - type AgentEvent, - type AgentEventBody, - type AgentRun, + type AgentRunHandle, type AgentRuntimePort, - type AgentUsage, - type BudgetKind, type CreateRunSpec, emptyUsage, type RunOutcome, @@ -35,8 +32,13 @@ export interface PiAgentRuntimeOptions { registry: CapabilityRegistry; /** 默认按设置构造真实上游;测试用 FakeModelProvider 覆盖 */ resolveModel?: PiModelResolver; - /** 每个事件产生时回调(server 用它写 agent_run_event 并推 SSE) */ - onEvent?: (event: AgentEvent) => void | Promise; + /** + * 每个事件产生时回调(server 用它写 agent_run_event 再推 SSE)。 + * 同一 run 内按 seq 顺序串行调用(前一个 Promise 结束才调下一个);回调抛错只记入 onEventError,不影响运行。 + * close() 会等待已排队的回调全部完成。 + */ + onEvent?: (event: AgentRunEvent) => void | Promise; + onEventError?: (error: unknown, event: AgentRunEvent) => void; now?: () => Date; } @@ -50,7 +52,7 @@ const SUMMARY_LIMIT = 600; export class PiAgentRuntime implements AgentRuntimePort { constructor(private readonly options: PiAgentRuntimeOptions) {} - async createRun(spec: CreateRunSpec): Promise { + async createRun(spec: CreateRunSpec): Promise { const ctx = BACKGROUND_CONTEXT; const settings = resolveAgentSettings(spec.settings); const capabilities = this.options.registry.resolve(spec.capabilities); @@ -86,22 +88,24 @@ export class PiAgentRuntime implements AgentRuntimePort { } } -type BudgetState = { kind: BudgetKind; message: string }; +type BudgetState = { kind: AgentBudgetKind; message: string }; -class PiAgentRun implements AgentRun { +class PiAgentRun implements AgentRunHandle { readonly runId: string; readonly sessionPath: string; pendingOperations = 0; - private readonly channel = new EventChannel(); + private readonly channel = new EventChannel(); private readonly allowed = new Map(); private readonly rejected = new Set(); private readonly abortController = new AbortController(); private harness!: AgentHarnessInstance; private lane!: AgentLane; - private seq = 0; + private readonly firstSeq: number; + private seq: number; + private delivery: Promise = Promise.resolve(); private toolCalls = 0; - private usage: AgentUsage = emptyUsage(); + private usage: AgentUsageTotals = emptyUsage(); private finalText = ""; private budget: BudgetState | undefined; private userAbort: string | undefined; @@ -119,6 +123,8 @@ class PiAgentRun implements AgentRun { ) { this.runId = spec.runId; this.sessionPath = sessionPath; + this.firstSeq = spec.firstSeq ?? 0; + this.seq = this.firstSeq; for (const c of capabilities) this.allowed.set(c.toolName, c.descriptor.id); } @@ -130,10 +136,10 @@ class PiAgentRun implements AgentRun { return { name: entry.toolName, label: descriptor.title, - description: `${descriptor.description}\n[能力 ${descriptor.id} v${descriptor.version};作用域 ${descriptor.scope};风险 ${descriptor.riskLevel}]`, + description: `${descriptor.description}\n[能力 ${descriptor.id} v${descriptor.version};类别 ${descriptor.kind};风险 ${descriptor.riskLevel}]`, parameters: registry.inputJsonSchema(descriptor.id) as unknown as TSchema, replay: descriptor.replay, - ...(descriptor.scope === "execute" ? { executionMode: "sequential" as const } : {}), + ...(descriptor.kind === "execute" ? { executionMode: "sequential" as const } : {}), execute: async (toolCallId, params, _onUpdate, _toolCtx, invocation, context) => { const signal = context.abortSignal ?? this.abortController.signal; try { @@ -141,7 +147,7 @@ class PiAgentRun implements AgentRun { actorId: actor.actorId, taskId: actor.taskId, scopeIds: actor.scopeIds, - ...(actor.allowedScopes ? { allowedScopes: actor.allowedScopes } : {}), + ...(actor.allowedKinds ? { allowedKinds: actor.allowedKinds } : {}), signal, source: "agent", invocationId: invocation.invocationId ?? toolCallId, @@ -251,11 +257,11 @@ class PiAgentRun implements AgentRun { ); } - events(fromSeq = 0): AsyncIterable { - return this.channel.iterate(fromSeq); + events(fromSeq = this.firstSeq): AsyncIterable { + return this.channel.iterate(fromSeq - this.firstSeq); } - history(): AgentEvent[] { + history(): AgentRunEvent[] { return this.channel.snapshot(); } @@ -291,6 +297,7 @@ class PiAgentRun implements AgentRun { await this.harness?.close(ctx).catch(() => undefined); await this.repo.close(ctx).catch(() => undefined); this.channel.close(); + await this.delivery; } // ---- internals ---- @@ -341,7 +348,7 @@ class PiAgentRun implements AgentRun { return outcome; } - private failed(reason: RunOutcome["reason"] & string, message: string, budget?: BudgetKind): RunOutcome { + private failed(reason: RunOutcome["reason"] & string, message: string, budget?: AgentBudgetKind): RunOutcome { return { status: "failed", reason, @@ -359,7 +366,7 @@ class PiAgentRun implements AgentRun { this.emit({ type: "tool_call_rejected", toolCallId, toolName, ...(capabilityId ? { capabilityId } : {}), reason }); } - private exceed(kind: BudgetKind, message: string): void { + private exceed(kind: AgentBudgetKind, message: string): void { if (this.budget) return; this.budget = { kind, message }; this.scheduleAbort(); @@ -392,13 +399,15 @@ class PiAgentRun implements AgentRun { this.deadlineTimer = undefined; } - private emit(body: AgentEventBody): void { - const event = { ...body, runId: this.runId, seq: this.seq++, at: new Date().toISOString() } as AgentEvent; + private emit(body: AgentRunEventBody): void { + const event = { ...body, runId: this.runId, seq: this.seq++, at: new Date().toISOString() } as AgentRunEvent; this.channel.push(event); - if (this.options.onEvent) { - Promise.resolve() - .then(() => this.options.onEvent?.(event)) - .catch(() => undefined); + const onEvent = this.options.onEvent; + if (onEvent) { + // 串行投递:落库顺序与 seq 一致 + this.delivery = this.delivery + .then(() => onEvent(event)) + .catch((error) => this.options.onEventError?.(error, event)); } } } @@ -427,7 +436,7 @@ function assistantText(message: AgentMessage): string | undefined { .join(""); } -function toAgentUsage(u: Usage): AgentUsage { +function toAgentUsage(u: Usage): AgentUsageTotals { return { inputTokens: u.input, outputTokens: u.output, diff --git a/packages/agent/src/runtime/port.ts b/packages/agent/src/runtime/port.ts index 217349c6..82cb5d2a 100644 --- a/packages/agent/src/runtime/port.ts +++ b/packages/agent/src/runtime/port.ts @@ -1,52 +1,14 @@ -import type { CapabilityScope } from "../capability/types.ts"; +import type { + AgentBudgetKind, + AgentFailureReason, + AgentRunEvent, + AgentUsageTotals, + CapabilityKind, +} from "@umt/contracts"; import type { AgentSettings } from "../settings.ts"; -/** 本包对外的用量统计(与 SDK 类型解耦) */ -export interface AgentUsage { - inputTokens: number; - outputTokens: number; - cacheReadTokens: number; - cacheWriteTokens: number; - totalTokens: number; - costUsd: number; -} - -export type AgentFailureReason = "budget" | "aborted" | "error"; - -export type AgentEventBody = - | { type: "run_started"; goal: string; resumed: boolean } - | { type: "text_delta"; delta: string } - | { type: "thinking_delta"; delta: string } - | { - type: "tool_call_start"; - toolCallId: string; - capabilityId: string; - /** 已脱敏、截断的输入摘要 */ - inputSummary: string; - recovery: boolean; - } - | { - type: "tool_call_end"; - toolCallId: string; - capabilityId: string; - isError: boolean; - /** 已脱敏、截断的输出摘要 */ - outputSummary: string; - recovery: boolean; - } - | { type: "tool_call_rejected"; toolCallId: string; toolName: string; capabilityId?: string; reason: string } - | { - type: "compaction"; - phase: "start" | "end"; - reason: "manual" | "threshold" | "overflow"; - status?: "completed" | "declined" | "aborted" | "failed"; - } - | { type: "run_completed"; finalText: string; usage: AgentUsage } - | { type: "run_failed"; reason: AgentFailureReason; message: string; usage: AgentUsage; budget?: BudgetKind }; - -export type BudgetKind = "tool_calls" | "duration" | "tokens" | "cost"; - -export type AgentEvent = AgentEventBody & { runId: string; seq: number; at: string }; +// 事件、用量、失败原因的类型统一取自 @umt/contracts(AgentRunEvent / AgentUsageTotals / AgentFailureReason / +// AgentBudgetKind);本文件只定义运行时端口。 /** 调用能力时使用的业务身份;授权回调据此到 automation_grant / approval 查询 */ export interface AgentActor { @@ -54,8 +16,8 @@ export interface AgentActor { /** 持久任务 ID(agent_run 所属任务) */ taskId: string; scopeIds: string[]; - /** 本次运行允许触达的作用域上限;"先给方案"模式传 ['read','preview','propose'] */ - allowedScopes?: CapabilityScope[]; + /** 本次运行允许触达的能力类别上限;"先给方案"模式传 ['read','preview','propose'] */ + allowedKinds?: CapabilityKind[]; } export interface CreateRunSpec { @@ -69,26 +31,32 @@ export interface CreateRunSpec { /** 应用数据目录下的会话目录;会话文件为 sessionDir/.jsonl */ sessionDir: string; actor: AgentActor; + /** + * 本进程第一个事件的 seq。续跑(崩溃恢复/多轮对话)时传该 run 已落库事件的 max(seq)+1 + * (agent_run_event 中无记录时为 0),保证 (run_id, seq) 不冲突;默认 0。 + */ + firstSeq?: number; } export interface RunOutcome { status: "completed" | "failed"; reason?: AgentFailureReason; - budget?: BudgetKind; + budget?: AgentBudgetKind; message?: string; finalText: string; - usage: AgentUsage; + usage: AgentUsageTotals; } -export interface AgentRun { +/** 一次运行的句柄(区别于 contracts 的 AgentRun 行) */ +export interface AgentRunHandle { readonly runId: string; readonly sessionPath: string; /** 打开已有会话时发现的未完成操作数;>0 时应调用 resume() */ readonly pendingOperations: number; /** 从 fromSeq(含)开始回放并持续推送事件;close() 后结束 */ - events(fromSeq?: number): AsyncIterable; - /** 已产生的全部事件快照 */ - history(): AgentEvent[]; + events(fromSeq?: number): AsyncIterable; + /** 本进程已产生的全部事件快照 */ + history(): AgentRunEvent[]; /** 发送一条用户消息并驱动到本轮结束;text 省略时使用 goal */ prompt(text?: string): Promise; /** 崩溃/重启后继续未完成的操作 */ @@ -100,9 +68,9 @@ export interface AgentRun { /** 领域/服务层只依赖这个端口,不直接接触 Pi SDK 类型 */ export interface AgentRuntimePort { - createRun(spec: CreateRunSpec): Promise; + createRun(spec: CreateRunSpec): Promise; } -export function emptyUsage(): AgentUsage { +export function emptyUsage(): AgentUsageTotals { return { inputTokens: 0, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0, totalTokens: 0, costUsd: 0 }; } diff --git a/packages/agent/test/helpers.ts b/packages/agent/test/helpers.ts index 127cd246..7cd70626 100644 --- a/packages/agent/test/helpers.ts +++ b/packages/agent/test/helpers.ts @@ -1,9 +1,9 @@ import { mkdtemp } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import type { AgentRunEvent } from "@umt/contracts"; import { z } from "zod"; import { - type AgentEvent, type AgentSettings, type AuthorizationChecker, type CapabilityAuditRecord, @@ -40,6 +40,7 @@ export function makeFixture(opts: { slowMs?: number } = {}): Fixture { description: "按标题查询本地媒体库是否已有该作品及已有版本。", inputSchema: z.object({ title: z.string().min(1), year: z.number().int().optional() }), outputSchema: z.object({ title: z.string(), inLibrary: z.boolean(), versions: z.array(z.string()) }), + kind: "read", scope: "read", riskLevel: "low", idempotent: true, @@ -56,7 +57,8 @@ export function makeFixture(opts: { slowMs?: number } = {}): Fixture { description: "提交下载计划到下载器。只代表已提交,不代表已入库。", inputSchema: z.object({ title: z.string(), quality: z.enum(["720p", "1080p", "2160p"]) }), outputSchema: z.object({ accepted: z.boolean(), idempotencyKey: z.string(), status: z.literal("submitted") }), - scope: "execute", + kind: "execute", + scope: "organize.execute", riskLevel: "medium", idempotent: true, supportsPreview: true, @@ -82,6 +84,7 @@ export function makeFixture(opts: { slowMs?: number } = {}): Fixture { description: "查看站点发布详情(返回原始字段,可能含私有链接)。", inputSchema: z.object({ releaseId: z.string() }), outputSchema: z.object({ releaseId: z.string(), torrentUrl: z.string(), cookie: z.string(), note: z.string() }), + kind: "read", scope: "read", riskLevel: "low", idempotent: true, @@ -127,6 +130,6 @@ export function settings(patch: Partial = {}): AgentSettings { return { ...FAKE_AGENT_SETTINGS, ...patch }; } -export function types(events: AgentEvent[]): string[] { +export function types(events: AgentRunEvent[]): string[] { return events.filter((e) => e.type !== "text_delta" && e.type !== "thinking_delta").map((e) => e.type); } diff --git a/packages/agent/test/registry.test.ts b/packages/agent/test/registry.test.ts index 2b0fbf7f..001a54bf 100644 --- a/packages/agent/test/registry.test.ts +++ b/packages/agent/test/registry.test.ts @@ -1,3 +1,4 @@ +import { CAPABILITY_IDS, capabilitySpec } from "@umt/contracts"; import { describe, expect, it } from "vitest"; import { z } from "zod"; import { @@ -32,6 +33,7 @@ describe("CapabilityRegistry.register", () => { description: "说明", inputSchema: z.object({}), outputSchema: z.object({}), + kind: "read" as const, scope: "read" as const, riskLevel: "low" as const, idempotent: true, @@ -44,13 +46,41 @@ describe("CapabilityRegistry.register", () => { expect(() => r.register({ ...base, id: "bad id" }, () => ({}))).toThrow(/域.动作/); r.register({ ...base, id: "catalog.lookup" }, () => ({})); expect(() => r.register({ ...base, id: "catalog.lookup" }, () => ({}))).toThrow(/已注册/); - expect(() => r.register({ ...base, id: "cache.cleanup", scope: "execute", idempotent: false }, () => ({}))).toThrow( + expect(() => r.register({ ...base, id: "cache.cleanup", kind: "execute", idempotent: false }, () => ({}))).toThrow( /replay/, ); expect(() => r.resolve(["catalog.lookup", "nope.nope"])).toThrow(/未注册/); expect(r.get("catalog.lookup")?.toolName).toBe("catalog__lookup"); }); + it("首批 14 个能力可直接用 contracts 的 capabilitySpec 注册", () => { + const r = new CapabilityRegistry(); + for (const id of CAPABILITY_IDS) r.register(capabilitySpec(id), () => ({}) as never); + expect(r.list().map((c) => c.descriptor.id)).toEqual([...CAPABILITY_IDS]); + expect(r.inputJsonSchema("release.search").type).toBe("object"); + }); + + it("approve 类能力同样必须经授权回调,且生成幂等键", async () => { + const seen: string[] = []; + const r = new CapabilityRegistry({ + authorize: ({ descriptor }) => { + seen.push(descriptor.id); + return { allowed: true, approvalId: "ap-1" }; + }, + }); + r.register({ ...base, id: "approval.decide", kind: "approve", replay: "never" }, (_i, c) => { + expect(c.idempotencyKey).toMatch(/^approval\.decide:v1:/); + return {}; + }); + await r.invoke("approval.decide", {}, ctx()); + expect(seen).toEqual(["approval.decide"]); + await expect( + new CapabilityRegistry() + .register({ ...base, id: "approval.decide", kind: "approve" }, () => ({})) + .invoke("approval.decide", {}, ctx()), + ).rejects.toMatchObject({ code: "unauthorized" }); + }); + it("输入 JSON Schema 可给模型使用(无 $schema)", () => { const { registry } = makeFixture(); const schema = registry.inputJsonSchema("download.submit"); @@ -97,7 +127,7 @@ describe("CapabilityRegistry.invoke", () => { expect(fx.audits.at(-1)).toMatchObject({ outcome: "ok", grantId: "grant:download.submit" }); }); - it("未注入授权回调:read 放行、execute 一律拒绝;allowedScopes 限制作用域", async () => { + it("未注入授权回调:read 放行、execute 一律拒绝;allowedKinds 限制能力类别", async () => { const r = new CapabilityRegistry(); const fx = makeFixture(); const inv = fx.registry.get("inventory.query")!; @@ -113,7 +143,7 @@ describe("CapabilityRegistry.invoke", () => { fx.registry.invoke( "download.submit", { title: "x", quality: "720p" }, - ctx({ allowedScopes: ["read", "preview"] }), + ctx({ allowedKinds: ["read", "preview"] }), ), ).rejects.toMatchObject({ code: "unauthorized" }); expect(fx.submitted).toHaveLength(0); @@ -158,6 +188,7 @@ describe("CapabilityRegistry.invoke", () => { description: "说明", inputSchema: z.object({}), outputSchema: z.object({ ok: z.boolean() }), + kind: "read" as const, scope: "read" as const, riskLevel: "low" as const, idempotent: true, diff --git a/packages/agent/test/runtime.test.ts b/packages/agent/test/runtime.test.ts index cf273c2b..f7f9ad77 100644 --- a/packages/agent/test/runtime.test.ts +++ b/packages/agent/test/runtime.test.ts @@ -1,10 +1,10 @@ import { spawn } from "node:child_process"; import { existsSync, readFileSync } from "node:fs"; import { join } from "node:path"; +import { AgentRunEvent } from "@umt/contracts"; import { afterEach, describe, expect, it } from "vitest"; import { - type AgentEvent, - type AgentRun, + type AgentRunHandle, type AgentSettings, checkToolCall, createFakeRuntime, @@ -13,7 +13,7 @@ import { import { crashScript, RUN_ID } from "./fixtures/crash-script.ts"; import { type Fixture, makeFixture, settings, tempDir, types } from "./helpers.ts"; -const opened: AgentRun[] = []; +const opened: AgentRunHandle[] = []; afterEach(async () => { await Promise.all(opened.splice(0).map((r) => r.close())); }); @@ -42,9 +42,20 @@ const dune: FakeScript = { async function start( fx: Fixture, scripts: FakeScript[], - opts: { capabilities?: string[]; settings?: Partial; runId?: string; sessionDir?: string } = {}, + opts: { + capabilities?: string[]; + settings?: Partial; + runId?: string; + sessionDir?: string; + firstSeq?: number; + onEvent?: (e: AgentRunEvent) => void | Promise; + } = {}, ) { - const { runtime, fake } = createFakeRuntime({ registry: fx.registry, scripts }); + const { runtime, fake } = createFakeRuntime({ + registry: fx.registry, + scripts, + ...(opts.onEvent ? { onEvent: opts.onEvent } : {}), + }); const sessionDir = opts.sessionDir ?? (await tempDir()); const run = await runtime.createRun({ runId: opts.runId ?? "run-1", @@ -54,6 +65,7 @@ async function start( settings: settings(opts.settings), sessionDir, actor: { actorId: "user:1", taskId: "task:1", scopeIds: ["media:dune"] }, + ...(opts.firstSeq !== undefined ? { firstSeq: opts.firstSeq } : {}), }); opened.push(run); return { run, fake, sessionDir }; @@ -106,12 +118,12 @@ describe("PiAgentRuntime + FakeModelProvider", () => { it("events() 可按 seq 回放并在 close 后结束", async () => { const fx = makeFixture(); const { run } = await start(fx, [{ match: "Dune", steps: [], finalText: "库里已有。" }]); - const collected: AgentEvent[] = []; + const collected: AgentRunEvent[] = []; const consumer = (async () => { for await (const e of run.events()) collected.push(e); })(); await run.prompt(); - const tail: AgentEvent[] = []; + const tail: AgentRunEvent[] = []; const replay = (async () => { for await (const e of run.events(1)) tail.push(e); })(); @@ -122,6 +134,31 @@ describe("PiAgentRuntime + FakeModelProvider", () => { expect(collected.at(-1)?.type).toBe("run_completed"); }); + it("onEvent 按 seq 串行投递 contracts AgentRunEvent;firstSeq 续接序号,close 等待投递完成", async () => { + const fx = makeFixture(); + const delivered: AgentRunEvent[] = []; + const { run } = await start(fx, [{ match: "Dune", steps: [], finalText: "库里已有。" }], { + firstSeq: 10, + // 故意让前面的事件落库更慢,验证不会乱序 + onEvent: async (e) => { + await new Promise((r) => setTimeout(r, e.seq === 10 ? 20 : 0)); + delivered.push(e); + }, + }); + await run.prompt(); + const tail: AgentRunEvent[] = []; + const replay = (async () => { + for await (const e of run.events(11)) tail.push(e); + })(); + await run.close(); + await replay; + const seqs = run.history().map((e) => e.seq); + expect(seqs[0]).toBe(10); + expect(delivered.map((e) => e.seq)).toEqual(seqs); + expect(tail[0]?.seq).toBe(11); + for (const e of delivered) expect(AgentRunEvent.safeParse(e).success, e.type).toBe(true); + }); + it("白名单:未注册工具(bash)与未列入白名单的已注册能力都被拒绝并产生事件,不执行", async () => { const fx = makeFixture(); fx.grants.set("user:1", new Set(["download.submit"])); diff --git a/packages/contracts/src/agent-events.ts b/packages/contracts/src/agent-events.ts new file mode 100644 index 00000000..bb7ebdb0 --- /dev/null +++ b/packages/contracts/src/agent-events.ts @@ -0,0 +1,188 @@ +import { z } from "zod"; +import type { AgentRunStatus } from "./agent.ts"; +import { Id, JsonObject, NonNegInt, Timestamp } from "./common.ts"; +import type { TypedEvent } from "./events.ts"; + +// Agent 运行事件(REQ §10.6):packages/agent 产生,server 先写 agent_run_event(带 seq)再推 SSE。 +// 这里是唯一定义;agent 包的 onEvent 回调直接收 AgentRunEvent。 + +export const AgentRunEventType = z.enum([ + "run_started", + "text_delta", + "thinking_delta", + "tool_call_start", + "tool_call_end", + "tool_call_rejected", + "compaction", + "run_completed", + "run_failed", +]); +export type AgentRunEventType = z.infer; + +/** 一次运行累计用量(数值都齐全;写 agent_run.usage 时可直接用,是 AgentUsage 的子集形状) */ +export const AgentUsageTotals = z.object({ + inputTokens: NonNegInt, + outputTokens: NonNegInt, + cacheReadTokens: NonNegInt, + cacheWriteTokens: NonNegInt, + totalTokens: NonNegInt, + costUsd: z.number().nonnegative(), +}); +export type AgentUsageTotals = z.infer; + +export const AgentFailureReason = z.enum(["budget", "aborted", "error"]); +export type AgentFailureReason = z.infer; +export const AgentBudgetKind = z.enum(["tool_calls", "duration", "tokens", "cost"]); +export type AgentBudgetKind = z.infer; + +const RunStarted = z.object({ type: z.literal("run_started"), goal: z.string(), resumed: z.boolean() }); +const TextDelta = z.object({ type: z.literal("text_delta"), delta: z.string() }); +const ThinkingDelta = z.object({ type: z.literal("thinking_delta"), delta: z.string() }); +const ToolCallStart = z.object({ + type: z.literal("tool_call_start"), + toolCallId: z.string(), + capabilityId: z.string(), + /** 已脱敏、截断的输入摘要 */ + inputSummary: z.string(), + /** 崩溃恢复后重放/补记的调用 */ + recovery: z.boolean(), +}); +const ToolCallEnd = z.object({ + type: z.literal("tool_call_end"), + toolCallId: z.string(), + capabilityId: z.string(), + isError: z.boolean(), + /** 已脱敏、截断的输出摘要 */ + outputSummary: z.string(), + recovery: z.boolean(), +}); +const ToolCallRejected = z.object({ + type: z.literal("tool_call_rejected"), + toolCallId: z.string(), + toolName: z.string(), + capabilityId: z.string().optional(), + reason: z.string(), +}); +const Compaction = z.object({ + type: z.literal("compaction"), + phase: z.enum(["start", "end"]), + reason: z.enum(["manual", "threshold", "overflow"]), + status: z.enum(["completed", "declined", "aborted", "failed"]).optional(), +}); +const RunCompleted = z.object({ type: z.literal("run_completed"), finalText: z.string(), usage: AgentUsageTotals }); +const RunFailed = z.object({ + type: z.literal("run_failed"), + reason: AgentFailureReason, + message: z.string(), + usage: AgentUsageTotals, + budget: AgentBudgetKind.optional(), +}); + +/** 事件主体(不含 runId/seq/at) */ +export const AgentRunEventBody = z.discriminatedUnion("type", [ + RunStarted, + TextDelta, + ThinkingDelta, + ToolCallStart, + ToolCallEnd, + ToolCallRejected, + Compaction, + RunCompleted, + RunFailed, +]); +export type AgentRunEventBody = z.infer; + +const EventMeta = { + /** = agent_run.id */ + runId: z.string().min(1), + /** 该 run 内从 0 起单调递增;续跑时从 agent_run.last_event_seq + 1 继续 */ + seq: NonNegInt, + at: Timestamp, +}; + +/** 完整事件:主体 + runId/seq/at */ +export const AgentRunEvent = z.discriminatedUnion("type", [ + RunStarted.extend(EventMeta), + TextDelta.extend(EventMeta), + ThinkingDelta.extend(EventMeta), + ToolCallStart.extend(EventMeta), + ToolCallEnd.extend(EventMeta), + ToolCallRejected.extend(EventMeta), + Compaction.extend(EventMeta), + RunCompleted.extend(EventMeta), + RunFailed.extend(EventMeta), +]); +export type AgentRunEvent = AgentRunEventBody & { runId: string; seq: number; at: string }; + +/** agent_run_event 行:(run_id, seq) 唯一;payload = 事件主体去掉 type */ +export const AgentRunEventRow = z.object({ + runId: Id, + seq: NonNegInt, + type: AgentRunEventType, + at: Timestamp, + payload: JsonObject, +}); +export type AgentRunEventRow = z.infer; + +export function agentRunEventToRow(e: AgentRunEvent): AgentRunEventRow { + const { runId, seq, at, type, ...payload } = e; + return { runId, seq, at, type, payload }; +} + +/** 行 → 事件;payload 不合法时抛错(表里只应有 agentRunEventToRow 写入的内容) */ +export function agentRunEventFromRow(row: AgentRunEventRow): AgentRunEvent { + return AgentRunEvent.parse({ ...row.payload, type: row.type, runId: row.runId, seq: row.seq, at: row.at }); +} + +/** 工具调用事件 → SSE agent.tool_call 的 phase */ +export function toolCallPhase(e: AgentRunEventBody): "started" | "succeeded" | "failed" | "denied" | null { + switch (e.type) { + case "tool_call_start": + return "started"; + case "tool_call_end": + return e.isError ? "failed" : "succeeded"; + case "tool_call_rejected": + return "denied"; + default: + return null; + } +} + +/** run_completed / run_failed / run_started → agent_run.status */ +export function agentRunStatusOf(e: AgentRunEventBody): AgentRunStatus | null { + switch (e.type) { + case "run_started": + return "running"; + case "run_completed": + return "succeeded"; + case "run_failed": + return e.reason === "budget" ? "budget_exceeded" : e.reason === "aborted" ? "cancelled" : "failed"; + default: + return null; + } +} + +/** + * Agent 运行事件 → 全局 SSE 事件(agent.run_status / agent.text_delta / agent.tool_call)。 + * thinking_delta 与 compaction 不上全局总线(只在 run 详情里按 agent_run_event 回放),返回 null。 + */ +export function agentRunEventToBusEvent(e: AgentRunEvent): TypedEvent | null { + const phase = toolCallPhase(e); + if (phase) { + let capabilityId = ""; + let summary = ""; + if (e.type === "tool_call_start") [capabilityId, summary] = [e.capabilityId, e.inputSummary]; + else if (e.type === "tool_call_end") [capabilityId, summary] = [e.capabilityId, e.outputSummary]; + else if (e.type === "tool_call_rejected") [capabilityId, summary] = [e.capabilityId ?? e.toolName, e.reason]; + return { type: "agent.tool_call", payload: { runId: e.runId, runSeq: e.seq, capabilityId, phase, summary } }; + } + if (e.type === "text_delta") { + return { type: "agent.text_delta", payload: { runId: e.runId, runSeq: e.seq, delta: e.delta } }; + } + const status = agentRunStatusOf(e); + if (status) { + const summary = e.type === "run_completed" ? e.finalText : e.type === "run_failed" ? e.message : null; + return { type: "agent.run_status", payload: { runId: e.runId, status, summary } }; + } + return null; +} diff --git a/packages/contracts/src/agent.ts b/packages/contracts/src/agent.ts index e72a14a3..360c0021 100644 --- a/packages/contracts/src/agent.ts +++ b/packages/contracts/src/agent.ts @@ -30,6 +30,7 @@ export const AgentUsage = z.object({ outputTokens: NonNegInt.optional(), cacheReadTokens: NonNegInt.optional(), cacheWriteTokens: NonNegInt.optional(), + totalTokens: NonNegInt.optional(), costUsd: z.number().nonnegative().optional(), durationMs: NonNegInt.optional(), }); diff --git a/packages/contracts/src/capabilities-io.ts b/packages/contracts/src/capabilities-io.ts index 9b8a2ea1..a77bfeb5 100644 --- a/packages/contracts/src/capabilities-io.ts +++ b/packages/contracts/src/capabilities-io.ts @@ -1,6 +1,6 @@ import { z } from "zod"; import { MediaAsset } from "./assets.ts"; -import type { KnownCapabilityId } from "./capability.ts"; +import { CAPABILITY_CATALOG, type CapabilitySpec, type KnownCapabilityId } from "./capability.ts"; import { Id, JsonObject, NonNegInt } from "./common.ts"; import { MediaEntity, MediaKind } from "./content.ts"; import { AcquisitionPlan, DownloadTask, Task, TaskStatus } from "./execution.ts"; @@ -188,6 +188,23 @@ export const CAPABILITY_IO = { "inventory.query": { input: InventoryQueryInput, output: InventoryQueryOutput }, } as const satisfies Record; +type CapabilityIo = typeof CAPABILITY_IO; +/** 首批能力 id 对应的运行时规格类型(input/output 为该能力的具体 schema) */ +export type KnownCapabilitySpec = CapabilitySpec< + CapabilityIo[K]["input"], + CapabilityIo[K]["output"] +>; + +/** + * 首批能力的运行时规格:CAPABILITY_CATALOG 的元数据 + CAPABILITY_IO 的 schema。 + * server 各模块注册处理器时一律用它,不要手写描述: + * `registry.register(capabilitySpec("release.search"), handler)` + */ +export function capabilitySpec(id: K): KnownCapabilitySpec { + const { inputSchema: _i, outputSchema: _o, ...meta } = CAPABILITY_CATALOG[id]; + return { ...meta, inputSchema: CAPABILITY_IO[id].input, outputSchema: CAPABILITY_IO[id].output }; +} + export type CatalogLookupInput = z.infer; export type CatalogLookupOutput = z.infer; export type ReleaseSearchInput = z.infer; diff --git a/packages/contracts/src/capability.ts b/packages/contracts/src/capability.ts index 5fdea40d..3729038c 100644 --- a/packages/contracts/src/capability.ts +++ b/packages/contracts/src/capability.ts @@ -24,37 +24,31 @@ export const MutationPlan = z.object({ }); export type MutationPlan = z.infer; -/** 能力风险级别 */ -export const CapabilityRisk = z.enum(["read", "preview", "write", "destructive"]); -export type CapabilityRisk = z.infer; - /** 能力 ID:`域.动作`,如 download.submit */ export const CapabilityId = z.string().regex(/^[a-z][a-z0-9]*\.[a-z][a-z0-9_]*$/); export type CapabilityId = z.infer; -/** 能力定义(注册中心元数据;处理器在 server 侧挂载) */ -export interface CapabilityDefinition { - id: CapabilityId; - version: number; - description: string; - input: I; - output: O; - scopes: string[]; - risk: CapabilityRisk; - idempotent: boolean; - supportsPreview: boolean; -} - // ── 能力描述符与首批能力(任务书 §3.1-9;REQ §10.3) ── /** 风险级别:low 只读;medium 预览/可撤回写入;high 触发下载、文件变更或策略发布 */ export const RiskLevel = z.enum(["low", "medium", "high"]); export type RiskLevel = z.infer; -/** 接口类别(REQ §12.3):read / preview / propose / approve / execute */ +/** + * 接口类别(REQ §12.3),由弱到强: + * - read:只读查询 + * - preview:生成预览/方案,不落任何变更 + * - propose:生成并保存变更计划(MutationPlan),等待授权 + * - approve:对计划作出授权决定 + * - execute:产生外部副作用(下载提交、Copy/Move、策略发布…) + */ export const CapabilityKind = z.enum(["read", "preview", "propose", "approve", "execute"]); export type CapabilityKind = z.infer; +/** 崩溃恢复时 harness 能否用持久化参数重跑:有副作用的一律 never,只读用 safe */ +export const CapabilityReplay = z.enum(["safe", "never"]); +export type CapabilityReplay = z.infer; + /** 首批能力 ID */ export const CAPABILITY_IDS = [ "catalog.lookup", @@ -78,7 +72,11 @@ export type KnownCapabilityId = z.infer; /** schema 引用:指向 contracts 中导出的 schema 名(如 "ReleaseSearchInput"),便于序列化给 UI/MCP 列表 */ export const SchemaRef = z.string().regex(/^[A-Z][A-Za-z0-9]*$/); -/** 能力注册中心的可序列化描述(处理器在 server 侧挂载;模型只能看到白名单内的描述) */ +/** + * 能力的可序列化描述(唯一定义;处理器在 server 侧挂载,模型只能看到白名单内的描述)。 + * 运行时注册用 `CapabilitySpec`(同字段,inputSchema/outputSchema 换成真实 zod schema), + * 首批能力的 spec 用 `capabilitySpec(id)` 从 CAPABILITY_CATALOG + CAPABILITY_IO 组装。 + */ export const CapabilityDescriptor = z.object({ id: CapabilityId, version: z.number().int().min(1), @@ -87,14 +85,24 @@ export const CapabilityDescriptor = z.object({ kind: CapabilityKind, inputSchema: SchemaRef, outputSchema: SchemaRef, - /** 调用所需 API scope */ + /** 经 HTTP/MCP 调用时所需的 API scope */ scope: ApiScope, riskLevel: RiskLevel, idempotent: z.boolean(), supportsPreview: z.boolean(), + replay: CapabilityReplay, }); export type CapabilityDescriptor = z.infer; +/** 运行时能力规格:描述符字段 + 真实的输入/输出 zod schema(注册中心、Pi 工具、HTTP 共用) */ +export type CapabilitySpec = Omit< + CapabilityDescriptor, + "inputSchema" | "outputSchema" +> & { inputSchema: I; outputSchema: O }; + +/** 需要走授权回调(automation_grant / approval)的类别 */ +export const AUTHORIZED_KINDS: readonly CapabilityKind[] = ["propose", "approve", "execute"]; + /** 首批能力的描述(inputSchema/outputSchema 名对应 capabilities-io.ts 中的导出) */ export const CAPABILITY_CATALOG: Record = { "catalog.lookup": d( @@ -276,5 +284,7 @@ function d( riskLevel, idempotent, supportsPreview, + // 有副作用的类别一律不可重放 + replay: kind === "read" || kind === "preview" ? "safe" : "never", }; } diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index b6b3e436..4057ce60 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -1,4 +1,5 @@ export * from "./agent.ts"; +export * from "./agent-events.ts"; export * from "./api.ts"; export * from "./assets.ts"; export * from "./capabilities-io.ts"; diff --git a/packages/contracts/test/agent-events.test.ts b/packages/contracts/test/agent-events.test.ts new file mode 100644 index 00000000..42c80e33 --- /dev/null +++ b/packages/contracts/test/agent-events.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from "vitest"; +import * as C from "../src/index.ts"; + +const RUN = "7f0f5d64-6a39-4a4e-9a86-0d3a2f1e8b11"; +const AT = "2026-09-27T00:00:00.000Z"; +const usage = { inputTokens: 1, outputTokens: 2, cacheReadTokens: 0, cacheWriteTokens: 0, totalTokens: 3, costUsd: 0 }; + +const samples: C.AgentRunEventBody[] = [ + { type: "run_started", goal: "g", resumed: false }, + { type: "text_delta", delta: "hi" }, + { type: "thinking_delta", delta: "..." }, + { type: "tool_call_start", toolCallId: "t1", capabilityId: "inventory.query", inputSummary: "{}", recovery: false }, + { + type: "tool_call_end", + toolCallId: "t1", + capabilityId: "inventory.query", + isError: true, + outputSummary: "x", + recovery: false, + }, + { type: "tool_call_rejected", toolCallId: "t2", toolName: "bash", reason: "不在白名单" }, + { type: "compaction", phase: "end", reason: "threshold", status: "completed" }, + { type: "run_completed", finalText: "done", usage }, + { type: "run_failed", reason: "budget", message: "超预算", usage, budget: "tokens" }, +]; + +describe("agent 运行事件", () => { + it("样例覆盖全部 9 种事件类型", () => { + expect(samples.map((s) => s.type).sort()).toEqual([...C.AgentRunEventType.options].sort()); + }); + + it("行往返无损,且行 schema 合法", () => { + samples.forEach((body, seq) => { + const e = { ...body, runId: RUN, seq, at: AT } as C.AgentRunEvent; + expect(C.AgentRunEvent.safeParse(e).success, body.type).toBe(true); + const row = C.agentRunEventToRow(e); + expect(C.AgentRunEventRow.safeParse(row).success, body.type).toBe(true); + expect(row.payload).not.toHaveProperty("type"); + expect(C.agentRunEventFromRow(row)).toEqual(e); + }); + }); + + it("映射到全局 SSE 事件:tool phase / run status,且 payload 合法", () => { + const bus = samples.map((body, seq) => + C.agentRunEventToBusEvent({ ...body, runId: RUN, seq, at: AT } as C.AgentRunEvent), + ); + expect( + bus.map((b) => + b + ? `${b.type}:${"phase" in b.payload ? b.payload.phase : "status" in b.payload ? b.payload.status : ""}` + : null, + ), + ).toEqual([ + "agent.run_status:running", + "agent.text_delta:", + null, + "agent.tool_call:started", + "agent.tool_call:failed", + "agent.tool_call:denied", + null, + "agent.run_status:succeeded", + "agent.run_status:budget_exceeded", + ]); + for (const b of bus) if (b) expect(C.parseEventPayload(b.type, b.payload), b.type).not.toBeNull(); + }); +}); diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index bce7846f..f4af6cef 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -40,6 +40,7 @@ const ROWS: Record AcquisitionPlan: { row: C.AcquisitionPlan }, AcquisitionSlot: { row: C.AcquisitionSlot }, AgentRun: { row: C.AgentRun }, + AgentRunEvent: { row: C.AgentRunEventRow }, ApiKey: { row: C.ApiKey, dbOnly: ["keyHash"] }, Approval: { row: C.Approval }, Artifact: { row: C.Artifact }, @@ -80,6 +81,10 @@ const ROWS: Record TitleAlias: { row: C.TitleAlias }, }; +/** 引擎自有表(@umt/tasks 内部使用,不经 API 暴露),不要求 Row schema */ +const INTERNAL_TABLES = ["TasksRunMeta", "TasksFact", "TasksOpLedger"]; +const INTERNAL_ENUMS = ["TasksOpLedgerStatus"]; + /** DB 枚举类型名 → contracts 枚举 */ const ENUMS: Record = { AcquisitionPlanPurpose: C.PlanPurpose, @@ -88,6 +93,7 @@ const ENUMS: Record = { AcquisitionSlotStatus: C.SlotStatus, AgentRunTrigger: C.AgentTrigger, AgentRunStatus: C.AgentRunStatus, + AgentRunEventType: C.AgentRunEventType, ApprovalStatus: C.ApprovalStatus, ArtifactKind: C.ArtifactKind, AssetLocationLocationKind: C.LocationKind, @@ -177,7 +183,7 @@ describe("contracts ⇔ db 表结构同步", () => { const tables = tableColumns(); it("每张表都有 Row schema", () => { - expect([...tables.keys()].sort()).toEqual(Object.keys(ROWS).sort()); + expect([...tables.keys()].filter((t) => !INTERNAL_TABLES.includes(t)).sort()).toEqual(Object.keys(ROWS).sort()); }); for (const [table, spec] of Object.entries(ROWS)) { @@ -190,6 +196,7 @@ describe("contracts ⇔ db 表结构同步", () => { it("每个 DB 枚举都有对应 contracts 枚举且取值一致", () => { const aliases = enumAliases(); + for (const internal of INTERNAL_ENUMS) aliases.delete(internal); expect([...aliases.keys()].sort()).toEqual(Object.keys(ENUMS).sort()); for (const [name, values] of aliases) { expect(ENUMS[name]!.options, name).toEqual(values); diff --git a/packages/contracts/test/schemas.test.ts b/packages/contracts/test/schemas.test.ts index f633593d..360b6990 100644 --- a/packages/contracts/test/schemas.test.ts +++ b/packages/contracts/test/schemas.test.ts @@ -184,6 +184,18 @@ describe("关键合同行为", () => { expect(isSchema(exported[cap.inputSchema]), cap.inputSchema).toBe(true); expect(isSchema(exported[cap.outputSchema]), cap.outputSchema).toBe(true); expect(exported[cap.inputSchema]).toBe(C.CAPABILITY_IO[cap.id as C.KnownCapabilityId].input); + expect(exported[cap.outputSchema]).toBe(C.CAPABILITY_IO[cap.id as C.KnownCapabilityId].output); + } + }); + + it("capabilitySpec 组装目录元数据与真实 schema;有副作用的类别不可重放", () => { + for (const id of C.CAPABILITY_IDS) { + const spec = C.capabilitySpec(id); + const { inputSchema: _i, outputSchema: _o, ...meta } = C.CAPABILITY_CATALOG[id]; + expect(spec).toMatchObject(meta); + expect(spec.inputSchema).toBe(C.CAPABILITY_IO[id].input); + expect(spec.outputSchema).toBe(C.CAPABILITY_IO[id].output); + if (spec.kind !== "read" && spec.kind !== "preview") expect(spec.replay, id).toBe("never"); } }); diff --git a/packages/tasks/migrations/0001_tasks_tables.ts b/packages/tasks/migrations/0001_tasks_tables.ts deleted file mode 100644 index e7ca169b..00000000 --- a/packages/tasks/migrations/0001_tasks_tables.ts +++ /dev/null @@ -1,57 +0,0 @@ -import { type Kysely, sql } from "kysely"; - -// @umt/tasks 自带的三张表(W2 合并进 packages/db 迁移流程): -// - tasks_run_meta:引擎 run 的业务侧元数据(traceId/优先级/取消标志),取消标志是 abortSignal 的来源 -// - tasks_fact:signal 之前先落库的事实;waitForFact 以它为准,信号只用来加速唤醒 -// - tasks_op_ledger:操作账本(fencing),记录 attempt_token,旧 attempt 的提交会被拒绝 - -// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 -export async function up(db: Kysely): Promise { - await db.schema - .createTable("tasks_run_meta") - .addColumn("run_id", "text", (c) => c.primaryKey()) - .addColumn("task_name", "text", (c) => c.notNull()) - .addColumn("task_version", "integer", (c) => c.notNull()) - .addColumn("idempotency_key", "text", (c) => c.notNull()) - .addColumn("trace_id", "text") - .addColumn("priority", "integer", (c) => c.notNull().defaultTo(0)) - .addColumn("cancel_requested_at", "timestamptz") - .addColumn("cancel_reason", "text") - .addColumn("created_at", "timestamptz", (c) => c.notNull().defaultTo(sql`now()`)) - .execute(); - await db.schema - .createIndex("tasks_run_meta_task_idx") - .on("tasks_run_meta") - .columns(["task_name", "created_at"]) - .execute(); - - await db.schema - .createTable("tasks_fact") - .addColumn("id", "bigserial", (c) => c.primaryKey()) - .addColumn("run_id", "text", (c) => c.notNull()) - .addColumn("name", "text", (c) => c.notNull()) - .addColumn("payload", "jsonb") - .addColumn("created_at", "timestamptz", (c) => c.notNull().defaultTo(sql`now()`)) - .execute(); - await db.schema.createIndex("tasks_fact_run_name_idx").on("tasks_fact").columns(["run_id", "name", "id"]).execute(); - - await db.schema - .createTable("tasks_op_ledger") - .addColumn("op_key", "text", (c) => c.primaryKey()) - .addColumn("run_id", "text") - .addColumn("attempt_token", "uuid", (c) => c.notNull()) - .addColumn("attempts", "integer", (c) => c.notNull().defaultTo(1)) - .addColumn("status", "text", (c) => c.notNull().check(sql`status in ('started','succeeded')`)) - .addColumn("result", "jsonb") - .addColumn("last_error", "text") - .addColumn("started_at", "timestamptz", (c) => c.notNull().defaultTo(sql`now()`)) - .addColumn("completed_at", "timestamptz") - .execute(); -} - -// biome-ignore lint/suspicious/noExplicitAny: 同上 -export async function down(db: Kysely): Promise { - await db.schema.dropTable("tasks_op_ledger").ifExists().execute(); - await db.schema.dropTable("tasks_fact").ifExists().execute(); - await db.schema.dropTable("tasks_run_meta").ifExists().execute(); -} diff --git a/packages/tasks/src/migrate.ts b/packages/tasks/src/migrate.ts deleted file mode 100644 index 76b439ed..00000000 --- a/packages/tasks/src/migrate.ts +++ /dev/null @@ -1,25 +0,0 @@ -import { promises as fs } from "node:fs"; -import path from "node:path"; -import { fileURLToPath } from "node:url"; -import { type Kysely, sql } from "kysely"; -import { FileMigrationProvider, type MigrationResultSet, Migrator } from "kysely/migration"; - -/** 本包自带迁移(tasks_run_meta / tasks_fact / tasks_op_ledger),W2 合并进 packages/db */ -export const TASKS_MIGRATIONS_DIR = fileURLToPath(new URL("../migrations", import.meta.url)); - -/** 迁移到最新;表建在连接的默认 schema(search_path)里,迁移记录表独立命名避免与 @umt/db 冲突 */ -// biome-ignore lint/suspicious/noExplicitAny: 迁移器不关心具体表类型 -export async function migrateTasksTables(db: Kysely): Promise { - // 显式指定迁移记录表的 schema:Kysely 默认按表名跨 schema 判断存在性,多个 schema 并存时会误判 - const { rows } = await sql<{ schema: string }>`SELECT current_schema() AS schema`.execute(db); - const migrator = new Migrator({ - db, - provider: new FileMigrationProvider({ fs, path, migrationFolder: TASKS_MIGRATIONS_DIR }), - migrationTableName: "tasks_migration", - migrationLockTableName: "tasks_migration_lock", - migrationTableSchema: rows[0]!.schema, - }); - const result = await migrator.migrateToLatest(); - if (result.error) throw result.error; - return result; -}