feat(agent): capability registry, Pi AgentRuntimePort, fake model provider and redaction guard (W1-5)
- CapabilityRegistry: descriptor(id/version/schema/scope/risk/idempotent/preview/replay), invoke = scope limit -> input schema -> server-side authorize -> idempotency key -> handler -> output schema -> audit; capability id <-> tool name mapping (download.submit <-> download__submit) - PiAgentRuntime (pi-agent-core AgentHarness 0.87.1): whitelist tools + before_tool re-check, redacted event stream, JSONL session at sessionDir/<runId>.jsonl, resume/abort/compact, tool-call/duration/token/cost budgets -> abort + run_failed(budget) - AgentSettings -> pi-ai createProvider (anthropic / openai-compatible), compaction mapping - FakeModelProvider + createFakeRuntime (scripted, transcript-derived, survives crash/resume) - buildSystemPrompt, redactForModel; 27 tests incl. SIGKILL + resume Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
rootandClaude Fable 5.1 committed
1 parent
1262340856
commit
9c881e737f
24 files changed
+2419
-8
No files matched your search
@@ -0,0 +1,209 @@
|
||||
import {
|
||||
type AssistantMessage,
|
||||
createModels,
|
||||
type FauxContentBlock,
|
||||
type FauxResponseFactory,
|
||||
fauxAssistantMessage,
|
||||
fauxProvider,
|
||||
fauxText,
|
||||
fauxThinking,
|
||||
fauxToolCall,
|
||||
type JsonObject,
|
||||
type Message,
|
||||
type TranscriptContext,
|
||||
} from "@earendil-works/pi-ai";
|
||||
import type { CapabilityRegistry } from "../capability/registry";
|
||||
import { capabilityIdToToolName, toolNameToCapabilityId } from "../capability/tool-name";
|
||||
import type { ResolvedPiModel } from "./model";
|
||||
import { PiAgentRuntime, type PiAgentRuntimeOptions } from "./runtime";
|
||||
|
||||
/** 前面各步工具调用的结果(按发生顺序),供脚本根据结果决定下一步 */
|
||||
export interface FakeToolResult {
|
||||
toolName: string;
|
||||
capabilityId?: string;
|
||||
isError: boolean;
|
||||
text: string;
|
||||
/** text 能解析成 JSON 时的值 */
|
||||
json?: unknown;
|
||||
}
|
||||
|
||||
export interface FakeToolCall {
|
||||
/** 能力 ID(自动映射为工具名) */
|
||||
capability?: string;
|
||||
/** 原始工具名(用于模拟模型调用白名单外的工具,如 bash) */
|
||||
toolName?: string;
|
||||
input: JsonObject | ((results: FakeToolResult[]) => JsonObject);
|
||||
}
|
||||
|
||||
export interface FakeStep {
|
||||
thinking?: string;
|
||||
text?: string;
|
||||
toolCalls?: FakeToolCall[];
|
||||
}
|
||||
|
||||
export interface FakeScript {
|
||||
name?: string;
|
||||
/** 匹配用户消息(通常就是 goal) */
|
||||
match: string | RegExp | ((goal: string) => boolean);
|
||||
steps: FakeStep[];
|
||||
finalText: string | ((results: FakeToolResult[]) => string);
|
||||
}
|
||||
|
||||
export interface FakeModelProviderOptions {
|
||||
contextWindow?: number;
|
||||
maxTokens?: number;
|
||||
/** 压缩请求时返回的摘要文本 */
|
||||
summaryText?: string;
|
||||
}
|
||||
|
||||
const SUMMARY_MARKERS = ["context summarization assistant", "structured context checkpoint summary"];
|
||||
|
||||
/**
|
||||
* 脚本化假模型:无状态地从 transcript 推导当前步骤(最近一条用户消息之后已有几条 assistant 消息),
|
||||
* 因此进程崩溃后在新进程里 resume 也能从正确的步骤继续。不联网、不需要 API Key。
|
||||
*/
|
||||
export class FakeModelProvider {
|
||||
private readonly faux;
|
||||
private readonly scripts: FakeScript[];
|
||||
private readonly summaryText: string;
|
||||
/** 已收到的模型请求数(含压缩摘要请求) */
|
||||
calls = 0;
|
||||
/** 最近一次请求看到的系统提示与工具名(测试断言用) */
|
||||
lastRequest: { transcript: string } | undefined;
|
||||
|
||||
constructor(scripts: FakeScript[], options: FakeModelProviderOptions = {}) {
|
||||
this.scripts = scripts;
|
||||
this.summaryText = options.summaryText ?? "## 摘要\n目标:见首条消息。已完成动作:见工具结果。剩余事项:继续执行。";
|
||||
this.faux = fauxProvider({
|
||||
provider: "umt-fake",
|
||||
models: [
|
||||
{
|
||||
id: "fake-model",
|
||||
contextWindow: options.contextWindow ?? 64_000,
|
||||
maxTokens: options.maxTokens ?? 4096,
|
||||
reasoning: true,
|
||||
},
|
||||
],
|
||||
tokensPerSecond: 0,
|
||||
});
|
||||
const factory: FauxResponseFactory = (context) => {
|
||||
this.faux.appendResponses([factory]);
|
||||
return this.respond(context);
|
||||
};
|
||||
this.faux.setResponses([factory]);
|
||||
}
|
||||
|
||||
/** @internal 供 PiAgentRuntime 使用的模型装配 */
|
||||
resolve(): ResolvedPiModel {
|
||||
const models = createModels();
|
||||
models.setProvider(this.faux.provider);
|
||||
return { models, model: this.faux.getModel() };
|
||||
}
|
||||
|
||||
private respond(context: TranscriptContext): AssistantMessage {
|
||||
this.calls += 1;
|
||||
const messages = context.messages as Message[];
|
||||
const transcript = messages.map(messageText).join("\n");
|
||||
this.lastRequest = { transcript };
|
||||
const head = messages[0] ? messageText(messages[0]) : "";
|
||||
if (SUMMARY_MARKERS.some((m) => head.includes(m) || lastUserText(messages).includes(m))) {
|
||||
return fauxAssistantMessage(this.summaryText);
|
||||
}
|
||||
|
||||
const lastUser = lastUserIndex(messages);
|
||||
const goal = lastUser >= 0 ? messageText(messages[lastUser] as Message) : "";
|
||||
const script = this.scripts.find((s) => matches(s.match, goal));
|
||||
if (!script) {
|
||||
return fauxAssistantMessage(`[fake] 没有脚本匹配目标:${goal.slice(0, 80)}`, {
|
||||
stopReason: "error",
|
||||
errorMessage: "no fake script matched",
|
||||
});
|
||||
}
|
||||
const after = messages.slice(lastUser + 1);
|
||||
const stepIndex = after.filter((m) => m.role === "assistant").length;
|
||||
const results = toolResults(after);
|
||||
const step = script.steps[stepIndex];
|
||||
if (!step) {
|
||||
const text = typeof script.finalText === "function" ? script.finalText(results) : script.finalText;
|
||||
return fauxAssistantMessage(text);
|
||||
}
|
||||
const blocks: FauxContentBlock[] = [];
|
||||
if (step.thinking) blocks.push(fauxThinking(step.thinking));
|
||||
if (step.text) blocks.push(fauxText(step.text));
|
||||
for (const call of step.toolCalls ?? []) {
|
||||
const name = call.toolName ?? capabilityIdToToolName(call.capability ?? "");
|
||||
blocks.push(fauxToolCall(name, typeof call.input === "function" ? call.input(results) : call.input));
|
||||
}
|
||||
const hasTools = (step.toolCalls?.length ?? 0) > 0;
|
||||
return fauxAssistantMessage(blocks, hasTools ? { stopReason: "toolUse" } : {});
|
||||
}
|
||||
}
|
||||
|
||||
export interface CreateFakeRuntimeOptions extends Omit<PiAgentRuntimeOptions, "resolveModel"> {
|
||||
registry: CapabilityRegistry;
|
||||
scripts: FakeScript[];
|
||||
fake?: FakeModelProviderOptions;
|
||||
}
|
||||
|
||||
/** e2e/测试便捷函数:脚本化假模型 + 真实 PiAgentRuntime(注册中心、白名单、预算、JSONL 全部走真实路径) */
|
||||
export function createFakeRuntime(options: CreateFakeRuntimeOptions): {
|
||||
runtime: PiAgentRuntime;
|
||||
fake: FakeModelProvider;
|
||||
} {
|
||||
const { scripts, fake: fakeOptions, ...rest } = options;
|
||||
const fake = new FakeModelProvider(scripts, fakeOptions);
|
||||
return { runtime: new PiAgentRuntime({ ...rest, resolveModel: () => fake.resolve() }), fake };
|
||||
}
|
||||
|
||||
/** 供 settings 使用的假设置(provider 字段不会真正联网,模型由 FakeModelProvider 覆盖) */
|
||||
export const FAKE_AGENT_SETTINGS = {
|
||||
provider: "anthropic",
|
||||
apiKey: "fake-key-not-used",
|
||||
model: "fake-model",
|
||||
} as const;
|
||||
|
||||
function matches(match: FakeScript["match"], goal: string): boolean {
|
||||
if (typeof match === "string") return goal.includes(match);
|
||||
if (match instanceof RegExp) return match.test(goal);
|
||||
return match(goal);
|
||||
}
|
||||
|
||||
function messageText(m: Message): string {
|
||||
if (typeof m.content === "string") return m.content;
|
||||
return (m.content as Array<{ type: string; text?: string }>)
|
||||
.map((c) => (c.type === "text" ? (c.text ?? "") : ""))
|
||||
.join("");
|
||||
}
|
||||
|
||||
function lastUserIndex(messages: Message[]): number {
|
||||
for (let i = messages.length - 1; i >= 0; i--) if (messages[i]?.role === "user") return i;
|
||||
return -1;
|
||||
}
|
||||
|
||||
function lastUserText(messages: Message[]): string {
|
||||
const i = lastUserIndex(messages);
|
||||
return i >= 0 ? messageText(messages[i] as Message) : "";
|
||||
}
|
||||
|
||||
function toolResults(messages: Message[]): FakeToolResult[] {
|
||||
const out: FakeToolResult[] = [];
|
||||
for (const m of messages) {
|
||||
if (m.role !== "toolResult") continue;
|
||||
const text = messageText(m);
|
||||
let json: unknown;
|
||||
try {
|
||||
json = JSON.parse(text);
|
||||
} catch {
|
||||
json = undefined;
|
||||
}
|
||||
const capabilityId = toolNameToCapabilityId(m.toolName);
|
||||
out.push({
|
||||
toolName: m.toolName,
|
||||
...(capabilityId ? { capabilityId } : {}),
|
||||
isError: m.isError,
|
||||
text,
|
||||
...(json !== undefined ? { json } : {}),
|
||||
});
|
||||
}
|
||||
return out;
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
import type { CompactionSettings } from "@earendil-works/pi-agent-core";
|
||||
import { type Api, createModels, createProvider, type Model, type MutableModels } from "@earendil-works/pi-ai";
|
||||
import { anthropicMessagesApi } from "@earendil-works/pi-ai/api/anthropic-messages.lazy";
|
||||
import { openAICompletionsApi } from "@earendil-works/pi-ai/api/openai-completions.lazy";
|
||||
import type { ResolvedAgentSettings } from "../settings";
|
||||
|
||||
/** Pi 侧需要的模型装配结果(仅在 src/pi/* 内部流转) */
|
||||
export interface ResolvedPiModel {
|
||||
models: MutableModels;
|
||||
model: Model<Api>;
|
||||
}
|
||||
|
||||
export type PiModelResolver = (settings: ResolvedAgentSettings) => ResolvedPiModel;
|
||||
|
||||
const ANTHROPIC_DEFAULT_BASE = "https://api.anthropic.com";
|
||||
const PROVIDER_ID = { anthropic: "umt-anthropic", "openai-compatible": "umt-openai-compatible" } as const;
|
||||
|
||||
/**
|
||||
* 设置 → pi-ai createProvider。apiKey 只在请求时经 auth.resolve 交给 SDK,不读环境变量。
|
||||
* Anthropic baseUrl 不带 /v1;OpenAI 兼容 baseUrl 带 /v1(与 spike 抓包结论一致)。
|
||||
*/
|
||||
export function resolveModelFromSettings(settings: ResolvedAgentSettings): ResolvedPiModel {
|
||||
const providerId = PROVIDER_ID[settings.provider];
|
||||
const common = {
|
||||
id: settings.model,
|
||||
name: settings.model,
|
||||
provider: providerId,
|
||||
reasoning: settings.thinkingLevel !== "off",
|
||||
input: ["text"] as ("text" | "image")[],
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
|
||||
contextWindow: settings.contextBudgetTokens,
|
||||
maxTokens: Math.min(settings.maxOutputTokens, Math.floor(settings.contextBudgetTokens / 2)),
|
||||
};
|
||||
const apiKey = settings.apiKey;
|
||||
const auth = { apiKey: { name: "umt-secret", resolve: async () => ({ auth: { apiKey } }) } };
|
||||
const models = createModels();
|
||||
let model: Model<Api>;
|
||||
if (settings.provider === "anthropic") {
|
||||
const m: Model<"anthropic-messages"> = {
|
||||
...common,
|
||||
api: "anthropic-messages",
|
||||
baseUrl: (settings.baseUrl ?? ANTHROPIC_DEFAULT_BASE).replace(/\/+$/, "").replace(/\/v1$/, ""),
|
||||
};
|
||||
models.setProvider(createProvider({ id: providerId, auth, models: [m], api: anthropicMessagesApi() }));
|
||||
model = m;
|
||||
} else {
|
||||
if (!settings.baseUrl) throw new Error("openai-compatible 上游必须配置 baseUrl(带 /v1)");
|
||||
const m: Model<"openai-completions"> = {
|
||||
...common,
|
||||
api: "openai-completions",
|
||||
baseUrl: settings.baseUrl.replace(/\/+$/, ""),
|
||||
compat: { supportsDeveloperRole: false },
|
||||
};
|
||||
models.setProvider(createProvider({ id: providerId, auth, models: [m], api: openAICompletionsApi() }));
|
||||
model = m;
|
||||
}
|
||||
return { models, model };
|
||||
}
|
||||
|
||||
/** 压缩设置:threshold 模式下占用达到 threshold 比例时触发;manual 只允许手动 compact() */
|
||||
export function compactionFromSettings(settings: ResolvedAgentSettings, contextWindow: number): CompactionSettings {
|
||||
const reserveTokens = Math.max(256, Math.round(contextWindow * (1 - settings.compaction.threshold)));
|
||||
return {
|
||||
enabled: settings.compaction.mode === "threshold",
|
||||
reserveTokens,
|
||||
keepRecentTokens: Math.max(256, Math.round(contextWindow * 0.25)),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,439 @@
|
||||
import {
|
||||
AgentHarness,
|
||||
type AgentHarness as AgentHarnessInstance,
|
||||
type AgentHarnessTool,
|
||||
type AgentLane,
|
||||
type AgentMessage,
|
||||
BACKGROUND_CONTEXT,
|
||||
type Context,
|
||||
type JsonlSessionRepo,
|
||||
type RunResult,
|
||||
} from "@earendil-works/pi-agent-core";
|
||||
import type { Usage } from "@earendil-works/pi-ai";
|
||||
import type { TSchema } from "typebox";
|
||||
import { isCapabilityError } from "../capability/errors";
|
||||
import type { CapabilityRegistry, RegisteredCapability } from "../capability/registry";
|
||||
import { toolNameToCapabilityId } from "../capability/tool-name";
|
||||
import { redactForModel, redactString, summarizeForLog } from "../redact";
|
||||
import { EventChannel } from "../runtime/event-channel";
|
||||
import {
|
||||
type AgentEvent,
|
||||
type AgentEventBody,
|
||||
type AgentRun,
|
||||
type AgentRuntimePort,
|
||||
type AgentUsage,
|
||||
type BudgetKind,
|
||||
type CreateRunSpec,
|
||||
emptyUsage,
|
||||
type RunOutcome,
|
||||
} from "../runtime/port";
|
||||
import { type ResolvedAgentSettings, resolveAgentSettings } from "../settings";
|
||||
import { compactionFromSettings, type PiModelResolver, resolveModelFromSettings } from "./model";
|
||||
import { openRunSession } from "./session";
|
||||
|
||||
export interface PiAgentRuntimeOptions {
|
||||
registry: CapabilityRegistry;
|
||||
/** 默认按设置构造真实上游;测试用 FakeModelProvider 覆盖 */
|
||||
resolveModel?: PiModelResolver;
|
||||
/** 每个事件产生时回调(server 用它写 agent_run_event 并推 SSE) */
|
||||
onEvent?: (event: AgentEvent) => void | Promise<void>;
|
||||
now?: () => Date;
|
||||
}
|
||||
|
||||
export const DEFAULT_COMPACTION_INSTRUCTIONS =
|
||||
"保留:本次目标、约束与授权范围、证据引用(对象 ID/记录)、已完成的动作及其结果(已提交≠已入库)、剩余事项。不要保留任何凭证。";
|
||||
|
||||
const LANE = "main";
|
||||
const SUMMARY_LIMIT = 600;
|
||||
|
||||
/** Pi(pi-agent-core AgentHarness)实现的 AgentRuntimePort */
|
||||
export class PiAgentRuntime implements AgentRuntimePort {
|
||||
constructor(private readonly options: PiAgentRuntimeOptions) {}
|
||||
|
||||
async createRun(spec: CreateRunSpec): Promise<AgentRun> {
|
||||
const ctx = BACKGROUND_CONTEXT;
|
||||
const settings = resolveAgentSettings(spec.settings);
|
||||
const capabilities = this.options.registry.resolve(spec.capabilities);
|
||||
const { models, model } = (this.options.resolveModel ?? resolveModelFromSettings)(settings);
|
||||
const opened = await openRunSession(spec.sessionDir, spec.runId, ctx);
|
||||
|
||||
const run = new PiAgentRun(spec, settings, capabilities, this.options, opened.path, opened.repo);
|
||||
try {
|
||||
const { harness, open } = await AgentHarness.create(
|
||||
{
|
||||
session: opened.session,
|
||||
models,
|
||||
model,
|
||||
systemPrompt: spec.systemPrompt,
|
||||
tools: capabilities.map((c) => run.toTool(c)),
|
||||
toolContext: {},
|
||||
thinkingLevel: settings.thinkingLevel,
|
||||
compaction: compactionFromSettings(settings, model.contextWindow),
|
||||
retry: { enabled: settings.maxRetries > 0, maxRetries: settings.maxRetries, baseDelayMs: 1000 },
|
||||
streamOptions: { timeoutMs: settings.requestTimeoutMs, maxRetries: settings.maxRetries },
|
||||
toolExecution: "sequential",
|
||||
},
|
||||
ctx,
|
||||
);
|
||||
const lane = await harness.lane(LANE, ctx);
|
||||
run.attach(harness, lane, open.filter((o) => o.lane === LANE).length);
|
||||
} catch (error) {
|
||||
await opened.session.close(ctx).catch(() => undefined);
|
||||
await opened.repo.close(ctx).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
return run;
|
||||
}
|
||||
}
|
||||
|
||||
type BudgetState = { kind: BudgetKind; message: string };
|
||||
|
||||
class PiAgentRun implements AgentRun {
|
||||
readonly runId: string;
|
||||
readonly sessionPath: string;
|
||||
pendingOperations = 0;
|
||||
|
||||
private readonly channel = new EventChannel<AgentEvent>();
|
||||
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 toolCalls = 0;
|
||||
private usage: AgentUsage = emptyUsage();
|
||||
private finalText = "";
|
||||
private budget: BudgetState | undefined;
|
||||
private userAbort: string | undefined;
|
||||
private startedAt: number | undefined;
|
||||
private deadlineTimer: NodeJS.Timeout | undefined;
|
||||
private closed = false;
|
||||
|
||||
constructor(
|
||||
private readonly spec: CreateRunSpec,
|
||||
private readonly settings: ResolvedAgentSettings,
|
||||
capabilities: RegisteredCapability[],
|
||||
private readonly options: PiAgentRuntimeOptions,
|
||||
sessionPath: string,
|
||||
private readonly repo: JsonlSessionRepo,
|
||||
) {
|
||||
this.runId = spec.runId;
|
||||
this.sessionPath = sessionPath;
|
||||
for (const c of capabilities) this.allowed.set(c.toolName, c.descriptor.id);
|
||||
}
|
||||
|
||||
/** 能力 → Pi 工具。执行时统一走注册中心 invoke(schema/作用域/授权/幂等/审计) */
|
||||
toTool(entry: RegisteredCapability): AgentHarnessTool<object> {
|
||||
const { descriptor } = entry;
|
||||
const registry = this.options.registry;
|
||||
const actor = this.spec.actor;
|
||||
return {
|
||||
name: entry.toolName,
|
||||
label: descriptor.title,
|
||||
description: `${descriptor.description}\n[能力 ${descriptor.id} v${descriptor.version};作用域 ${descriptor.scope};风险 ${descriptor.riskLevel}]`,
|
||||
parameters: registry.inputJsonSchema(descriptor.id) as unknown as TSchema,
|
||||
replay: descriptor.replay,
|
||||
...(descriptor.scope === "execute" ? { executionMode: "sequential" as const } : {}),
|
||||
execute: async (toolCallId, params, _onUpdate, _toolCtx, invocation, context) => {
|
||||
const signal = context.abortSignal ?? this.abortController.signal;
|
||||
try {
|
||||
const output = await registry.invoke(descriptor.id, params, {
|
||||
actorId: actor.actorId,
|
||||
taskId: actor.taskId,
|
||||
scopeIds: actor.scopeIds,
|
||||
...(actor.allowedScopes ? { allowedScopes: actor.allowedScopes } : {}),
|
||||
signal,
|
||||
source: "agent",
|
||||
invocationId: invocation.invocationId ?? toolCallId,
|
||||
agentRunId: this.runId,
|
||||
});
|
||||
return {
|
||||
content: [{ type: "text", text: JSON.stringify(redactForModel(output) ?? null) }],
|
||||
details: undefined,
|
||||
};
|
||||
} catch (error) {
|
||||
if (isCapabilityError(error)) {
|
||||
throw new Error(JSON.stringify({ error: { code: error.code, message: redactString(error.message) } }));
|
||||
}
|
||||
throw new Error(redactString(error instanceof Error ? error.message : String(error)));
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
attach(harness: AgentHarnessInstance<object>, lane: AgentLane, pendingOperations: number): void {
|
||||
this.harness = harness;
|
||||
this.lane = lane;
|
||||
this.pendingOperations = pendingOperations;
|
||||
const { events, hooks } = harness;
|
||||
|
||||
// 白名单二次校验 + 工具调用次数预算
|
||||
hooks.on("before_tool", (event) => {
|
||||
const decision = checkToolCall(event.toolName, this.allowed);
|
||||
if (!decision.allowed) {
|
||||
this.reject(event.toolCallId, event.toolName, decision.reason);
|
||||
return { block: { reason: decision.reason } };
|
||||
}
|
||||
this.toolCalls += 1;
|
||||
if (this.toolCalls > this.settings.maxToolCalls) {
|
||||
const reason = `工具调用次数超出预算(${this.settings.maxToolCalls})`;
|
||||
this.exceed("tool_calls", reason);
|
||||
this.rejected.add(event.toolCallId);
|
||||
this.emit({
|
||||
type: "tool_call_rejected",
|
||||
toolCallId: event.toolCallId,
|
||||
toolName: event.toolName,
|
||||
capabilityId: decision.capabilityId,
|
||||
reason,
|
||||
});
|
||||
return { block: { reason, terminate: true } };
|
||||
}
|
||||
return undefined;
|
||||
});
|
||||
hooks.on("before_request", () => {
|
||||
this.checkDeadline();
|
||||
if (this.budget) this.scheduleAbort();
|
||||
return undefined;
|
||||
});
|
||||
|
||||
events.on("message_update", (e) => {
|
||||
if (e.event.type === "text_delta") this.emit({ type: "text_delta", delta: e.event.delta });
|
||||
else if (e.event.type === "thinking_delta") this.emit({ type: "thinking_delta", delta: e.event.delta });
|
||||
});
|
||||
events.on("message_end", (e) => {
|
||||
const text = assistantText(e.message);
|
||||
if (text?.trim()) this.finalText = text;
|
||||
});
|
||||
events.on("tool_start", (e) => {
|
||||
if (this.rejected.has(e.toolCallId)) return;
|
||||
const capabilityId = this.allowed.get(e.toolName);
|
||||
if (!capabilityId) {
|
||||
// 模型调用了未注册的工具(例如 bash):harness 已合成 unavailable 错误,这里只产生拒绝事件
|
||||
this.reject(e.toolCallId, e.toolName, `工具 ${e.toolName} 不在本次白名单内`);
|
||||
return;
|
||||
}
|
||||
this.emit({
|
||||
type: "tool_call_start",
|
||||
toolCallId: e.toolCallId,
|
||||
capabilityId,
|
||||
inputSummary: summarizeForLog(e.args, SUMMARY_LIMIT),
|
||||
recovery: e.recovery === true,
|
||||
});
|
||||
});
|
||||
events.on("tool_end", (e) => {
|
||||
if (this.rejected.has(e.toolCallId)) return;
|
||||
const capabilityId = this.allowed.get(e.toolName) ?? e.toolName;
|
||||
const text = e.result.content
|
||||
.map((c) => (c.type === "text" ? c.text : `[${c.type}]`))
|
||||
.join("")
|
||||
.trim();
|
||||
this.emit({
|
||||
type: "tool_call_end",
|
||||
toolCallId: e.toolCallId,
|
||||
capabilityId,
|
||||
isError: e.isError,
|
||||
outputSummary: summarizeForLog(text, SUMMARY_LIMIT),
|
||||
recovery: e.recovery === true,
|
||||
});
|
||||
});
|
||||
events.on("usage", (e) => {
|
||||
this.usage = toAgentUsage(e.totals);
|
||||
const cost = this.settings.costBudget;
|
||||
if (cost?.maxTotalTokens !== undefined && this.usage.totalTokens > cost.maxTotalTokens) {
|
||||
this.exceed("tokens", `token 用量 ${this.usage.totalTokens} 超出预算 ${cost.maxTotalTokens}`);
|
||||
} else if (cost?.maxCostUsd !== undefined && this.usage.costUsd > cost.maxCostUsd) {
|
||||
this.exceed("cost", `费用 $${this.usage.costUsd.toFixed(4)} 超出预算 $${cost.maxCostUsd}`);
|
||||
}
|
||||
});
|
||||
events.on("compaction_start", (e) => this.emit({ type: "compaction", phase: "start", reason: e.reason }));
|
||||
events.on("compaction_end", (e) =>
|
||||
this.emit({ type: "compaction", phase: "end", reason: e.reason, status: e.status }),
|
||||
);
|
||||
}
|
||||
|
||||
events(fromSeq = 0): AsyncIterable<AgentEvent> {
|
||||
return this.channel.iterate(fromSeq);
|
||||
}
|
||||
|
||||
history(): AgentEvent[] {
|
||||
return this.channel.snapshot();
|
||||
}
|
||||
|
||||
async prompt(text?: string): Promise<RunOutcome> {
|
||||
const message = text ?? this.spec.goal;
|
||||
return this.drive(false, () => this.lane.prompt(message, undefined, BACKGROUND_CONTEXT));
|
||||
}
|
||||
|
||||
async resume(): Promise<RunOutcome> {
|
||||
return this.drive(true, async () => {
|
||||
const r = await this.lane.resume(BACKGROUND_CONTEXT);
|
||||
if (!r.ok && r.error._tag === "NothingToResume") return undefined;
|
||||
return r as RunResult;
|
||||
});
|
||||
}
|
||||
|
||||
async abort(reason = "用户取消"): Promise<void> {
|
||||
this.userAbort = reason;
|
||||
this.abortController.abort(reason);
|
||||
await this.lane.abort(BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
}
|
||||
|
||||
async compact(customInstructions = DEFAULT_COMPACTION_INSTRUCTIONS): Promise<void> {
|
||||
const r = await this.lane.compact({ customInstructions }, BACKGROUND_CONTEXT);
|
||||
if (!r.ok && r.error._tag !== "NothingToCompact") throw new Error(`压缩失败: ${r.error._tag}`);
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
this.clearDeadline();
|
||||
const ctx: Context = BACKGROUND_CONTEXT;
|
||||
await this.harness?.close(ctx).catch(() => undefined);
|
||||
await this.repo.close(ctx).catch(() => undefined);
|
||||
this.channel.close();
|
||||
}
|
||||
|
||||
// ---- internals ----
|
||||
|
||||
private async drive(resumed: boolean, start: () => Promise<RunResult | undefined>): Promise<RunOutcome> {
|
||||
if (this.closed) throw new Error(`run ${this.runId} 已关闭`);
|
||||
this.startedAt ??= Date.now();
|
||||
this.emit({ type: "run_started", goal: this.spec.goal, resumed });
|
||||
this.checkDeadline();
|
||||
if (this.budget) return this.finish(undefined);
|
||||
this.armDeadline();
|
||||
let result: RunResult | undefined;
|
||||
try {
|
||||
result = await start();
|
||||
} finally {
|
||||
this.clearDeadline();
|
||||
}
|
||||
if (result && !result.ok) throw new Error(`Pi 拒绝了本次操作: ${result.error._tag} ${result.error.message}`);
|
||||
this.pendingOperations = 0;
|
||||
return this.finish(result);
|
||||
}
|
||||
|
||||
private finish(result: RunResult | undefined): RunOutcome {
|
||||
const record = result?.ok ? result.value : undefined;
|
||||
const status = record && "status" in record ? record.status : undefined;
|
||||
let outcome: RunOutcome;
|
||||
if (this.budget) {
|
||||
outcome = this.failed("budget", this.budget.message, this.budget.kind);
|
||||
} else if (status === "aborted" || this.userAbort) {
|
||||
outcome = this.failed("aborted", this.userAbort ?? "已中止");
|
||||
} else if (status === "failed") {
|
||||
const err = record && "error" in record ? record.error : undefined;
|
||||
outcome = this.failed("error", redactString(err ? `${err.code}: ${err.message}` : "运行失败"));
|
||||
} else if (status === "suspended") {
|
||||
outcome = this.failed("error", "上游进入延迟模式(deferred),v0.1 不支持");
|
||||
} else {
|
||||
outcome = { status: "completed", finalText: this.finalText, usage: this.usage };
|
||||
this.emit({ type: "run_completed", finalText: outcome.finalText, usage: this.usage });
|
||||
return outcome;
|
||||
}
|
||||
this.emit({
|
||||
type: "run_failed",
|
||||
reason: outcome.reason ?? "error",
|
||||
message: outcome.message ?? "",
|
||||
usage: this.usage,
|
||||
...(outcome.budget ? { budget: outcome.budget } : {}),
|
||||
});
|
||||
return outcome;
|
||||
}
|
||||
|
||||
private failed(reason: RunOutcome["reason"] & string, message: string, budget?: BudgetKind): RunOutcome {
|
||||
return {
|
||||
status: "failed",
|
||||
reason,
|
||||
message,
|
||||
finalText: this.finalText,
|
||||
usage: this.usage,
|
||||
...(budget ? { budget } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
private reject(toolCallId: string, toolName: string, reason: string): void {
|
||||
if (this.rejected.has(toolCallId)) return;
|
||||
this.rejected.add(toolCallId);
|
||||
const capabilityId = toolNameToCapabilityId(toolName);
|
||||
this.emit({ type: "tool_call_rejected", toolCallId, toolName, ...(capabilityId ? { capabilityId } : {}), reason });
|
||||
}
|
||||
|
||||
private exceed(kind: BudgetKind, message: string): void {
|
||||
if (this.budget) return;
|
||||
this.budget = { kind, message };
|
||||
this.scheduleAbort();
|
||||
}
|
||||
|
||||
/** 不能在事件/钩子回调里同步等待 abort(会与事件投递互相等待),放到下一个宏任务 */
|
||||
private scheduleAbort(): void {
|
||||
setTimeout(() => {
|
||||
this.abortController.abort("budget");
|
||||
void this.lane.abort(BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
}, 0);
|
||||
}
|
||||
|
||||
private checkDeadline(): void {
|
||||
if (this.startedAt !== undefined && Date.now() - this.startedAt > this.settings.maxDurationMs) {
|
||||
this.exceed("duration", `运行时长超出预算(${this.settings.maxDurationMs}ms)`);
|
||||
}
|
||||
}
|
||||
|
||||
private armDeadline(): void {
|
||||
const remaining = (this.startedAt ?? Date.now()) + this.settings.maxDurationMs - Date.now();
|
||||
this.deadlineTimer = setTimeout(
|
||||
() => this.exceed("duration", `运行时长超出预算(${this.settings.maxDurationMs}ms)`),
|
||||
Math.max(0, remaining),
|
||||
);
|
||||
}
|
||||
|
||||
private clearDeadline(): void {
|
||||
if (this.deadlineTimer) clearTimeout(this.deadlineTimer);
|
||||
this.deadlineTimer = undefined;
|
||||
}
|
||||
|
||||
private emit(body: AgentEventBody): void {
|
||||
const event = { ...body, runId: this.runId, seq: this.seq++, at: new Date().toISOString() } as AgentEvent;
|
||||
this.channel.push(event);
|
||||
if (this.options.onEvent) {
|
||||
Promise.resolve()
|
||||
.then(() => this.options.onEvent?.(event))
|
||||
.catch(() => undefined);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** before_tool 白名单判定(纯函数,便于单测) */
|
||||
export function checkToolCall(
|
||||
toolName: string,
|
||||
allowed: ReadonlyMap<string, string>,
|
||||
): { allowed: true; capabilityId: string } | { allowed: false; reason: string; capabilityId?: string } {
|
||||
const capabilityId = allowed.get(toolName);
|
||||
if (capabilityId) return { allowed: true, capabilityId };
|
||||
const mapped = toolNameToCapabilityId(toolName);
|
||||
return {
|
||||
allowed: false,
|
||||
reason: `工具 ${toolName} 不在本次白名单内`,
|
||||
...(mapped ? { capabilityId: mapped } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function assistantText(message: AgentMessage): string | undefined {
|
||||
const m = message as { role?: string; content?: unknown };
|
||||
if (m.role !== "assistant" || !Array.isArray(m.content)) return undefined;
|
||||
return (m.content as Array<{ type: string; text?: string }>)
|
||||
.filter((c) => c.type === "text")
|
||||
.map((c) => c.text ?? "")
|
||||
.join("");
|
||||
}
|
||||
|
||||
function toAgentUsage(u: Usage): AgentUsage {
|
||||
return {
|
||||
inputTokens: u.input,
|
||||
outputTokens: u.output,
|
||||
cacheReadTokens: u.cacheRead,
|
||||
cacheWriteTokens: u.cacheWrite,
|
||||
totalTokens: u.totalTokens,
|
||||
costUsd: u.cost.total,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
import { mkdir, readFile, rename, rmdir, stat } from "node:fs/promises";
|
||||
import { dirname, join, resolve } from "node:path";
|
||||
import { type Context, type JsonlSessionMetadata, JsonlSessionRepo, type Session } from "@earendil-works/pi-agent-core";
|
||||
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/harness/env/nodejs";
|
||||
|
||||
export interface OpenedSession {
|
||||
session: Session<JsonlSessionMetadata>;
|
||||
repo: JsonlSessionRepo;
|
||||
path: string;
|
||||
created: boolean;
|
||||
}
|
||||
|
||||
const SAFE_RUN_ID = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/;
|
||||
|
||||
export function sessionPathFor(sessionDir: string, runId: string): string {
|
||||
if (!SAFE_RUN_ID.test(runId) || runId.includes("..")) throw new Error(`runId 不能用作文件名: ${runId}`);
|
||||
return join(resolve(sessionDir), `${runId}.jsonl`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 打开/创建 sessionDir/<runId>.jsonl。
|
||||
* JsonlSessionRepo 默认会把文件放到 `<root>/--cwd--/<时间戳>_<id>.jsonl`;这里创建后把空会话文件
|
||||
* (只有 header)移到固定路径,再按元数据打开,之后所有写入都追加到固定路径。
|
||||
* NodeExecutionEnv 只当作会话目录的 FileSystem 使用,它的 exec/shell 从不暴露给模型。
|
||||
*/
|
||||
export async function openRunSession(sessionDir: string, runId: string, ctx: Context): Promise<OpenedSession> {
|
||||
const root = resolve(sessionDir);
|
||||
await mkdir(root, { recursive: true });
|
||||
const target = sessionPathFor(root, runId);
|
||||
const fileSystem = new NodeExecutionEnv({ cwd: root });
|
||||
const repo = new JsonlSessionRepo({ fileSystem, sessionsRoot: root });
|
||||
|
||||
const existing = await stat(target).catch(() => undefined);
|
||||
if (existing) {
|
||||
const firstLine = (await readFile(target, "utf8")).split("\n", 1)[0] ?? "";
|
||||
const header = JSON.parse(firstLine) as { id: string; createdAt: number; storageVersion: number; cwd: string };
|
||||
if (header.id !== runId) throw new Error(`会话文件 ${target} 属于 ${header.id},不是 ${runId}`);
|
||||
const metadata: JsonlSessionMetadata = {
|
||||
id: header.id,
|
||||
createdAt: header.createdAt,
|
||||
storageVersion: header.storageVersion,
|
||||
cwd: header.cwd,
|
||||
path: target,
|
||||
modifiedAt: existing.mtimeMs,
|
||||
};
|
||||
const session = await repo.open(metadata, ctx);
|
||||
return { session, repo, path: target, created: false };
|
||||
}
|
||||
|
||||
const draft = await repo.create({ id: runId, cwd: root }, ctx);
|
||||
const draftPath = draft.metadata.path;
|
||||
await draft.close(ctx);
|
||||
await rename(draftPath, target);
|
||||
await rmdir(dirname(draftPath)).catch(() => undefined);
|
||||
const info = await stat(target);
|
||||
const session = await repo.open({ ...draft.metadata, path: target, modifiedAt: info.mtimeMs }, ctx);
|
||||
return { session, repo, path: target, created: true };
|
||||
}
|
||||
Reference in new issue
Block a user