From 003815591cf658b8b4eac1a4d2715718d0eb2ee7 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 8 Oct 2026 13:42:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E8=BF=90=E8=A1=8C=E6=97=B6?= =?UTF-8?q?=E4=BB=8E=20pi-agent-core=200.99.2=20=E8=BF=81=E5=88=B0=20pi-du?= =?UTF-8?q?rable=201.1.0=20+=20pi-ai=201.1.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PiAgentRuntime 改用 pi-durable Harness:每个 run 一个 JSONL 存储目录 sessionDir// + root 对话; 系统提示 / 能力工具 / 白名单与预算钩子装在每次运行自己的扩展里;watchEvents 翻译成 AgentRunEvent - 挂起(工具内 close)立即封口,调用保持未完成,重开后 resume 重放;replayGate 用调用级 memo - 旧格式 .jsonl 打开时抛 LegacySessionError;skills 加载与 块格式内置(源自 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 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- CHANGELOG.md | 12 + deno.json | 10 +- deno.lock | 66 +- docs/adr/ADR-003-pi-integration.md | 17 +- docs/adr/ADR-004-deno-runtime.md | 1 + docs/design/pi-providers.md | 2 + docs/design/webtop-remote-browser.md | 2 +- docs/runbook/agent.md | 3 +- packages/agent/README.md | 12 +- packages/agent/src/index.ts | 2 +- packages/agent/src/pi/catalog.ts | 16 +- packages/agent/src/pi/model.ts | 10 +- packages/agent/src/pi/runtime.ts | 647 ++++++++++++------ packages/agent/src/pi/session.ts | 87 ++- packages/agent/src/prompt/skills.ts | 103 ++- packages/agent/src/prompt/types.ts | 2 +- packages/agent/src/runtime/port.ts | 4 +- packages/agent/test/helpers.ts | 1 + packages/agent/test/runtime.test.ts | 90 ++- packages/agent/test/settings.test.ts | 7 +- packages/agent/test/skills.test.ts | 2 +- packages/contracts/src/agent-events.ts | 2 + packages/contracts/src/pi.ts | 6 +- .../db/migrations/0093_pi_azure_provider.ts | 15 + 24 files changed, 756 insertions(+), 363 deletions(-) create mode 100644 packages/db/migrations/0093_pi_azure_provider.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index f355ee95..c0b951b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ 版本号遵循 [SemVer](https://semver.org/lang/zh-CN/)。所有成员包(`apps/*`、`packages/*`)与根 `deno.json` 使用同一个版本号;`/api/v1/health` 与备份清单里的版本取自 `apps/server/deno.json`。 +## [Unreleased] + +迁移 `0093`。 + +### 调整 +- 智能体运行时升级到 Pi 1.1（pi-durable + pi-ai 1.1.0）：会话的每一步（模型回复、工具调用、用量）都先写盘再显示，进程中途退出后续跑更稳；工具调用结束事件多了执行耗时（毫秒）。 +- 多轮对话和续跑后的 token / 费用累计改为整段会话的真实总量（之前换进程续跑后从 0 重新计，界面上的用量偏低）。 +- 模型服务的 Azure 提供方随 Pi 改名为 `azure`（原 `azure-openai-responses`），迁移 `0093` 自动改已保存的模型服务，配置项不变。 + +### 注意 +- 会话存储格式变了：会话文件从 `agent-sessions/<会话 ID>.jsonl` 改为 `agent-sessions/<会话 ID>/` 目录。升级前已经结束的会话照常查看；升级时还在运行或挂起等待（审批、选方案、等子会话）的旧会话不能继续，会以「这个会话由旧版本创建……」失败，需要的话重新发起。 + ## [0.3.40] - 2026-10-08 迁移 `0091`。 diff --git a/deno.json b/deno.json index 2d060e64..1b02d7dc 100644 --- a/deno.json +++ b/deno.json @@ -15,10 +15,12 @@ "nodeModulesDir": "auto", "imports": { "@biomejs/biome": "npm:@biomejs/biome@2.5.15", - "@earendil-works/pi-agent-core": "npm:@earendil-works/pi-agent-core@0.99.2", - "@earendil-works/pi-agent-core/": "npm:/@earendil-works/pi-agent-core@0.99.2/", - "@earendil-works/pi-ai": "npm:@earendil-works/pi-ai@0.99.2", - "@earendil-works/pi-ai/": "npm:/@earendil-works/pi-ai@0.99.2/", + "@earendil-works/chord": "npm:@earendil-works/chord@1.1.0", + "@earendil-works/chord/": "npm:/@earendil-works/chord@1.1.0/", + "@earendil-works/pi-ai": "npm:@earendil-works/pi-ai@1.1.0", + "@earendil-works/pi-ai/": "npm:/@earendil-works/pi-ai@1.1.0/", + "@earendil-works/pi-durable": "npm:@earendil-works/pi-durable@1.1.0", + "@earendil-works/pi-durable/": "npm:/@earendil-works/pi-durable@1.1.0/", "@hono/swagger-ui": "npm:@hono/swagger-ui@0.6.1", "@hono/zod-openapi": "npm:@hono/zod-openapi@1.6.3", "@modelcontextprotocol/sdk": "npm:@modelcontextprotocol/sdk@1.31.0", diff --git a/deno.lock b/deno.lock index 45828f7e..7b7f4429 100644 --- a/deno.lock +++ b/deno.lock @@ -7,8 +7,9 @@ "npm:@babel/parser@7.29.9": "7.29.9", "npm:@biomejs/biome@*": "2.5.15", "npm:@biomejs/biome@2.5.15": "2.5.15", - "npm:@earendil-works/pi-agent-core@0.99.2": "0.99.2_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5", - "npm:@earendil-works/pi-ai@0.99.2": "0.99.2_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5", + "npm:@earendil-works/chord@1.1.0": "1.1.0", + "npm:@earendil-works/pi-ai@1.1.0": "1.1.0_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5", + "npm:@earendil-works/pi-durable@1.1.0": "1.1.0_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5", "npm:@fontsource-variable/inter-tight@^5.3.0": "5.3.0", "npm:@fontsource-variable/space-grotesk@^5.3.0": "5.3.0", "npm:@hono/swagger-ui@0.6.1": "0.6.1_hono@4.13.12", @@ -91,8 +92,8 @@ ], "bin": true }, - "@anthropic-ai/sdk@0.124.0_zod@4.6.5": { - "integrity": "sha512-cN5O8i9UVxHeOQAzj/XjshWXG8KiibJDw9OGpH2Z/eR3n/RBxdoLxDJOcfqAJWvjaMDFfHTBADU04hWRJVkDyA==", + "@anthropic-ai/sdk@0.129.0_zod@4.6.5": { + "integrity": "sha512-MH7LB20kNpGLUpPTe2OG5XvL/KhX8HUO9XyBaFjgFk6TLS4Xjt0VPIuMKywe7POKKPslX+QxSlVok9e+8kG5Nw==", "dependencies": [ "json-schema-to-ts", "standardwebhooks", @@ -424,8 +425,8 @@ ], "bin": true }, - "@babel/runtime@7.29.7": { - "integrity": "sha512-Nq8OhGWiZIZGV6hLHoyAKLLcJihP/xFeBMGJoUrxTX2psI8dCifzLhZISFb+VWS3wFMRDmCGw5R+dOySCqPLhw==" + "@babel/runtime@7.29.10": { + "integrity": "sha512-NpugTzbKgGrNe45Ui32cN8vCFFMxVtP/9rtfwv0nP5FRDAjvYBxJcLVpYA9mr1jxP6cnIT7x9LUzuDTbWYYNRA==" }, "@babel/template@7.29.7": { "integrity": "sha512-puq+Gf35oI24FeN11LkoUQFqv9uwNeWpxXZi/Ji3rRIoKAzKnxRaZ+Gkj0vKS9ZCiTESfng1N9LyOyXvo+m+Gg==", @@ -511,26 +512,14 @@ "@date-fns/tz@1.5.0": { "integrity": "sha512-lwYN/vDPeNRULcepoE/LO2Pgx+7/RV+S9ARfbc9lr2DtGkOD7pAiruHvbR1RX3Qyf6ja47EWJDMsNK5vK08DJg==" }, - "@earendil-works/chord@0.99.2": { - "integrity": "sha512-2dJWiNsOJm/ioBl1fhpJmw0hIjFUhM0CAteEnVkiVG0Kg4LimR7Asq8PfjclEvrbCPsDntzy7rGKhTZObI9TZw==", + "@earendil-works/chord@1.1.0": { + "integrity": "sha512-gsHzKfyQ3t0ZIQ3cLBjJ5Vqy1TYnFyruHIV3g3BwmulCDYe5BW4yT9vaJk9QzN6YANILIO75uzuq+Gdxtzz1og==", "dependencies": [ "esbuild" ] }, - "@earendil-works/pi-agent-core@0.99.2_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5": { - "integrity": "sha512-VX0QMcg8HKBsv1bYP2NwmutBYa44OrwmtqR6Dq7QC3Ln8X/IlpHhUxB5L1JgnXNUUkTm8Wg/HZhzS797NBDMCQ==", - "dependencies": [ - "@earendil-works/chord", - "@earendil-works/pi-ai", - "@earendil-works/pi-telemetry", - "diff", - "ignore", - "typebox@1.3.27", - "yaml@2.9.0" - ] - }, - "@earendil-works/pi-ai@0.99.2_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5": { - "integrity": "sha512-9RFOEdY+ZTJ1AI+UuAFs4RM0tF2Tje/R/CEv9gWTJHt0iTu8XHZs64U/EIZxNSllFmec/4HfwsOxYj1qQek9bg==", + "@earendil-works/pi-ai@1.1.0_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5": { + "integrity": "sha512-1T7LAkc/5Bvc0v6w4vAGVdCrli0o/E0pEmYKTnixu95vSFArBjvbhS/G4ZwI0RUePgf0Imcu0VyqlM4EcXxqfw==", "dependencies": [ "@anthropic-ai/sdk", "@aws-sdk/client-bedrock-runtime", @@ -545,8 +534,17 @@ ], "bin": true }, - "@earendil-works/pi-telemetry@0.99.2": { - "integrity": "sha512-WThYU4XM6jjjzH4FzLdreN7QAPS9SDdokL0W0Ldheg1ssCIJkfpK7PgebFm7/PoQub7OiXx+SRF6B/oF7z7lXA==" + "@earendil-works/pi-durable@1.1.0_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5": { + "integrity": "sha512-NtuB3RY24vQ+0/rWYpeHupyFWBfXuKq17to3VMYUYqfD/uuGnz6F6UTPBEm51VSxxA6LZwRcDwLX2LtRehnIgQ==", + "dependencies": [ + "@earendil-works/chord", + "@earendil-works/pi-ai", + "diff", + "typebox@1.3.27" + ] + }, + "@earendil-works/pi-telemetry@1.1.0": { + "integrity": "sha512-8gAK05/2pPozZZz6hCLOi8x2vsqzvsmO9gG92kbvuvTPm0qMHaFlaXX98ndXsrOi5pmzha7vNLguuG4+0Rgxjw==" }, "@esbuild/aix-ppc64@0.28.2": { "integrity": "sha512-XExcO+dvLKvVtNTibSTBej1NCAbaGhWn9Ww1ZPx80qsahhPFe/8jgWP0IchNe0F3HwkU7n8ejhH8bjonqht8mQ==", @@ -708,7 +706,7 @@ "@fontsource-variable/space-grotesk@5.3.0": { "integrity": "sha512-2IxmvfB08i9vnGB3Ym/AXvhRE+8XOjWMXIyDum03c+tPwH0FUoMNQfGpU8NXPxjbws0Vvss3AH0Zqt4oJBBAdw==" }, - "@google/genai@2.21.0_@modelcontextprotocol+sdk@1.31.0_zod@4.6.5": { + "@google/genai@2.21.0_@modelcontextprotocol+sdk@1.31.0": { "integrity": "sha512-+PDtco2/Z0ONdzCGekCoCT+O1VJS9xJQNN4XzQpXG/t3El/SWWMkCWlFRO1KmivOHPa4Q0VjUYu1HBKCZ/v33Q==", "dependencies": [ "@modelcontextprotocol/sdk", @@ -3298,9 +3296,6 @@ "safer-buffer" ] }, - "ignore@7.0.8": { - "integrity": "sha512-YYNsSlXBjMk92SKnkwvB5LOVSa6OznlFUGcsvrFgNJbJCd0M1XKeFVRc8ZByeCqz32FivYNHJVooLmdqrmvp/Q==" - }, "immer@11.1.18": { "integrity": "sha512-EQyQtLiYW029lyoczMl/Hh4Xu7cDecSc58JRYpHyL4tIAu3eqd1yJzQX04d2BZHDkzFFvm6qJEJWOtfDSWAXbQ==" }, @@ -3681,7 +3676,7 @@ "openapi3-ts@4.6.1": { "integrity": "sha512-XW9MOldkhoICNeXVzzmXzmOW5G73ppOEGmh7fLCqHjgfdEYCGGN+00MlVCeUZgovjjfC56j9tvtDt1zGabNjjA==", "dependencies": [ - "yaml@2.9.1" + "yaml" ] }, "openworkflow@0.10.1_postgres@3.4.9": { @@ -4555,14 +4550,14 @@ "postcss", "rolldown", "tinyglobby", - "yaml@2.9.1" + "yaml" ], "optionalDependencies": [ "fsevents" ], "optionalPeers": [ "@types/node", - "yaml@2.9.1" + "yaml" ], "bin": true }, @@ -4632,10 +4627,6 @@ "yallist@3.1.1": { "integrity": "sha512-a4UGQaWPH59mOXUYnAG2ewncQS4i4F43Tv3JoAM+s2VDAmS9NsK8GpDMLrCHPksFT7h3K6TOoUNn2pb7RoXx4g==" }, - "yaml@2.9.0": { - "integrity": "sha512-2AvhNX3mb8zd6Zy7INTtSpl1F15HW6Wnqj0srWlkKLcpYl/gMIMJiyuGq2KeI2YFxUPjdlB+3Lc10seMLtL4cA==", - "bin": true - }, "yaml@2.9.1": { "integrity": "sha512-3NxN8+78OdzbT7C/WjGsyfPAtJaN3FNDsWxv7Y7mcDsT/oOmgW8BpyQQFFBnvZE3j9Y2Sdz1ULFLezL7Eb2yFw==", "bin": true @@ -4654,8 +4645,9 @@ "dependencies": [ "npm:7z-wasm@1.2.0", "npm:@biomejs/biome@2.5.15", - "npm:@earendil-works/pi-agent-core@0.99.2", - "npm:@earendil-works/pi-ai@0.99.2", + "npm:@earendil-works/chord@1.1.0", + "npm:@earendil-works/pi-ai@1.1.0", + "npm:@earendil-works/pi-durable@1.1.0", "npm:@hono/swagger-ui@0.6.1", "npm:@hono/zod-openapi@1.6.3", "npm:@modelcontextprotocol/sdk@1.31.0", diff --git a/docs/adr/ADR-003-pi-integration.md b/docs/adr/ADR-003-pi-integration.md index 697009c1..fa64d44f 100644 --- a/docs/adr/ADR-003-pi-integration.md +++ b/docs/adr/ADR-003-pi-integration.md @@ -1,6 +1,6 @@ # ADR-003:Pi 接入采用 pi-agent-core `AgentHarness`,封装在 `AgentRuntimePort` 之后 -- 状态:已采纳(2026-09-27,W0-B spike 后) +- 状态:已采纳(2026-09-27,W0-B spike 后);2026-10-08 运行时迁到 `@earendil-works/pi-durable`,见文末「2026-10-08 迁到 pi-durable」,下文的 AgentHarness 细节保留为历史 - 关联:REQ §10、§11.4;plan §1、§3.1-9;验证记录 spike `w0b-pi`（已删除，见 git `ceb66f8`）;代码 `spikes/pi/`（已删除，见 git `ceb66f8`） ## 决定 @@ -91,3 +91,18 @@ faux.setResponses([ - Pi 不提供权限系统(REQ §10.2)。进程、容器、网络白名单要在部署层落实。 - 模型流中途崩溃后按"合成 error + 重试"恢复,只读了规范,未实测;v0.1 的 e2e 要补上。 - 真实上游只验证了请求形状(本地抓包),没有联网调用。 + +## 2026-10-08 迁到 pi-durable + +pi-agent-core 1.0 删掉了 harness(AgentHarness、会话存储、压缩、skills、系统提示等),只剩 `Agent` 与循环;harness 搬到 `@earendil-works/pi-durable`(实验性,版本之间 API 会变)。我们改为: + +- 依赖锁精确版本:`@earendil-works/pi-durable@1.1.0`、`@earendil-works/pi-ai@1.1.0`、`@earendil-works/chord@1.1.0`(pi-durable 的 Context / 文档状态),**移除** pi-agent-core。Pi 的类型仍只出现在 `packages/agent/src/pi/*`(另有 `prompt/skills.ts` 的格式约定),对外只有 `AgentRuntimePort`,端口与 `AgentRunEvent` 合同不变(`tool_call_end` 只新增可选 `durationMs`)。 +- 一个 run = 一个 pi-durable 存储:`sessionDir//` 下的 JSONL 存储(`openNodeJsonlStorage`),里面用 root 对话。每次 `createRun` 新建一个 `Harness` + 一个只给本次用的 registry,装一个 `umt` 扩展:系统提示是一个不加标签的 section,能力是 tools,白名单 / 工具次数预算在 `ToolTask.beforeTool`,时长 / 费用预算在 `GenerationTask.beforeRequest`。 +- **单写者**:pi-durable 不做跨进程加锁,同一存储同一时刻只能由一个进程打开。runAgent 每次运行在一个持久步骤里打开一次会话、结束时关闭;同一 run 同一时刻只由持有任务引擎租约的那个 worker 驱动(租约过期才会被别的 worker 接手),挂起后由下一次 runAgent 重新打开。旧版 JSONL 会话也是同样的前提。 +- 事件:用 `watchEvents` 订阅 root 对话的提交事件,翻译成我们的 `AgentRunEvent`(文本 / 思考增量、工具开始 / 结束 / 拒绝、用量、压缩)。增量按提交粒度(默认 100ms)到达。用量取 `pi.usage` 文档的合计,是整段会话的总量(换进程续跑不再从 0 计)。 +- prompt = `root.submit({type:"input"})` + `wait()`;resume = `harness.resume()` 后等未完成的输入;abort = `root.abort()`;手动压缩 = `root.compact()` + 等任务;自动压缩用 `compaction: { enabled, reserveTokens, keepRecentTokens, backgroundTokens: 0 }`(不做后台提前压缩,与旧版阈值压缩一致)。 +- 重放:能力 `replay=safe` → 工具 `replay:"safe"`,`never` → `"unsafe"`(崩溃后得到 interrupted 错误结果);带 `replayGate` 的审批类能力对 harness 声明 safe,用调用级 memo 区分首次与重放。调用 ID 用 `<工具任务 ID>:`,重放时不变。 +- 挂起(agent.await / 审批 / 选方案):工具内部调用 `handle.close()` 时立即封口 Harness(之后的提交一律被拒),该调用在存储里保持未完成;重新打开后 `pendingOperations>0`,`resume()` 重放这次调用。 +- Skills:pi-durable 不再提供 skills 加载器,`prompt/skills.ts` 保留最小实现(`/SKILL.md` 的发现、frontmatter 解析与校验、Pi 的 `` 调用块格式),来源 pi-agent-core 0.99.2(MIT)。 +- 工具输出上限:pi-durable 默认把工具文本截到 50KB / 2000 行,我们放宽到 8MB,沿用旧版不截断的行为。 +- 旧会话:0.99.2 的 `sessionDir/.jsonl` 不能续跑;只有旧文件、没有新目录时 `createRun` 抛 `LegacySessionError`,runAgent 以清楚的错误结束该 run(已结束的会话照常从 PG 事件查看)。 diff --git a/docs/adr/ADR-004-deno-runtime.md b/docs/adr/ADR-004-deno-runtime.md index 852cae5e..06687134 100644 --- a/docs/adr/ADR-004-deno-runtime.md +++ b/docs/adr/ADR-004-deno-runtime.md @@ -45,6 +45,7 @@ "kysely": "npm:kysely@0.29.6", "kysely/": "npm:/kysely@0.29.6/", "pg": "npm:pg@8.23.0", "@types/pg": "npm:@types/pg@8.23.1", "postgres": "npm:postgres@3.4.9", "openworkflow": "npm:openworkflow@0.10.1", "openworkflow/": "npm:/openworkflow@0.10.1/", + // 2026-10-08 起 pi-agent-core 换成 @earendil-works/pi-durable@1.1.0 + @earendil-works/chord@1.1.0(见 ADR-003) "@earendil-works/pi-agent-core": "npm:@earendil-works/pi-agent-core@0.87.1", "@earendil-works/pi-agent-core/": "npm:/@earendil-works/pi-agent-core@0.87.1/", "@earendil-works/pi-ai": "npm:@earendil-works/pi-ai@0.87.1", "typebox": "npm:typebox@1.3.27", diff --git a/docs/design/pi-providers.md b/docs/design/pi-providers.md index a6de1c04..6d88aeb8 100644 --- a/docs/design/pi-providers.md +++ b/docs/design/pi-providers.md @@ -1,5 +1,7 @@ # Pi 模型上游与订阅登录(pi-ai / pi-coding-agent 0.87.1) +> 2026-10-08 升级补充：Agent 运行时换成 `@earendil-works/pi-durable` 1.1.0 + `pi-ai` 1.1.0（+ `@earendil-works/chord` 1.1.0），不再依赖 pi-agent-core（1.0 起它只剩 `Agent` 与循环，harness 移到 pi-durable）。provider 目录仍是 42 个内置；pi-ai 1.0.3 把 `azure-openai-responses` 改名为 `azure`（api id 与 `AZURE_OPENAI_*` 配置名不变），合同枚举随之改名，迁移 `0093` 改已存的模型服务。登录流程、能力表与设置页映射不变。运行时的变化见 ADR-003「2026-10-08 迁到 pi-durable」。 +> > 2026-10-01 升级补充：当前接入目标已更新为 pi-agent-core／pi-ai 0.99.2，执行结果见 全面升级记录（已删除，见 git `ceb66f8`）。下文保留 0.87.1 的研究基线，不用历史计数验收新版。 > > 新版实际目录为 42 个内置 provider（新增 TypeSafe 分类服务），合同和目录回归已同步；TypeSafe 没有 Agent 聊天模型，目录说明已标明。OAuth 共 9 个 provider：新增 `openai` 原生 Sign in with ChatGPT，浏览器流程与回调端口 1455；现有 `openai-codex` 仍保留 device／browser 两种模式。新版 `openai` 是动态注册的公开客户端流程，没有照搬旧 Codex 客户端身份提示。来源：[0.99.2 OpenAI provider](https://github.com/earendil-works/pi/blob/v0.99.2/packages/ai/src/providers/openai.ts)、[原生 ChatGPT OAuth](https://github.com/earendil-works/pi/blob/v0.99.2/packages/ai/src/auth/oauth/openai-chatgpt.ts)。 diff --git a/docs/design/webtop-remote-browser.md b/docs/design/webtop-remote-browser.md index f9b1d0fa..32c93097 100644 --- a/docs/design/webtop-remote-browser.md +++ b/docs/design/webtop-remote-browser.md @@ -227,7 +227,7 @@ ENV CAMOUFOX_EXECUTABLE_PATH=/opt/camoufox/camoufox-bin CAMOUFOX_INSTALL_DIR=/op 另追加一条带 `image_url` 的 user 消息;Google Gemini 3+ 嵌在 `functionResponse.parts`,更早版本另起 user 轮;Bedrock 转 image 块; Mistral 视模型而定。全部以 `model.input.includes("image")` 为门,不支持视觉的模型得到占位文本。 - **我们的 runtime 目前把能力输出 `JSON.stringify` 成单个 text 块**(`packages/agent/src/pi/runtime.ts` L193–196),要加图片通道。 - pi-agent-core 有 `transformContext`,可在每轮请求前把较早的截图替换为占位文本,控制 token 与缓存前缀。 + pi-durable 的 `GenerationTask.beforeRequest` 钩子(原 pi-agent-core `transformContext`)可在每轮请求前把较早的截图替换为占位文本,控制 token 与缓存前缀。 - **camofox 1.17.0 没有坐标鼠标端点**;适配器现在的坐标点击是 `elementFromPoint` → 选择器点击,对 canvas 滑块、 跨源 iframe(Turnstile 复选框)与拖拽无效。截图是 `page.screenshot({type:'png'})`,像素 = 视口 CSS 像素 × DPR,runbook 实测 DPR=1。 diff --git a/docs/runbook/agent.md b/docs/runbook/agent.md index d17bdc32..3c4e6d60 100644 --- a/docs/runbook/agent.md +++ b/docs/runbook/agent.md @@ -64,7 +64,8 @@ AG-3 补的只读摘要(类别 read,scope `read`,都在后台调查 / 巡检白 1. `drive` 步骤: - 取上游(provider / baseUrl / 模型 / 思考档位来自 `pi_upstream`,凭证由上游的凭证存储在请求时解密注入,不落日志),从设置读上下文预算、压缩和预算。 - 不设每日次数 / 费用上限(2026-10-06 起;`budget.agentMaxRunsPerDay` / `agentMaxCostUsdPerDay` 旧值兼容读取、不生效),花费在设置 → 智能体「用量与限制」与运行状态里可见(`GET /agent/usage`)。 - - 打开 `AGENT_SESSION_DIR/.jsonl`。`pendingOperations>0` 说明上次崩溃:先按幂等键对账,再 `resume()`;否则就 `prompt(goal)`。最后逐条消费追加的消息。 + - 打开 `AGENT_SESSION_DIR//`(pi-durable JSONL 存储;同一时刻只由一个进程打开)。`pendingOperations>0` 说明上次崩溃或挂起:先按幂等键对账,再 `resume()`;否则就 `prompt(goal)`。最后逐条消费追加的消息。 + - 0.3.40 及以前留下的 `.jsonl`(旧格式)不能续跑:这类 run 被唤醒时直接失败,错误写明「这个会话由旧版本创建」;已结束的照常查看。 2. `goal-settled` 步骤:只在会话正常结束、且不是后台只读调查(内部 `agent_run.mode=plan_only`)时执行。每隔 `agent.goalPollSec` 检查一次,直到本 run 触达的计划、下载和整理记录全部落定,最长等 `agent.goalWaitHours`。等待期间 run 状态为 `waiting`,汇报写的是"已提交≠已入库"。 3. `finalize` 步骤:从业务表汇总计划、下载任务和整理记录,生成 `AgentGoalReport`(artifact kind=report),并据此决定终态: - 入库完成 → `succeeded`,汇报带整理记录 ID; diff --git a/packages/agent/README.md b/packages/agent/README.md index 05f9a622..5eedb9d1 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -1,20 +1,22 @@ # @umt/agent Pi Agent 接入层:能力注册中心、`AgentRuntimePort` 及其 Pi 实现、设置映射、敏感信息守卫、系统提示模板,以及测试用的脚本化假模型。 -依赖锁定为 `@earendil-works/pi-agent-core@0.87.1`、`@earendil-works/pi-ai@0.87.1`、`typebox@1.3.27`(见 ADR-003)。Pi SDK 的类型只出现在 `src/pi/*` 中,其它包只依赖 `src/index.ts` 导出的自有接口。 +依赖锁定为 `@earendil-works/pi-durable@1.1.0`、`@earendil-works/pi-ai@1.1.0`、`@earendil-works/chord@1.1.0`(见 ADR-003「2026-10-08 迁到 pi-durable」)。Pi SDK 的类型只出现在 `src/pi/*` 中,其它包只依赖 `src/index.ts` 导出的自有接口。 ## 边界 - 模型只能调用本次 `capabilities` 白名单里的能力。白名单有三层保护: - 1. 注册层:只把白名单能力注册成 Pi 工具,其它工具名(例如 `bash`)由 harness 合成 unavailable 错误; - 2. `before_tool` 钩子再校验一次,白名单外的调用一律 block,并产生 `tool_call_rejected` 事件; + 1. 注册层:只把白名单能力注册成 Pi 工具,其它工具名(例如 `bash`)由 harness 合成 tool_unavailable 错误,并产生 `tool_call_rejected` 事件; + 2. `ToolTask.beforeTool` 钩子再校验一次,白名单外的调用一律 block,并产生 `tool_call_rejected` 事件; 3. 每次执行都走 `CapabilityRegistry.invoke`,依次做能力类别上限(allowedKinds)、schema、授权、幂等键、输出 schema 和审计。 -- **不提供** bash/进程执行、任意 SQL、任意文件读写、任意网络/HTTP 代理、读取 Cookie/Token、安装依赖或修改代码的工具。`NodeExecutionEnv` 只作为会话目录的 FileSystem 使用,它的 exec 不会暴露给模型。 +- **不提供** bash/进程执行、任意 SQL、任意文件读写、任意网络/HTTP 代理、读取 Cookie/Token、安装依赖或修改代码的工具。Harness 不配执行环境(`env`),pi-durable 自带的 read/write/edit/bash 工具不安装。 - 授权只在服务端判定:`authorize` 回调查 automation_grant/approval。模型参数里写的"用户已同意"不起作用。没有注入 `authorize` 时,propose/approve/execute 一律拒绝。 - 幂等键由服务端按操作语义生成(默认是 `id:v版本:sha256(规范化输入)`,可以注入自定义生成器),通过 `ctx.idempotencyKey` 交给 handler,不采用模型给的值。 - 工具的输入输出在进入模型上下文、事件和审计之前,都要经过 `redactForModel`(cookie/passkey/token/apiKey/authorization/私有下载 URL/URL 内嵌账号密码)处理。`apiKey` 只在请求时经 provider `auth.resolve` 交给 SDK。 - 子代理:`subagentsEnabled` 在 v0.1 中只做记录。SDK 本身不提供子代理,以后由 `agent.delegate` 能力实现,并继承父任务的授权与预算。 -- JSONL 会话(`sessionDir/.jsonl`)只用于转写和恢复点。业务真相放在 PG,压缩摘要不作为唯一状态。 +- 会话存储(`sessionDir//`,pi-durable JSONL)只用于转写和恢复点。业务真相放在 PG,压缩摘要不作为唯一状态。 +- 单写者:同一会话存储同一时刻只能由一个进程打开(pi-durable 不做跨进程加锁);runAgent 每次运行只打开一次,结束或挂起时关闭。 +- 0.3.40 及以前的 `sessionDir/.jsonl`(pi-agent-core 0.99 格式)不能续跑,打开时抛 `LegacySessionError`。 ## 新增一个能力 diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 3fe5eacb..120c6df0 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -38,7 +38,7 @@ export { type PiAgentRuntimeOptions, toolContent, } from "./pi/runtime.ts"; -export { sessionPathFor } from "./pi/session.ts"; +export { LegacySessionError, sessionPathFor } from "./pi/session.ts"; export { classifyUpstreamError, PiUpstreamClient, diff --git a/packages/agent/src/pi/catalog.ts b/packages/agent/src/pi/catalog.ts index d8b157ca..83754a2f 100644 --- a/packages/agent/src/pi/catalog.ts +++ b/packages/agent/src/pi/catalog.ts @@ -12,10 +12,10 @@ import type { } from "@umt/contracts"; // pi-ai 内置 provider → 能力表。认证方式、模型表直接读 builtinProviders(); -// 登录方式、env 字段、条款提示是研究文档(docs/design/pi-providers.md §1.3/§2/§5)及 0.99.2 升级核对的静态表, +// 登录方式、env 字段、条款提示是研究文档(docs/design/pi-providers.md §1.3/§2/§5)及 0.99.2 / 1.1.0 升级核对的静态表, // 升级 pi-ai 时随文档一起复核。 -export const PI_AI_VERSION = "0.99.2"; +export const PI_AI_VERSION = "1.1.0"; let cached: readonly Provider[] | null = null; /** 内置 provider 实例(无状态的共享;Radius 这类动态 provider 按上游另建新实例) */ @@ -71,7 +71,7 @@ const env = (name: string, label: string, required: boolean, hint: string | null /** provider 需要的非密配置(存 extra.env;请求时经 pi 的 AuthResult.env 下发) */ export const ENV_FIELDS: Record = { - "azure-openai-responses": [ + azure: [ env("AZURE_OPENAI_BASE_URL", "Azure 端点 URL", false, "与资源名二选一"), env("AZURE_OPENAI_RESOURCE_NAME", "Azure 资源名", false, "与端点 URL 二选一"), env("AZURE_OPENAI_API_VERSION", "API 版本", false, null), @@ -95,13 +95,7 @@ export const ENV_FIELDS: Record = { /** 用服务端环境凭据(不存密钥)的 provider:authKind none */ const AMBIENT = new Set(["amazon-bedrock", "google-vertex"]); -const CLOUD = new Set([ - "amazon-bedrock", - "google-vertex", - "azure-openai-responses", - "cloudflare-ai-gateway", - "cloudflare-workers-ai", -]); +const CLOUD = new Set(["amazon-bedrock", "google-vertex", "azure", "cloudflare-ai-gateway", "cloudflare-workers-ai"]); /** baseUrl 由 token / env 决定,不允许覆盖 */ const NO_BASE_OVERRIDE = new Set([ "openai-codex", @@ -109,7 +103,7 @@ const NO_BASE_OVERRIDE = new Set([ "radius", "amazon-bedrock", "google-vertex", - "azure-openai-responses", + "azure", "cloudflare-ai-gateway", "cloudflare-workers-ai", ]); diff --git a/packages/agent/src/pi/model.ts b/packages/agent/src/pi/model.ts index 817f5b45..0ef74d10 100644 --- a/packages/agent/src/pi/model.ts +++ b/packages/agent/src/pi/model.ts @@ -1,7 +1,7 @@ -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 { CompactionPolicy } from "@earendil-works/pi-durable"; import type { ResolvedAgentSettings } from "../settings.ts"; /** Pi 侧需要的模型装配结果(仅在 src/pi/* 内部流转) */ @@ -62,12 +62,16 @@ export function resolveModelFromSettings(settings: ResolvedAgentSettings): Resol return { models, model }; } -/** 压缩设置:threshold 模式下占用达到 threshold 比例时触发;manual 只允许手动 compact() */ -export function compactionFromSettings(settings: ResolvedAgentSettings, contextWindow: number): CompactionSettings { +/** + * 压缩设置:threshold 模式下占用达到 threshold 比例时阻塞压缩;manual 只允许手动 compact()。 + * backgroundTokens=0:不做 pi-durable 的后台提前压缩,行为与 pi-agent-core 0.99 的阈值压缩一致。 + */ +export function compactionFromSettings(settings: ResolvedAgentSettings, contextWindow: number): CompactionPolicy { 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)), + backgroundTokens: 0, }; } diff --git a/packages/agent/src/pi/runtime.ts b/packages/agent/src/pi/runtime.ts index 2343264d..15baac9a 100644 --- a/packages/agent/src/pi/runtime.ts +++ b/packages/agent/src/pi/runtime.ts @@ -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// 下的 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(); private readonly allowed = new Map(); private readonly rejected = new Set(); + /** 已发出 tool_call_start 的调用 */ + private readonly started = new Set(); + /** 打开会话时仍在执行中的调用(崩溃 / 挂起前未完成):恢复后的事件带 recovery */ + private readonly recovering = new Map(); + /** 模型给出的调用参数(未经过 intent 的调用,结束时补发 start 用) */ + private readonly callArgs = new Map(); private readonly abortController = new AbortController(); - private harness!: AgentHarnessInstance; - private lane!: AgentLane; + private harness!: Harness; + private root!: Conversation; + private stream: AgentEventStream | undefined; private readonly firstSeq: number; private seq: number; private delivery: Promise = 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 | 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 { + 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 => { + 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(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, lane: AgentLane, pendingOperations: number): void { + async attach(harness: Harness, root: Conversation): Promise { + 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 { + 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 { + 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 { @@ -325,21 +491,34 @@ class PiAgentRun implements AgentRunHandle { async prompt(text?: string): Promise { 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 { 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 { 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 { - 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 { - 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): Promise { + private async drive( + resumed: boolean, + start: () => Promise, + ): Promise { 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 { + 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; } diff --git a/packages/agent/src/pi/session.ts b/packages/agent/src/pi/session.ts index 8bd1b167..7f64934f 100644 --- a/packages/agent/src/pi/session.ts +++ b/packages/agent/src/pi/session.ts @@ -1,58 +1,55 @@ -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"; +import { stat } from "node:fs/promises"; +import { join, resolve } from "node:path"; +import type { Context } from "@earendil-works/chord"; +import { type JsonlStorage, openNodeJsonlStorage } from "@earendil-works/pi-durable/storage/jsonl/node"; -export interface OpenedSession { - session: Session; - repo: JsonlSessionRepo; +export interface OpenedStorage { + storage: JsonlStorage; + /** 会话存储目录 sessionDir// */ 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 { +function checkRunId(runId: string): void { if (!SAFE_RUN_ID.test(runId) || runId.includes("..")) throw new Error(`runId 不能用作文件名: ${runId}`); +} + +/** pi-durable JSONL 存储目录:sessionDir//(一个 run 一个目录,目录里是 pi-durable 自己的追加文件) */ +export function sessionPathFor(sessionDir: string, runId: string): string { + checkRunId(runId); + return join(resolve(sessionDir), runId); +} + +/** 0.3.40 及以前(pi-agent-core 0.99.2)的会话文件 sessionDir/.jsonl,格式与 pi-durable 不兼容 */ +export function legacySessionPathFor(sessionDir: string, runId: string): string { + checkRunId(runId); return join(resolve(sessionDir), `${runId}.jsonl`); } -/** - * 打开/创建 sessionDir/.jsonl。 - * JsonlSessionRepo 默认会把文件放到 `/--cwd--/<时间戳>_.jsonl`;这里创建后把空会话文件 - * (只有 header)移到固定路径,再按元数据打开,之后所有写入都追加到固定路径。 - * NodeExecutionEnv 只当作会话目录的 FileSystem 使用,它的 exec/shell 从不暴露给模型。 - */ -export async function openRunSession(sessionDir: string, runId: string, ctx: Context): Promise { - 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 }; +/** 旧版本留下的会话无法续跑:给出明确原因(已结束的会话照常从数据库事件查看) */ +export class LegacySessionError extends Error { + constructor(readonly legacyPath: string) { + super("这个会话由旧版本创建（会话格式已升级），不能继续运行；已有的记录照常可看，需要继续请新开一个会话"); + this.name = "LegacySessionError"; } +} - 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 }; +/** + * 打开/创建 sessionDir// 下的 pi-durable JSONL 存储。 + * + * 单写者:pi-durable 不做跨进程加锁,同一存储同一时刻只能由一个进程打开。runAgent 每次运行只打开一次会话 + * (同一 run 同一时刻只由持有任务引擎租约的 worker 驱动,见 docs/adr/ADR-003-pi-integration.md),关闭后另一个进程才可以接手。 + * + * 只有旧格式文件 .jsonl、没有新目录时抛 LegacySessionError(不尝试解析旧格式)。 + */ +export async function openRunStorage(sessionDir: string, runId: string, ctx: Context): Promise { + const path = sessionPathFor(sessionDir, runId); + const current = await stat(path).catch(() => undefined); + if (!current) { + const legacy = legacySessionPathFor(sessionDir, runId); + if (await stat(legacy).catch(() => undefined)) throw new LegacySessionError(legacy); + } + const storage = await openNodeJsonlStorage(path, ctx); + return { storage, path }; } diff --git a/packages/agent/src/prompt/skills.ts b/packages/agent/src/prompt/skills.ts index b308b4b1..0c5405c3 100644 --- a/packages/agent/src/prompt/skills.ts +++ b/packages/agent/src/prompt/skills.ts @@ -1,36 +1,57 @@ +import { readdir, readFile } from "node:fs/promises"; +import { join } from "node:path"; import { fileURLToPath } from "node:url"; -import { BACKGROUND_CONTEXT, formatSkillInvocation, loadSkills, type Skill } from "@earendil-works/pi-agent-core"; -import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/harness/env/nodejs"; import { CAPABILITY_CATALOG } from "@umt/contracts"; import { parse as parseYaml } from "yaml"; import type { PromptSection, SessionKind, SkillDoc } from "./types.ts"; -// Skill(AG-5):packages/agent/skills//SKILL.md。发现、解析与校验用 Pi harness 的 loadSkills -// (agentskills.io 格式:frontmatter name / description + 正文;name 必须等于目录名)。 +// Skill(AG-5):packages/agent/skills//SKILL.md(agentskills.io 格式:frontmatter name / description + 正文; +// name 必须等于目录名)。 // // Pi 的模型侧用法是把 索引放进系统提示,由模型用文件读取工具自己读 SKILL.md; // 我们的智能体没有(也不该有)文件读取工具,所以改成服务端按意图选:frontmatter 的 triggers 命中目标或追加消息时, -// 首轮把正文注入系统提示的「Skill」层,后续消息用 Pi 的 formatSkillInvocation 附在用户消息前(见 runner)。 +// 首轮把正文注入系统提示的「Skill」层,后续消息用 Pi 的 skill 块格式附在用户消息前(见 runner)。 // 本项目的 frontmatter 扩展:title(中文名)、triggers(正则)、capabilities(用到的能力)、sessions(适用会话类型)。 +// +// 发现 / 解析 / 校验与 skill 块格式取自 pi-agent-core 0.99.2 的 harness/skills.ts(MIT,© Earendil Works); +// pi 1.x 把 harness 移到 pi-durable 后不再提供 skills,这里只保留本项目用到的部分: +// 一层目录下的 /SKILL.md,不读 .gitignore、不收根目录的散 .md、不跟随符号链接。 /** 仓库内的 Skill 目录;deno compile 用 --include 嵌入(apps/server/deno.json 的 compile 任务) */ export const SKILLS_DIR = fileURLToPath(new URL("../../skills", import.meta.url)); const DEFAULT_SESSIONS: SessionKind[] = ["chat", "patrol", "investigate", "child"]; const SESSION_KINDS = new Set(["chat", "patrol", "investigate", "child", "vision"]); +const MAX_NAME_LENGTH = 64; +const MAX_DESCRIPTION_LENGTH = 1024; export interface SkillLoadResult { skills: SkillDoc[]; - /** 格式问题(Pi 的诊断 + 扩展字段校验);测试要求为空 */ + /** 格式问题(Pi 同款校验 + 扩展字段校验);测试要求为空 */ diagnostics: string[]; } -function frontmatter(raw: string): Record { +/** frontmatter + 正文(与 Pi 的 parseFrontmatter 一致:没有 frontmatter 时整篇是正文) */ +function parseSkillFile(raw: string): { frontmatter: Record; body: string } { const text = raw.replace(/\r\n?/g, "\n"); - if (!text.startsWith("---")) return {}; + if (!text.startsWith("---")) return { frontmatter: {}, body: text }; const end = text.indexOf("\n---", 3); - if (end < 0) return {}; - return (parseYaml(text.slice(4, end)) ?? {}) as Record; + if (end < 0) return { frontmatter: {}, body: text }; + return { + frontmatter: (parseYaml(text.slice(4, end)) ?? {}) as Record, + body: text.slice(end + 4).trim(), + }; +} + +/** Pi 的 name 规则:等于目录名、≤64、只含小写字母数字与连字符、不以连字符开头结尾、不连续 */ +function validateName(name: string, dirName: string): string[] { + const errors: string[] = []; + if (name !== dirName) errors.push(`name "${name}" 与目录名 "${dirName}" 不一致`); + if (name.length > MAX_NAME_LENGTH) errors.push(`name 超过 ${MAX_NAME_LENGTH} 个字符`); + if (!/^[a-z0-9-]+$/.test(name)) errors.push("name 只能用小写字母、数字和连字符"); + if (name.startsWith("-") || name.endsWith("-")) errors.push("name 不能以连字符开头或结尾"); + if (name.includes("--")) errors.push("name 不能有连续连字符"); + return errors; } const strings = (v: unknown): string[] => @@ -38,25 +59,46 @@ const strings = (v: unknown): string[] => /** 读目录下全部 Skill;失败的项进 diagnostics,不抛错(运行时缺 Skill 只少一层提示,不影响会话) */ export async function loadSkillDocs(dir: string = SKILLS_DIR): Promise { - const env = new NodeExecutionEnv({ cwd: dir }); const diagnostics: string[] = []; - let skills: Skill[] = []; + let entries: { name: string; isDirectory(): boolean }[]; try { - const r = await loadSkills(env, dir, BACKGROUND_CONTEXT); - skills = r.skills; - for (const d of r.diagnostics) diagnostics.push(`${d.path}: ${d.message}`); + entries = await readdir(dir, { withFileTypes: true }); } catch (err) { return { skills: [], diagnostics: [`${dir}: ${(err as Error).message}`] }; } const out: SkillDoc[] = []; - for (const s of skills) { - const raw = await env.readTextFile(s.filePath, BACKGROUND_CONTEXT); - const fm = raw.ok ? frontmatter(raw.value) : {}; + for (const entry of entries) { + if (!entry.isDirectory() || entry.name.startsWith(".") || entry.name === "node_modules") continue; + const filePath = join(dir, entry.name, "SKILL.md"); + const where = `${entry.name}/SKILL.md`; + let raw: string; + try { + raw = await readFile(filePath, "utf8"); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") diagnostics.push(`${where}: ${(err as Error).message}`); + continue; + } + let fm: Record; + let body: string; + try { + ({ frontmatter: fm, body } = parseSkillFile(raw)); + } catch (err) { + diagnostics.push(`${where}: frontmatter 解析失败：${(err as Error).message}`); + continue; + } + const description = typeof fm.description === "string" ? fm.description : ""; + if (!description.trim()) { + diagnostics.push(`${where}: 缺少 description`); + continue; + } + if (description.length > MAX_DESCRIPTION_LENGTH) + diagnostics.push(`${where}: description 超过 ${MAX_DESCRIPTION_LENGTH} 个字符`); + const name = typeof fm.name === "string" && fm.name ? fm.name : entry.name; + for (const e of validateName(name, entry.name)) diagnostics.push(`${where}: ${e}`); const triggers = strings(fm.triggers); const capabilities = strings(fm.capabilities); const sessions = strings(fm.sessions) as SessionKind[]; - const title = typeof fm.title === "string" ? fm.title : s.name; - const where = `${s.name}/SKILL.md`; + const title = typeof fm.title === "string" ? fm.title : name; if (triggers.length === 0) diagnostics.push(`${where}: 缺少 triggers`); for (const t of triggers) { try { @@ -68,14 +110,14 @@ export async function loadSkillDocs(dir: string = SKILLS_DIR): Promise a.name.localeCompare(b.name)); @@ -153,16 +195,17 @@ export function skillsLayer(loaded: readonly SkillDoc[], available: readonly Ski return { layer: "skills", title: "Skill", lines }; } +/** Pi 的 skill 调用块格式(pi-agent-core 0.99.2 formatSkillInvocation) */ +function formatSkillInvocation(skill: { name: string; filePath: string; content: string }): string { + const dir = skill.filePath.replace(/[\\/]+$/, "").replace(/[\\/][^\\/]*$/, "") || "/"; + return `\nReferences are relative to ${dir}.\n\n${skill.content}\n`; +} + /** 追加消息命中新 Skill 时,用 Pi 的格式把 Skill 附在消息前(模型当作本条消息的做事说明) */ export function withSkillInvocation(skills: readonly SkillDoc[], text: string): string { if (skills.length === 0) return text; const blocks = skills.map((s) => - formatSkillInvocation({ - name: s.name, - description: s.description, - content: s.content, - filePath: `skills/${s.name}/SKILL.md`, - }), + formatSkillInvocation({ name: s.name, content: s.content, filePath: `skills/${s.name}/SKILL.md` }), ); return `${blocks.join("\n\n")}\n\n${text}`; } diff --git a/packages/agent/src/prompt/types.ts b/packages/agent/src/prompt/types.ts index 223c3c50..6b1c5db4 100644 --- a/packages/agent/src/prompt/types.ts +++ b/packages/agent/src/prompt/types.ts @@ -92,7 +92,7 @@ export interface LayeredPromptInput { constraints?: readonly string[]; } -/** 一个 Skill(Pi harness Skill + 本项目的 frontmatter 扩展) */ +/** 一个 Skill(Pi 的 agentskills.io 格式 + 本项目的 frontmatter 扩展) */ export interface SkillDoc { /** 目录名,小写字母 / 数字 / 连字符 */ name: string; diff --git a/packages/agent/src/runtime/port.ts b/packages/agent/src/runtime/port.ts index 1ae3c509..c6057819 100644 --- a/packages/agent/src/runtime/port.ts +++ b/packages/agent/src/runtime/port.ts @@ -46,14 +46,14 @@ export interface AgentActor { } export interface CreateRunSpec { - /** = agent_run.id,同时作为 Pi 会话 ID 与 JSONL 文件名 */ + /** = agent_run.id,同时作为会话存储目录名 */ runId: string; goal: string; systemPrompt: string; /** 本次允许的能力 ID 白名单 */ capabilities: string[]; settings: AgentSettings; - /** 应用数据目录下的会话目录;会话文件为 sessionDir/.jsonl */ + /** 应用数据目录下的会话目录;本次会话存储在 sessionDir//(pi-durable JSONL) */ sessionDir: string; actor: AgentActor; /** diff --git a/packages/agent/test/helpers.ts b/packages/agent/test/helpers.ts index 7cd70626..c8f05501 100644 --- a/packages/agent/test/helpers.ts +++ b/packages/agent/test/helpers.ts @@ -106,6 +106,7 @@ export function makeFixture(opts: { slowMs?: number } = {}): Fixture { export function sleep(ms: number, signal?: AbortSignal): Promise { return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(new Error("aborted by signal")); const t = setTimeout(resolve, ms); signal?.addEventListener( "abort", diff --git a/packages/agent/test/runtime.test.ts b/packages/agent/test/runtime.test.ts index 9a2167a1..e15fc7d5 100644 --- a/packages/agent/test/runtime.test.ts +++ b/packages/agent/test/runtime.test.ts @@ -1,19 +1,29 @@ import { spawn } from "node:child_process"; -import { existsSync, readFileSync } from "node:fs"; +import { existsSync, readdirSync, readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { AgentRunEvent } from "@umt/contracts"; import { afterEach, describe, expect, it } from "vitest"; +import { z } from "zod"; import { type AgentRunHandle, type AgentSettings, checkToolCall, createFakeRuntime, type FakeScript, + LegacySessionError, } from "../src/index.ts"; import { crashScript, RUN_ID } from "./fixtures/crash-script.ts"; import { type Fixture, makeFixture, settings, tempDir, types } from "./helpers.ts"; const opened: AgentRunHandle[] = []; + +/** 会话存储目录下全部文件的文本(pi-durable JSONL 主文件 + 附属文件) */ +function sessionText(dir: string): string { + return readdirSync(dir, { recursive: true, withFileTypes: true }) + .filter((f) => f.isFile()) + .map((f) => readFileSync(join(f.parentPath, f.name), "utf8")) + .join("\n"); +} afterEach(async () => { await Promise.all(opened.splice(0).map((r) => r.close())); }); @@ -72,7 +82,7 @@ async function start( } describe("PiAgentRuntime + FakeModelProvider", () => { - it("完整闭环:事件顺序、usage、最终文本、JSONL 位于 sessionDir/.jsonl", async () => { + it("完整闭环:事件顺序、usage、最终文本、会话存储位于 sessionDir//", async () => { const fx = makeFixture(); fx.grants.set("user:1", new Set(["download.submit"])); const { run, sessionDir } = await start(fx, [dune]); @@ -104,9 +114,11 @@ describe("PiAgentRuntime + FakeModelProvider", () => { const last = events.at(-1)!; expect(last).toMatchObject({ type: "run_completed", finalText: outcome.finalText }); - expect(run.sessionPath).toBe(join(sessionDir, "run-1.jsonl")); + expect(run.sessionPath).toBe(join(sessionDir, "run-1")); expect(existsSync(run.sessionPath)).toBe(true); - expect(readFileSync(run.sessionPath, "utf8")).toContain("download__submit"); + expect(sessionText(run.sessionPath)).toContain("download__submit"); + const end = events.find((e) => e.type === "tool_call_end"); + expect(end).toMatchObject({ durationMs: expect.any(Number) }); // 审计来自 agent,带 invocationId 与 runId expect(fx.audits.find((a) => a.capabilityId === "download.submit")).toMatchObject({ source: "agent", @@ -233,7 +245,7 @@ describe("PiAgentRuntime + FakeModelProvider", () => { for (const secret of ["abcdef123456", "deadbeef", "sk-live-XXXX"]) { expect(transcript).not.toContain(secret); expect(JSON.stringify(run.history())).not.toContain(secret); - expect(readFileSync(run.sessionPath, "utf8")).not.toContain(secret); + expect(sessionText(run.sessionPath)).not.toContain(secret); } }); @@ -382,4 +394,72 @@ describe("PiAgentRuntime + FakeModelProvider", () => { expect(recovered).toMatchObject({ capabilityId: "download.submit", isError: true, recovery: true }); expect(run.history()[0]).toMatchObject({ type: "run_started", resumed: true }); }, 30_000); + + it("挂起:工具执行中关闭会话(结果不落盘),换进程重开后 resume 重放该调用并完成", async () => { + const sessionDir = await tempDir(); + const fx = makeFixture(); + let handle: AgentRunHandle | undefined; + const calls: Array<{ inv: string; resumed: boolean }> = []; + let suspended = true; + fx.registry.register( + { + id: "agent.await", + version: 1, + title: "等待子会话", + description: "等子会话落定。", + inputSchema: z.object({ runIds: z.array(z.string()) }), + outputSchema: z.object({ allDone: z.boolean() }), + kind: "read", + scope: "read", + riskLevel: "low", + idempotent: true, + supportsPreview: false, + replay: "safe", + }, + async (_input, c) => { + calls.push({ inv: c.invocationId ?? "", resumed: !suspended }); + if (suspended) { + // 与 server 的 suspendRun 一样:在工具内部关闭会话,返回值不会进入模型上下文 + await new Promise((r) => setTimeout(r, 0)); + await handle!.close(); + return { allDone: false }; + } + return { allDone: true }; + }, + ); + const script: FakeScript = { + match: "Dune", + steps: [{ toolCalls: [{ capability: "agent.await", input: { runIds: ["child-1"] } }] }], + finalText: (r) => `子会话:${r.at(-1)?.text}`, + }; + const first = await start(fx, [script], { sessionDir, runId: "run-sus", capabilities: ["agent.await"] }); + handle = first.run; + await expect(first.run.prompt()).rejects.toThrow(); + await first.run.close(); + expect(first.run.history().some((e) => e.type === "tool_call_end")).toBe(false); + + suspended = false; + const second = await start(fx, [script], { + sessionDir, + runId: "run-sus", + capabilities: ["agent.await"], + firstSeq: first.run.history().length, + }); + expect(second.run.pendingOperations).toBe(1); + const out = await second.run.resume(); + expect(out).toMatchObject({ status: "completed", finalText: '子会话:{"allDone":true}' }); + expect(calls).toHaveLength(2); + expect(calls[1]).toMatchObject({ inv: calls[0]!.inv, resumed: true }); + const ends = second.run.history().filter((e) => e.type === "tool_call_end"); + expect(ends).toEqual([expect.objectContaining({ capabilityId: "agent.await", isError: false, recovery: true })]); + }); + + it("旧版(pi-agent-core 0.99)会话文件:打开时给出明确错误,不崩溃", async () => { + const sessionDir = await tempDir(); + writeFileSync(join(sessionDir, "run-old.jsonl"), '{"type":"header","id":"run-old","storageVersion":4}\n'); + const fx = makeFixture(); + const err = await start(fx, [dune], { sessionDir, runId: "run-old" }).catch((e: unknown) => e); + expect(err).toBeInstanceOf(LegacySessionError); + expect((err as Error).message).toContain("旧版本"); + }); }); diff --git a/packages/agent/test/settings.test.ts b/packages/agent/test/settings.test.ts index 625a050d..4aa76366 100644 --- a/packages/agent/test/settings.test.ts +++ b/packages/agent/test/settings.test.ts @@ -38,7 +38,12 @@ describe("AgentSettings → pi-ai", () => { }); const { model } = resolveModelFromSettings(s); expect(model).toMatchObject({ api: "openai-completions", baseUrl: "http://llm.local/v1", reasoning: false }); - expect(compactionFromSettings(s, 32_000)).toEqual({ enabled: true, reserveTokens: 8000, keepRecentTokens: 8000 }); + expect(compactionFromSettings(s, 32_000)).toEqual({ + enabled: true, + reserveTokens: 8000, + keepRecentTokens: 8000, + backgroundTokens: 0, + }); const manual = resolveAgentSettings({ ...s, compaction: { mode: "manual", threshold: 0.8 } }); expect(compactionFromSettings(manual, 32_000).enabled).toBe(false); expect(() => resolveAgentSettings({ ...base, model: "" })).toThrow(); diff --git a/packages/agent/test/skills.test.ts b/packages/agent/test/skills.test.ts index 9dbaa21a..e4983e79 100644 --- a/packages/agent/test/skills.test.ts +++ b/packages/agent/test/skills.test.ts @@ -23,7 +23,7 @@ beforeAll(async () => { const REQUIRED_SECTIONS = ["## 适用意图", "## 步骤", "## 需要批准", "## 汇报模板", "## 不要做"]; -describe("加载(Pi harness loadSkills)", () => { +describe("加载(skills//SKILL.md)", () => { it("至少 5 个 Skill,没有格式诊断", () => { expect(diagnostics).toEqual([]); expect(skills.length).toBeGreaterThanOrEqual(5); diff --git a/packages/contracts/src/agent-events.ts b/packages/contracts/src/agent-events.ts index a589aee9..15f596ec 100644 --- a/packages/contracts/src/agent-events.ts +++ b/packages/contracts/src/agent-events.ts @@ -63,6 +63,8 @@ const ToolCallEnd = z.object({ /** 已脱敏、截断的输出摘要 */ outputSummary: z.string(), recovery: z.boolean(), + /** 工具执行耗时(毫秒,单调时钟;没真正执行的调用没有) */ + durationMs: z.number().int().nonnegative().optional(), }); const ToolCallRejected = z.object({ type: z.literal("tool_call_rejected"), diff --git a/packages/contracts/src/pi.ts b/packages/contracts/src/pi.ts index 379f1ca1..25158848 100644 --- a/packages/contracts/src/pi.ts +++ b/packages/contracts/src/pi.ts @@ -3,14 +3,14 @@ import { Id, Timestamp } from "./common.ts"; // Pi 模型上游(W7-A):一个上游 = 一个 pi-ai provider 的一份凭据(API Key / 订阅登录 token / 环境凭据)。 // 同一 provider 可以建多个上游(例如两个 Codex 账号);Agent 运行按上游 id 取凭据。 -// 研究结论与映射见 docs/design/pi-providers.md;provider 列表锁定 pi-ai 0.99.2 的 builtinProviders()。 +// 研究结论与映射见 docs/design/pi-providers.md;provider 列表锁定 pi-ai 1.1.0 的 builtinProviders()(1.0.3 起 azure-openai-responses 改名 azure,迁移 0093 已改存量)。 -/** pi-ai 0.99.2 内置的 42 个 provider id(含 TypeSafe 分类服务) */ +/** pi-ai 1.1.0 内置的 42 个 provider id(含 TypeSafe 分类服务) */ export const PI_BUILTIN_PROVIDER_IDS = [ "amazon-bedrock", "ant-ling", "anthropic", - "azure-openai-responses", + "azure", "baseten", "cerebras", "cloudflare-ai-gateway", diff --git a/packages/db/migrations/0093_pi_azure_provider.ts b/packages/db/migrations/0093_pi_azure_provider.ts new file mode 100644 index 00000000..9288f5c6 --- /dev/null +++ b/packages/db/migrations/0093_pi_azure_provider.ts @@ -0,0 +1,15 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// pi-ai 1.0.3 把 Azure provider 从 azure-openai-responses 改名为 azure(api id 与 AZURE_OPENAI_* 配置名不变)。 +// 已存的模型服务跟着改名,否则按 provider 查不到 pi-ai 的内置 provider。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl(db, `update pi_upstream set provider = 'azure' where provider = 'azure-openai-responses';`); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl(db, `update pi_upstream set provider = 'azure-openai-responses' where provider = 'azure';`); +}