feat(agent): 运行时从 pi-agent-core 0.99.2 迁到 pi-durable 1.1.0 + pi-ai 1.1.0
- PiAgentRuntime 改用 pi-durable Harness:每个 run 一个 JSONL 存储目录 sessionDir/<runId>/ + root 对话; 系统提示 / 能力工具 / 白名单与预算钩子装在每次运行自己的扩展里;watchEvents 翻译成 AgentRunEvent - 挂起(工具内 close)立即封口,调用保持未完成,重开后 resume 重放;replayGate 用调用级 memo - 旧格式 <runId>.jsonl 打开时抛 LegacySessionError;skills 加载与 <skill> 块格式内置(源自 pi-agent-core,MIT) - tool_call_end 新增可选 durationMs;用量取 pi.usage 合计(整段会话总量) - Azure provider 改名 azure,迁移 0093 改存量;移除 pi-agent-core,锁 pi-durable/pi-ai/chord 1.1.0 - 文档:ADR-003、pi-providers、runbook、agent README、CHANGELOG Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
dd0f6e2c09
commit
003815591c
24 files changed
+756
-363
No files matched your search
+436
-211
@@ -1,15 +1,24 @@
|
||||
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
|
||||
import type { AssistantMessage, Message, ToolResultMessage } from "@earendil-works/pi-ai";
|
||||
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";
|
||||
type AgentEvent,
|
||||
type AgentEventStream,
|
||||
type Conversation,
|
||||
createRegistry,
|
||||
defineExtension,
|
||||
type Extension,
|
||||
GenerationTask,
|
||||
Harness,
|
||||
type HarnessSettings,
|
||||
hook,
|
||||
type SettledSubmissionRecord,
|
||||
section,
|
||||
type ToolExecutionResult,
|
||||
type ToolRegistration,
|
||||
ToolTask,
|
||||
type UsageState,
|
||||
watchEvents,
|
||||
} from "@earendil-works/pi-durable";
|
||||
import {
|
||||
type AgentBudgetKind,
|
||||
type AgentRunEvent,
|
||||
@@ -33,7 +42,7 @@ import {
|
||||
} from "../runtime/port.ts";
|
||||
import { type ResolvedAgentSettings, resolveAgentSettings } from "../settings.ts";
|
||||
import { compactionFromSettings, type PiModelResolver, resolveModelFromSettings } from "./model.ts";
|
||||
import { openRunSession } from "./session.ts";
|
||||
import { openRunStorage } from "./session.ts";
|
||||
import {
|
||||
PiUpstreamClient,
|
||||
type UpstreamClientOptions,
|
||||
@@ -63,12 +72,18 @@ export interface PiAgentRuntimeOptions {
|
||||
export const DEFAULT_COMPACTION_INSTRUCTIONS =
|
||||
"保留:本次目标、约束与授权范围、证据引用(对象 ID/记录)、已完成的动作及其结果(已提交≠已入库)、剩余事项。不要保留任何凭证。";
|
||||
|
||||
const LANE = "main";
|
||||
/** 可挂起重放的能力:首次执行时写入的调用级 memo(重放时据此区分) */
|
||||
const STARTED_MEMO = "umt.started";
|
||||
const SUMMARY_LIMIT = 600;
|
||||
/** pi-durable 默认把工具文本截到 50KB / 2000 行;能力输出已自带分页与脱敏,沿用 0.99 时代不截断的行为 */
|
||||
const TOOL_OUTPUT_LIMITS = { maxBytes: 8 * 1024 * 1024, maxLines: 1_000_000 } as const;
|
||||
|
||||
/** Pi(pi-agent-core AgentHarness)实现的 AgentRuntimePort */
|
||||
/**
|
||||
* Pi(pi-durable Harness)实现的 AgentRuntimePort。
|
||||
*
|
||||
* 一个 run = 一个 pi-durable 存储(sessionDir/<runId>/ 下的 JSONL)+ 其中的 root 对话。
|
||||
* Harness 只在本进程打开一次(单写者),close() 后其他进程才能接手续跑。
|
||||
*/
|
||||
export class PiAgentRuntime implements AgentRuntimePort {
|
||||
constructor(private readonly options: PiAgentRuntimeOptions) {}
|
||||
|
||||
@@ -84,32 +99,29 @@ export class PiAgentRuntime implements AgentRuntimePort {
|
||||
: resolveModelFromSettings(settings);
|
||||
const { models, model } = resolved;
|
||||
const thinkingLevel = resolved.thinkingLevel ?? settings.thinkingLevel;
|
||||
const opened = await openRunSession(spec.sessionDir, spec.runId, ctx);
|
||||
const opened = await openRunStorage(spec.sessionDir, spec.runId, ctx);
|
||||
|
||||
const run = new PiAgentRun(spec, settings, capabilities, this.options, opened.path, opened.repo);
|
||||
const run = new PiAgentRun(spec, settings, capabilities, this.options, opened.path);
|
||||
run.supportsImages = Array.isArray(model.input) && model.input.includes("image");
|
||||
const registry = createRegistry();
|
||||
registry.install(run.extension());
|
||||
const harnessSettings: HarnessSettings = {
|
||||
stream: { timeoutMs: settings.requestTimeoutMs, maxRetries: settings.maxRetries },
|
||||
retry: { enabled: settings.maxRetries > 0, maxRetries: settings.maxRetries, baseDelayMs: 1000 },
|
||||
compaction: compactionFromSettings(settings, model.contextWindow),
|
||||
toolExecution: "sequential",
|
||||
};
|
||||
let harness: Harness | undefined;
|
||||
try {
|
||||
const { harness, open } = await AgentHarness.create(
|
||||
{
|
||||
session: opened.session,
|
||||
models,
|
||||
model,
|
||||
systemPrompt: spec.systemPrompt,
|
||||
tools: capabilities.map((c) => run.toTool(c)),
|
||||
toolContext: {},
|
||||
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);
|
||||
harness = await Harness.open(opened.storage, { models, registry, settings: harnessSettings }, ctx);
|
||||
const agent = { model: { provider: model.provider, modelId: model.id }, thinkingLevel };
|
||||
const root = await harness.root(ctx, { agent });
|
||||
// 已有会话:模型 / 思考档位跟随本次设置(换了上游或模型后继续)
|
||||
await root.configure(agent, ctx);
|
||||
await run.attach(harness, root);
|
||||
} catch (error) {
|
||||
await opened.session.close(ctx).catch(() => undefined);
|
||||
await opened.repo.close(ctx).catch(() => undefined);
|
||||
if (harness) await harness.close(ctx).catch(() => undefined);
|
||||
else await opened.storage.close(ctx).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
return run;
|
||||
@@ -117,6 +129,7 @@ export class PiAgentRuntime implements AgentRuntimePort {
|
||||
}
|
||||
|
||||
type BudgetState = { kind: AgentBudgetKind; message: string };
|
||||
type ContentBlock = AssistantMessage["content"][number];
|
||||
|
||||
class PiAgentRun implements AgentRunHandle {
|
||||
readonly runId: string;
|
||||
@@ -128,9 +141,16 @@ class PiAgentRun implements AgentRunHandle {
|
||||
private readonly channel = new EventChannel<AgentRunEvent>();
|
||||
private readonly allowed = new Map<string, string>();
|
||||
private readonly rejected = new Set<string>();
|
||||
/** 已发出 tool_call_start 的调用 */
|
||||
private readonly started = new Set<string>();
|
||||
/** 打开会话时仍在执行中的调用(崩溃 / 挂起前未完成):恢复后的事件带 recovery */
|
||||
private readonly recovering = new Map<string, { toolName: string; taskId: unknown }>();
|
||||
/** 模型给出的调用参数(未经过 intent 的调用,结束时补发 start 用) */
|
||||
private readonly callArgs = new Map<string, unknown>();
|
||||
private readonly abortController = new AbortController();
|
||||
private harness!: AgentHarnessInstance<object>;
|
||||
private lane!: AgentLane;
|
||||
private harness!: Harness;
|
||||
private root!: Conversation;
|
||||
private stream: AgentEventStream | undefined;
|
||||
private readonly firstSeq: number;
|
||||
private seq: number;
|
||||
private delivery: Promise<void> = Promise.resolve();
|
||||
@@ -144,14 +164,21 @@ class PiAgentRun implements AgentRunHandle {
|
||||
private startedAt: number | undefined;
|
||||
private deadlineTimer: NodeJS.Timeout | undefined;
|
||||
private closed = false;
|
||||
private closing: Promise<void> | undefined;
|
||||
/** 正在执行的工具数:工具内部触发 close()(挂起等待)时不能等 Harness 关闭,它要等这个工具结束 */
|
||||
private activeTools = 0;
|
||||
/** 流式中的 assistant 内容(由 message_start / message_update 维护)与各块已发出的字数 */
|
||||
private partial: ContentBlock[] | undefined;
|
||||
private shown: number[] = [];
|
||||
private batches = 0;
|
||||
private consuming = false;
|
||||
|
||||
constructor(
|
||||
private readonly spec: CreateRunSpec,
|
||||
private readonly settings: ResolvedAgentSettings,
|
||||
capabilities: RegisteredCapability[],
|
||||
private readonly capabilities: RegisteredCapability[],
|
||||
private readonly options: PiAgentRuntimeOptions,
|
||||
sessionPath: string,
|
||||
private readonly repo: JsonlSessionRepo,
|
||||
) {
|
||||
this.runId = spec.runId;
|
||||
this.sessionPath = sessionPath;
|
||||
@@ -160,8 +187,29 @@ class PiAgentRun implements AgentRunHandle {
|
||||
for (const c of capabilities) this.allowed.set(c.toolName, c.descriptor.id);
|
||||
}
|
||||
|
||||
/** 本次运行的 pi-durable 扩展:系统提示 + 能力工具 + 白名单 / 预算钩子 */
|
||||
extension(): Extension {
|
||||
const systemPrompt = this.spec.systemPrompt;
|
||||
return defineExtension({
|
||||
name: "umt",
|
||||
sections: [section("umt", () => systemPrompt, { tag: false })],
|
||||
tools: this.capabilities.map((c) => this.toTool(c)),
|
||||
hooks: [
|
||||
hook(ToolTask, { beforeTool: (call) => this.beforeTool(call.id, call.name) }),
|
||||
hook(GenerationTask, {
|
||||
beforeRequest: () => {
|
||||
this.checkDeadline();
|
||||
this.checkCost();
|
||||
if (this.budget) this.scheduleAbort();
|
||||
return undefined;
|
||||
},
|
||||
}),
|
||||
],
|
||||
});
|
||||
}
|
||||
|
||||
/** 能力 → Pi 工具。执行时统一走注册中心 invoke(schema/作用域/授权/幂等/审计) */
|
||||
toTool(entry: RegisteredCapability): AgentHarnessTool<object> {
|
||||
toTool(entry: RegisteredCapability): ToolRegistration {
|
||||
const { descriptor } = entry;
|
||||
const registry = this.options.registry;
|
||||
const actor = this.spec.actor;
|
||||
@@ -170,149 +218,267 @@ class PiAgentRun implements AgentRunHandle {
|
||||
const gated = descriptor.replay === "never" && !!gate;
|
||||
return {
|
||||
name: entry.toolName,
|
||||
label: descriptor.title,
|
||||
description: `${descriptor.description}\n[能力 ${descriptor.id} v${descriptor.version};类别 ${descriptor.kind};风险 ${descriptor.riskLevel}]`,
|
||||
parameters: registry.inputJsonSchema(descriptor.id) as unknown as TSchema,
|
||||
replay: gated ? "safe" : descriptor.replay,
|
||||
replay: gated || descriptor.replay === "safe" ? "safe" : "unsafe",
|
||||
outputLimits: TOOL_OUTPUT_LIMITS,
|
||||
...(descriptor.kind === "execute" ? { executionMode: "sequential" as const } : {}),
|
||||
execute: async (toolCallId, params, _onUpdate, _toolCtx, invocation, context) => {
|
||||
const signal = context.abortSignal ?? this.abortController.signal;
|
||||
const invocationId = invocation.invocationId ?? toolCallId;
|
||||
if (gated) {
|
||||
const replayed = (await invocation.getMemo(STARTED_MEMO)) === true;
|
||||
if (!replayed) await invocation.setMemo(STARTED_MEMO, true);
|
||||
else if (!gate(descriptor.id, invocationId))
|
||||
throw new Error(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
code: "interrupted",
|
||||
message: "执行被中断,外部结果未知;先用查询工具核对实际状态,不要直接重复提交",
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
execute: async (params, api, context): Promise<ToolExecutionResult> => {
|
||||
const signals = [this.abortController.signal, ...(context.abortSignal ? [context.abortSignal] : [])];
|
||||
const signal = AbortSignal.any(signals);
|
||||
// 稳定的调用 ID:同一存储内唯一,挂起 / 崩溃后重放时不变(审批、子会话等待按它对应)
|
||||
const invocationId = `${api.taskId}:${api.callId}`;
|
||||
this.activeTools += 1;
|
||||
try {
|
||||
const output = await registry.invoke(descriptor.id, params, {
|
||||
actorId: actor.actorId,
|
||||
taskId: actor.taskId,
|
||||
scopeIds: actor.scopeIds,
|
||||
...(actor.allowedKinds ? { allowedKinds: actor.allowedKinds } : {}),
|
||||
signal,
|
||||
source: "agent",
|
||||
invocationId,
|
||||
agentRunId: this.runId,
|
||||
modelImages: this.supportsImages,
|
||||
});
|
||||
const content = toolContent(output, this.supportsImages);
|
||||
const notice = this.budgetNotice();
|
||||
if (notice) content.push({ type: "text", text: notice });
|
||||
return { content, details: undefined };
|
||||
} catch (error) {
|
||||
const notice = this.budgetNotice();
|
||||
const tail = notice ? `\n${notice}` : "";
|
||||
if (isCapabilityError(error)) {
|
||||
throw new Error(
|
||||
JSON.stringify({ error: { code: error.code, message: redactString(error.message) } }) + tail,
|
||||
);
|
||||
if (gated) {
|
||||
const replayed = (await api.memo<boolean>(STARTED_MEMO, context)) === true;
|
||||
if (!replayed) await api.memo(STARTED_MEMO, true, context);
|
||||
else if (!gate(descriptor.id, invocationId))
|
||||
return errorResult(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
code: "interrupted",
|
||||
message: "执行被中断,外部结果未知;先用查询工具核对实际状态,不要直接重复提交",
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
throw new Error(redactString(error instanceof Error ? error.message : String(error)) + tail);
|
||||
try {
|
||||
const output = await registry.invoke(descriptor.id, params, {
|
||||
actorId: actor.actorId,
|
||||
taskId: actor.taskId,
|
||||
scopeIds: actor.scopeIds,
|
||||
...(actor.allowedKinds ? { allowedKinds: actor.allowedKinds } : {}),
|
||||
signal,
|
||||
source: "agent",
|
||||
invocationId,
|
||||
agentRunId: this.runId,
|
||||
modelImages: this.supportsImages,
|
||||
});
|
||||
// 会话已在本调用内关闭(挂起等待):结果不落盘,恢复后重放本调用
|
||||
if (this.closed) throw new Error("会话已挂起");
|
||||
const content = toolContent(output, this.supportsImages);
|
||||
const notice = this.budgetNotice();
|
||||
if (notice) content.push({ type: "text", text: notice });
|
||||
return { content };
|
||||
} catch (error) {
|
||||
// 中止 / 挂起:交给 harness(中止写 aborted 结果;挂起时调用保持未完成)
|
||||
if (this.closed || context.abortSignal?.aborted) throw error;
|
||||
const notice = this.budgetNotice();
|
||||
const tail = notice ? `\n${notice}` : "";
|
||||
if (isCapabilityError(error))
|
||||
return errorResult(
|
||||
JSON.stringify({ error: { code: error.code, message: redactString(error.message) } }) + tail,
|
||||
);
|
||||
return errorResult(redactString(error instanceof Error ? error.message : String(error)) + tail);
|
||||
}
|
||||
} finally {
|
||||
this.activeTools -= 1;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
attach(harness: AgentHarnessInstance<object>, lane: AgentLane, pendingOperations: number): void {
|
||||
async attach(harness: Harness, root: Conversation): Promise<void> {
|
||||
const ctx = BACKGROUND_CONTEXT;
|
||||
this.harness = harness;
|
||||
this.lane = lane;
|
||||
this.pendingOperations = pendingOperations;
|
||||
const { events, hooks } = harness;
|
||||
this.root = root;
|
||||
const inspection = await harness.inspect(ctx);
|
||||
const inputs = inspection.submissions.filter((s) => s.conversationId === root.id && s.type === "input");
|
||||
this.pendingOperations = inputs.length > 0 ? inputs.length : inspection.tasks.length > 0 ? 1 : 0;
|
||||
const stream = await watchEvents(harness, root.id, ctx);
|
||||
for (const slot of stream.snapshot.tools)
|
||||
if (slot.status === "running") this.recovering.set(slot.callId, { toolName: slot.name, taskId: slot.taskId });
|
||||
this.usage = usageTotals(stream.snapshot.usage);
|
||||
this.stream = stream;
|
||||
stream.start(async (events) => {
|
||||
this.batches += 1;
|
||||
this.consuming = true;
|
||||
try {
|
||||
for (const event of events) await this.onAgentEvent(event);
|
||||
} finally {
|
||||
this.consuming = false;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// 白名单二次校验 + 工具调用次数预算
|
||||
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;
|
||||
const extCalls = this.spec.externalUsage?.().toolCalls ?? 0;
|
||||
if (this.toolCalls + extCalls > this.settings.maxToolCalls) {
|
||||
const reason =
|
||||
extCalls > 0
|
||||
? `工具调用次数超出预算(${this.settings.maxToolCalls},含子会话 ${extCalls} 次)`
|
||||
: `工具调用次数超出预算(${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();
|
||||
this.checkCost();
|
||||
if (this.budget) this.scheduleAbort();
|
||||
return undefined;
|
||||
});
|
||||
/** 白名单二次校验 + 工具调用次数预算(ToolTask.beforeTool) */
|
||||
private beforeTool(toolCallId: string, toolName: string): { block: string } | undefined {
|
||||
const decision = checkToolCall(toolName, this.allowed);
|
||||
if (!decision.allowed) {
|
||||
this.reject(toolCallId, toolName, decision.reason);
|
||||
return { block: decision.reason };
|
||||
}
|
||||
if (this.budget) {
|
||||
// 已超预算、中止尚未落地:不再执行任何工具
|
||||
this.reject(toolCallId, toolName, this.budget.message);
|
||||
return { block: this.budget.message };
|
||||
}
|
||||
this.toolCalls += 1;
|
||||
const extCalls = this.spec.externalUsage?.().toolCalls ?? 0;
|
||||
if (this.toolCalls + extCalls > this.settings.maxToolCalls) {
|
||||
const reason =
|
||||
extCalls > 0
|
||||
? `工具调用次数超出预算(${this.settings.maxToolCalls},含子会话 ${extCalls} 次)`
|
||||
: `工具调用次数超出预算(${this.settings.maxToolCalls})`;
|
||||
this.exceed("tool_calls", reason);
|
||||
this.rejected.add(toolCallId);
|
||||
this.emit({
|
||||
type: "tool_call_rejected",
|
||||
toolCallId,
|
||||
toolName,
|
||||
capabilityId: decision.capabilityId,
|
||||
reason,
|
||||
});
|
||||
return { block: reason };
|
||||
}
|
||||
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;
|
||||
this.finalIsTail = true;
|
||||
}
|
||||
});
|
||||
events.on("tool_start", (e) => {
|
||||
this.finalIsTail = false;
|
||||
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} 不在本次白名单内`);
|
||||
/** pi-durable 事件(每次提交一批)→ AgentRunEvent */
|
||||
private async onAgentEvent(e: AgentEvent): Promise<void> {
|
||||
switch (e.type) {
|
||||
case "message_start": {
|
||||
if (e.message.role !== "assistant") return;
|
||||
this.partial = structuredClone(e.message.content);
|
||||
this.shown = [];
|
||||
this.flushDeltas(this.partial);
|
||||
return;
|
||||
}
|
||||
this.emit({
|
||||
type: "tool_call_start",
|
||||
toolCallId: e.toolCallId,
|
||||
capabilityId,
|
||||
inputSummary: summarizeForLog(e.args, SUMMARY_LIMIT),
|
||||
recovery: e.recovery === true,
|
||||
});
|
||||
case "message_update": {
|
||||
for (const c of e.changes) {
|
||||
if (c.type === "message") {
|
||||
this.partial = structuredClone(c.message.content);
|
||||
continue;
|
||||
}
|
||||
if (!this.partial) continue;
|
||||
if (
|
||||
c.type === "text_start" ||
|
||||
c.type === "thinking_start" ||
|
||||
c.type === "toolcall_start" ||
|
||||
c.type === "block"
|
||||
)
|
||||
this.partial[c.contentIndex] = structuredClone(c.block);
|
||||
else if (c.type === "text_delta" || c.type === "thinking_delta") {
|
||||
const block = this.partial[c.contentIndex];
|
||||
if (block?.type === "text") block.text += c.delta;
|
||||
else if (block?.type === "thinking") block.thinking += c.delta;
|
||||
}
|
||||
}
|
||||
if (this.partial) this.flushDeltas(this.partial);
|
||||
return;
|
||||
}
|
||||
case "message_end": {
|
||||
const message = e.entry.model?.[0] as Message | undefined;
|
||||
if (message?.role !== "assistant") return;
|
||||
this.flushDeltas(message.content);
|
||||
this.partial = undefined;
|
||||
this.shown = [];
|
||||
const text = assistantText(message);
|
||||
const calls = message.content.filter((c) => c.type === "toolCall");
|
||||
for (const c of calls) this.callArgs.set(c.id, c.arguments);
|
||||
if (text.trim()) {
|
||||
this.finalText = text;
|
||||
this.finalIsTail = calls.length === 0;
|
||||
} else if (calls.length > 0) this.finalIsTail = false;
|
||||
return;
|
||||
}
|
||||
case "tool_execution_start": {
|
||||
if (this.rejected.has(e.toolCallId) || this.started.has(e.toolCallId)) return;
|
||||
const capabilityId = this.allowed.get(e.toolName);
|
||||
if (!capabilityId) return;
|
||||
this.emitToolStart(e.toolCallId, capabilityId, e.args);
|
||||
return;
|
||||
}
|
||||
case "tool_execution_end": {
|
||||
if (this.rejected.has(e.toolCallId)) return;
|
||||
const capabilityId = this.allowed.get(e.toolName);
|
||||
if (!capabilityId) {
|
||||
// 模型调用了本次没提供的工具(例如 bash):harness 已合成 tool_unavailable 错误,这里只产生拒绝事件
|
||||
this.reject(e.toolCallId, e.toolName, `工具 ${e.toolName} 不在本次白名单内`);
|
||||
return;
|
||||
}
|
||||
// 参数校验失败、恢复后才结束的调用没有 start:先补一条
|
||||
if (!this.started.has(e.toolCallId))
|
||||
this.emitToolStart(e.toolCallId, capabilityId, this.callArgs.get(e.toolCallId) ?? {});
|
||||
const result = e.entry?.model?.[0] as Message | undefined;
|
||||
const tool = result?.role === "toolResult" ? (result as ToolResultMessage) : undefined;
|
||||
const text = tool
|
||||
? tool.content
|
||||
.map((c) => (c.type === "text" ? c.text : `[${c.type}]`))
|
||||
.join("")
|
||||
.trim()
|
||||
: "工具调用没有结果(执行被中断)";
|
||||
this.emit({
|
||||
type: "tool_call_end",
|
||||
toolCallId: e.toolCallId,
|
||||
capabilityId,
|
||||
isError: tool?.isError ?? true,
|
||||
outputSummary: summarizeForLog(text, SUMMARY_LIMIT),
|
||||
recovery: this.recovering.has(e.toolCallId),
|
||||
...(tool?.durationMs !== undefined ? { durationMs: tool.durationMs } : {}),
|
||||
});
|
||||
return;
|
||||
}
|
||||
case "usage_changed":
|
||||
this.usage = usageTotals(e.usage);
|
||||
this.checkCost();
|
||||
return;
|
||||
case "compaction_start":
|
||||
this.emit({ type: "compaction", phase: "start", reason: e.reason });
|
||||
return;
|
||||
case "compaction_end": {
|
||||
const task = await this.harness.getTask(e.taskId, BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
const state = task?.state;
|
||||
const outcome = state && state.status === "terminal" ? state.outcome.status : undefined;
|
||||
const status =
|
||||
outcome === "completed" ? "completed" : outcome === "aborted" ? "aborted" : outcome ? "failed" : undefined;
|
||||
this.emit({ type: "compaction", phase: "end", reason: e.reason, ...(status ? { status } : {}) });
|
||||
return;
|
||||
}
|
||||
default:
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/** 流式文本:按块补发尚未发出的部分 */
|
||||
private flushDeltas(content: readonly ContentBlock[]): void {
|
||||
content.forEach((block, i) => {
|
||||
if (block.type !== "text" && block.type !== "thinking") return;
|
||||
const text = block.type === "text" ? block.text : block.thinking;
|
||||
const done = this.shown[i] ?? 0;
|
||||
if (text.length <= done) return;
|
||||
this.shown[i] = text.length;
|
||||
const delta = text.slice(done);
|
||||
this.emit(block.type === "text" ? { type: "text_delta", delta } : { type: "thinking_delta", delta });
|
||||
});
|
||||
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,
|
||||
});
|
||||
}
|
||||
|
||||
private emitToolStart(toolCallId: string, capabilityId: string, args: unknown): void {
|
||||
this.started.add(toolCallId);
|
||||
this.finalIsTail = false;
|
||||
this.emit({
|
||||
type: "tool_call_start",
|
||||
toolCallId,
|
||||
capabilityId,
|
||||
inputSummary: summarizeForLog(args, SUMMARY_LIMIT),
|
||||
recovery: this.recovering.has(toolCallId),
|
||||
});
|
||||
events.on("usage", (e) => {
|
||||
this.usage = toAgentUsage(e.totals);
|
||||
this.checkCost();
|
||||
});
|
||||
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 }),
|
||||
);
|
||||
}
|
||||
|
||||
/** 恢复时为打开会话时仍在执行的调用补发 start(带 recovery),参数取自工具任务的 intent */
|
||||
private async announceRecovery(): Promise<void> {
|
||||
for (const [callId, slot] of this.recovering) {
|
||||
if (this.started.has(callId) || this.rejected.has(callId)) continue;
|
||||
const capabilityId = this.allowed.get(slot.toolName) ?? toolNameToCapabilityId(slot.toolName) ?? slot.toolName;
|
||||
let args: unknown = this.callArgs.get(callId) ?? {};
|
||||
if (slot.taskId !== undefined) {
|
||||
const task = await this.harness.getTask(slot.taskId as never, BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
const checkpoint = task?.state.checkpoint as { arguments?: unknown } | undefined;
|
||||
if (checkpoint?.arguments) args = checkpoint.arguments;
|
||||
}
|
||||
this.emitToolStart(callId, capabilityId, args);
|
||||
}
|
||||
}
|
||||
|
||||
events(fromSeq = this.firstSeq): AsyncIterable<AgentRunEvent> {
|
||||
@@ -325,21 +491,34 @@ class PiAgentRun implements AgentRunHandle {
|
||||
|
||||
async prompt(text?: string): Promise<RunOutcome> {
|
||||
const message = text ?? this.spec.goal;
|
||||
return this.drive(false, () => this.lane.prompt(message, undefined, BACKGROUND_CONTEXT));
|
||||
return this.drive(false, async () => {
|
||||
const submission = await this.root.submit({ type: "input", content: message }, BACKGROUND_CONTEXT);
|
||||
return submission.wait(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;
|
||||
const ctx = BACKGROUND_CONTEXT;
|
||||
const inspection = await this.harness.inspect(ctx);
|
||||
const inputs = inspection.submissions.filter((s) => s.conversationId === this.root.id && s.type === "input");
|
||||
if (inspection.tasks.length === 0 && inputs.length === 0) return undefined;
|
||||
await this.announceRecovery();
|
||||
this.harness.resume();
|
||||
let last: SettledSubmissionRecord | undefined;
|
||||
for (const s of inputs) {
|
||||
const submission = await this.harness.submission(s.id, ctx);
|
||||
if (submission) last = await submission.wait(ctx);
|
||||
}
|
||||
if (!last) await this.root.waitForIdle(ctx);
|
||||
return last;
|
||||
});
|
||||
}
|
||||
|
||||
async abort(reason = "用户取消"): Promise<void> {
|
||||
this.userAbort = reason;
|
||||
this.abortController.abort(reason);
|
||||
await this.lane.abort(BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
await this.root?.abort(BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
}
|
||||
|
||||
appendEvent(body: InjectedEventBody): boolean {
|
||||
@@ -353,54 +532,80 @@ class PiAgentRun implements AgentRunHandle {
|
||||
}
|
||||
|
||||
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}`);
|
||||
const ctx = BACKGROUND_CONTEXT;
|
||||
const id = await this.root.compact(customInstructions, ctx);
|
||||
const settled = await this.harness.waitForTask(id, ctx);
|
||||
const outcome = settled.state.outcome;
|
||||
if (outcome.status === "failed") throw new Error(`压缩失败: ${outcome.error.message}`);
|
||||
if (outcome.status === "completed" && outcome.result.submissionId !== undefined) {
|
||||
const placed = await this.harness.submission(outcome.result.submissionId, ctx);
|
||||
await placed?.wait(ctx);
|
||||
}
|
||||
await this.settleEvents();
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭会话(Harness + 存储)。挂起等待时由工具内部调用:此时 Harness 关闭要等这个工具结束,
|
||||
* 所以只封口(之后的提交一律被拒,本调用在存储里保持未完成)就返回,由外层再次 close() 等待真正关完。
|
||||
*/
|
||||
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();
|
||||
await this.delivery;
|
||||
if (!this.closing) {
|
||||
this.closed = true;
|
||||
this.clearDeadline();
|
||||
// 同步调用:立即封口,工具随后的结果提交会被拒绝
|
||||
const harnessClosed = this.harness ? this.harness.close(BACKGROUND_CONTEXT) : Promise.resolve();
|
||||
this.closing = (async () => {
|
||||
await harnessClosed.catch(() => undefined);
|
||||
await this.stream?.stop().catch(() => undefined);
|
||||
this.channel.close();
|
||||
await this.delivery;
|
||||
})();
|
||||
}
|
||||
if (this.activeTools > 0) return;
|
||||
await this.closing;
|
||||
}
|
||||
|
||||
// ---- internals ----
|
||||
|
||||
private async drive(resumed: boolean, start: () => Promise<RunResult | undefined>): Promise<RunOutcome> {
|
||||
private async drive(
|
||||
resumed: boolean,
|
||||
start: () => Promise<SettledSubmissionRecord | 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;
|
||||
let record: SettledSubmissionRecord | undefined;
|
||||
try {
|
||||
result = await start();
|
||||
record = await start();
|
||||
} finally {
|
||||
this.clearDeadline();
|
||||
}
|
||||
if (result && !result.ok) throw new Error(`Pi 拒绝了本次操作: ${result.error._tag} ${result.error.message}`);
|
||||
await this.settleEvents();
|
||||
this.pendingOperations = 0;
|
||||
return this.finish(result);
|
||||
return this.finish(record);
|
||||
}
|
||||
|
||||
private finish(result: RunResult | undefined): RunOutcome {
|
||||
const record = result?.ok ? result.value : undefined;
|
||||
const status = record && "status" in record ? record.status : undefined;
|
||||
/** 等事件流把已提交的批次处理完(批次在提交时同步生成,经微任务投递) */
|
||||
private async settleEvents(): Promise<void> {
|
||||
for (let i = 0; i < 1000; i++) {
|
||||
const before = this.batches;
|
||||
await new Promise((r) => setTimeout(r, 0));
|
||||
while (this.consuming) await new Promise((r) => setTimeout(r, 1));
|
||||
if (this.batches === before) return;
|
||||
}
|
||||
}
|
||||
|
||||
private finish(record: SettledSubmissionRecord | undefined): RunOutcome {
|
||||
let outcome: RunOutcome;
|
||||
if (this.budget) {
|
||||
outcome = this.failed("budget", this.budget.message, this.budget.kind);
|
||||
} else if (status === "aborted" || this.userAbort) {
|
||||
} else if (this.userAbort || (record?.status === "unanswered" && record.reason === "aborted")) {
|
||||
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 if (record?.status === "unanswered") {
|
||||
outcome = this.failed("error", redactString(unansweredMessage(record)));
|
||||
} else {
|
||||
outcome = { status: "completed", finalText: this.finalText, usage: this.usage };
|
||||
this.emit({ type: "run_completed", finalText: outcome.finalText, usage: this.usage });
|
||||
@@ -445,7 +650,7 @@ class PiAgentRun implements AgentRunHandle {
|
||||
private scheduleAbort(): void {
|
||||
setTimeout(() => {
|
||||
this.abortController.abort("budget");
|
||||
void this.lane.abort(BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
void this.root.abort(BACKGROUND_CONTEXT).catch(() => undefined);
|
||||
}, 0);
|
||||
}
|
||||
|
||||
@@ -563,22 +768,42 @@ export function checkToolCall(
|
||||
};
|
||||
}
|
||||
|
||||
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 }>)
|
||||
function errorResult(text: string): ToolExecutionResult {
|
||||
return { content: [{ type: "text", text }], isError: true };
|
||||
}
|
||||
|
||||
/** 没有得到回答的输入 → 失败原因 */
|
||||
function unansweredMessage(record: SettledSubmissionRecord): string {
|
||||
const detail = typeof record.detail === "string" ? record.detail : record.detail ? JSON.stringify(record.detail) : "";
|
||||
switch (record.reason) {
|
||||
case "model_error":
|
||||
return `模型调用失败: ${detail || "未知错误"}`;
|
||||
case "no_model":
|
||||
return "模型不可用:检查模型服务与模型设置";
|
||||
case "faulted":
|
||||
return `运行出错: ${detail || "未知错误"}`;
|
||||
default:
|
||||
return `本轮没有完成(${record.reason}${detail ? `: ${detail}` : ""})`;
|
||||
}
|
||||
}
|
||||
|
||||
function assistantText(message: AssistantMessage): string {
|
||||
return message.content
|
||||
.filter((c) => c.type === "text")
|
||||
.map((c) => c.text ?? "")
|
||||
.map((c) => c.text)
|
||||
.join("");
|
||||
}
|
||||
|
||||
function toAgentUsage(u: Usage): AgentUsageTotals {
|
||||
return {
|
||||
inputTokens: u.input,
|
||||
outputTokens: u.output,
|
||||
cacheReadTokens: u.cacheRead,
|
||||
cacheWriteTokens: u.cacheWrite,
|
||||
totalTokens: u.totalTokens,
|
||||
costUsd: u.cost.total,
|
||||
};
|
||||
/** pi.usage(按 provider/model 与工具分桶)→ 合计 */
|
||||
function usageTotals(u: UsageState): AgentUsageTotals {
|
||||
const t = emptyUsage();
|
||||
for (const x of [...Object.values(u.models), ...Object.values(u.tools)]) {
|
||||
t.inputTokens += x.input;
|
||||
t.outputTokens += x.output;
|
||||
t.cacheReadTokens += x.cacheRead;
|
||||
t.cacheWriteTokens += x.cacheWrite;
|
||||
t.totalTokens += x.totalTokens;
|
||||
t.costUsd += x.cost.total;
|
||||
}
|
||||
return t;
|
||||
}
|
||||
Reference in new issue
Block a user