refactor(contracts,agent): single capability spec and agent run event definitions in contracts

B3: CapabilityDescriptor gains replay; CapabilitySpec + capabilitySpec(id) assemble
CAPABILITY_CATALOG + CAPABILITY_IO; drop CapabilityDefinition/CapabilityRisk. agent
registry registers contracts CapabilitySpec (kind/allowedKinds instead of its own
scope enum; approve also requires authorize). AgentRun port renamed AgentRunHandle.

B9: contracts AgentRunEvent/AgentRunEventBody/AgentRunEventRow/AgentUsageTotals with
row and SSE mapping helpers; agent onEvent receives AgentRunEvent, delivered serially
in seq order, firstSeq continues numbering across processes.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-27 19:15:49 +08:00
1 parent 8f59bf3772
commit db5352eecd
18 files changed
+553 -293

No files matched your search

+41 -28
View File
@@ -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<I extends z.ZodType, O extends z.ZodType>(
descriptor: CapabilityDescriptor<I, O>,
descriptor: CapabilitySpec<I, O>,
handler: CapabilityHandler<I, O>,
): 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);
}
+11 -36
View File
@@ -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<I extends z.ZodType = z.ZodType, O extends z.ZodType = z.ZodType> {
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<I extends z.ZodType = z.ZodType, O extends z.ZodTy
) => Promise<z.input<O>> | z.input<O>;
export interface AuthorizationRequest {
descriptor: CapabilityDescriptor;
descriptor: CapabilitySpec;
input: unknown;
ctx: InvokeContext;
}
@@ -74,15 +49,15 @@ export type AuthorizationChecker = (
request: AuthorizationRequest,
) => Promise<AuthorizationDecision> | 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"];
+37 -28
View File
@@ -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<void>;
/**
* 每个事件产生时回调(server 用它写 agent_run_event 再推 SSE)。
* 同一 run 内按 seq 顺序串行调用(前一个 Promise 结束才调下一个);回调抛错只记入 onEventError,不影响运行。
* close() 会等待已排队的回调全部完成。
*/
onEvent?: (event: AgentRunEvent) => void | Promise<void>;
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<AgentRun> {
async createRun(spec: CreateRunSpec): Promise<AgentRunHandle> {
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<AgentEvent>();
private readonly channel = new EventChannel<AgentRunEvent>();
private readonly allowed = new Map<string, string>();
private readonly rejected = new Set<string>();
private readonly abortController = new AbortController();
private harness!: AgentHarnessInstance<object>;
private lane!: AgentLane;
private seq = 0;
private readonly firstSeq: number;
private seq: number;
private delivery: Promise<void> = 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<AgentEvent> {
return this.channel.iterate(fromSeq);
events(fromSeq = this.firstSeq): AsyncIterable<AgentRunEvent> {
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,
+25 -57
View File
@@ -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/<runId>.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<AgentEvent>;
/** 已产生的全部事件快照 */
history(): AgentEvent[];
events(fromSeq?: number): AsyncIterable<AgentRunEvent>;
/** 本进程已产生的全部事件快照 */
history(): AgentRunEvent[];
/** 发送一条用户消息并驱动到本轮结束;text 省略时使用 goal */
prompt(text?: string): Promise<RunOutcome>;
/** 崩溃/重启后继续未完成的操作 */
@@ -100,9 +68,9 @@ export interface AgentRun {
/** 领域/服务层只依赖这个端口,不直接接触 Pi SDK 类型 */
export interface AgentRuntimePort {
createRun(spec: CreateRunSpec): Promise<AgentRun>;
createRun(spec: CreateRunSpec): Promise<AgentRunHandle>;
}
export function emptyUsage(): AgentUsage {
export function emptyUsage(): AgentUsageTotals {
return { inputTokens: 0, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0, totalTokens: 0, costUsd: 0 };
}