/** * root run 收敛 → task / goal 状态回写 + OKR 写一次 + report_to 唤醒(27-P3C 缺陷 F1 / F2)。 * * ── 为什么要有这个模块(F1)── * 批 1 落地时,`updateTask` 的全部调用方只做三件事:把 task 置 running、写 archived_at、 * 用户 PATCH。**没有任何一条路径写 completed / failed / blocked 或 finished_at。** * 于是一条正常跑完的 run 留下的 task 永远停在 `running`: * · `DELETE /:cid/goals/:goalId` 永久 409「下还有正在执行的 task」, * 而它给的自救 hint(`POST /api/graph-runs/{id}/stop`)对一条 completed 的 run 会因 * `RUN_TRANSITIONS.completed = ['ready']` 抛 InvalidTransitionError → 400。自救路径本身是死路; * · 列表里所有历史任务永远显示「执行中」—— 正是 27-P3C 开篇要修的「历史任务看不见」的镜像版本。 * * ── 为什么挂在 `graph-run-sse.transitionRun` 上而不是枚举路径(§五.1.1)── * 设计文档已经论证过:`stopRun` 与「预算耗尽 → blocked」这两条最常见的释放路径**都不走 settleRun** * (`stopRun` 的注释自己写着「不经过 settleRun」)。枚举一次漏一次,而漏掉的症状是 * 「任务静默躺着、界面上跟正常执行一模一样」。`transitionRun` 是全仓库 run 状态转换的唯一漏斗。 * * ── 为什么 OKR 要挪到 goal 层(F2)── * §四.7 禁令一让新 run 不再带 `okr_node_id`(否则一个 goal 下 N 个 task 会反复覆写同一个 OKR 节点, * 一条 failed 的尾巴能把已经 done 的节点翻掉)。但批 1 只做了「不写」的那一半:`goals.okr_node_id` * 有写侧、有展示,**零读侧** —— 调用方传了、响应回显了(假确认)、run 跑完了,OKR 节点原地不动, * 日志一行都没有。这里补上另一半:goal 收敛时按 goal 的 okr_node_id 写**一次**。 * * 全模块的副作用**一律由 `work-graph-db` 的条件更新把闸门**,谁拿到 changes=1 谁才继续往下做: * · `settleTaskFromRun` —— run 终态 → task 回写; * · `settleGoalStatus` —— 并行收敛的 N 条 task 里选出唯一一个收敛 goal 的; * · `claimGoalSlot` —— 认领 verdict / reported_at 这两格(唯一 token,见下方常量处); * · `settleGoalVerdict` —— 验收结论回写,只有认领者写得进去,也只有它有资格汇报。 * 并发面不止「同一个进程里的多条 task」:pm2 cluster / 滚动重启下有多个实例在跑同一张表, * 所以判据一律放进 SQL 的 WHERE 里,**不做「先读一下再写」**(那在跨进程下不是原子的)。 */ /** * run 终态 → task 状态。 * `stopped → cancelled`:人工主动停止不是执行失败,把它记成 failed 会污染 goal 的验收判定 * (下面 settleGoal 只认「全部 completed 才 done」,两者都进 failed 分支,但 task 行上要分得清)。 * * `blocked → blocked` 保留,是因为界面必须看得见「卡住了,等人加预算」—— * 但它**不是终态**(见 `work-graph-db.TASK_TERMINAL_STATUS`):它只写 task 行,不触发 goal 收敛, * run 恢复后跑完还会被 completed 覆盖回去。 */ const RUN_TO_TASK_STATUS = { completed: 'completed', failed: 'failed', stopped: 'cancelled', blocked: 'blocked', }; /** * ── 两格占位 token:验收(verdict)与汇报(reported_at)── * * **验收那一格**(第三轮复核 缺陷 2):CAS 闸门必须把 goal 写进**终态**(占位值不进终态 = * 没有闸门,见 settleGoalIfAllTasksDone),可验收是分钟级的异步动作。于是有一段窗口: * `status='done'` 已经落库,判词还没有。这段窗口里进程一重启,goal 就永久停在 * 「done + verdict 为空 + validation_tokens=0 + 从没汇报过」,而 done ∈ GOAL_TERMINAL_STATUS、 * CAS 已闭,再也回不去。最要命的不是丢,是**查不出来**:`verdict IS NULL` 也正是 * 「只有 1 条 task、按规则跳过验收」那条合法路径的形态,两者在库里长得一模一样。 * * **汇报那一格**(第二轮复核):判词落库那一刻,这条 goal 就永远退出了验收重推的扫描面 * (扫描只认占位值),而紧接着的 `reportGoalOutcome` 是可能失败的(fetch 5xx / 进程被杀 / * OKR 目录被删)。于是「验收完成但汇报丢了」在库里**和一次完美收敛一模一样** —— * 判词漂亮、状态终态,而 OKR 永远没写、上级永远没醒。同一个病,换了一格。 * * 两格用同一套形状,让每种情况在库里各有各的形态(三态可分辨): * · 列为 NULL → 这一步压根没发生过(验收:本来就没验;汇报:还没报) * · 前缀匹配的占位值 → 这一步认领了但没确认完成(进行中,或中途丢了 → 重推补) * · 其它非空值 → 已定案(真判词 / 真汇报时间戳) * * ⚠️ 占位值**每次认领都必须是新的唯一 token**,这不是洁癖: * 认领靠 `UPDATE … WHERE 列 = 上一个值` 的 changes 决出唯一赢家,而 SQLite 的 changes 数的是 * 「匹配到的行」、不管值变没变。所以只要认领写进去的值**自己也在 WHERE 认的那一组里**, * 已被认领的行就照样匹配、changes 恒为 1 —— 闸门原地失效,两个实例各起一次 org-validator、 * 各写一次 OKR、各叫醒一次上级(全是真花钱且不可重放的)。上一版的 * `WHERE verdict IN (PENDING, RESUMING)` + `SET verdict = RESUMING` 恰好就是这个形态。 */ const VERDICT_PENDING_PREFIX = '⏳ goal 级验收中(结论未定)#'; const REPORT_PENDING_PREFIX = '⏳ goal 收敛汇报中#'; /** * 唯一 token 的两段:进程随机段 + 进程内自增段。 * 只用自增会在多实例间撞号(每个进程都从 1 开始),只用随机要靠概率;两段拼起来 * 「同一进程内绝不重复 + 跨进程几乎不可能重复」两件事各有一个确定的机制兜着。 */ const PROC_TAG = require('crypto').randomBytes(5).toString('hex'); let claimSeq = 0; const newClaim = (prefix) => `${prefix}${PROC_TAG}-${(claimSeq += 1)}@${Date.now()}`; /** * 认领租约(第三轮复核实跑复现的那条)。 * * `verdict` 那一格有两道闸:认领 + 回写决胜(`WHERE verdict = @claim`),所以跨进程抢跑 * 最多多烧一次 LLM、副作用一个都做不了。**`reported_at` 只有认领这一道**,而认领原来 * 接受「任意占位值」当旧值 —— 于是 B 实例启动时会把 A 实例**手上还活着**的汇报认领抢走, * 上级 agent 就同一条 goal 被叫醒两次(两轮真 LLM 会话),A 那句收尾 claim 静默失败、 * 它自己还以为报成了。`inFlight` 只挡本进程,跨进程挡不住。 * * 修法是给占位值加时间戳做租约:**活着的认领不许抢,过期的才准抢**。 * 这样「窟窿仍然补得上」与「在飞的不被抢」两件事同时成立 —— 只认 null 会让死进程留下的 * 占位永远补不回来,那是把一个可恢复的洞换成不可恢复的洞。 * * 解析不出时间戳(旧格式的占位值)一律按**已过期**处理:那是上一版代码留下的, * 它的主人早就不在了。 */ const CLAIM_LEASE_MS = Number(process.env.GOAL_CLAIM_LEASE_MS) || 10 * 60 * 1000; function claimAgeMs(value) { const at = Number(String(value).split('@').pop()); return Number.isFinite(at) ? Date.now() - at : Infinity; } const isStaleClaim = (value, prefix) => isPending(value, prefix) && claimAgeMs(value) > CLAIM_LEASE_MS; const isPending = (v, prefix) => typeof v === 'string' && v.startsWith(prefix); /** 判词是不是「还没有结论」的占位形态(给调用方 / 测试用,别在别处重复写 startsWith) */ const isPendingVerdict = (v) => isPending(v, VERDICT_PENDING_PREFIX); /** * 本进程**此刻正在飞**的认领 token。 * * 上一版的注释断言「没有任何 in-flight 验收能活过一次重启,所以此刻还挂着占位值的全是丢掉的」—— * 这句话**只对单进程成立**。pm2 cluster / 滚动重启下,B 实例启动时 A 实例还活着、验收还在飞, * 那些 goal 的 verdict 就是占位值,B 会把它们当「丢了」全部抢走。 * 跨进程没法凭库里的一个字符串判断对方死没死(那要租约 + 心跳,另排);但**本进程自己**认领的 * 哪些还在飞是确知的,所以至少把「自己抢自己」这一半彻底堵死。 * 跨进程那一半由后面的条件回写兜底:抢跑的那个回写时 changes=0,副作用一次都做不了。 */ const inFlight = new Set(); /** * 一次启动最多重推几条。org-validator 是真起 CLI、真花钱的, * 一次异常停机堆下 200 条待验收 goal 时,无上限的补跑会在开机瞬间烧掉一大笔。 * * ponytail: 上限是**每次启动**,超出的那些留着占位判词、下次启动接着补(占位值是幂等的, * 不会丢),但如果堆积速度超过每次启动补跑的速度就永远追不上。真到那天把重推从 * 「启动一次性」挪到心跳里按固定速率消费(`heartbeat-service` 已有这套节奏)。 */ const MAX_RESUME_PER_BOOT = 20; /** 判词写进 task 行时的长度上限 —— 那一行是任务列表里的一条小字,不是日志窗口 */ const BLOCK_NOTE_MAX = 120; /** * 「这条 run 卡在哪」→ 一句人话,写给任务列表看。 * * 取第一个 blocked/fail 的节点:root run 收敛成 blocked 时按定义只会有这么一个(其余是 * pending),多出来也只挑最靠前的那个 —— 这是**留痕**,不是完整归因,完整的在节点详情里。 * * 三段都可能缺,缺了就跳过,绝不因此返回空串:`fail_reason` 合法地为 null(走兜底边)是常态, * 而「有节点卡住了」这件事本身就值得写上去。 * * **不 throw**:调用方已经把 task 落库了,这里抛会连带吃掉后面的 goal 收敛。 */ function describeRunBlock(runId) { try { const wdb = require('./work-graph-db'); const state = wdb.getRunState(runId); if (!state) return null; const stuck = state.nodes.find((n) => n.status === 'blocked' || n.status === 'fail'); if (!stuck) return null; const who = stuck.name || stuck.defNodeId; const why = stuck.failReason ? `${stuck.failReason}` : '未给出 fail_reason(走兜底边)'; const note = (stuck.remediationNote || '').replace(/\s+/g, ' ').trim(); const tail = note ? `:${note.length > BLOCK_NOTE_MAX ? `${note.slice(0, BLOCK_NOTE_MAX)}…` : note}` : ''; return `节点 ${who} 卡住(${why})${tail}`; } catch (err) { // 吞掉异常但**不吞掉日志**:这条链路坏了的症状是任务列表少一行小字,不会报错、不会有人发现 console.error(`[task-settle] describeRunBlock 失败 run=${runId}: ${err.message}`); return null; } } /** * 一条 root run 收敛了:回写它的 task,必要时再收敛整个 goal。 * * **不 throw** 是调用方的硬要求(钩子挂在状态转换的尾巴上),但本函数只做同步 DB 写, * 异常统一由 `graph-run-sse` 那层的 try/catch 兜住 —— 在这里再吞一次会让真 bug 连日志都没有。 * * @param {object} run graph_runs 的整行(调用方已保证 parent_run_id 为空 = root run) * @param {string} to 转入的 run 状态 * @returns {{task: string, goal: string|null}|null} null = 这次没有产生任何回写(幂等 no-op) */ function settleRootRun(run, to) { const wdb = require('./work-graph-db'); const taskStatus = RUN_TO_TASK_STATUS[to]; // task_id 为空的 root run 是存量数据(goals/tasks 表出现之前建的),没有回写对象 if (!taskStatus || !run || !run.task_id) return null; if (!wdb.settleTaskFromRun(run.task_id, run.id, taskStatus)) return null; // blocked 是**唯一一个要人动手**的回写结果,所以它必须把「卡在哪、为什么」一起带上: // 任务列表那一行只有一个橙色的「挂起」,原因行读的是 tasks.blocked_reason,而这条路径 // 一直没写它 —— 于是从任务列表看不出任何线索,必须逐层点进 org 图 → work 图 → 节点详情 // 才能知道发生了什么。noteTaskBlockedReason 是留痕不是状态机(条件更新,不会盖掉别人的处置)。 if (taskStatus === 'blocked') { const why = describeRunBlock(run.id); if (why) wdb.noteTaskBlockedReason(run.task_id, run.id, why); } const task = wdb.getTask(run.task_id); return { task: taskStatus, goal: task && task.goal_id ? settleGoalIfAllTasksDone(task.goal_id) : null }; } /** * goal 级收敛:这个 goal 的 task 是不是都到终态了?是则定判词并做**一次**对外汇报。 * * 状态口径: * · 全部 completed → `done` * · 有任何一条不是 completed(failed / cancelled)→ `failed` * draft 的 task 算「还没到终态」,所以批 1 那种「一次只跑一条、其余留 draft」的 goal * 会一直等到最后一条也跑完才收敛 —— 这正是想要的:需求没做完就不该判它达成了。 * * **`blocked` 同样算「还没到终态」**(第二轮复核 缺陷 2,靠 `TASK_TERMINAL_STATUS` 不含它实现): * 挂起等人加预算 ≠ 需求失败。收敛一次是**不可撤销**的 —— 它写 OKR 节点、唤醒上级 agent, * 而 goal 收敛后进 GOAL_TERMINAL_STATUS,人加完预算把 run 跑完也再改不回 done。 * 所以有 blocked 在就一条都不写:goal 停在 `active`,等它恢复后跑完再一次性收敛。 * * ── goal 级验收挂在哪(§四.10)── * **一定在 CAS 闸门之后**。闸门之前跑验收,并行收敛的 N 个 task 会各调一次 org-validator * (真花钱、还可能给出 N 个不一致的判词),而其中 N-1 次的结论最后连库都写不进去。 * 闸门之后只剩一个赢家,验收和它带的两个不可重放副作用(写 OKR、唤醒上级)都归它做。 * * 三条出口,只有第三条调 agent: * · 有任何一条 task ≠ completed → 直接 `failed`,不调(显然没交付,省一次调用); * · 全 completed 且只有 1 条 task → 跳过,直接采用该 task 的结论(单 task 的 goal, * 两级验收问的是同一件事,再调一次纯浪费); * · 全 completed 且 ≥2 条 → 调一次 org-validator,判词落 `goals.verdict`。 * * @returns {string|null} 本次**初步**定下的 goal 状态;null = 还没到终态,或收敛权被别人抢走。 * ⚠️ 第三条出口返回的 `'done'` 是 CAS 占位值:验收是异步的(要起 agent,分钟级), * 判 fail 时会在几十秒后被改写成 `failed`。之所以先写终态而不是留个中间态:CAS 闸门 * 靠「status 已进终态」挡住重入,占位值不进终态就等于没有闸门。 * 同一次还会把 `verdict` 认领成一个唯一占位 token,好让这段窗口在库里认得出来 * (否则窗口内重启 = 静默丢一次验收,且与「跳过验收」同形,见常量处那段)。 */ function settleGoalIfAllTasksDone(goalId) { const wdb = require('./work-graph-db'); const tasks = wdb.listGoalTasks(goalId); if (!tasks.length) return null; if (tasks.some((t) => !wdb.TASK_TERMINAL_STATUS.includes(t.status))) return null; const allCompleted = tasks.every((t) => t.status === 'completed'); const status = allCompleted ? 'done' : 'failed'; // CAS:并行收敛的多个 task 会同时算出「全都到终态了」,只有抢到的这一个才做下面的副作用 if (!wdb.settleGoalStatus(goalId, status)) return null; let goal = wdb.getGoal(goalId); if (!goal) return status; if (allCompleted && tasks.length >= 2) { // ponytail: CAS 写终态与占位判词是两条 SQL,中间有一条指令宽的窗口 —— 恰好卡在这里被 // SIGKILL 的话,goal 会留下「done + verdict 为空」,和「1 条 task 跳过验收」再次同形, // 重推的**验收**分支扫不到它。上限就是这一条指令:两句之间没有任何 await / IO 让出, // 而且重推的**汇报**分支(reported_at IS NULL)还兜得住它,至少 OKR / 上级不会漏。 // 真要根治得把两次写包进一个 better-sqlite3 事务,那要动 settleGoalStatus 的签名,另排。 // // 认领用的是**唯一 token**、且 WHERE 精确认「我刚读到的那个旧值」(通常是 NULL; // goal 从 failed 回 active 重跑时会是上一轮的真判词)。主路径同样要过闸门 —— // 认领与回写在主路径和重推路径上必须是同一套东西,否则重推那边守得再严, // 主路径一条无闸门的回写就能把 OKR 写第二次。 const claim = newClaim(VERDICT_PENDING_PREFIX); let claimed = 0; try { claimed = wdb.claimGoalSlot(goalId, 'verdict', { from: goal.verdict === undefined ? null : goal.verdict, to: claim }); } catch (e) { console.error(`[TaskSettle] goal ${goalId} 占位判词写入失败: ${e.message}`); } if (!claimed) { // 判词在这一条指令的窗口里被别人改了。验收照跑(结论有用),但回写会被 WHERE 挡掉、 // 也就不会汇报 —— 那条 goal 的汇报由启动重推的 `reported_at IS NULL` 分支补。 console.error(`[TaskSettle] goal ${goalId} 占位判词认领失败(判词已被别处改写),本次验收将不回写、不汇报`); } goal = wdb.getGoal(goalId) || goal; inFlight.add(claim); // 出栈再跑:本函数在 `graph-run-sse.afterRunTransition`(「整段绝不 throw」的同步钩子) // 里被调用,而验收是分钟级的异步动作。在这里 await 就是把一次 run 状态转换挂在 LLM 上。 // 汇报(OKR / report_to)一并推迟到验收之后 —— §四.10 约束 5:对外说的必须是**最终**结论。 // // `.catch` 不是防御性冗余(第三轮复核 缺陷 3):runGoalValidation 是 async, // 它内部任何逃逸的异常在这里就是一个 **unhandledRejection**,Node ≥15 默认直接杀进程 —— // 一条 goal 的验收出错,赔上整个服务。对照 company.js 里那个 setImmediate 也是这么挂的。 const pending = goal; setImmediate(() => { runGoalValidation(pending, tasks, { claim }) .catch((e) => console.error(`[TaskSettle] goal ${goalId} 验收流程逃逸异常: ${e && e.stack ? e.stack : e}`)); }); return status; } // 跳过验收的两条出口:汇报本身现在是 async(要等 fetch 的响应才敢盖「已汇报」戳), // 但 OKR 那条分支在第一个 await 之前就把 updateNode 调掉了,所以对同步调用方 // (transitionRun 尾巴上的钩子)来说这里的可见行为没变。`.catch` 是必需的: // 这是个 async 函数,异常不会同步抛给 graph-run-sse 的 try/catch,只会变成 unhandledRejection。 reportGoalOutcome(goal, status, tasks) .catch((e) => console.error(`[TaskSettle] goal ${goalId} 收敛汇报失败: ${e && e.stack ? e.stack : e}`)); return status; } /** * 异步跑一次 goal 级验收,然后回写 + 汇报。**本函数绝不 throw**(没有调用方接得住它: * 它跑在 setImmediate 里,抛出去就是一个 unhandledRejection)。 * * 失败兜底(§四.10 约束 4):**验收出问题不能把 goal 卡住**。走到这里已经是「全部 task * 都 completed」这个既成事实,所以保守收敛成 `done`,但把「验收没执行 / 为什么」写进 verdict —— * 让界面上的这个 done 看得出它没验过。反过来把读不到判词当 fail 是不行的: * failed 是终态、会写 OKR、会唤醒上级,误杀一次真交付不可撤销。 * * @param {object} goal CAS 抢到收敛权时读到的 goal 行(status 已是占位的 'done') * @param {object[]} tasks 该 goal 的全部 task 行 * @param {{claim: string}} opts `claim` **两条路径都必传**:回写要求 `goals.verdict` 仍是本次 * 认领的那个唯一 token,否则整段放弃(别人已经定案了)。没有它就没有闸门, * 而闸门后面挂着两个不可重放的副作用(写 OKR、唤醒上级)。 */ async function runGoalValidation(goal, tasks, opts = {}) { const wdb = require('./work-graph-db'); const claim = opts.claim; if (!claim) { // 调用方少传 claim 时宁可什么都不做:无闸门地跑一遍验收,就是把「同一个 OKR 节点写两次」 // 这类事故的概率原样交还给运气。两个调用方都在本文件里,红了就是编码错误。 console.error(`[TaskSettle] goal ${goal.id} 的验收缺少认领 token,拒绝执行(没有闸门就不做副作用)`); return; } let status = 'done'; let verdict; let tokens = 0; try { try { const r = await require('./goal-validation').validateGoal({ companyId: goal.company_id, goal, tasks, }); // `pass === false` 才算不过:实现返回个残缺对象时不该被读成「判 fail」 status = r && r.pass === false ? 'failed' : 'done'; verdict = (r && r.verdict) || `org-validator 判 ${status === 'done' ? 'pass' : 'fail'}(未给判词)`; tokens = (r && r.tokens) || 0; } catch (e) { status = 'done'; verdict = `⚠️ goal 级验收未执行(${e.message});已按「${tasks.length} 条 task 全部 completed」保守判 done`; // 抛出前挂在 error 上的那部分 token 是真烧掉了的,一样要记账(否则拆分/验收失败的花费在账上蒸发) tokens = e.tokens || 0; console.error(`[TaskSettle] goal ${goal.id} 级验收失败,保守收敛为 done: ${e.message}`); } // 状态 + 判词 + 记账**一条 SQL 落库**,靠 `WHERE verdict = @claim` 决出赢家 // (原来是 getGoal 检查 + updateGoal 回写:读和写之间隔着一整轮事件循环, // 跨进程根本不是原子的 —— 两边都读到「还是我的占位值」,然后互相覆盖、各汇报一次)。 let won = 0; try { won = wdb.settleGoalVerdict(goal.id, { claim, status, verdict, addValidationTokens: tokens }); } catch (e) { console.error(`[TaskSettle] goal ${goal.id} 验收结果回写失败: ${e.message}`); } if (!won) { // 两种情况合并成一条出口:① 判词已被别处定案(人在界面上看到的就是那句,绝不能覆盖); // ② 回写本身炸了。**都不汇报** —— 上一版「回写炸了也要汇报」是错的:库里还挂着占位值, // 下次启动重推会再验一次再报一次,而汇报不可重放(OKR 写两次、上级叫醒两次)。 // 该被补的是那一次汇报,由重推按 reported_at 补,不是在这里赌一把。 console.log(`[TaskSettle] goal ${goal.id} 的验收结论未能落库(已被别处定案或回写失败),本次不汇报`); // 但**钱要记**:这一次 org-validator 是真跑了、token 是真烧了。不记的话这半边花费 // 永远看不见 —— 与本模块(以及 settleGoalPlanning 那条「没命中也补记账」的分支) // 反复申明的口径自相矛盾。判词可以输,账不能丢。 if (tokens > 0) { try { wdb.bumpValidationTokens(goal.id, tokens); } catch (e) { console.error(`[TaskSettle] goal ${goal.id} 补记验收 token 失败: ${e.message}`); } } return; } // 缺陷 3:本函数自称「绝不 throw」,可这一句原来在 try 之外裸着。reportGoalOutcome 有两条 // **同步**抛出面 —— `org-graph-service.terminalAgentId()` 读库、`OkrService.updateNode()` // 底下 `_saveObjective` / `_rebuildIndex` 的 fs 写(ENOENT / EACCES / 目录被删)。 // 批 1 走同步路径时这些异常由 `graph-run-sse.afterRunTransition` 的 try/catch 兜住,只丢一行日志; // 这条路跑在 setImmediate 的 async 函数里,同样的异常会变成 unhandledRejection 杀进程。 try { await reportGoalOutcome(wdb.getGoal(goal.id) || { ...goal, status, verdict }, status, tasks); } catch (e) { console.error(`[TaskSettle] goal ${goal.id} 收敛汇报失败: ${e && e.stack ? e.stack : e}`); } } finally { // 认领的 token 不再在飞:本进程的启动重推从此可以(也应该)把它当成「丢了」重新认领 inFlight.delete(claim); } } /** * goal 收敛的对外汇报 —— 粒度是 **goal**:「人提的这个需求到底算不算做到了」。 * 上级 agent 收到的 taskId 是 `G-xxx`。 * * 沿用它那个 either/or:有 `report_to` 就走 `/api/agent/task/report`(那条链路自己会更新 OKR 节点、 * 推进 Pipeline、唤醒上级),没有就直接 `OkrService.updateNode`。两条分支都会写到 OKR, * 所以这里绝不能写成「先写 OKR 再汇报」—— 那就是同一个节点被写两次。 * * **这是全系统唯一的对外汇报出口**(第二轮复核 缺陷 3)。company.js 里那条 run 级的 * `reportRunOutcome` 已经删了:批 1 的唯一形态是 1 goal : 1 task(`POST /:cid/tasks` 自动建 goal、 * `createTask` 继承 `goal.report_to`),两条汇报会先后打到同一个 `/api/agent/task/report`, * 而它第 5 步会 `setImmediate` 自调 `/api/agent/{to}/chat` —— 同一件事把上级叫醒两次、 * 白烧一轮 LLM,其中一次带的还是它从没见过的 `G-` id。留 goal 级的理由见 company.js 那段注释。 * * `from` = **交付者**,取 org graph 的末节点(`terminalAgentId`),不是 `created_by`: * `created_by` 是「谁下的单」,`POST /:cid/tasks` 里默认就是字符串 `'human'`, * 而下游拿它去 `readStateFile('human','todo.md')` / `findActiveRunStep('human')` 全部落空 —— * 不报错,只是汇报里的执行者信息整段是假的。图为空时退回 `report_to`(自己汇报给自己, * 至少是个真实存在的 agent),绝不退回 `created_by`。 * * @returns {Promise} true = 这次真报出去了(`reported_at` 已盖上真时间戳); * false = 没轮到我报(别处已认领 / 已报过),或报失败了(占位值留在库上,等启动重推补) */ async function reportGoalOutcome(goal, status, tasks) { const wdb = require('./work-graph-db'); // ── 汇报闸门(第二轮复核:「查不出来」从验收那一格搬到了汇报那一格)── // 先认领 reported_at,报完再把认领 token 换成真时间戳。于是库里三态可分: // NULL = 没报过 / 占位 = 报到一半丢了(重推补)/ 时间戳 = 报成了。 // 认领的旧值只认「NULL 或上一次留下的占位」:已经是时间戳就说明报过了,谁也不许再报一次。 const claim = newClaim(REPORT_PENDING_PREFIX); // ⚠️ 只抢**过期**的占位(租约见 CLAIM_LEASE_MS)。抢活着的那一份 = 跨进程重复汇报。 const prev = isStaleClaim(goal.reported_at, REPORT_PENDING_PREFIX) ? goal.reported_at : null; if (!wdb.claimGoalSlot(goal.id, 'reported_at', { from: prev, to: claim })) { console.log(`[TaskSettle] goal ${goal.id} 的收敛汇报已被别处认领或完成,本次跳过`); return false; } inFlight.add(claim); const bad = tasks.filter((t) => t.status !== 'completed'); const summary = ( `goal ${goal.id}「${goal.title}」收敛为 ${status}:共 ${tasks.length} 个 task,` + (bad.length ? `未完成 ${bad.map((t) => `${t.id}=${t.status}`).join('、')}` : '全部完成') // 全部 completed 却收敛成 failed,只能是 goal 级验收判的(§四.10)。不带判词的话, // 上级 agent 收到的是一句自相矛盾的话:「全部完成,结论 failed」。 + (goal.verdict ? `。验收判词:${goal.verdict}` : '') ).slice(0, 500); // okr_node_id 形如 O-001.1.1,第一段就是 objectiveId(同 OkrService.parseNodeId 口径) const okrNodeId = goal.okr_node_id || null; const objectiveId = okrNodeId ? String(okrNodeId).split('.')[0] : null; try { if (goal.report_to) { const from = require('./org-graph-service').terminalAgentId(goal.company_id) || goal.report_to; const PORT = require('../config').port || 3000; // 原来这里是**发射后不管**:`.then` 只对 !ok 打一行 error,函数立刻 return。 // 进程在这几十毫秒到几秒里被杀、或端点回 5xx,上级就永远不会被叫醒,而库里这条 goal // 长得和一次完美收敛一模一样。现在等到结果再决定盖不盖「已汇报」的戳。 const ok = await fetch(`http://127.0.0.1:${PORT}/api/agent/task/report`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ taskId: goal.id, from, to: goal.report_to, status, summary, objectiveId, nodeId: okrNodeId, }), }) // fetch 不因 4xx/5xx reject,不显式看 res.ok 会把地址写错这类问题完全静默吞掉 .then((r) => { if (!r.ok) console.error(`[TaskSettle] goal 汇报失败: HTTP ${r.status} goal=${goal.id}`); return !!r.ok; }) .catch((e) => { console.error(`[TaskSettle] goal 汇报失败(网络): ${e.message}`); return false; }); // 没报成就把占位值留在 reported_at 上:这正是重推要找的形状,下次启动接着补。 if (!ok) return false; } else if (objectiveId && okrNodeId) { const { OkrService } = require('./okr-service'); new OkrService(goal.company_id).updateNode(objectiveId, okrNodeId, { status, summary }); console.log(`[TaskSettle] 🎯 ${goal.company_id} OKR ${okrNodeId} → ${status}(goal ${goal.id})`); } // 既没有 report_to 也没有 OKR 节点时同样盖戳:「没有人要通知」也是一次做完了的汇报, // 不盖的话重推会永远把这条 goal 当成待补的窟窿,每次启动重扫一遍。 // // ponytail: 语义是 at-least-once —— 请求真到了对面、响应在回来的路上丢了的话, // 重推会再报一次。要做到 exactly-once 得让 `/api/agent/task/report` 认幂等键 // (goal id + 收敛时间),那是接收侧的改动,另排。 wdb.claimGoalSlot(goal.id, 'reported_at', { from: claim, to: new Date().toISOString() }); return true; } finally { inFlight.delete(claim); } } /** * 启动时补跑收敛链路上被打断的两步:**验收**和**汇报**(第三轮复核 缺陷 2 + 第二轮复核)。 * * 占位值只让事故**可查**,查得出来还得有人补 —— 否则库里多一列证据,goal 照样永远停在 * 「done 但没验 / 没汇报」。挂在 `server/index.js` 既有的「恢复中断任务」旁边(同一个延迟块, * 等 HTTP 就绪之后:`reportGoalOutcome` 的 report_to 分支会打 `127.0.0.1` 的自家接口)。 * * 两类窟窿、两种补法: * · **验收丢了** —— `verdict` 还是占位形态:重认领 → 重跑 org-validator → 回写 → 汇报; * · **只有汇报丢了** —— 判词已经是真结论(或本来就跳过验收),但 `reported_at` 是 NULL / * 占位形态:**只补汇报**,绝不重跑验收(那是白烧一轮 LLM,而且可能给出与库里不同的判词)。 * 这一类是判词落库到汇报之间那几十毫秒到几秒的窗口,以及 fetch 回 5xx 的那些。 * * ⚠️ 上一版这里写着「没有任何 in-flight 验收能活过一次重启,所以此刻挂着占位值的全是丢掉的」—— * **这句话只对单进程成立**。pm2 cluster / 滚动重启下 B 起来时 A 还活着、验收还在飞。 * 所以现在的幂等是三层,跨进程也站得住: * 1. 认领是条件 UPDATE,且认领值是**唯一 token**、WHERE 只认「我刚扫到的那个旧值」—— * 上一版把自己要写的 RESUMING 也放进了 WHERE,已被认领的行照样匹配,changes 恒为 1; * 2. 本进程还在飞的 token 直接跳过(inFlight),自己不会抢自己; * 3. 真正的兜底在回写那一句:`WHERE verdict = @claim`(`settleGoalVerdict`)。跨进程抢跑的 * 那一个即使多烧了一次 org-validator,回写 changes=0 → 一个副作用都做不了。 * * 逐条 await 是故意的:并发补跑 = 同时起 N 个 Claude CLI,开机瞬间把机器和额度一起打满。 * * @returns {Promise} 本次真正补跑了几条(验收 + 只补汇报,合计) */ async function resumePendingGoalValidations() { const wdb = require('./work-graph-db'); const db = wdb.getDb(); // 汇报那一类只扫 done / failed:cancelled / archived 是人手动收掉的,从来就没有要汇报的东西, // 把它们算进来等于开机给上级 agent 群发一堆「这条被取消了」。 const stuck = db .prepare(` SELECT id, verdict, reported_at FROM goals WHERE ( ( status IN (${wdb.GOAL_TERMINAL_STATUS.map((s) => `'${s}'`).join(', ')}) AND verdict LIKE @vp ) OR ( status IN ('done', 'failed') AND (reported_at IS NULL OR reported_at LIKE @rp) ) ) -- 归档的不补汇报:DELETE /goals 只盖 archived_at、**不改 status**,所以上面那句 -- 「只扫 done/failed 就等于排除了人手动收掉的」并不成立(实测:done + archived 的 -- goal 会被开机重推汇报一次)。注释描述的防线不存在,这里把它真的建起来。 AND archived_at IS NULL ORDER BY updated_at ASC LIMIT @limit `) .all({ vp: `${VERDICT_PENDING_PREFIX}%`, rp: `${REPORT_PENDING_PREFIX}%`, limit: MAX_RESUME_PER_BOOT }); if (!stuck.length) return 0; console.log(`[TaskSettle] 🔄 ${stuck.length} 条 goal 的收敛链路在窗口里被打断,补跑中…`); let done = 0; for (const row of stuck) { try { // 本进程正在飞的那些不是窟窿,是活的(见 inFlight 那段注释) if (inFlight.has(row.verdict) || inFlight.has(row.reported_at)) continue; if (isPendingVerdict(row.verdict)) { const claim = newClaim(VERDICT_PENDING_PREFIX); // WHERE 只认扫到的那个旧值;claim 全局唯一,永远不可能等于 row.verdict if (!wdb.claimGoalSlot(row.id, 'verdict', { from: row.verdict, to: claim })) continue; inFlight.add(claim); const goal = wdb.getGoal(row.id); if (!goal) { inFlight.delete(claim); continue; } await runGoalValidation(goal, wdb.listGoalTasks(row.id), { claim }); done += 1; continue; } // 只补汇报:闸门在 reportGoalOutcome 自己身上(认领 reported_at),这里不需要再判一次 const goal = wdb.getGoal(row.id); if (!goal) continue; if (await reportGoalOutcome(goal, goal.status, wdb.listGoalTasks(row.id))) done += 1; } catch (e) { // 一条补跑失败不该拖垮其余的;它的占位值还在,下次启动接着补 console.error(`[TaskSettle] goal ${row.id} 收敛重推失败: ${e.message}`); } } console.log(`[TaskSettle] ✅ 收敛重推完成 ${done}/${stuck.length}`); return done; } module.exports = { settleRootRun, settleGoalIfAllTasksDone, resumePendingGoalValidations, RUN_TO_TASK_STATUS, // 占位值不再是常量而是「前缀 + 唯一 token」,所以对外给的是前缀和判定函数: // 任何地方想认出「这条还没有结论」,都必须走 isPendingVerdict,别再去比字符串相等。 VERDICT_PENDING_PREFIX, REPORT_PENDING_PREFIX, isPendingVerdict, isStaleClaim, CLAIM_LEASE_MS, };