diff --git a/apps/server/src/modules/agent/challenge.ts b/apps/server/src/modules/agent/challenge.ts index f1e3f0b2..0d9049e6 100644 --- a/apps/server/src/modules/agent/challenge.ts +++ b/apps/server/src/modules/agent/challenge.ts @@ -24,7 +24,7 @@ import { BrowserServiceError, type ChallengeVision, registerChallengeVision } fr import { resolveRunUpstream } from "../pi/index.ts"; import type { ServerContext } from "../types.ts"; import { parseFakeScripts } from "./fake-scripts.ts"; -import { type AgentEnv, agentSettings, eventSink, publishStatus, sessionDirOf } from "./runner.ts"; +import { type AgentEnv, agentSettings, eventSink, publishStatus, runtimeReporter, sessionDirOf } from "./runner.ts"; // VIS-1 browser.solve_challenge:有界的视觉子流程(REQ §4.4、§10)。 // - 服务端起一个受限的子 agent_run(父会话 = 调用它的 Agent 运行;白名单只有截图 / 坐标点击 / 输入 / 滚动 / 等待 / 读页面), @@ -305,6 +305,7 @@ async function runChild( const onEvent = eventSink(env); const onEventError = (err: unknown) => ctx.logger.warn({ runId: child.id, err: { message: (err as Error).message } }, "视觉子流程事件落库失败"); + const onReport = runtimeReporter(ctx); let scripts: FakeScript[] = [FALLBACK_SCRIPT]; if (fake) { try { @@ -315,11 +316,12 @@ async function runChild( } } const runtime = fake - ? createFakeRuntime({ registry, scripts, onEvent, onEventError, fake: { vision: true } }).runtime + ? createFakeRuntime({ registry, scripts, onEvent, onEventError, onReport, fake: { vision: true } }).runtime : new PiAgentRuntime({ registry, onEvent, onEventError, + onReport, upstream: { spec: upstream.spec, credentials: upstream.credentials }, }); const systemPrompt = buildSystemPrompt({ diff --git a/apps/server/src/modules/agent/runner.ts b/apps/server/src/modules/agent/runner.ts index dc565b41..09f95718 100644 --- a/apps/server/src/modules/agent/runner.ts +++ b/apps/server/src/modules/agent/runner.ts @@ -10,6 +10,7 @@ import { type FakeScript, PiAgentRuntime, type RunOutcome, + type RuntimeReport, type SessionKind, type SkillDoc, selectSkills, @@ -399,6 +400,15 @@ export async function buildRunPrompt( return { systemPrompt, kind, skills, available }; } +/** + * 运行时内部错误(本次运行扩展的系统提示 / 能力工具 / 白名单与预算钩子抛错,pi-durable 的 onReport)→ warn 日志。 + * message 已由运行时脱敏;只记位置与信息,不记原始错误对象。 + */ +export function runtimeReporter(ctx: ServerContext): (r: RuntimeReport) => void { + return (r) => + ctx.logger.warn({ runId: r.runId, source: r.source, err: { message: r.message } }, "agent 运行时内部错误"); +} + /** 读 agent_run_event 里最后一轮的结束事件(崩溃发生在"模型已结束、业务状态未写"之间时复用) */ async function lastOutcomeEvent(env: AgentEnv, runId: string) { const r = await env.ctx.db @@ -460,6 +470,7 @@ export async function drive( const onEvent = eventSink(env); const onEventError = (err: unknown, e: AgentRunEvent) => ctx.logger.warn({ runId, seq: e.seq, err: { message: (err as Error).message } }, "agent 事件落库失败"); + const onReport = runtimeReporter(ctx); let scripts: FakeScript[] = DEMO_SCRIPTS; if (fake) { try { @@ -475,11 +486,12 @@ export async function drive( } } const runtime = fake - ? createFakeRuntime({ registry, scripts, onEvent, onEventError, fake: { vision: true } }).runtime + ? createFakeRuntime({ registry, scripts, onEvent, onEventError, onReport, fake: { vision: true } }).runtime : new PiAgentRuntime({ registry, onEvent, onEventError, + onReport, ...(upstream ? { upstream: { spec: upstream.spec, credentials: upstream.credentials } } : {}), }); const maxSeq = await ctx.db diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 120c6df0..c24f66eb 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -36,6 +36,7 @@ export { DEFAULT_COMPACTION_INSTRUCTIONS, PiAgentRuntime, type PiAgentRuntimeOptions, + type RuntimeReport, toolContent, } from "./pi/runtime.ts"; export { LegacySessionError, sessionPathFor } from "./pi/session.ts"; diff --git a/packages/agent/src/pi/runtime.ts b/packages/agent/src/pi/runtime.ts index 15baac9a..5922ebe0 100644 --- a/packages/agent/src/pi/runtime.ts +++ b/packages/agent/src/pi/runtime.ts @@ -1,3 +1,4 @@ +import type { Context } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import type { AssistantMessage, Message, ToolResultMessage } from "@earendil-works/pi-ai"; import { @@ -13,6 +14,7 @@ import { hook, type SettledSubmissionRecord, section, + type ToolExecutionApi, type ToolExecutionResult, type ToolRegistration, ToolTask, @@ -66,9 +68,23 @@ export interface PiAgentRuntimeOptions { */ onEvent?: (event: AgentRunEvent) => void | Promise; onEventError?: (error: unknown, event: AgentRunEvent) => void; + /** + * 运行时内部错误(不影响本次调用结果、但应该看见的):本次运行扩展里的系统提示 section、能力工具、白名单 / 预算钩子抛错, + * 以及 pi-durable 的 onReport(提交 / 进度写入等失败)。message 已脱敏。不能抛错。 + */ + onReport?: (report: RuntimeReport) => void; now?: () => Date; } +export interface RuntimeReport { + runId: string; + /** 出错的位置:section:umt、tool:<工具名>、hook:beforeTool、hook:beforeRequest、harness(pi-durable onReport) */ + source: string; + /** 已脱敏的错误信息 */ + message: string; + error: unknown; +} + export const DEFAULT_COMPACTION_INSTRUCTIONS = "保留:本次目标、约束与授权范围、证据引用(对象 ID/记录)、已完成的动作及其结果(已提交≠已入库)、剩余事项。不要保留任何凭证。"; @@ -113,7 +129,11 @@ export class PiAgentRuntime implements AgentRuntimePort { }; let harness: Harness | undefined; try { - harness = await Harness.open(opened.storage, { models, registry, settings: harnessSettings }, ctx); + harness = await Harness.open( + opened.storage, + { models, registry, settings: harnessSettings, onReport: (error) => run.report("harness", error) }, + ctx, + ); const agent = { model: { provider: model.provider, modelId: model.id }, thinkingLevel }; const root = await harness.root(ctx, { agent }); // 已有会话:模型 / 思考档位跟随本次设置(换了上游或模型后继续) @@ -192,27 +212,58 @@ class PiAgentRun implements AgentRunHandle { const systemPrompt = this.spec.systemPrompt; return defineExtension({ name: "umt", - sections: [section("umt", () => systemPrompt, { tag: false })], + sections: [ + section( + "umt", + this.reported("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(ToolTask, { + beforeTool: this.reported("hook:beforeTool", (call: { id: string; name: string }) => + this.beforeTool(call.id, call.name), + ), + }), hook(GenerationTask, { - beforeRequest: () => { + beforeRequest: this.reported("hook:beforeRequest", () => { this.checkDeadline(); this.checkCost(); if (this.budget) this.scheduleAbort(); return undefined; - }, + }), }), ], }); } + /** 记一条运行时内部错误(脱敏后交给 onReport;onReport 自身出错忽略) */ + report(source: string, error: unknown): void { + try { + const message = redactString(error instanceof Error ? error.message : String(error)); + this.options.onReport?.({ runId: this.runId, source, message, error }); + } catch { + // 上报失败不影响运行 + } + } + + /** 包一层:抛错时先上报(带位置)再原样抛出,pi-durable 的处理语义不变 */ + private reported(source: string, fn: (...args: A) => R): (...args: A) => R { + return (...args: A) => { + try { + return fn(...args); + } catch (error) { + this.report(source, error); + throw error; + } + }; + } + /** 能力 → Pi 工具。执行时统一走注册中心 invoke(schema/作用域/授权/幂等/审计) */ toTool(entry: RegisteredCapability): ToolRegistration { const { descriptor } = entry; const registry = this.options.registry; - const actor = this.spec.actor; const gate = this.spec.replayGate; // 可挂起重放(见 CreateRunSpec.replayGate):对 harness 声明 safe,执行时用 memo 判断是否重放 const gated = descriptor.replay === "never" && !!gate; @@ -223,62 +274,78 @@ class PiAgentRun implements AgentRunHandle { replay: gated || descriptor.replay === "safe" ? "safe" : "unsafe", outputLimits: TOOL_OUTPUT_LIMITS, ...(descriptor.kind === "execute" ? { executionMode: "sequential" as const } : {}), - execute: async (params, api, context): Promise => { - 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 { - if (gated) { - const replayed = (await api.memo(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: "执行被中断,外部结果未知;先用查询工具核对实际状态,不要直接重复提交", - }, - }), - ); - } - 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; - } - }, + execute: (params, api, context) => + this.executeTool(entry, gated, params, api, context).catch((error: unknown) => { + if (!this.closed && !context.abortSignal?.aborted) this.report(`tool:${entry.toolName}`, error); + throw error; + }), }; } + private async executeTool( + entry: RegisteredCapability, + gated: boolean, + params: unknown, + api: ToolExecutionApi, + context: Context, + ): Promise { + const { descriptor } = entry; + const registry = this.options.registry; + const actor = this.spec.actor; + const gate = this.spec.replayGate; + 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 { + if (gated) { + const replayed = (await api.memo(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: "执行被中断,外部结果未知;先用查询工具核对实际状态,不要直接重复提交", + }, + }), + ); + } + 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; + } + } + async attach(harness: Harness, root: Conversation): Promise { const ctx = BACKGROUND_CONTEXT; this.harness = harness; diff --git a/packages/agent/test/runtime.test.ts b/packages/agent/test/runtime.test.ts index e15fc7d5..a5cde541 100644 --- a/packages/agent/test/runtime.test.ts +++ b/packages/agent/test/runtime.test.ts @@ -11,6 +11,7 @@ import { createFakeRuntime, type FakeScript, LegacySessionError, + type RuntimeReport, } from "../src/index.ts"; import { crashScript, RUN_ID } from "./fixtures/crash-script.ts"; import { type Fixture, makeFixture, settings, tempDir, types } from "./helpers.ts"; @@ -462,4 +463,43 @@ describe("PiAgentRuntime + FakeModelProvider", () => { expect(err).toBeInstanceOf(LegacySessionError); expect((err as Error).message).toContain("旧版本"); }); + + it("扩展钩子抛错:上报 onReport(带 runId 与位置、已脱敏),pi-durable 照常按 block 处理", async () => { + const fx = makeFixture(); + const reports: RuntimeReport[] = []; + const { runtime } = createFakeRuntime({ + registry: fx.registry, + scripts: [ + { + match: "Dune", + steps: [{ toolCalls: [{ capability: "inventory.query", input: { title: "Dune" } }] }], + finalText: (r) => `结果:${r.at(-1)?.isError}`, + }, + ], + onReport: (r) => reports.push(r), + }); + const run = await runtime.createRun({ + runId: "run-report", + goal: "Dune", + systemPrompt: "x", + capabilities: ["inventory.query"], + settings: settings(), + sessionDir: await tempDir(), + actor: { actorId: "user:1", taskId: "t", scopeIds: [] }, + externalUsage: () => { + throw new Error("子会话用量读取失败 passkey=abcdef123456"); + }, + }); + opened.push(run); + const out = await run.prompt(); + expect(out).toMatchObject({ status: "completed", finalText: "结果:true" }); + expect(reports).toEqual([ + expect.objectContaining({ + runId: "run-report", + source: "hook:beforeTool", + message: expect.stringContaining("子会话用量读取失败"), + }), + ]); + expect(reports[0]?.message).not.toContain("abcdef123456"); + }); });