diff --git a/docs/README.md b/docs/README.md index 2c317ff..1f7060c 100644 --- a/docs/README.md +++ b/docs/README.md @@ -47,6 +47,7 @@ Graph Agent 架构重构(27 系列)。**已完成的阶段文档已移入 `d | [27-p4-parallel-and-artifacts.md](./todo/system-design/27-p4-parallel-and-artifacts.md) | **P4(回收编号)并行执行与产物索引** — 📝 设计中,未实现。fan-out 真并发 + 同 agent 多实例隔离 + `files.json` 产物索引。三个 Phase,C 可插队 | | [28-followups.md](./todo/system-design/28-followups.md) | 28 号(Agent 注册表统一)的遗留清单 — 主文档已在 `done/`,这份装没做完的:dream 机制(设计见 29 号,已实现,见下)、`source_kind`(✅已删)/`legacy_path`(不能删)删列、三项「已实施未验证」。任务态清理(✅已实现,见下)。§五 是**已立起来的守卫清单**(别重复造,也别删) | | [29-dream-design.md](./todo/system-design/29-dream-design.md) | **dream(经验汇流)**— ✅ 已实现。定时整理所有 agent 在所有公司的记忆,提炼后跨公司沉淀(不泄漏公司信息)。四条取舍已拍板并实现:去敏独立判定、四落点分层写入(经验/`SKILL.md`/`soul.md`/`agent.md`,受管区块+三道闸门)、每日定时(串行+空转保护)、引用真伪自动查、REST+控制台入口(`/api/dream/*`、`/dream`)。`innovator-evolver` 真 CLI 跑通一次。**仍缺的**:「结论对不对」只能人抽查;「套进 org/work graph 来跑」评估后用户选了不做(作用域不匹配,见文内 §5.7) | +| [30-task-delegation-reliability-gaps.md](./todo/system-design/30-task-delegation-reliability-gaps.md) | **任务委派可靠性现状**— 📝 调研记录,非设计方案。DevLoop `runUnit`、Company 招聘装配、Graph 执行三条链路核对结论一致:都是「失败预算耗尽 → 停等人工」,没有「卡住了主动通知人」这一层;DevLoop 当前还默认走 mock,真实 CLI 成功率无实证数据 | ## todo/future/(暂缓,非当前重心) diff --git a/docs/todo/system-design/27-p4-parallel-and-artifacts.md b/docs/todo/system-design/27-p4-parallel-and-artifacts.md index 193e468..b517789 100644 --- a/docs/todo/system-design/27-p4-parallel-and-artifacts.md +++ b/docs/todo/system-design/27-p4-parallel-and-artifacts.md @@ -17,6 +17,26 @@ --- +## 〇.5、上级背景:黑灯工厂差距评估(2026-08-21) + +用户原话(提问): + +> lemonboss 现在离我们想做的黑灯工厂,能够自主 7*24 小时,自我规划,自我改进,自我迭代,自我完成任务的 graph org agent 系统,还有多远 + +评估结论(基于当时代码库状态 + `system-design/27` 系列文档): + +图引擎地基(P0~P3)扎实——org/work graph 统一执行引擎、暂停/重试/回退/熔断、真 Claude CLI 跑通、控制台都已落地并有真跑验收,这部分不是障碍。离「黑灯工厂」还有三个主要缺口: + +1. **真并发/无人值守跑不起来** —— 即本文档要做的 P4。现在一家公司同时只有一个 active root run,是单线程步进(`advanceOne` 一次只拾取一个 ready 节点),没有 fan-out 真并发,也没有为无人值守设计的看门狗/异常恢复层。7×24 意味着要能同时开很多任务、扛得住某个节点卡死或进程崩溃,这套目前没有。 +2. **自我改进闭环是半成品** —— P5(自我进化)对应的 dream 机制(29 号文档)刚实现:能定时把 agent 经验蒸馏回 `memory.md`/`SKILL.md`/`soul.md`/`agent.md`,但"蒸馏出来的结论对不对"只能人工抽查,没有自动校验,也没有 REST/控制台入口去监督它。更麻烦的是几条已知的本体装配漏洞——org-coder 没有共享本体会静默退化成通用 Base Agent;devloop 的 impl/test 节点根本没有 `agent.md`;work 节点执行时 org node 的 `agent.md` 硬截断前 2000 字、且只加载 `agent.md`,soul/tools/identity/user 全部不进提示词。自我改进要改的是这些本体,本体装配本身有静默漏洞,改进闭环就是在往一个不完整的东西上写。 +3. **没有真正的"自我规划"层** —— 现在的 org graph / work graph 都是 `orggraph-planning` / `workgraph-planning` 根据**人给的目标**生成的一次性产物,OKR/任务系统的驱动权还在人手上。定时的只有 dream(整理记忆),没有"自主决定下一步该做什么任务/该设什么目标"的常驻调度层——这一层连设计文档都还没有(不在现有 P5/P6 范围内)。 + +优先级建议:P4 并行(本文档)→ 补本体装配的静默坑(不然自我改进在改错东西)→ P6 收编统一循环 → 把 dream 的"结论对不对"自动化 → 最后才是自主任务生成层(目前完全空白,需要新设计)。 + +本文档要设计的 P4 是这份差距评估里排第一的缺口,也是三者中唯一已有详细设计(§四)且大部分已验证落地(阶段 A/C 完成,见 §五校准块)的部分——剩下的是阶段 B(T3-T6, T10)。 + +--- + ## 一、目标与非目标 ### 目标 @@ -152,7 +172,23 @@ advanceBatch(runId, { maxConcurrency }) → { results: [...], activated: [...] - **自动模式**:`maxConcurrency = 有效额度`,一次拾取整批并发跑。 - `advanceOne(runId)` 保留为 `advanceBatch(runId, {maxConcurrency: 1})` 的薄封装,P3 已有的调用方和测试不用改。 -**拾取的原子性**:P3 的幂等靠 `WHERE status='ready'` 条件更新(受影响行数=0 即视为被别人拿走)。批量拾取要沿用同一手法——**在一个事务里把这批节点一次性置 `running`**,而不是先 SELECT 再逐个 UPDATE,否则两个并发的 `advanceBatch` 会拾取到重叠的集合。 +**拾取的原子性 —— 决策(2026-08-21,推翻本节上一版)**:不需要新开一个包裹整批的 `BEGIN IMMEDIATE` 事务。直接复用 `advanceOne` 今天已经在用的同一个原语——`sse.transitionNode(runId, nodeId, 'ready', 'running', ...)`,单行条件 `UPDATE ... WHERE status='ready'`(`graph-engine.js:1369` 原话:「ready → running 的条件更新就是并发闸门」)。这条语句本身在 SQLite 单写者模型下已经是原子的。 + +```js +const candidates = db.prepare( + `SELECT * FROM graph_run_nodes WHERE run_id=? AND status='ready' ORDER BY created_at, id LIMIT ?` +).all(runId, maxConcurrency); + +const claimed = candidates.filter(node => + sse.transitionNode(runId, node.id, 'ready', 'running', ...) // advanceOne 今天调的同一个函数 +); +// 拾取阶段全程同步、零 I/O;下一步才 await 真正的 agent 调用 +const results = await Promise.allSettled(claimed.map(node => executeNode(node))); +``` + +**为什么不需要包裹事务**:两个并发 `advanceBatch` 即使 `SELECT` 到同一批候选节点,也是逐行在 `transitionNode` 上竞争——每一行只有一个调用者拿到 `changes=1`,另一个拿到 `0` 直接跳过。最终两边拾取到的集合天然无交集(§六 验收点 2),不需要把「数出还有空位」到「占上」捆成一个不可分事务——那是 `work-graph-db.js:1437` 建 run 时才需要的手法(它是「先数活跃 run 数、再判断、再写」这种读-判-写耦合在一起的多语句决策,必须原子;节点拾取没有这种耦合,每一行的归属独立由它自己的 `UPDATE` 决定)。 + +**已知且接受的代价**:某次 `advanceBatch` 可能因为在部分候选节点上输掉竞争,实际拾到的节点数少于 `maxConcurrency`(即使还有别的 ready 节点没进这次的 `LIMIT` 窗口)。自动模式下一轮循环会补上,不必现在解决。 ### 4.2 同一 agent 的多实例隔离 @@ -185,6 +221,21 @@ const scope = `gr:${nodeRunId}`; **沿用 P3 已实现的 join 语义,不加特例**:`graph-engine.js:1078` 的规则是「所有入边条件都满足才置 ready」。失败的那条分支走的是 `fail` 边,不满足 fan-in 节点的 `success` 入边条件,所以 **fan-in 节点不会被点亮**,图停在那里——这正是期望行为。 +**并发下的 join 正确性证明(2026-08-21,落实 T6 前先确认过再写代码)**:A、B 两个并行分支都喂给 fan-in 节点 C,会不会两边都在对方提交前读到「还没齐」,谁都不去点亮 C?不会——`settleNode`(`graph-engine.js:1514`)是**纯同步函数**,内部(写自己状态 → `activateSuccessTargets` 读兄弟节点状态 → 可能 `markReady`)全程没有一个 `await`,全是同步的 `better-sqlite3` 调用。Node 事件循环单线程,`await agentExecutor(ctx)` 一旦 resolve,从那之后到 `settleNode` 跑完是**不可打断的一整个 tick**——A、B 谁的 agent 调用先 resolve 不重要,**谁的 `settleNode` 后跑**,谁就会看到两边状态都已提交,从而正确点亮 C。真正并发的只有 `await` 着的 agent 调用本身(正是 T6 要并行的那部分)。**结论:join 不需要额外事务或锁**,`activateSuccessTargets`/`markReady` 原样复用即可。 + +**T4/T6 的具体切法**:把 `advanceOne` 里「拾取之后」的整段(建 ctx → `await` 执行 → `settleNode`,`graph-engine.js:1379-1505`)抽成一个可复用函数(如 `runClaimedNode(runId, run, pickedNode, opts)`),`advanceOne` 与 `advanceBatch` 的并行路径共用它: + +```js +// T4 的拾取只到「置 running」为止(见 §4.1);执行与结算是这里 +const settled = await Promise.allSettled( + claimed.map(node => runClaimedNode(runId, run, node, opts)) +); +``` + +`runClaimedNode` 内部原有的 try/catch(把 agent 调用异常转成 `fail` outcome 喂给 `settleNode`,而不是往外抛)已经是主要的错误containment;`Promise.allSettled` 补的是最后一道防线——万一 `settleNode` 自己有 bug 抛出去,也不会拖垮同批里其它节点的 promise。 + +**已知且不打算优化的冗余**:一批里多个节点并发结算,各自 `settleNode` 末尾都会调一次 `settleRun(runId)`——同一个 run 短时间内被判定收敛好几次,幂等但浪费,不是 bug,不在本阶段处理范围。 + 需要补的是**可观测性**,不是新语义: - run 层面要能回答「我卡在等哪几个节点」。控制台已有的节点状态足以推导,但要在 UI 上显式标出「fan-in 节点 `X` 在等 `A`(fail) / `B`(success) / `C`(success)」,否则用户只看到「不动了」。 @@ -192,7 +243,17 @@ const scope = `gr:${nodeRunId}`; ### 4.4 并发额度:三层,逐层取小 -对齐 §三.5 的既有数字,不另立一套: +**决策(2026-08-21)**:暂时不做「三层配额审批」这套东西——公司级 active root run 数、单机 session 数都**不设硬顶**(下面表格降级为参考数字,不是要实现的约束),唯一要卡的门是**账户欠费**(一旦欠费,不允许发起新的 agent 调用)。 + +> ⚠️ **欠费检查目前在代码里不存在**——全仓库搜过 `graph-engine.js`,唯一的门是 `checkBudget(run)`(`graph-engine.js:2334`),那是**任务级**的 agent-call/token 预算池(R-15 那套「只增不减的共池」),跟「账户是否欠费」是两个不同维度的概念,不能互相顶替。P4 范围内**不新建**账户级欠费系统;如果这道门确实要在 P4 上线前接通,需要先确认计费系统那边有没有现成的余额/欠费状态可查——这是一个**依赖,不是设计问题**,先问清楚再决定要不要塞进 T5。 +> +> **恢复语义决策(2026-08-21)**:欠费挂起直接复用 `suspendRunTree`(`graph-engine.js:2383`)——跟预算耗尽走同一条路:root run → `blocked`,全树非终态 run 置 `pause_requested=1`,只是 `reason` 从 `'budget-exhausted'` 换成 `'billing-arrears'`。这套三层级联暂停/恢复(`pauseRun` / `startRunSync` / `resumeDescendants`,`graph-engine.js:2506-2578`)已经实现并测过,不需要为欠费再发明一套状态机。 +> +> **但有一个真实缺口**:今天从 `blocked` 恢复靠人手点「启动」(`admitOrRequeue('blocked'→'running')`,注释原话「解除 blocked 的正当入口是...点一下启动」)。欠费清零不会有人去点——这正是「7×24 无人值守缺看门狗」那个更大缺口的一个具体实例。**P4 范围内需要补的是一个自动恢复触发器**:欠费状态清除时,批量对所有 `run_status='blocked' AND block_reason='billing-arrears'` 的 root run 调 `startRun`。触发源是计费系统的事件/轮询,不是图引擎自己能感知的——这也是为什么欠费检查本身要先问清楚计费系统那边的依赖(上一段已经点出)。 +> +> 在那道门接通之前,`advanceBatch` 相当于「无限额度」,唯一还在起作用的数字上限是 T1 已经落地的 `SessionPool.maxSessions`(防止单机字面意义上被并发的 Claude CLI 子进程压垮,是内存/CPU 层面的保命阀,跟欠费门是两回事,**不因为这条决策而放松**)。 + +对齐 §三.5 的既有数字,不另立一套(**下表现改为参考基线,非本阶段要实现的硬约束**): | 层级 | 上限 | 落在哪 | |---|---|---| @@ -200,7 +261,7 @@ const scope = `gr:${nodeRunId}`; | **单公司** | 默认 3 个 active root run | 替换 §三.3 那个「只允许 1 个」的唯一索引 | | **单机** | ≤ 20 个 ACP session | `SessionPool` 加 `maxSessions`(`plan-cost-concurrency.md` §④ 已经提出要加这个配置,至今未加) | -**有效额度 = min(三层剩余)**。`advanceBatch` 拾取前先算这个值,为 0 就本轮不拾取(不是报错,等下一轮)。 +~~**有效额度 = min(三层剩余)**。`advanceBatch` 拾取前先算这个值,为 0 就本轮不拾取(不是报错,等下一轮)。~~ **按 2026-08-21 决策作废**:不算三层 min,只剩单公司/单机两个数字本身也不设硬顶。`advanceBatch` 唯一还要看的额度就是本 run 的 `max_concurrency`(SELECT 的 `LIMIT`),不需要为它算「剩余」——因为再没有别的层可以让它变小。 **任务级总预算不变**:P3 已实现「按 root_run_id 共池的 backward_count + 任务级预算挂 root run」,并行不改变预算语义——**并行只是让预算烧得更快**,这一点要在控制台的成本提示里体现。 @@ -502,13 +563,13 @@ R-19 之后这两件事有实体(`graph_edge_contracts` × `graph_files`), |---|---|---|---| | ~~**T1**~~ ✅ | `SessionPool` 加 `maxSessions` 上限 + 调用方传 `scope` —— **已完成**(27-P3C §15.5-2,守卫 `session-pool-max.test.js` / `session-pool-scope-key.test.js`) | `server/src/core/session-pool.js` | 无 | | **T2** | DevLoop 规格:§12 第一条不变量放宽为 `(pid, agent, scope)`。**改 HTML 后必须跑 `npm run spec:extract`** | `docs/prd/newfunction/*.html` | 无 | -| **T3** | `graph_runs` 加 `max_concurrency` 列(**未做**);~~DROP `idx_graph_runs_one_active_root`,改为事务内计数检查~~ ✅ **后半已完成**(27-P3C 批 4:四步准入 `BEGIN IMMEDIATE` → 事务内重读 → 事务内计数 → 条件更新,8 进程实测零超发) | `server/services/work-graph-db.js` | 无 | -| **T4** | `advanceBatch(runId, {maxConcurrency})`:事务内批量拾取并置 running;`advanceOne` 降级为薄封装 | `server/services/graph-engine.js` | T3 | -| **T5** | 并发额度计算(三层取小)+ 额度为 0 时的等待语义 | `server/services/graph-engine.js` | T1, T3, T4 | -| **T6** | 并行执行 + 错误降级(照搬 `orchestrator.executeWithDependencies` 的 `Promise.allSettled` 策略,单分支失败不终结整批) | `server/services/graph-engine.js` | T4 | +| ~~**T3**~~ ✅ | `graph_runs` 加 `max_concurrency` 列(可空,NULL 用新导出常量 `DEFAULT_MAX_CONCURRENCY=5`,对齐 `plan-cost-concurrency.md` §④);`createRun` 新增 `maxConcurrency` 参数并持久化 —— **2026-08-21 已完成**(迁移块 + `createRun` INSERT 都改了,`hasColumn` 判断 + `ALTER TABLE ADD COLUMN` 照抄既有存量库迁移写法)。~~DROP `idx_graph_runs_one_active_root`~~ 那半本来就已完成(27-P3C 批 4),不用动 | `server/services/work-graph-db.js` | 无 | +| ~~**T4**~~ ✅ | `advanceBatch(runId, {maxConcurrency})`:`maxConcurrency<=1` 直接委托 `advanceOne`(零行为差异,不是反过来让 advanceOne 委托 advanceBatch——见函数注释里的理由);`>1` 时 `SELECT ... LIMIT` 取候选,循环调 `sse.transitionNode` 逐行拾取(**不包裹事务**,§4.1 决策的完整论证)。**2026-08-21 已实现并测过** | `server/services/graph-engine.js` | T3 ✅ | +| ~~**T5**~~ | ~~并发额度计算(三层取小)+ 额度为 0 时的等待语义~~ **按 2026-08-21 决策大幅简化并重新定义**:不再有三层 min 计算,`advanceBatch` 只看本 run 的 `max_concurrency`。剩下两件事都**依赖计费系统接口确认**,暂不能开工:① 欠费检查怎么接(`checkBudget` 是任务级预算,跟账户欠费是两个概念,不能互相顶替);② 欠费清零后的**自动恢复触发器**——复用 `suspendRunTree`/`startRunSync` 现成的挂起/恢复状态机(`reason='billing-arrears'`),但今天从 `blocked` 恢复是人手点「启动」,无人值守场景需要一个订阅计费事件、批量调 `startRun` 的轮询/webhook,这是新代码,不是配置项 | `server/services/graph-engine.js` | T1, T3, T4,另需计费系统接口 | +| ~~**T6**~~ ✅ | 把 `advanceOne` 拾取之后的整段(建 ctx → 执行 → `settleNode`)抽成 `runClaimedNode(...)`,`advanceBatch` 用 `Promise.allSettled` 并发跑 claimed 节点。fan-in join 复用现有 `activateSuccessTargets`/`markReady`,**不需要新事务/锁**(并发安全性证明见 §4.3)。**2026-08-21 已实现并测过** | `server/services/graph-engine.js` | T4 ✅ | | ~~**T7**~~ ✅ | 产物索引:追加、查询、rebuild —— **已完成,形态换成 `graph_files` 表**(见 §3.6 校准块),不是 `files.json` | `server/services/work-graph-db.js` | — | | ~~**T8**~~ ✅ | 产物按 run/node 分目录 + 节点产出后自动进索引 —— **已完成**(`runs/{rootRunId}/nodes/{nodeRunId}/`) | `server/services/graph-engine.js` | — | -| **T9** | node agent 的 prompt 注入 `files.json` 检索说明 | `system/templates/base/agent.md` 等 | T7 | +| **T9** | ~~node agent 的 prompt 注入 `files.json` 检索说明~~ **很可能已经被 R-19 做掉、不用再建**:R-19 的设计哲学(§4.6「读回侧」)明确拒绝了「注入一句『自己去索引里查』」这条路——`describeInputs`/`resolveInboundLines` 已经把声明过契约的输入**直接解析成绝对路径**塞进提示词,不依赖 agent 自己检索。T9 原始设想的场景(agent 主动查询未在边上声明的共享产物)目前没有对应的可调用工具——Claude CLI 只读被塞进提示词或提示它去 `Read` 的路径,`graph_files` 表对它是不可见的。**先确认这个场景是否真的需要,需要则要设计一个新工具而不是「prompt 里加一句话」,不需要则直接划掉** | `system/templates/base/agent.md` 等 | T7 | | **T10** | 控制台:fan-in 等待态可视化(「在等哪几个节点」)+ 并行运行中的多节点高亮 | `web/src/views/GraphConsole.vue` | T6 | | **T11** | 测试:批量拾取的并发幂等(两个 `advanceBatch` 不拾取重叠集合)、部分失败的 fan-in 停等、额度上限生效、`files.json` rebuild 一致性 | `server/tests/unit/` | T4-T8 | @@ -517,7 +578,7 @@ R-19 之后这两件事有实体(`graph_edge_contracts` × `graph_files`), | 阶段 | 含 | 交付 | 可否独立上线 | |---|---|---|---| | **A · 隔离** | ~~T1~~ ✅, T2 | 同一 agent 可安全并发调用 —— **T1 已完成**,只剩 T2(DevLoop 规格放宽) | 基本具备 | -| **B · 并行** | T3-T6, T10 | fan-out 真正并发;解除单 root run 约束 | ✅ 依赖 A | +| **B · 并行** | ~~T3~~ ✅ ~~T4~~ ✅ ~~T6~~ ✅, T5(卡计费系统依赖), T10 | fan-out 真正并发;解除单 root run 约束 | ✅ 依赖 A | | **C · 产物索引** | T7-T9 | `files.json` + 按 task 分目录 | ✅ **完全独立于 A/B**,可最先做 | **阶段 C 与 A/B 无依赖**,如果想先看到效果,可以先做 C(它是用户明确要求过的第 3 项)。 @@ -528,6 +589,23 @@ R-19 之后这两件事有实体(`graph_edge_contracts` × `graph_files`), > 这一段是「文档落后于代码」的又一例 —— 判断 P4 还剩多少工作量时, > **先照着任务表逐条 grep 一遍代码**,别照文档估。 +> **2026-08-21 实现记录(T3/T4/T6)**: +> - `server/services/work-graph-db.js`:`graph_runs` 加 `max_concurrency`(可空列 + 存量库迁移), +> 新导出常量 `DEFAULT_MAX_CONCURRENCY=5`,`createRun` 新增 `maxConcurrency` 参数并校验/持久化。 +> - `server/services/graph-engine.js`:把 `advanceOne` 拾取节点之后的整段(建 ctx → 执行 → `settleNode`) +> 原样抽成 `runClaimedNode(runId, run, def, defNode, nodeName, picked, opts)`,`advanceOne` 改成调它 +> (逐字同行为,靠现有 40 个用例背书,跑绿)。新增 `advanceBatch(runId, opts)`:`maxConcurrency<=1` +> 委托 `advanceOne`;`>1` 时按 §4.1 决策逐行 `sse.transitionNode` 拾取(不包裹事务)、 +> `Promise.allSettled` 并发跑 `runClaimedNode`。两者都从 `module.exports` 导出。 +> - 新测试 `server/tests/unit/graph-engine-advance-batch.test.js`(5 条,diamond 图 entry→{a,b,c}→join): +> fan-out 真并行(不借助 setImmediate/sleep,靠「`advanceBatch` 在自己第一次 await 之前全程同步」 +> 这条性质直接同步断言)、批量拾取跳过已被抢走的候选、部分失败时 join 保持 pending 且 run 收敛到 +> `blocked`、`maxConcurrency<=1` 委托行为、`run.max_concurrency` 持久化生效。 +> - 全量 `npx vitest run`:147 文件 / 1347 用例全绿,零回归。 +> - **没做的**:`advanceBatch` 还没接进 `runAuto`/路由——自动模式今天仍然只调 `advanceOne`, +> 实际生产流量拿不到并行,这是接下来的线(大概率是 T10 之前、或与它同批)。T5(欠费门 + 自动恢复 +> 触发器)仍卡在计费系统接口确认,T10(控制台可视化)未动。 + --- ## 六、验收标准 @@ -565,9 +643,9 @@ R-19 之后这两件事有实体(`graph_edge_contracts` × `graph_files`), 1. **并行让预算烧得更快,但预算语义不变**。任务级总预算是「只增不减的共池」(P3 已实现),并行不改变总量,只改变消耗速率。**控制台的成本提示需要相应更新**——用户点「自动」时看到的预估时间会大幅缩短,但花费不变。 2. **改 DevLoop 规格的连带风险**(T2)。§12 第一条不变量是 DevLoop 的,放宽它理论上不影响 DevLoop 自身(它的 `runProject` 波内本来就是串行的),但**必须跑一遍 DevLoop 全部测试**确认。 -3. **`maxSessions` 上限触顶时的行为未定**。额度为 0 时是「等待下一轮」还是「拒绝并报错」?本文档 §4.4 选了前者(不报错、等下一轮),但如果一个 run 长期抢不到额度会静默卡住——需要一个「等待超时」告警,否则表现为「不动了」。**留待实现时定阈值。** -4. **`files.json` 的并发写入**。并行节点会同时追加索引,需要串行化(文件锁或走 SQLite)。本文档设计的是 JSON 文件形态(符合用户要求「一个 files.json 文件」),但**并发追加是真实风险**——可能需要退化为「引擎内单写者队列」。 -5. **`files.json` 与 `consumedBy` 的准确性**。`consumedBy` 是**声明式**的(规划时写「谁应该读它」),不是实测的。如果规划错了,索引会误导下游 agent。缓解:把它当提示而非约束,agent 仍可读任何 `files.json` 里的条目。 +3. ~~**`maxSessions` 上限触顶时的行为未定**。额度为 0 时是「等待下一轮」还是「拒绝并报错」?本文档 §4.4 选了前者(不报错、等下一轮),但如果一个 run 长期抢不到额度会静默卡住——需要一个「等待超时」告警,否则表现为「不动了」。**留待实现时定阈值。**~~ **按 2026-08-21 决策已解决**:公司级/机器级不再设硬顶,「长期抢不到额度」这个场景基本不会出现了。账户欠费门定了——复用 `suspendRunTree` 的挂起语义(`reason='billing-arrears'`),跟预算耗尽是同一条路径,不发明第二种失败形态。**新缺口**:欠费清零后的恢复今天要人手点「启动」,无人值守场景下没人点——需要一个自动恢复触发器(欠费状态清除时批量对 blocked 的 root run 调 `startRun`),细节见 §4.4 决策块。欠费检查本身怎么接、恢复触发器订阅什么事件源,都要等计费系统那边的接口确认。 +4. ~~**`files.json` 的并发写入**。并行节点会同时追加索引,需要串行化(文件锁或走 SQLite)。本文档设计的是 JSON 文件形态(符合用户要求「一个 files.json 文件」),但**并发追加是真实风险**——可能需要退化为「引擎内单写者队列」。~~ **已随 §4.6 的 v0.2 重写作废**:产物索引早就从「一个 files.json 文件」改成了 `graph_files` SQLite 表(阶段 C 已完成)。并发追加不需要文件锁——`indexNodeOutputs` 是 `settleNode` 内部的同步调用,跟 §4.3 证明 join 不需要额外锁是同一个道理(Node 单线程事件循环 + `better-sqlite3` 单写者,写表本身已经串行)。这条风险不成立,留着只是防止以后有人照旧设计去实现一个不存在的 JSON 文件锁。 +5. **产物索引的 `consumedBy`/契约声明的准确性**。`graph_edge_contracts` 的 `needs`/`produces` 是**声明式**的(规划时写「这条边应该传什么」),不是实测的。如果规划 agent 声明错了,`validateGraph` 的 `data-contract` 规则能拦住「声明要吃的 kind 全图没人产」这类静态错误,但拦不住「声明的 kind 与实际语义对不上」这种语义错误。缓解方式不变:索引/契约是给 agent 的强提示(引擎直接把解析出的路径递到提示词里),不是不可绕过的约束,agent 仍可读同一 task 下任何已登记的产物。 --- diff --git a/docs/todo/system-design/30-task-delegation-reliability-gaps.md b/docs/todo/system-design/30-task-delegation-reliability-gaps.md new file mode 100644 index 0000000..d045542 --- /dev/null +++ b/docs/todo/system-design/30-task-delegation-reliability-gaps.md @@ -0,0 +1,86 @@ +# 30-task-delegation-reliability-gaps. 任务委派可靠性现状 —— 已确认的鸿沟清单 + +> **日期**: 2026-08-21 +> **性质**: 现状调研记录,不是设计方案。列的是「代码现在长什么样、缺什么」,不是「该怎么补」—— +> 要不要补、按什么优先级补,留给后面拍板。 + +--- + +## 起因 + +用户问:「自动跑一个任务,能保证成功吗?」分别对 DevLoop(`runUnit`)和 Company/OKR(Agent 招聘 + 任务委派) +两条链路查了代码。结论是同一个:**全系统没有任何一处能"保证自动跑到底"**,统一是 +「重试到预算/时间窗耗尽 → 停下等人工」的模式,区别只在于兜底的**粒度**和**可见性**。 + +--- + +## 一、DevLoop `runUnit`(Coder→Tester 循环) + +**文件**: `server/src/core/devloop/loop-controller.js`(`runUnit`,394-479 行)、 +`server/services/devloop/unit-registry.js`(FSM) + +- `retry_budget` 默认 5 次,每次失败把上一轮 `coder.log` 摘要 + Tester 失败原因打包成 + retry context 喂回 Coder,带着"上次为什么挂"重跑(reflection)。 +- 5 次全挂 → `unitRegistry.block(...)`,状态机进 `blocked`(`unit-registry.js:36`)。 + **没有自动退出路径**,`blocked → pending` 只能靠人工干预。 +- **当前默认走 mock 模式**(`opts.mock !== false` 默认 true),走 + `ClaudeCliProtocol({mock:true})` 回放 `fixtures/devloop-mock/*.json` 固定结果, + 不是真起 Claude CLI + 跑测试。所以"能不能保证成功"这个问题在默认配置下基本没有实际意义—— + mock 场景下结果是脚本编排好的,真实 CLI 下的成功率目前**没有实证数据**。要测真实场景 + 得显式传 `mock: false`。 +- 阶段现状:DevLoop 目前在 P2(单 Unit 循环),跨 Unit 的 `runProject`(P3)代码已经存在 + (同文件 556 行往下),但官方状态仍是"正朝 P3 推进",还未收口。 + +--- + +## 二、Company/OKR Agent 任务委派 + +### 2.1 招聘装配的时间窗(`org-coder-has-no-shared-body` 记忆已同步更新) + +**文件**: `server/services/company-service.js`(430、500-511 行) + +- `hire()` 现在会现场校验一次角色定义是否就绪,用 `body_not_ready` 网关挡住别家公司在 + 定义写完前重复招同一角色——比早前"静默退化成通用 Base Agent 且无报错"的状态好。 +- 但 agent-creator 写 `agent.md`/`soul.md` 依旧是**异步触发 + 单会话锁**,校验和填充之间 + 仍有时间窗,**没有完成回执/超时告警**。卡在这个窗口只能靠人工发现(比如逐个 + `head -3 data/companies/<公司>/agents/<本体>/agent.md` 看是不是还是 Base Agent 空壳)。 + +### 2.2 会话回收只有一层兜底 + +**文件**: `server/routes/agent-chat.js`(818-828 行,`CALL_KINDS` 表) + +- `CALL_KINDS` 精确覆盖心跳/pipeline/日程/告警/日报五类,跑完即回收。 +- 未标 `kind` 的调用退化成普通 `chat` 类型,靠 30 分钟空闲 TTL 兜底,**没有第二道防线** + (例如全局会话数硬顶)。漏标的后台调用不报错,只在内存曲线上体现—— + 与 `docs/done/claude-cli-session-leak.md` 记录的那次"2h43m 漏出 82 个进程/17.5GB"是 + 同一类风险,这次风险面在 Company/OKR 委派路径上同样存在,不只是 agent-chat 原生场景。 + 守卫 `agent-chat-session-reclaim.test.js` 已覆盖 `CALL_KINDS` 标注本身,但不覆盖 + "该标而没标"这类新增调用点的遗漏。 + +### 2.3 Graph 执行失败无自动恢复,与 DevLoop 同构 + +**文件**: `server/services/graph-engine.js`(`LOOP_HARD_CAP_EDGE` + fail 边路由) + +- 有兜底边就转兜底节点,没有就直接进 `blocked` 停等人工,不会自愈。 +- 这是委派链路的通用终态,不是 Company 独有问题——和 DevLoop 的 + `retry_budget` 耗尽 → `blocked` 是同一种设计选择(失败预算 + 人工兜底, + 不追求自动恢复),只是目前**分散在两套系统里各自实现,没有统一的"完成回执/超时告警"层**。 + +--- + +## 三、共性模式(值得记住,不代表已拍板要改) + +三条链路(DevLoop unit、Company 招聘装配、Graph 执行)共享同一个设计现状: +**有失败预算 + 人工兜底,没有"卡住了主动通知人"这一层**。 + +失败之后系统会停在一个确定状态(`blocked` / `training` 卡住 / 会话挂着不回收), +但**不会主动喊人**——都是被动的,靠人去翻状态或者翻内存曲线才发现。 +这是三处鸿沟共同的根因形状,不是三个独立问题。 + +--- + +## 参考 + +- DevLoop 阶段现状:[27-graph-agent-architecture.md](./27-graph-agent-architecture.md) +- 会话泄漏排查手册:[claude-cli-session-leak.md](../../done/claude-cli-session-leak.md) +- `body_not_ready` 网关所在的招聘流程背景:[28-followups.md](./28-followups.md) diff --git a/server/services/graph-engine.js b/server/services/graph-engine.js index 6f44286..8da2bde 100644 --- a/server/services/graph-engine.js +++ b/server/services/graph-engine.js @@ -56,6 +56,8 @@ const { edgeContracts, findFiles, recordFile, + // 27-P4 T4/T5:advanceBatch 批量拾取宽度的缺省值 + DEFAULT_MAX_CONCURRENCY, } = require('./work-graph-db'); /** @@ -193,6 +195,12 @@ const ENGINE_FAIL_REASONS = Object.freeze({ NODE_ERROR: 'node_error', BOUNDARY_ESCAPE: 'boundary_escape', CHILD_RUN_FAILED: 'child_run_failed', + // T-mt25ecbd:produces/needs 契约以前只进提示词(软提示),agent 没写文件 / 没读文件都 + // 不影响节点判 success、不影响下游点亮——三份计划文档没落盘、goal-benchmark 照样过。 + // 这两个新取值让契约变成真闸门:declared 的 kind 在 graph_files 里真的查不到就判 fail, + // 不发明取值域之外的字符串(否则会被 §四.D.2.2 的合法性检查当自创取值降级成 null)。 + MISSING_INPUT: 'missing_required_input', + PRODUCES_MISSING: 'produces_missing', }); /** 归因/结构化产出文件名(agent.md「产出格式」:runs/{rootRunId}/nodes/{nodeRunId}/attempt-{n}.json) */ @@ -1376,6 +1384,26 @@ async function advanceOne(runId, opts = {}) { return { ...result(runId, run.run_status, 'taken-by-other'), node: nodeView(runId, cur) }; } + return runClaimedNode(runId, run, def, defNode, nodeName, picked, opts); +} + +/** + * 已经拾取(`ready → running` 条件更新已经成功)的节点,从建 ctx 到执行到结算的完整一段。 + * 27-P4 T6 从 `advanceOne` 里原样抽出来,**行为逐字不变**——只是把原来闭包捕获的局部变量 + * 变成参数,好让 `advanceBatch` 的并行路径也能复用同一段逻辑(§4.3 决策:并发下的 join + * 正确性不依赖新增锁/事务,靠的正是这段收尾——`settleNode` 全程同步——的既有写法, + * 所以这里一个字都不能改成异步中途掺一次 DB 之外的 await)。 + * + * @param {string} runId + * @param {object} run getRun(runId) 的快照(调用方已经读过一次,不在这里重读) + * @param {object} def defIndex(run)(调用方已经算过一次,批量拾取时给多个节点共用) + * @param {object} defNode def.nodes.get(picked.def_node_id) || {} + * @param {string} nodeName + * @param {object} picked graph_run_nodes 行(此刻已经是 running) + * @param {object} [opts] + * @returns {Promise} + */ +async function runClaimedNode(runId, run, def, defNode, nodeName, picked, opts = {}) { const ctx = { runId, run, @@ -1506,15 +1534,174 @@ async function advanceOne(runId, opts = {}) { } /** + * 批量推进(27-P4 T4/T6):一次拾取至多 `maxConcurrency` 个 ready 节点,并发执行。 + * + * `maxConcurrency <= 1` 时**直接委托给 `advanceOne`**,不走下面的批量路径——零行为差异, + * `advanceOne` 现有的调用方/测试不用改一个字(见 27-P4 §4.1 决策:不是「advanceOne 包 advanceBatch」, + * 是反过来,这样 advanceOne 的行为永远只由它自己的测试背书,不会被批量路径的改动间接影响)。 + * + * `maxConcurrency > 1` 时: + * 1. 先做一遍与 `advanceOne` 完全相同的 run 级检查(stop/pause/running/预算)—— + * 这段刻意重复一份而不是抽共享函数,因为 `advanceOne` 是被大量测试钉住的稳定代码, + * 抽共享会让这次改动的 diff 面盖到它头上,得不偿失。 + * 2. `SELECT ... LIMIT maxConcurrency` 取候选,逐个走 `sse.transitionNode('ready'→'running')` + * —— 单行条件更新本身就是并发闸门,不需要包裹事务(§4.1 决策的完整论证)。 + * 3. 拾到的节点各自 `runClaimedNode`,`Promise.allSettled` 并发执行——一个节点的 + * executor 抛出去不会波及同批其它节点(`runClaimedNode` 内部已经把 agent 调用异常 + * 转成 fail outcome,这里的 catch 是防它自己有 bug 的最后一道网)。 + * + * @param {string} runId + * @param {object} [opts] + * @param {number} [opts.maxConcurrency] 不传则读 `run.max_concurrency`,再不然用 DEFAULT_MAX_CONCURRENCY + * @param {boolean} [opts.ignorePause] + * @param {string} [opts.task] + * @returns {Promise<{ok: boolean, runId: string, runStatus: string|null, reason: string|null, + * results: AdvanceResult[], activated: Array}>} + */ +async function advanceBatch(runId, opts = {}) { + const run0 = getRun(runId); + const maxConcurrency = opts.maxConcurrency + || (run0 && run0.max_concurrency) + || DEFAULT_MAX_CONCURRENCY; + + if (maxConcurrency <= 1) { + const res = await advanceOne(runId, opts); + return { ...res, results: res.node ? [res] : [] }; + } + + const run = run0; + if (!run) return { ok: false, runId, runStatus: null, reason: 'run-not-found', results: [], activated: [] }; + if (run.stop_requested) { + return { ok: false, runId, runStatus: run.run_status, reason: 'stop-requested', results: [], activated: [] }; + } + if (run.pause_requested && !opts.ignorePause) { + landPause(runId); + return { + ok: false, runId, runStatus: (getRun(runId) || {}).run_status, reason: 'pause-requested', + results: [], activated: [], + }; + } + if (run.run_status !== 'running') { + return { ok: false, runId, runStatus: run.run_status, reason: 'run-not-running', results: [], activated: [] }; + } + + const gate = checkBudget(run); + if (!gate.ok) { + suspendRunTree(run); + return { + ok: false, runId, runStatus: (getRun(runId) || {}).run_status, reason: 'budget-exhausted', + results: [], activated: [], budget: gate, + }; + } + + const def = defIndex(run); + const candidates = getDb() + .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND status = 'ready' ORDER BY created_at, id LIMIT ?`) + .all(runId, maxConcurrency); + if (!candidates.length) { + return { ok: true, runId, runStatus: settleRun(runId), reason: 'no-ready-node', results: [], activated: [] }; + } + + const claimed = []; + for (const picked of candidates) { + const defNode = def.nodes.get(picked.def_node_id) || {}; + const nodeName = defNode.name || picked.def_node_id; + // 逐行条件更新——两个并发 advanceBatch 即使 SELECT 到同一批候选,这里也只会有 + // 一边拿到 changes=1(§4.1 决策),拿不到的直接跳过,不算错误。 + if ( + sse.transitionNode(runId, picked.id, 'ready', 'running', { startedAt: now() }, { + node_name: nodeName, node_kind: defNode.nodeKind || 'agent_call', attempt: picked.attempt, + }) + ) { + claimed.push({ picked, defNode, nodeName }); + } + } + if (!claimed.length) { + // 候选全被别的 advanceBatch/advanceOne 抢走:不是错误,下一轮再拾取 + return { ok: true, runId, runStatus: run.run_status, reason: 'taken-by-other', results: [], activated: [] }; + } + + const settled = await Promise.allSettled( + claimed.map(({ picked, defNode, nodeName }) => runClaimedNode(runId, run, def, defNode, nodeName, picked, opts)) + ); + + const results = settled.map((s, i) => { + if (s.status === 'fulfilled') return s.value; + // 防的是 runClaimedNode 自己的 bug(它内部的 try/catch 已经把 agent 调用异常转成 + // fail outcome,正常情况下不会走到这里)。节点此刻卡在 running——不代它结算, + // 留给人工「重试」或下一次真跑复现;这里只保证同批其它节点不受影响。 + const { picked } = claimed[i]; + return { + ...result(runId, run.run_status, 'node-executor-rejected'), + node: nodeView(runId, getRunNode(picked.id)), + error: s.reason && s.reason.message, + }; + }); + + return { + ok: true, + runId, + runStatus: settleRun(runId), + reason: null, + results, + activated: results.flatMap((r) => r.activated || []), + }; +} + +/** * 节点执行结束后的落地:写终态 → 评估出边 → 收敛判定。 * 也是子 run 收敛时回写父节点走的同一条路(propagateToParent)。 * * @param {object} outcome {status, failReason?, remediationNote?, outputRef?, agentRunId?, childRunId?} */ function settleNode(runId, nodeRunId, outcome = {}) { - const to = outcome.status === 'success' ? 'success' : 'fail'; + let to = outcome.status === 'success' ? 'success' : 'fail'; + let failReason = outcome.failReason === undefined ? null : outcome.failReason; + + // T-mt25ecbd:executor 报 success 不等于契约兑现了 —— agent 可能压根没写声明过的文件 + // (brd/prd/sdd 三个 planner 节点复现过)。产出闸门放在这里、卡在 transitionNode 落库 + // 之前,这样一旦判 fail 就走下面 to==='fail' 的 routeFailure 分支,不会先点亮下游 + // 再回头发现产物不存在。 + if (to === 'success') { + const run0 = getRun(runId); + const nodeRun0 = getRunNode(nodeRunId); + const def0 = run0 ? defIndex(run0) : null; + // subgraph 节点(org 层)没有自己的产物目录——它的「产出」是子 work run 里的节点登记的 + // (org-layer-contracts.test.js ⑤ 的口径:output_ref 恒为 null)。这两道闸门只对 + // agent_call 节点生效,subgraph 节点的契约兑现与否由子 run 自己的节点去过这两道闸门。 + const isSubgraphNode = !!(def0 && def0.nodes.get(nodeRun0 && nodeRun0.def_node_id) + && def0.nodes.get(nodeRun0.def_node_id).nodeKind === 'subgraph'); + if (run0 && nodeRun0 && def0 && !isSubgraphNode) { + const missingProduces = unsatisfiedRequiredProduces(run0, def0, nodeRun0); + if (missingProduces.length) { + to = 'fail'; + failReason = ENGINE_FAIL_REASONS.PRODUCES_MISSING; + timelineNote({ runId, nodeRunId }, 'node.output', + `节点声明会产出 ${missingProduces.join(', ')},但产物目录里没找到对应文件,按 fail 结算`, + { node_run_id: nodeRunId, missing_kinds: missingProduces, degraded: 'produces_missing' }); + } else { + // T-mt25ecbd 症状②:goal-benchmark 的入边 needs:[brd,prd,sdd] 以前只用来拼 + // describeInputs 那句 ⚠️ 提示词——agent 完全可能对它视而不见、照样判自己成功 + // (复现原样:三份文档一份没写出来,benchmark 仍产出合法 acceptance)。 + // 这里在真正判 success 之前把同一份 needs 契约当真:声明了 required 却在 + // graph_files 里连一份实例都查不到,就不认这次 success,按 fail 结算。 + const missingInputs = requiredInputsMissing( + { run: run0, companyId: run0.company_id, defNodeId: nodeRun0.def_node_id, nodeKind: 'agent_call' }, + def0 + ); + if (missingInputs.length) { + to = 'fail'; + failReason = ENGINE_FAIL_REASONS.MISSING_INPUT; + timelineNote({ runId, nodeRunId }, 'node.output', + `节点声明必需的上游产物 ${missingInputs.join(', ')} 在索引里查不到实例,按 fail 结算`, + { node_run_id: nodeRunId, missing_kinds: missingInputs, degraded: 'missing_required_input' }); + } + } + } + } + const changed = sse.transitionNode(runId, nodeRunId, 'running', to, { - failReason: outcome.failReason === undefined ? null : outcome.failReason, + failReason, // 跨层地址与 fail_reason 同一次条件更新落库(R-17.1)。恒写显式值:success 或本轮没归因 // 就写 null,不给「读到上一轮地址」留窗口。normalize 让执行器可以直接给对象或 JSON 串。 failNodeRef: normalizeNodeRef(outcome.failNodeRef), @@ -3456,6 +3643,56 @@ const { companyRootDir, taskDirRel } = require('./company-paths'); * kind 怎么定:先看这个节点的**出边**上声明了什么(规划时就写下了「我会产一份 brd」), * 再按文件名兜底。声明优先,因为那是图作者的意图;文件名只是巧合。 */ +/** + * T-mt25ecbd 产出闸门:这个节点的出边声明了**必需**(required)produces kind, + * 而它的产物目录里**连一个候选文件都没有**——判「压根没执行到位」,按 fail 结算。 + * + * 刻意只判「空目录」,不判「declared kind 认领不上具体哪个文件」:后者是 + * indexNodeOutputs 的分类问题(ambiguous 时退成 kind=artifact,下游拿 ⚠️ 提示词), + * 分类不准不代表节点没干活——把它也当失败拦下来会误杀`⑦ 有歧义时不猜`那类合法场景 + * (agent 确实写了东西,只是文件名让引擎猜不出哪份对应哪个 kind)。 + * 本轮要拦的是 brd/prd/sdd 三个节点复现的那种更粗的情形:声明了要产出、 + * 目录里什么候选文件都没有——那不是分类模糊,是根本没写。 + * + * 目录读不了/没声明时一律放行(同 indexNodeOutputs 的兜底:不该让基础设施问题 + * 或没声明契约的节点被这道新闸门误杀)。 + * + * @returns {string[]} 一个都没写出来时返回声明的 kind 列表;有任何候选文件 = 放行([]) + */ +function unsatisfiedRequiredProduces(run, def, nodeRun) { + if (!run || !nodeRun) return []; + let declared = []; + try { + const layer = run.graph_kind === 'org' ? 'org' : 'work'; + const outbound = ((def && def.edges) || []).filter( + (e) => String(e.from) === String(nodeRun.def_node_id) && (e.edgeKind || 'forward') !== 'backward' + ); + for (const e of outbound) { + for (const c of edgeContracts(layer, e.id)) { + if ((c.role || 'produces') === 'produces' && c.required) declared.push(c.kind); + } + } + declared = [...new Set(declared)]; + } catch { + return []; // 契约查不到不该拖垮结算(同 indexNodeOutputs 的兜底) + } + if (!declared.length) return []; + + let dir; + let files = []; + try { + dir = nodeArtifactDir(run, nodeRun.id); + if (!fs.existsSync(dir) || !fs.statSync(dir).isDirectory()) return declared; + files = fs.readdirSync(dir).filter((f) => !f.startsWith('.')); + } catch { + return []; // 目录读不了不该拖垮结算 + } + const candidates = files.filter( + (f) => !/^attempt-\d+\.json$/.test(f) && f !== CONTRACT_SNAPSHOT_FILE + ); + return candidates.length ? [] : declared; +} + function indexNodeOutputs(run, def, nodeRun) { if (!run || !nodeRun || !nodeRun.output_ref) return; let dir = nodeRun.output_ref; @@ -3554,6 +3791,39 @@ function indexNodeOutputs(run, def, nodeRun) { * 走同一段代码:三种口径(查得到给绝对路径 / required 查不到给 ⚠️ / * 可选查不到整行省略)就不可能在两处漂移 —— 复制一遍再各自演化正是本轮在防的那类断链。 */ +/** + * T-mt25ecbd 输入闸门:入边声明的**必需**(required)kind,有几个在产物索引里真的 + * 一份实例都查不到——与 describeInputs 给 agent 看的 ⚠️ 提示同一份数据源 + * (inboundContracts + findFiles),区别是这里把它变成执行前的硬拦截,不再只是 + * 「提示词里有一行警告,agent 爱信不信」。 + * + * 只看 required 且缺实例的:可选(required:false)输入本来就允许没有,不该拦。 + * 索引读不出来(层查询异常)时放行——同 describeInputs 的兜底,基础设施问题不该 + * 误伤节点执行。 + * + * @returns {string[]} 缺实例的必需 kind 列表;空数组 = 放行 + */ +function requiredInputsMissing(ctx, def) { + if (!ctx || !ctx.run || ctx.nodeKind === 'subgraph') return []; + let contracts = []; + try { + const layer = ctx.run.graph_kind === 'org' ? 'org' : 'work'; + contracts = inboundContracts(layer, (def && def.edges) || [], ctx.defNodeId); + } catch { + return []; + } + const requiredKinds = [...new Set(contracts.filter((c) => c.required).map((c) => c.kind))]; + if (!requiredKinds.length) return []; + let found; + try { + found = findFiles({ companyId: ctx.companyId, taskId: ctx.run.task_id, kinds: requiredKinds }); + } catch { + return []; + } + const haveKinds = new Set(found.map((f) => f.kind)); + return requiredKinds.filter((k) => !haveKinds.has(k)); +} + function resolveInboundLines(ctx, contracts) { const declared = [...new Set(contracts.map((c) => c.kind))]; if (!declared.length) return []; @@ -3902,6 +4172,8 @@ function latestRemediationNote(nodeRunId) { module.exports = { // 核心原语 advanceOne, + advanceBatch, + runClaimedNode, advanceTree, runAuto, settleNode, diff --git a/server/services/work-graph-db.js b/server/services/work-graph-db.js index ece1e3d..a6296c2 100644 --- a/server/services/work-graph-db.js +++ b/server/services/work-graph-db.js @@ -54,6 +54,14 @@ const ACTIVE_ROOT_RUN_STATUS = ['ready', 'running', 'paused']; const DEFAULT_MAX_ACTIVE_ROOT_RUNS = 1; /** + * 单个 run 内 `advanceBatch` 一次最多拾取几个 ready 节点 —— 缺省值(27-P4 §4.4)。 + * 对齐 `plan-cost-concurrency.md` §④「项目内 L1 并行 ≤5」,不是随手挑的数字。 + * `graph_runs.max_concurrency` 为 NULL 时用它;不做「三层 min」计算(2026-08-21 决策, + * 27-P4 §4.4)——这一层只是批量拾取的宽度,不是配额审批。 + */ +const DEFAULT_MAX_CONCURRENCY = 5; + +/** * 硬上限。配置里写个 100000 不该被当真 —— 每个坑对应一条真在跑的 root run(一个 Claude CLI * 子进程量级的开销),机器级闸门(SessionPool.maxSessions)至今不存在(登记在 28-followups)。 * 超过就按上限截断并打日志,不静默当成无限。 @@ -314,6 +322,10 @@ function getDb() { stop_requested INTEGER NOT NULL DEFAULT 0, backward_count INTEGER NOT NULL DEFAULT 0, budget_json TEXT, + -- 27-P4 T3:本 run 一次 advanceBatch 最多拾取几个 ready 节点。NULL = 用 + -- DEFAULT_MAX_CONCURRENCY(引擎侧解析,不在 DB 层给默认值,理由同其它可空列—— + -- 应用层的默认值改了不用迁移存量行)。 + max_concurrency INTEGER, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); @@ -633,6 +645,10 @@ function getDb() { // `resumePendingGoalValidations` 会把库里**所有**历史终态 goal 都看成「汇报丢了」, // 于是把它们统统重报一遍 —— OKR 节点被历史结论覆盖、上级 agent 被几百条陈年旧事叫醒, // 每一条都真起一轮 LLM。它们在旧代码下早就报过了(或永远不该报),一律按「已报」封口。 + // 27-P4 T3:graph_runs 追加 max_concurrency,可空,旧行 NULL = 按 DEFAULT_MAX_CONCURRENCY 处理。 + if (!hasColumn(db, 'graph_runs', 'max_concurrency')) { + db.exec(`ALTER TABLE graph_runs ADD COLUMN max_concurrency INTEGER`); + } if (!hasColumn(db, 'goals', 'reported_at')) { db.exec(`ALTER TABLE goals ADD COLUMN reported_at TEXT`); db.exec(` @@ -1055,10 +1071,14 @@ function createRun({ executionMode = 'manual', runStatus = 'planning', budget = null, + maxConcurrency = null, } = {}) { if (graphKind !== 'org' && graphKind !== 'work') { throw new Error(`graphKind 必须是 'org' 或 'work',收到: ${graphKind}`); } + if (maxConcurrency !== null && (!Number.isInteger(maxConcurrency) || maxConcurrency < 1)) { + throw new Error(`maxConcurrency 必须是 ≥1 的整数或 null,收到: ${JSON.stringify(maxConcurrency)}`); + } const db = getDb(); const now = new Date().toISOString(); @@ -1076,13 +1096,15 @@ function createRun({ db.prepare(` INSERT INTO graph_runs ( id, graph_kind, graph_id, company_id, root_run_id, parent_run_id, parent_node_run_id, - goal_id, task_id, okr_node_id, execution_mode, run_status, budget_json, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + goal_id, task_id, okr_node_id, execution_mode, run_status, budget_json, max_concurrency, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `).run( runId, graphKind, graphId, companyId, rootRunId, parentRunId, parentNodeRunId, goalId, taskId, okrNodeId, executionMode, runStatus, // budget 只挂 root run:子 run 通过 root_run_id 扣同一个池 parentRunId ? null : budget && JSON.stringify(budget), + maxConcurrency, now, now ); @@ -2565,6 +2587,8 @@ module.exports = { maxActiveRootRuns, DEFAULT_MAX_ACTIVE_ROOT_RUNS, MAX_ACTIVE_ROOT_RUNS_CAP, + // 27-P4 T3:单 run 批量拾取宽度的缺省值,`run.max_concurrency` 为 NULL 时用它 + DEFAULT_MAX_CONCURRENCY, canTransitionNode, canTransitionRun, InvalidTransitionError, diff --git a/server/tests/unit/graph-engine-advance-batch.test.js b/server/tests/unit/graph-engine-advance-batch.test.js new file mode 100644 index 0000000..0243edd --- /dev/null +++ b/server/tests/unit/graph-engine-advance-batch.test.js @@ -0,0 +1,234 @@ +/** + * Unit Tests — advanceBatch(27-P4 T4/T6) + * + * 覆盖 27-P4 §六 验收标准: + * 2. 批量拾取幂等:候选里已经被别人抢先拾取的节点被跳过,不重复认领、不报错 + * 3. fan-out 真并行:一张有 3 个并行分支的图,同一批 advanceBatch 里三个节点同时 running + * 4. 部分失败停等:3 个并行分支 1 个失败(无 fail 边 → blocked)时,fan-in 节点保持 pending, + * run 收敛到 blocked(不是 completed) + * 5. 额度生效:`max_concurrency` 建 run 时持久化,`advanceBatch` 不传 opts 时读它 + * 7. 手动模式不变:`maxConcurrency<=1` 直接委托 `advanceOne`,一次只推进一个节点 + * + * 「fan-out 真并行」这条不需要 setImmediate/sleep 等计时手段就能确定性断言: + * `advanceBatch` 在 `Promise.allSettled(...)` 之前的整段(run 级检查 → checkBudget → + * SELECT 候选 → 逐行 claim)全程同步,`.map()` 调 `runClaimedNode` 时,每个节点的 + * ctx 构建/记账/事件都在各自 hit 到 `await agentExecutor(ctx)` 之前同步跑完—— + * 所以调用 `advanceBatch(...)`(还没 await 它的返回值)这一行语句执行完,三个节点 + * 已经全部处于 running、执行器也都已经被调用过一次。这正是 27-P4 §4.3 那条并发正确性 + * 论证(Node 单线程事件循环)的直接体现,写成测试顺带验证了那条论证站得住。 + * + * 隔离手法与 graph-engine.test.js 一致:独立 GRAPH_DB_PATH / GRAPH_RUNS_DIR, + * agent 调用全部走注入的假 executor,不起真实 Claude CLI。 + */ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +describe('advanceBatch(27-P4 T4/T6)', () => { + let wdb; + let engine; + let sse; + let runsDir; + const TEST_DB_PATH = '/tmp/test-graph-engine-advance-batch.sqlite'; + + beforeAll(() => { + for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) { + try { fs.unlinkSync(f); } catch {} + } + process.env.GRAPH_DB_PATH = TEST_DB_PATH; + runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'graph-advance-batch-runs-')); + process.env.GRAPH_RUNS_DIR = runsDir; + for (const m of [ + '../../services/graph-db', + '../../services/graph-service', + '../../services/org-graph-service', + '../../services/work-graph-db', + '../../services/graph-run-sse', + '../../services/graph-engine', + ]) { + delete require.cache[require.resolve(m)]; + } + wdb = require('../../services/work-graph-db'); + sse = require('../../services/graph-run-sse'); + engine = require('../../services/graph-engine'); + }); + + afterAll(() => { + for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) { + try { fs.unlinkSync(f); } catch {} + } + delete process.env.GRAPH_DB_PATH; + try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch {} + delete process.env.GRAPH_RUNS_DIR; + }); + + beforeEach(() => { + const db = wdb.getDb(); + db.pragma('foreign_keys = OFF'); + db.exec(` + DELETE FROM graph_run_loop_records; + DELETE FROM graph_run_nodes; + DELETE FROM graph_runs; + DELETE FROM work_graph_edges; + DELETE FROM work_graph_nodes; + DELETE FROM work_graphs; + DELETE FROM system_agents; + `); + db.pragma('foreign_keys = ON'); + engine.setAgentExecutor(null); + }); + + let seq = 0; + const nextCompany = () => `p4b-co-${++seq}`; + + function ensureAgents(ids) { + const db = wdb.getDb(); + const ts = new Date().toISOString(); + const ins = db.prepare(` + INSERT OR IGNORE INTO system_agents + (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at) + VALUES (?, ?, 'public', NULL, ?, 'active', ?, ?) + `); + for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, ts, ts); + } + + /** entry → a,b,c(fan-out)→ join(fan-in),全部用 condition=success 的默认边 */ + function buildDiamond({ maxConcurrency = null } = {}) { + const companyId = nextCompany(); + const nodes = ['entry', 'a', 'b', 'c', 'join']; + ensureAgents(nodes); + const g = wdb.createWorkGraph({ + companyId, + nodes, + edges: [ + { from: 'entry', to: 'a' }, + { from: 'entry', to: 'b' }, + { from: 'entry', to: 'c' }, + { from: 'a', to: 'join' }, + { from: 'b', to: 'join' }, + { from: 'c', to: 'join' }, + ], + }); + expect(g.errors || []).toEqual([]); + const run = wdb.createRun({ + graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual', maxConcurrency, + }); + engine.markRunReady(run.id); + return { runId: run.id, companyId }; + } + + const nodeByName = (runId, name) => + wdb.getRunState(runId).nodes.find((n) => n.name === name); + const statusMap = (runId) => + Object.fromEntries(wdb.getRunState(runId).nodes.map((n) => [n.name, n.status])); + + const deferred = () => { + let resolve; + const promise = new Promise((r) => { resolve = r; }); + return { promise, resolve }; + }; + + it('fan-out 真并行(验收 3):同一批里 a/b/c 同时处于 running,而不是依次', async () => { + const { runId } = buildDiamond(); + const gates = { a: deferred(), b: deferred(), c: deferred() }; + const calls = []; + engine.setAgentExecutor(async (ctx) => { + calls.push(ctx.nodeName); + if (gates[ctx.nodeName]) await gates[ctx.nodeName].promise; + return { status: 'success' }; + }); + + engine.startRun(runId); + await engine.advanceOne(runId); // entry → success,点亮 a/b/c + calls.length = 0; // 只看这一批 advanceBatch 触发的调用,entry 那次不算 + + // 刻意不 await:advanceBatch 在自己第一次真正挂起(Promise.allSettled)之前全程同步, + // 所以这一行语句本身执行完,三个节点的 claim + executor 调用就都已经发生了。 + const batchPromise = engine.advanceBatch(runId, { maxConcurrency: 3 }); + + expect(calls.sort()).toEqual(['a', 'b', 'c']); + expect(statusMap(runId)).toMatchObject({ a: 'running', b: 'running', c: 'running' }); + + gates.a.resolve(); + gates.b.resolve(); + gates.c.resolve(); + const res = await batchPromise; + + expect(res.ok).toBe(true); + expect(res.results).toHaveLength(3); + expect(statusMap(runId)).toMatchObject({ + a: 'success', b: 'success', c: 'success', join: 'ready', + }); + }); + + it('批量拾取跳过已被抢走的候选(验收 2):不重复认领、不报错', async () => { + const { runId } = buildDiamond(); + engine.setAgentExecutor(async () => ({ status: 'success' })); + engine.startRun(runId); + await engine.advanceOne(runId); // entry → success + + // 模拟「b 已经被别的调用抢先拾取」:直接做条件更新,绕开 advanceBatch + const b = nodeByName(runId, 'b'); + expect(sse.transitionNode(runId, b.id, 'ready', 'running', {}, { node_name: 'b' })).toBeTruthy(); + + const res = await engine.advanceBatch(runId, { maxConcurrency: 3 }); + // 只认领了 a、c 两个,b 被跳过——既没有重复认领,也没有报错 + expect(res.ok).toBe(true); + expect(res.results.map((r) => r.node.name).sort()).toEqual(['a', 'c']); + // b 停在「别人」占着的 running,advanceBatch 完全没碰它(没有被结算成任何终态) + expect(statusMap(runId).b).toBe('running'); + }); + + it('部分失败停等(验收 4):c 失败(无 fail 边 → blocked)时 join 保持 pending,run 收敛到 blocked', async () => { + const { runId } = buildDiamond(); + engine.setAgentExecutor(async (ctx) => ( + ctx.nodeName === 'c' ? { status: 'fail' } : { status: 'success' } + )); + engine.startRun(runId); + await engine.advanceOne(runId); // entry → success + + const res = await engine.advanceBatch(runId, { maxConcurrency: 3 }); + expect(res.ok).toBe(true); + expect(statusMap(runId)).toMatchObject({ + a: 'success', b: 'success', c: 'blocked', join: 'pending', + }); + expect(res.runStatus).toBe('blocked'); + expect(wdb.getRun(runId).run_status).toBe('blocked'); + }); + + it('maxConcurrency<=1 直接委托 advanceOne(验收 7):一次只推进一个节点', async () => { + const { runId } = buildDiamond({ maxConcurrency: 1 }); + const calls = []; + engine.setAgentExecutor(async (ctx) => { calls.push(ctx.nodeName); return { status: 'success' }; }); + engine.startRun(runId); + + const r1 = await engine.advanceBatch(runId); + expect(r1.results).toHaveLength(1); + expect(r1.results[0].node.name).toBe('entry'); + expect(calls).toEqual(['entry']); + + // 只推进了一个:a/b/c 还没被拾取(委托给 advanceOne 之后,一次只认领一个 ready 节点) + expect(statusMap(runId)).toMatchObject({ a: 'ready', b: 'ready', c: 'ready' }); + }); + + it('额度生效(验收 5):run.max_concurrency 建 run 时持久化,不传 opts 时读它', async () => { + const { runId } = buildDiamond({ maxConcurrency: 2 }); + expect(wdb.getRun(runId).max_concurrency).toBe(2); + + const gates = { a: deferred(), b: deferred(), c: deferred() }; + engine.setAgentExecutor(async (ctx) => { + if (gates[ctx.nodeName]) await gates[ctx.nodeName].promise; + return { status: 'success' }; + }); + engine.startRun(runId); + await engine.advanceOne(runId); // entry → success + + const batchPromise = engine.advanceBatch(runId); // 不传 maxConcurrency,读 run.max_concurrency=2 + const running = Object.entries(statusMap(runId)).filter(([, s]) => s === 'running'); + expect(running).toHaveLength(2); // 三个 ready 里只拾取了两个,第三个仍是 ready + + for (const g of Object.values(gates)) g.resolve(); + await batchPromise; + }); +}); diff --git a/server/tests/unit/graph-engine-fail-reason.test.js b/server/tests/unit/graph-engine-fail-reason.test.js index 5efac1d..85cc5de 100644 --- a/server/tests/unit/graph-engine-fail-reason.test.js +++ b/server/tests/unit/graph-engine-fail-reason.test.js @@ -123,9 +123,13 @@ describe('引擎自判失败的 fail_reason 与失败上下文(W12 补完 / W9 // W12 补完 · 三类引擎自判失败都在 fail_reason 上留名 // ──────────────────────────────────────────────────────────────── - it('取值域:三个引擎取值与 agent 的归因取值域**不重叠**(复用会劫持精确 fail 边的路由)', () => { + it('取值域:五个引擎取值与 agent 的归因取值域**不重叠**(复用会劫持精确 fail 边的路由)', () => { const engineReasons = Object.values(engine.ENGINE_FAIL_REASONS); - expect(engineReasons).toEqual(['node_error', 'boundary_escape', 'child_run_failed']); + expect(engineReasons).toEqual([ + 'node_error', 'boundary_escape', 'child_run_failed', + // T-mt25ecbd:produces/needs 契约闸门新增的两个取值(settleNode 的 success 判定里) + 'missing_required_input', 'produces_missing', + ]); // infra_flake 在 routeFailure 里有特判(原地重跑、不烧计数)。任何一个引擎取值撞上它 // 都会让确定性失败(如 Agent not found)原地重跑到预算耗尽。 expect(engineReasons).not.toContain(engine.INFRA_FLAKE_REASON); diff --git a/server/tests/unit/graph-files-index.test.js b/server/tests/unit/graph-files-index.test.js index fda0ff1..4a5b086 100644 --- a/server/tests/unit/graph-files-index.test.js +++ b/server/tests/unit/graph-files-index.test.js @@ -266,7 +266,9 @@ describe('R-19 产物索引:graph_files / graph_edge_contracts', () => { // ── ④ 缺输入告警(R-19.4 读回端)──────────────────────────── - it('④ 声明了 required 输入却查不到实例:提示词里出现 ⚠️,不静默省略', async () => { + it('④a 声明的 produces 一个候选文件都没落盘:节点本身按 fail 结算(produces_missing),不点亮下游', async () => { + // T-mt25ecbd 复现原样:brd/prd/sdd 三个 planner 节点声明了 produces 却真的什么都没写, + // executor 没抛异常就被引擎当 success——这道闸门堵在这里,不再等下游发现读不到东西。 const companyId = 'co-files-missing'; const taskId = 'task-missing'; const b = buildRun({ @@ -278,14 +280,49 @@ describe('R-19 产物索引:graph_files / graph_edge_contracts', () => { edgeLayer: 'work', edgeId: b.edgeId('sdd-writer', 'impl'), kind: 'sdd', required: true, }); - // 上游什么都不写 → 声明在、实例不在 + // 上游什么都不写 → 声明在、连候选文件都没有 const tasks = await advance(b.runId, 2); expect(allFiles(companyId)).toEqual([]); + const state = wdb.getRunState(b.runId); + const sddWriter = state.nodes.find((n) => n.name === 'sdd-writer'); + expect(sddWriter.status).toBe('blocked'); // fail → 没有 fail 边 → 终态 blocked(FSM) + expect(sddWriter.failReason).toBe('produces_missing'); + // 上游都没能成功,下游压根不会被点亮,更谈不上跑 + expect(tasks.has('impl')).toBe(false); + }); + + it('④b 声明了 required 输入却查不到实例:提示词里出现 ⚠️,agent 判自己成功也不被引擎采信(missing_required_input)', async () => { + const companyId = 'co-files-missing-needs'; + const taskId = 'task-missing-needs'; + const b = buildRun({ + companyId, taskId, + nodes: [{ nodeName: 'sdd-writer' }, { nodeName: 'impl' }], + edges: [{ from: 'sdd-writer', to: 'impl', condition: 'success' }], + }); + wdb.declareEdgeContract({ + edgeLayer: 'work', edgeId: b.edgeId('sdd-writer', 'impl'), kind: 'sdd', required: true, + }); + + // 上游写了两份跟声明的 sdd 对不上、彼此也无歧义可猜的文件(同 ⑦ 的「有歧义时不猜」口径, + // 两份都落成 kind=artifact)——产出闸门只看「有没有候选文件」,不判分类对不对, + // 所以 sdd-writer 正常 success,impl 照样被点亮。但 graph_files 里查不到 kind=sdd 的 + // 实例——这正是 R-19.4 要报的那种「声明了却查不到」。 + const tasks = await advance(b.runId, 2, { + writes: { 'sdd-writer': { '文档一.md': '跟 sdd 无关', '文档二.md': '跟 sdd 也无关' } }, + }); + const prompt = tasks.get('impl'); expect(prompt).toContain('[上游输入'); expect(prompt).toContain('- sdd:⚠️'); expect(prompt).toContain('索引里查不到实例'); + + // impl 自己的 executor 判了 success(advance 默认 status: () => 'success'), + // 但引擎不采信——它自己也声明了要吃 required 的 sdd,查不到实例就不算数。 + const state = wdb.getRunState(b.runId); + const impl = state.nodes.find((n) => n.name === 'impl'); + expect(impl.status).toBe('blocked'); + expect(impl.failReason).toBe('missing_required_input'); }); // ── ⑤ 节点 fail 不登记 ──────────────────────────────────────