/** * Work Graph DB — P3 图执行引擎的数据层(T1 + T14) * * 设计见 docs/todo/system-design/27-p3-p4-work-graph-console.md * §四.A.1 分层(定义层 ≠ 运行时层) * §四.A.2 五张表(本文件照抄那段 SQL,含两个唯一索引与 graph_runs.backward_count) * §四.A.4 两张状态转换表 * §四.A.6 幂等(UPDATE ... WHERE status = ? 的条件更新,靠受影响行数判定) * §三.7 + §四.D.2.2 写入前五项校验 * * 分层要点: * 定义层 work_graphs / work_graph_nodes / work_graph_edges ← 本文件新建 * org_graph_nodes / org_graph_edges ← P2 已建,本文件只读 * 运行时层 graph_runs / graph_run_nodes / graph_run_loop_records ← 两层共用,引擎只认这个 * * DB 文件复用 P0/P2 的 graph.sqlite:直接用 graph-db.js 的 getDb(),同一个连接、同一个进程内单例。 * * 本文件只管「存」,不管「跑」。advanceOne / executor / 熔断计数逻辑在 graph-engine.js(T2/T7/T8)。 */ const crypto = require('crypto'); const { getDb: getGraphDb } = require('./graph-db'); // §三.7 明确要求 work graph 的写入校验**复用** P2 这一套,不另写一份 // R-19:normalizeContracts 也从那边拿 —— 边上的 produces / needs 两层通用, // 两处各写一份归一逻辑就是「差一个字母静默失效」的温床 const { validateGraph, normalizeContracts } = require('./org-graph-service'); // ──────────────────────────────────────────────────────────────── // 活跃 root run 取值域(27-P3C §五.1) // ──────────────────────────────────────────────────────────────── /** * 「这条 root run 正占着公司的执行坑」的取值域。 * * 全仓库唯一的定义处,调度器 / 路由 / 准入事务复用它而不是再抄一份字面量。 * 为什么强调这件事:`canDispatch` 那次事故就是前端把这三个值复述了一遍、还多写了一个 * `blocked`,于是公司失败一次之后再也下不了任务 —— 取值域一旦有第二个持有者就会分叉。 * * ⚠️ **`queued` 故意不在里面**:排队中的 run 不占坑,一家公司可以同时躺着任意多条。 * 这是「入队不占坑、出队才占坑」的全部依据 —— 出队那一跳 `queued → ready` 是「从活跃集外 * 进入活跃集」,会被 `updateRunStatus` 的准入事务数一次坑。 * 改这个数组之前先读 27-P3C §四.3 与 §五.2。 * * 批 4 之前这个数组还兼任 `idx_graph_runs_one_active_root` 的 WHERE 子句(拼进建表 SQL)。 * 那条索引已 DROP(见 getDb 的迁移块),现在它只有一个用途:准入事务的判据。 */ const ACTIVE_ROOT_RUN_STATUS = ['ready', 'running', 'paused']; /** * 一家公司同时能占几个执行坑 —— 缺省值。 * **缺省必须是 1**:读不到配置的公司(没建目录的单测、被 delete() 改名的公司、`_unknown`) * 一律回落到批 3 的行为。回落成 0 是「永远不出队 = 静默全停」,回落成 Infinity 是超发。 */ 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)。 * 超过就按上限截断并打日志,不静默当成无限。 */ const MAX_ACTIVE_ROOT_RUNS_CAP = 20; /** 同一份非法配置不必每次准入都刷一行日志 */ const warnedMaxParallel = new Set(); function warnMaxParallelOnce(companyId, raw, why) { const key = `${companyId}:${raw}`; if (warnedMaxParallel.has(key)) return; warnedMaxParallel.add(key); console.warn(`[work-graph-db] company ${companyId} 的 max_parallel_tasks=${JSON.stringify(raw)} ${why}`); } /** * 这家公司同时能占几个执行坑(27-P3C 批 4 / 27-P4 §一.3)。 * * **调用时读,不在模块加载时快照**。`company-paths.js` 顶上那条教训就是这个:镜像时代两个 * 模块各自快照 GRAPH_RUNS_DIR,加载顺序不同就算出不同的根。这里同理 —— 公司配置随时可改, * 快照一次就意味着「改了配置要重启才生效」,而且不报错。 * * 读的是 `companyRootDir(companyId)/company.json` 的 `max_parallel_tasks`(snake_case, * 与 27-P4 §一.3 的字段名逐字一致;**不认 camelCase 的 maxParallelTasks** —— 两个键并存 * 就是又一个双持有者)。走 `company-paths.companyRootDir` 而不是 `company-service.get`: * · 前者认 GRAPH_RUNS_DIR,后者不认。准入判据一旦去读 company.json,设了 GRAPH_RUNS_DIR * 的单测就会去读**真仓库**的 data/companies/,撞上同名公司就读到真配置(§六 缺陷 A 同型); * · 顺带避开 `company-service → graph-service → graph-db` 那条 require 环。 * 两者在生产下算出的是同一个路径(都是 REPO_ROOT/data/companies/{cid})。 * * 非法值(0 / 负数 / 小数 / 非数字 / 文件坏了)一律回落 DEFAULT 并打一次日志: * 「配置写错了就当没配」必须是能看见的,否则就是一次静默降级。 * * ponytail: 每次准入读一次盘(一个几百字节的 JSON)。准入是低频事件(一次出队/恢复一次), * 没加缓存。真到高频时加一层 mtime 缓存,别改成模块级快照。 */ function maxActiveRootRuns(companyId) { if (!companyId) return DEFAULT_MAX_ACTIVE_ROOT_RUNS; let raw; try { const { companyRootDir } = require('./company-paths'); const fp = require('path').join(companyRootDir(companyId), 'company.json'); const fs = require('fs'); if (!fs.existsSync(fp)) return DEFAULT_MAX_ACTIVE_ROOT_RUNS; raw = JSON.parse(fs.readFileSync(fp, 'utf-8')).max_parallel_tasks; } catch (err) { // company.json 坏了不该让一次出队变成 500:回落缺省,但必须留声 warnMaxParallelOnce(companyId, null, `读取失败(${err.message}),按 ${DEFAULT_MAX_ACTIVE_ROOT_RUNS} 处理`); return DEFAULT_MAX_ACTIVE_ROOT_RUNS; } if (raw === undefined || raw === null) return DEFAULT_MAX_ACTIVE_ROOT_RUNS; if (!Number.isInteger(raw) || raw < 1) { warnMaxParallelOnce(companyId, raw, `不是 ≥1 的整数,按 ${DEFAULT_MAX_ACTIVE_ROOT_RUNS} 处理`); return DEFAULT_MAX_ACTIVE_ROOT_RUNS; } if (raw > MAX_ACTIVE_ROOT_RUNS_CAP) { warnMaxParallelOnce(companyId, raw, `超过硬上限,按 ${MAX_ACTIVE_ROOT_RUNS_CAP} 处理`); return MAX_ACTIVE_ROOT_RUNS_CAP; } return raw; } /** * 坑满:一条 root run 想从活跃集外进入活跃集,但公司的执行坑已经占满。 * * 单独一个类型,不复用别的: * · 不是 `InvalidTransitionError` —— 转换本身完全合法(queued→ready 就是出队), * 错的是「这一刻没位置」,换个时间同一条转换会成功; * · 不是那条 UNIQUE 的 SqliteError —— 那是**兜底**那一层的产物,语义正确但报文是 * 一句 SQLite 原文,调用方要靠正则去猜。 * 调度器用它判「这轮没抢到」(正常结局),路由层用它出 409。 * * ⚠️ 这里以前挂着一条「message 里的 `constraint` 一词不许清掉,因为 `routes/company.js` * 有两处(dispatch / POST tasks)用 `/UNIQUE|constraint/i` 嗅探错误文本来翻 409」的告示。 * **批 3 把那两处 catch 整段删了**(下达不再因坑满失败,改成入队),告示描述的调用方 * 已经不存在,却一直留着 —— 一条注释在替一段不存在的代码索要保证,是本文档反复点名的 * 那个形状(§十四.3 第 3 条:注释是对的、代码没兑现;这次反过来)。 * 现在文本不再是任何人的判据:唯一的判据是 `isSlotRaceError`(下面那个函数),按类型判。 * message 里那句 `active-root-run constraint` 只当人读的说明留着,改它不会弄坏任何分支。 * * ⚠️ **待拆(批 4 留下的)**:批 4 已 DROP 掉同名的唯一索引,`constraint` 这个词会让读者 * 以为 DB 层还有一道约束 —— 该换成 `limit`。本批**故意不动这个词**:路由层正由另一个 agent * 并行改造,万一那边还留着按文本嗅探的分支,改早了会把一次 409 退化成 500。等路由那批落地、 * 确认全仓只剩 `instanceof` / `isSlotRaceError` 之后换掉它并删本段。 */ class RunSlotBusyError extends Error { /** * @param {string} companyId * @param {string} runId 想进坑而没进去的那条 run * @param {string[]} heldBy 此刻占着坑的 root run id —— **必须带上**:路由层要把它翻成 * 一句人能看懂的 409(「被 run X 占着」),少了它界面上只剩「坑满」,人无从判断该等谁。 * @param {number} max */ constructor(companyId, runId, heldBy, max) { const held = heldBy.length ? heldBy.join(', ') : '(未知)'; super(`company ${companyId} 的执行坑已满(${heldBy.length}/${max},占用者 ${held}):` + `run ${runId} 无法进入活跃集(active-root-run constraint)`); this.name = 'RunSlotBusyError'; this.companyId = companyId; this.runId = runId; this.heldBy = heldBy; this.active = heldBy.length; this.max = max; } } /** * 「这次没抢到坑」—— 竞争的正常结局,不是故障。**全仓库唯一的判据持有者。** * * 为什么必须只有一个:批 3 落地时它有两份,而且两份不一样 —— * `task-scheduler.isRaceError` 按 `/UNIQUE constraint failed/i` 嗅报错文本, * `routes/company.js#isSlotRace` 按 `err.code`,后者的注释还写着「口径同 task-scheduler」。 * 两个持有者迟早分叉,这一份从落地那天起就已经分叉了:改一句报错文案,调度器那边会把 * 一次正常竞争当成故障去 `abandonRun`(run 收成 stopped、task 收成 cancelled = 丢任务), * 而路由那边照常 409。这正是 `canDispatch` 那次事故的形状,只是换了个取值域。 * * **现在只有一种形态**:`RunSlotBusyError` —— 准入事务在 `BEGIN IMMEDIATE` 里数出坑已满。 * * ⚠️ 批 3 那版还把 `InvalidTransitionError` 算作竞态,注释写着「这条 run 已经被别人从 queued * 推走了」。**探针实测证伪了这句话**:被别人推走时 `updateRunStatus` 走的是条件更新 * `WHERE id=? AND run_status=@from`,**返回 0、不抛异常**。真正抛 `InvalidTransitionError` 的 * 只有一种情况 —— **调用方声明了一个非法转换**(`expected = from` 是调用方传进来的声明值, * `canTransitionRun` judge 的是 `expected → to`,不是 `cur.run_status → to`)。 * 也就是说那一支把「调用方写错了转换」当成「这轮没抢到」吞掉,还不留日志: * 调度器会 `continue`,任务永远排队,界面与正常排队一模一样 —— 又一条「写不进去但不报错」。 * 已删除该分支,理由与下面删 UNIQUE 那支完全相同:让它吵起来。 * * ⚠️ 批 4 **删掉了 `err.code === 'SQLITE_CONSTRAINT_UNIQUE'` 那一支**,这是一次故意的行为改变。 * 它当初的语义是「兜底那道唯一索引拒了」。索引 DROP 之后,这一跳再吐出 UNIQUE 冲突只剩一个 * 可能:**索引被建回来了**(旧版本进程启动时又跑了一遍它的 `CREATE UNIQUE INDEX IF NOT EXISTS`, * 见 getDb 迁移块那段告示)。那种情况下 max>1 会被悄悄压回 1 —— 留着这一支的话,调度器把它 * 当「这轮没抢到」`continue` 掉,任务永远排队、日志一行没有、界面和正常排队一模一样, * 正是要消灭的「写不进去但不报错」。删掉之后它会一路抛到 `abandonRun` / 500,吵得人能看见。 * * 判类型而不是判 message:改一句报错文案不该改变任何分支(这是批 3 那次分叉的根因)。 */ function isSlotRaceError(err) { if (!err) return false; return err instanceof RunSlotBusyError; } // ──────────────────────────────────────────────────────────────── // ID 生成 // ──────────────────────────────────────────────────────────────── /** 前缀 + 12 位 hex。不引入 nanoid 依赖,crypto 够用(碰撞概率 2^-48 量级)。 */ function newId(prefix) { return `${prefix}_${crypto.randomBytes(6).toString('hex')}`; } /** * goal / task 的 id:`G-{base36 毫秒}-{4 位 hex}`(27-P3C §四.5)。 * * 为什么不复用上面的 newId:`T-` 这个前缀与短横线分隔在 company.js / agent-chat.js 已经用开了, * 换成 `T_xxx` 会让两套 id 看上去不是同一类东西。 * * 为什么不能沿用现有的 `T-${Date.now().toString(36)}`:那个格式**同一毫秒内生成的两个 id * 完全相同**。今天没炸只是因为一次只建一条;一个 goal 一次性拆出 N 条 task 是**循环建行**, * 同毫秒撞是必然事件 —— 而 tasks.id 是主键,撞了就是 500。 * 随机后缀是仓库既有约定(_spec/CONVENTIONS.md §2:时间戳部分不保证全局唯一,靠后缀随机位避碰)。 */ function genId(prefix) { return `${prefix}-${Date.now().toString(36)}-${crypto.randomBytes(2).toString('hex')}`; } const genGoalId = () => genId('G'); const genTaskId = () => genId('T'); /** * 用新生成的 id 插一行,主键撞了就换一个再来(最多 5 次)。 * * 4 位 hex 只有 65536 种:同毫秒批量建行时,生日碰撞是**百分之几**量级,不是「绝不会」。 * id 格式按 §四.5 定死不动(它已经写进文档、也已经是对外可见的形状), * 所以补的是重试而不是加长随机位 —— 撞一次的代价是再跑一次 INSERT, * 而不补的代价是「一个 goal 拆 20 条 task,偶尔有一次整体 500」这种最难复现的故障。 * * 只吞这张表主键的 UNIQUE 冲突,别的约束照常抛出去。 */ function insertWithFreshId(table, gen, insert) { for (let attempt = 0; ; attempt += 1) { const id = gen(); try { insert(id); return id; } catch (err) { if (attempt >= 4 || !new RegExp(`UNIQUE constraint failed: ${table}\\.id`).test(err.message)) throw err; } } } // ──────────────────────────────────────────────────────────────── // 建表(幂等,可反复调用) // ──────────────────────────────────────────────────────────────── let schemaEnsured = false; /** 表是否已有某列(org_graph_edges.fail_reason 由 T-P3 加,本文件读时要能容忍它还没加) */ function hasColumn(db, table, column) { return db.pragma(`table_info(${table})`).some((c) => c.name === column); } function getDb() { const db = getGraphDb(); // 复用 graph.sqlite 的同一连接(WAL + foreign_keys=ON 已由它设好) if (schemaEnsured) return db; db.exec(` -- ── 定义层:work graph ────────────────────────────────────── CREATE TABLE IF NOT EXISTS work_graphs ( id TEXT PRIMARY KEY, company_id TEXT NOT NULL, owner_org_node_id INTEGER REFERENCES org_graph_nodes(id), planned_by TEXT NOT NULL DEFAULT 'workgraph-planning', goal_id TEXT, task_id TEXT, okr_node_id TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS work_graph_nodes ( id TEXT PRIMARY KEY, work_graph_id TEXT NOT NULL REFERENCES work_graphs(id), node_name TEXT NOT NULL, system_agent_id TEXT NOT NULL REFERENCES system_agents(id), node_kind TEXT NOT NULL DEFAULT 'agent_call' CHECK (node_kind IN ('agent_call','subgraph')), created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS work_graph_edges ( id TEXT PRIMARY KEY, work_graph_id TEXT NOT NULL REFERENCES work_graphs(id), from_node_id TEXT NOT NULL REFERENCES work_graph_nodes(id), to_node_id TEXT NOT NULL REFERENCES work_graph_nodes(id), condition TEXT NOT NULL CHECK (condition IN ('success','fail')), fail_reason TEXT, edge_kind TEXT NOT NULL DEFAULT 'forward' CHECK (edge_kind IN ('forward','backward')), created_at TEXT NOT NULL ); -- ── 运行时层:org / work 两层共用 ─────────────────────────── CREATE TABLE IF NOT EXISTS graph_runs ( id TEXT PRIMARY KEY, graph_kind TEXT NOT NULL CHECK (graph_kind IN ('org','work')), graph_id TEXT NOT NULL, company_id TEXT NOT NULL, root_run_id TEXT NOT NULL, parent_run_id TEXT REFERENCES graph_runs(id), parent_node_run_id TEXT REFERENCES graph_run_nodes(id), goal_id TEXT, task_id TEXT, okr_node_id TEXT, execution_mode TEXT NOT NULL DEFAULT 'manual' CHECK (execution_mode IN ('manual','auto')), run_status TEXT NOT NULL DEFAULT 'planning', pause_requested INTEGER NOT NULL DEFAULT 0, 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, -- 27-P4 T5:root run 挂起(run_status='blocked')的机器可读原因,只有 -- suspendRunTree 写它('budget-exhausted' | 'billing-arrears'),startRunSync -- 「启动」时无条件清空。只在 root run 上有意义——suspendRunTree 只把树的 root -- 转成 blocked,子 run 只收到 pause_requested,不动它们的 block_reason(它们本来就没有)。 -- 存在的意义是让「欠费清零后自动恢复」这类扫描能按原因过滤(run_status/block_reason -- 复合索引),不用去解析 timeline 里的 event 文本。 block_reason TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); -- 批 4 之前这里建过 idx_graph_runs_one_active_root(UNIQUE(company_id) WHERE 活跃且是 root), -- 即「一家公司同时只有一条活跃 root run」那条 v1 硬约束。它与 max_parallel_tasks > 1 直接冲突, -- 已删除,并在下面的迁移块里显式 DROP。**别把它加回来**:加回来 = 把并行悄悄压回 1 -- (症状见 isSlotRaceError 那段注释)。守卫现在只有一处,在 updateRunStatus 的准入事务里。 CREATE TABLE IF NOT EXISTS graph_run_nodes ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES graph_runs(id), def_node_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', attempt INTEGER NOT NULL DEFAULT 0, agent_run_id TEXT, child_run_id TEXT REFERENCES graph_runs(id), output_ref TEXT, fail_reason TEXT, -- R-17.1 跨层节点地址:归因**指向**的那个 work graph node,形如 -- {"orgNode":"org-coder","workNode":"impl-a"} -- 语义是「这一行的失败,根因在 orgNode 那张 work graph 的 workNode 上」。 -- 与 fail_reason 同源同轨:同一份 attempt-{n}.json 产出、同一次条件更新落库、 -- 同一条 deriveChildOutcome 透传上去。**不复用 fail_reason 承载**(它是选边的匹配键, -- 掺进地址会让所有精确 fail 边失配),也**不放在边上**(边是定义层, -- 而 work graph 是运行时才由 workgraph-planning 规划出来的,定义层根本写不出这个 id)。 fail_node_ref TEXT, -- child_run_id 那张子 run 是为**本节点的哪一次 attempt** 起/复用的(R-17.1 幂等判据)。 -- 判据从「子 run 是不是还活着」改成「它服务的是不是本轮」:R-17.1 要求复用旧子 run, -- 只看终态就会让「上一轮留下的终态」和「这一轮刚重置过」长得一样——那正是第二轮 -- 那个永久死锁的成因。存代次是唯一能把两者分开的信息。 child_run_attempt INTEGER, -- 「留给这个节点下一轮的修复建议」。载体必须是节点行而不是 graph_run_loop_records: -- 后者只在命中 backward 边时才有记录,走 forward fail 边 / 死路 blocked / infra_flake -- 原地重跑这三条路径上归因节点写下的 remediation_note 就地蒸发(R4-F1)。 -- 且 loop record 是熔断账本(loop_count 按 (run_id, edge_id) 记录数算), -- 借它存判词会直接污染熔断计数。 remediation_note TEXT, started_at TEXT, completed_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_run_nodes_unique ON graph_run_nodes(run_id, def_node_id); CREATE TABLE IF NOT EXISTS graph_run_loop_records ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES graph_runs(id), edge_id TEXT NOT NULL, from_node_run_id TEXT NOT NULL REFERENCES graph_run_nodes(id), to_node_run_id TEXT NOT NULL REFERENCES graph_run_nodes(id), loop_count INTEGER NOT NULL, verdict TEXT NOT NULL CHECK (verdict IN ('registered','fixed','not_fixed','force_terminated')), remediation_note TEXT, -- R-17.1:这次回退**实际落到**的跨层地址(解析后的结果,含 childRunId / nodeRunId)。 -- 与 graph_run_nodes.fail_node_ref 的分工:那边是归因的**主张**,这边是引擎的**动作**。 -- 主张可能解析不了(图换过、节点名不存在、与选中的边不是同一个 org node), -- 那时这一列为空 = 「退化成整张 work graph 重跑」,控制台据此显示回退粒度。 target_node_ref TEXT, created_at TEXT NOT NULL ); -- ── R-19:产物索引(27-P4 §4.6)──────────────────────────────────── -- 补的是「同一张 work graph 内部上游到下游没有数据通路」这个断点:边一直只承载 -- 控制流(condition / edge_kind / fail_reason),而「边的输入输出」此前只是 -- 从 output_ref **推断**出来的 UI 展示、没有实体,推断的那一端是节点私有目录, -- 下游读不到。第四次真跑就死在这:prd 节点的归因原话是 -- 「不是上游没产出,是它的产出结构上到不了我手上」。 -- 产物**实例**:运行时由**引擎**追加。不靠 agent 自觉写 —— 否则索引与实际文件必然漂移 -- (本轮反复验证:凡「靠 agent 自觉」的环节几乎全部落空过)。 -- rel_path 是**相对公司根**的路径,根由代码推导不入库:存绝对路径会烂, -- W13 清掉的那 10 条 legacy_path 悬空行就是这个病。 CREATE TABLE IF NOT EXISTS graph_files ( id TEXT PRIMARY KEY, company_id TEXT NOT NULL, task_id TEXT, rel_path TEXT NOT NULL, kind TEXT NOT NULL, produced_by_run_id TEXT, produced_by_node_run_id TEXT, created_at TEXT NOT NULL, superseded_by TEXT ); -- **部分**唯一:只约束「当前版本」(superseded_by IS NULL),历史行可以同路径共存。 -- 与 idx_org_graph_nodes_company_role_active 同一个模式 —— 想要「同一键只有一个活的」 -- 又要留历史,就得把条件写进索引,而不是靠删行。 CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_files_current ON graph_files(company_id, COALESCE(task_id, ''), rel_path) WHERE superseded_by IS NULL; CREATE INDEX IF NOT EXISTS idx_graph_files_lookup ON graph_files(company_id, COALESCE(task_id, ''), kind); -- 边上声明的数据**契约**:规划时写,只有 kind 没有实例。 -- edge_layer 区分两层边表 —— org_graph_edges.id 是 INTEGER、work_graph_edges.id 是 TEXT, -- 与 graph_run_nodes.def_node_id 同一处类型不整齐(见 27-P3/P4 §四.A.2 的注), -- 所以这里存字符串形式、不建外键,引用完整性靠服务层。 CREATE TABLE IF NOT EXISTS graph_edge_contracts ( id TEXT PRIMARY KEY, edge_layer TEXT NOT NULL CHECK (edge_layer IN ('org','work')), edge_id TEXT NOT NULL, kind TEXT NOT NULL, required INTEGER NOT NULL DEFAULT 1, -- produces = 产出方在自己的出边上声明「我产这个 kind 交给下游」 -- needs = 消费方在自己的入边上声明「我要吃这个 kind」,产出方不必是紧邻上游 -- 这个非对称必须落库:规则 6(data-contract)靠「有人 needs 但没人 produces」判错, -- 只存 kind 的话读回来就分不清谁该产、谁只是要吃,校验在库这一侧永远算不出来。 role TEXT NOT NULL DEFAULT 'produces' CHECK (role IN ('produces','needs')), created_at TEXT NOT NULL ); CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_edge_contracts_dedupe ON graph_edge_contracts(edge_layer, edge_id, kind); -- ── 27-P3C:goal / task 两级模型(27-p3c-goal-task-model.md §四.1 / §四.2)── -- 这不是新概念,是给一对**已经存在的悬空外键**补上它们的表:graph_runs.goal_id / -- task_id 从 P3 起就一直在写,却没有任何表接得住,于是「一个需求拆成几件活」 -- 在库里根本不存在,界面上也就画不出来。 -- 三级:goal(人提的一个需求,description 即验收边界)→ task(一件可独立跑的活) -- → run(task 的一次执行,已有)。 -- **没有 budget_json**:预算闸门是按 root run 算的(DEFAULT_TASK_BUDGET 只挂 root run, -- 子 run 一律上溯 root_run_id 扣同一个池),goal 级预算没有任何执行机制 —— -- 加一个没人消费的额度列,就是本仓库反复被咬的「实现在、零调用方」。 -- 但 task-planning 的拆分调用跑在任何 run 存在之前,那是一笔真花钱、又没有池子可扣的 -- 开销(goal 层「隐形的一半」),所以它只**记账**(planning_tokens / validation_tokens), -- 不设闸门;总额控制另排期。 CREATE TABLE IF NOT EXISTS goals ( id TEXT PRIMARY KEY, -- G-{base36}-{4hex},见 §四.5 company_id TEXT NOT NULL, title TEXT NOT NULL, -- 短标题,列表显示用;可改 description TEXT NOT NULL, -- 人类需求原文 = 验收边界;下达后不可改 status TEXT NOT NULL DEFAULT 'draft', -- draft | planning | active | done | failed | cancelled | archived -- planning = task-planning 正在拆分,是个真会持续几十秒的状态 -- failed = goal 级验收没过,或有 task 未能 completed;可回 active 补 task 重下 priority TEXT NOT NULL DEFAULT 'P1', okr_node_id TEXT, -- 可选,挂到 OKR 度量树的哪个节点(O-001.1.1) created_by TEXT, -- 'human' | agent id(Master/CEO 代下单) report_to TEXT, -- 收敛后要唤醒的上级 agent,见 §四.7 禁令二 planning_tokens INTEGER NOT NULL DEFAULT 0, -- task-planning 拆分烧掉的 token validation_tokens INTEGER NOT NULL DEFAULT 0, -- goal 级 org-validator 烧掉的 token verdict TEXT, -- goal 级验收判词(org-validator 的原话) planning_error TEXT, -- 非空 = task-planning 拆分失败、已降级成单 task(§四.9) -- 「这个 goal 的收敛已经对外汇报过了」这件事的**可查证据**(第二轮复核:汇报那一格)。 -- 判词落库和汇报(写 OKR / 唤醒上级)是两件事,中间那几十毫秒到几秒里进程被杀、 -- 或端点回 5xx,库里这条 goal 会长得**和一次完美收敛一模一样** —— 判词漂亮、状态终态, -- 而 OKR 永远没写、上级永远没醒。没有这一列就没有任何查询能把两者分开,也就补不了。 -- 三态(与 verdict 同一套形状): -- NULL → 从没汇报过 -- '⏳ …#{token}' → 认领了但没确认成功(正在报 / 报到一半丢了),启动重推会补 -- ISO 时间戳 → 真报成功了 reported_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, dispatched_at TEXT, -- 首次下达时间;非空 = 描述已冻结(§四.4) archived_at TEXT ); CREATE INDEX IF NOT EXISTS idx_goals_company ON goals(company_id, status); -- goal_id 是**真外键**:tasks 是新表,可以直接写上(graph_runs 那两列只能是逻辑外键 —— -- SQLite 不支持给已有列追加外键,重建 graph_runs 要连带 graph_run_nodes 的真外键一起动, -- 代价远大于收益,见 §四.6)。 CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, -- T-{base36}-{4hex} goal_id TEXT NOT NULL REFERENCES goals(id), company_id TEXT NOT NULL, -- 冗余一份:调度器按公司选任务,避免每次 JOIN seq INTEGER NOT NULL, -- goal 内序号,列表排序用(1,2,3…) title TEXT NOT NULL, description TEXT NOT NULL, -- 落进 task-goal.json 的那句;下达后不可改 status TEXT NOT NULL DEFAULT 'draft', -- draft | queued | running | completed | failed | blocked | cancelled -- queued 现在就在取值域里,但排队逻辑(调度器 / run_status='queued')另批实现: -- 取值域是数据形状的一部分,先钉死它,免得批 3 再来改一次列注释 execution_mode TEXT NOT NULL DEFAULT 'manual' CHECK (execution_mode IN ('manual','auto')), depends_on TEXT NOT NULL DEFAULT '[]', -- JSON 数组,同 goal 内的 task id budget_json TEXT, current_run_id TEXT, -- 最近一次 root run(逻辑外键,见 §四.6) attempt INTEGER NOT NULL DEFAULT 0, -- 起过几次 run(任务级重跑,≠ 节点级 retry) report_to TEXT, -- 从 goal 继承或调用方指定;见 §四.7 禁令二 created_by TEXT, -- 'task-planning' | 'human':拆出来的还是人加的 blocked_reason TEXT, -- status='blocked' 时的原因,如 upstream_failed:T-xxx priority TEXT NOT NULL DEFAULT 'P1', -- ── 批 5:两套任务体系统一(§十二)──────────────────────────── -- agent 之间派单(POST /api/agent/task/dispatch)与「跑一张 org graph」是两种执行载体, -- 硬说成一回事是骗自己:前者直接起一次 agent 对话,后者走调度器出队 + 图执行。 -- 所以显式建模,而不是让调度器去猜。调度器只管 executor='graph'。 -- ⚠️ 本段刻意不用反引号:这整块是 db.exec 的模板字符串,注释里的反引号会把它提前闭合。 executor TEXT NOT NULL DEFAULT 'graph' CHECK (executor IN ('graph','agent')), assignee_agent_id TEXT, -- executor='agent' 时派给谁(dispatch 的 to) -- 下面两列只对 executor='agent' 有意义(图任务用 current_run_id / task-goal.json): summary TEXT, -- 汇报摘要(/task/report 的 summary) execution_conv_id TEXT, -- 派单起的那次对话 id(task-{taskId}) -- ⚠️ 旧 meta 里还有个 sourceConvId,全仓**零读取方**,借这次统一删掉 —— -- 「实现在、零调用方」是本仓库反复立案的惯犯模式,不再往新表里搬一份。 -- 派单链:A 让 B 干活、B 又拆给 C。与 depends_on 不是一回事 —— -- depends_on 是同级任务之间的先后,parent_task_id 是上下级的派生。 parent_task_id TEXT REFERENCES tasks(id), created_at TEXT NOT NULL, updated_at TEXT NOT NULL, queued_at TEXT, -- 入队时间;非空 = 描述已冻结(§四.4) started_at TEXT, finished_at TEXT, archived_at TEXT ); CREATE INDEX IF NOT EXISTS idx_tasks_goal ON tasks(goal_id, seq); -- 列序与调度器的选取口径(company → status → priority → created_at)对齐 CREATE INDEX IF NOT EXISTS idx_tasks_queue ON tasks(company_id, status, priority, created_at); `); // 存量库迁移(写法照抄 graph-db.js migrate():hasColumn 判断 + ALTER TABLE ADD COLUMN, // 新建库走上面的 CREATE TABLE 就已经是目标形态,这里是 no-op)。 // ── 批 4:DROP 掉「一家公司只许一条活跃 root run」的唯一索引 ──────────────── // // 从建表块里删掉那条 CREATE **对存量库无效**(`IF NOT EXISTS` 只管不重复建,不会把已有的 // 删掉),所以必须在这里显式 DROP —— 与 graph-db.js:76 那处 // `DROP INDEX IF EXISTS idx_org_graph_nodes_company_role` 同一写法。 // 不套 hasColumn:那个函数 pragma 的是 table_info,承载不了索引;`DROP INDEX IF EXISTS` // 本身就是幂等的,重复跑、并发跑都安全(SQLite 把 DDL 序列化,慢的那个等 busy_timeout)。 // // ⚠️ 删索引 = 删掉最后一道 DB 兜底。删之前先确认主守卫是**真的**主守卫 —— // `updateRunStatus` 的准入是四步一体、缺一步就重新出现超发窗口: // 1. `BEGIN IMMEDIATE`:读之前就拿写锁,杜绝两个连接各自数出「还有空位」; // 2. 事务内**重读** run_status:状态已被别人推走时返回 0(幂等 no-op),不误报坑满; // 3. 事务内 `listActiveRootRunIds` 计数,判据是「从活跃集外进入活跃集」(集内移动不重复占坑); // 4. 同一事务内做条件更新 `WHERE id=? AND run_status=@from`(写进去的 to 不在 WHERE 里)。 // 三个测试文件钉着这四步:task-scheduler.test.js「准入守卫」那组、task-queue-acceptance.test.js // 的超发用例、graph-run-slot-recovery.test.js 的拒绝-退队列路径。 // // ⚠️ **上线要求全停再起**:DROP 之后只要有**旧版本代码**的进程再调一次 getDb(), // 它那条 `CREATE UNIQUE INDEX IF NOT EXISTS` 会把索引原样建回来,于是 max>1 的第二条准入 // 吃一句 SQLite 原文的 500(isSlotRaceError 已不再吞它,就是为了让这一幕吵起来)。 db.exec(`DROP INDEX IF EXISTS idx_graph_runs_one_active_root`); // ── DROP 掉 graph_files 的**旧无条件**唯一索引 idx_graph_files_path ────────── // // 它是产物索引加「版本」之前的形态:`UNIQUE(company_id, COALESCE(task_id,''), rel_path)`, // **没有** `WHERE superseded_by IS NULL`。后来加了部分索引 idx_graph_files_current, // 却没删掉旧的 —— 于是 recordFile 的「先把旧行 superseded_by 指向新行、再插新行」 // 这套版本化写法在**存量库上整个失效**:旧索引不认 superseded_by,同路径第二行直接撞。 // // 症状是静默的:recordFile 抛,被 writeFileIndex 降级成一条 timeline // (`产物索引写入失败 xxx: UNIQUE constraint failed: index 'idx_graph_files_path'`), // 节点照样 success。后果是**产物索引在任何有重试的 run 上都不全** —— 而下游节点 // 正是靠这份索引按 kind 找上游产物的。 // // 2026-08-20 真跑实测:一条 run 里每个节点的产物索引写入**全部**失败(19 次调用、 // 每次两三个文件),因为重试让同一条 rel_path 被产出了第二次。 // // 新建库没有这条索引(建表块里早就只剩 current 那条),所以**只有存量库会踩** —— // 又一次「新建库的路径由单测天然覆盖、存量库那条得另外安排人走」。 db.exec(`DROP INDEX IF EXISTS idx_graph_files_path`); // ── 批 5:tasks 追加三列(§十二.3)────────────────────────────── // 旧行的落点都对:executor 缺省 'graph'(批 1-4 建的全是图任务), // assignee_agent_id / parent_task_id 为 NULL(图任务没有派单对象、也不在派单链上)。 // ⚠️ `ALTER TABLE ADD COLUMN` 加不了 CHECK 也加不了外键(SQLite 限制),所以存量库上 // `executor` 的取值域**只由应用层守**,与新建库的 CHECK 不同源 —— 这是已知的不对称, // 写在这里免得下一个人以为库层拦得住。真要对齐得重建表,代价远大于收益。 if (!hasColumn(db, 'tasks', 'executor')) { db.exec(`ALTER TABLE tasks ADD COLUMN executor TEXT NOT NULL DEFAULT 'graph'`); } if (!hasColumn(db, 'tasks', 'assignee_agent_id')) { db.exec(`ALTER TABLE tasks ADD COLUMN assignee_agent_id TEXT`); } if (!hasColumn(db, 'tasks', 'parent_task_id')) { db.exec(`ALTER TABLE tasks ADD COLUMN parent_task_id TEXT`); } if (!hasColumn(db, 'tasks', 'summary')) { db.exec(`ALTER TABLE tasks ADD COLUMN summary TEXT`); } if (!hasColumn(db, 'tasks', 'execution_conv_id')) { db.exec(`ALTER TABLE tasks ADD COLUMN execution_conv_id TEXT`); } // ⚠️ 这条索引**必须建在 ALTER 之后**,不能放进上面的建表块: // 建表块整体先于迁移块执行,而存量库上 `CREATE TABLE IF NOT EXISTS tasks` 是 no-op —— // 表还是批 4 的形态、没有 assignee_agent_id,那一句 CREATE INDEX 会直接抛 // 「no such column」,`getDb()` 起不来 = **所有存量库一启动就炸**。实测踩过。 db.exec(`CREATE INDEX IF NOT EXISTS idx_tasks_assignee ON tasks(company_id, assignee_agent_id, created_at)`); // R4-F1:graph_run_nodes 追加 remediation_note,可空,旧行 NULL = 「没有留给下一轮的话」。 if (!hasColumn(db, 'graph_run_nodes', 'remediation_note')) { db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN remediation_note TEXT`); } // R-19.A2:graph_edge_contracts 追加 role。旧行一律回填 'produces' 而不是留空: // 规则 6 里 NULL 既不算产出也不算需求,一条本该是 produces 的旧行会让下游的 needs // 突然找不到产出方 —— 那是**误拒一张合法的图**。校验闸门宁可漏判不可错拦, // 且 produces 是压倒性多数(forward 边基本都写它),所以回填它是更接近事实的一侧。 if (!hasColumn(db, 'graph_edge_contracts', 'role')) { db.exec(`ALTER TABLE graph_edge_contracts ADD COLUMN role TEXT`); db.exec(`UPDATE graph_edge_contracts SET role = 'produces' WHERE role IS NULL`); } // R-17.1:跨层地址 + 子 run 代次。旧行两列都是 NULL—— // fail_node_ref = NULL → 没有跨层归因,回退退化成整张图重跑(与返工前同语义) // child_run_attempt = NULL → 代次未知,executeSubgraph 按「不是本轮的」处理(复用并重置), // 最坏情况多重跑一轮,绝不会返回 pending 卡死。 if (!hasColumn(db, 'graph_run_nodes', 'fail_node_ref')) { db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN fail_node_ref TEXT`); } if (!hasColumn(db, 'graph_run_nodes', 'child_run_attempt')) { db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN child_run_attempt INTEGER`); } if (!hasColumn(db, 'graph_run_loop_records', 'target_node_ref')) { db.exec(`ALTER TABLE graph_run_loop_records ADD COLUMN target_node_ref TEXT`); } // 27-P3C §四.9(批 2):自动拆分失败时的降级留痕。 // 降级后的 goal 与「本来就只有一条 task」的 goal 在库里长得**一模一样**(1 条 task、 // created_by='human'、depends_on=[]),没有这一列就分不出「这是拆出来的」和 // 「这是拆失败后兜底的」——界面也就没法按验收 20 标注「自动拆分失败,已按单任务处理」。 // 可空:NULL = 没有失败过(含从没拆过的存量 goal 与同步路径建的 goal)。 if (!hasColumn(db, 'goals', 'planning_error')) { db.exec(`ALTER TABLE goals ADD COLUMN planning_error TEXT`); } // 27-P3C 第二轮复核:汇报的可查性(语义见 CREATE TABLE 里那段)。 // **回填是这次迁移的重点,不是附赠品**:不回填的话,加完列之后第一次启动, // `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`); } // 27-P4 T5:graph_runs 追加 block_reason,可空,旧行 NULL = 「不是因为欠费/预算挂起的 // blocked run」(批 3 时代 blocked 只有熔断/死路一种成因,回填 NULL 如实反映当时没有这个概念)。 if (!hasColumn(db, 'graph_runs', 'block_reason')) { db.exec(`ALTER TABLE graph_runs ADD COLUMN block_reason TEXT`); } // 欠费自动恢复的扫描入口:`WHERE run_status='blocked' AND block_reason='billing-arrears' // AND parent_run_id IS NULL`,加个复合索引配这个查询。 db.exec(`CREATE INDEX IF NOT EXISTS idx_graph_runs_block_reason ON graph_runs(run_status, block_reason) WHERE parent_run_id IS NULL`); if (!hasColumn(db, 'goals', 'reported_at')) { db.exec(`ALTER TABLE goals ADD COLUMN reported_at TEXT`); db.exec(` UPDATE goals SET reported_at = updated_at WHERE reported_at IS NULL AND status IN (${GOAL_TERMINAL_STATUS.map((s) => `'${s}'`).join(', ')}) `); } // 墙钟预算改成「只算活跃时长」之后的存量重锚(graph-engine.wallclockElapsedMs)。 // // 老形态:spent.started_at = 第一次花钱的绝对时刻,没有 active_ms。那个锚点里混着 // **挂起等人**的时间,真实活跃时长复原不出来 —— 按 0 起算、一笔勾销。另外两维 // (agent_calls / tokens)是精确的且没受影响,真正兜住花钱的是那两维。 // // **为什么必须在开库时做,而不是交给转换钩子**:钩子只在「进 / 出 running」时触发, // 而升级那一刻正停在 running 的 run 不会再进一次 running —— 旧锚点一直挂着, // 下一次 checkBudget 直接判耗尽。 // // 放在**这里**而不是 graph-db.js 的 migrate():graph_runs 是本文件建的表, // graph-db 的迁移块跑在它之前,写在那边会在新库上直接 `no such table: graph_runs` // (踩过了 —— 和 §16.2 那次「加索引写进建表块」是同一类顺序问题)。 backfillWallclockAnchor(db); schemaEnsured = true; return db; } /** * 存量墙钟锚点重置:active_ms 补 0;running 的重新锚到现在,其余断开锚点。 * 只碰配了 wallclock_ms 的行 —— 没配的本来就不限墙钟,凭空加字段只会让人以为限了。 * 判据是 active_ms 缺席,所以第二遍是零行更新(幂等)。 */ function backfillWallclockAnchor(db) { const rows = db .prepare(`SELECT id, run_status, budget_json FROM graph_runs WHERE budget_json IS NOT NULL`) .all(); const upd = db.prepare(`UPDATE graph_runs SET budget_json = ? WHERE id = ?`); const nowIso = new Date().toISOString(); for (const r of rows) { let b; try { b = JSON.parse(r.budget_json); } catch { continue; } // 坏字段不该把开库卡死 if (!b || b.wallclock_ms == null) continue; if (!b.spent) b.spent = {}; if (b.spent.active_ms != null) continue; // 已经是新形态 b.spent.active_ms = 0; b.spent.started_at = r.run_status === 'running' ? nowIso : null; upd.run(JSON.stringify(b), r.id); } } // ──────────────────────────────────────────────────────────────── // §四.A.4 状态机(模式照抄 devloop/unit-registry.js:显式转换表 + 非法转换抛错) // ──────────────────────────────────────────────────────────────── class InvalidTransitionError extends Error { constructor(scope, from, to, id) { super(`Invalid ${scope} transition for ${id}: ${from} → ${to}`); this.name = 'InvalidTransitionError'; this.scope = scope; // 'node' | 'run' this.from = from; this.to = to; this.id = id; } } /** graph_run_nodes.status 转换表 */ const NODE_TRANSITIONS = { pending: ['ready', 'skipped'], ready: ['running', 'skipped'], running: ['success', 'fail'], // fail → ready 是命中 backward 边或人工「重试」;fail → blocked 是死路 / 硬顶耗尽 fail: ['ready', 'blocked'], // success 是终态,唯一出路是人工「重试」强制打回 ready success: ['ready'], // blocked 唯一走出方式就是人工「重试」 blocked: ['ready'], skipped: [], }; /** graph_runs.run_status 转换表 */ const RUN_TRANSITIONS = { planning: ['ready'], // 27-P3C §四.3:排队态。入队时 createRun 直接建成 queued(不经过 planning), // 出队就是 `updateRunStatus(runId,'queued','ready')` 这一条 CAS,**不走 markRunReady** // (它把 from 写死成 'planning',对 queued run 是空转:WHERE run_status='planning' 匹配不到)。 // // 出口只有两条,两条都必不可少: // · ready —— 出队。之后照走既有的 ready → running,起跑路径全仓库仍然只有一条。 // · stopped —— 撤出队列 / 出队后又发现抢不到坑要收回。**漏了这条的后果是静默的**: // `graph-engine.stopRun` 对每条 run 都先问一句 `canTransitionRun(cur,'stopped')`, // 为 false 就跳过 —— 于是「停止」一个排队任务只会写个 stop_requested=1, // run 永远躺在队里,界面上和正常排队一模一样。 queued: ['ready', 'stopped'], ready: ['running', 'stopped'], // running → failed:§四.A.4 的转换列表里没写这一条,但 run_status 枚举含 failed, // 且 §四.D.0 明确「子 run blocked/failed → 父节点 fail」,说明 failed 是可达态,故补上。 running: ['paused', 'blocked', 'completed', 'failed', 'stopped'], paused: ['running', 'stopped'], // queued 这条出口是**恢复路径的退路**(27-P3C 批 3 复核 F1):人加完预算点「启动」时坑 // 可能刚被队里下一条占走,那时唯一能让这条 run 还有将来的动作就是把它放回队列, // 由调度器在坑空时按位次接回来 —— 否则它永远躺在 blocked(非终态),goal 也就永不收敛。 // 唯一实现处是 `requeueBlockedRootRun`(run 行与 task 行必须同一个事务,见那里)。 blocked: ['running', 'stopped', 'queued'], // R-17.1:completed / failed 不再是绝对终态。跨层寻址允许**上一层**的回退边点名一个 // 已经收敛的子 run 里的节点(org-validator 归因到 org-coder 的某个 work graph node, // 而 org-coder 那张子 run 早就 completed 了),复用旧子 run 就必须能把它重新拉起来。 // // **出口只开到 ready,不开到 running**:`ready` 的语义正是「图还在、已就绪、等启动」, // 重开一张收敛过的 run 就是这个状态;随后走既有的 `ready → running` 启动转换, // 于是「怎么起跑」全仓库仍然只有一条路径。给终态直开一条到 running 的捷径既多一条 // 语义重复的边,也会让「run 是被谁拉起来的」在 timeline 上分不清。 completed: ['ready'], failed: ['ready'], // stopped 仍是绝对终态:它是**人工主动终止**的结果,自动流程不许把它复活(见 reopenChildRun) stopped: [], }; function canTransition(table, from, to) { return Boolean(table[from]) && table[from].includes(to); } const canTransitionNode = (from, to) => canTransition(NODE_TRANSITIONS, from, to); const canTransitionRun = (from, to) => canTransition(RUN_TRANSITIONS, from, to); // ──────────────────────────────────────────────────────────────── // 写入前校验:§三.7 四项(复用)+ 第五项 fail-fallback-edge(T14) // ──────────────────────────────────────────────────────────────── /** * 第五项校验(§四.D.2.2 末尾 + 31-work-graph-fail-fallback-coverage): * **除图的唯一入口节点外**,每个节点都必须至少留一条 fail_reason IS NULL 的兜底边。 * * 判据从 27-P3P4/T14 的旧版收紧过一次(2026-08-21,见 31 号文档 §三决策 A): * 旧版只在节点**已经画了别的 fail 边**时才检查兜底边齐不齐;新版不再要求这个前提—— * 一条 fail 边都没有的非入口节点,现在同样算违规。 * * **为什么按"入口节点"豁免,不按节点名字/agent 类型**:本仓库所有本体共用 `_base` 的 * RESEARCH→PLAN→EXECUTE→VERIFY 四段式协议,VERIFY 阶段任何 agent_call 节点都可能写一份 * `attempt-N.json` 自判失败(`graph-engine.readNodeProduct` 对所有节点一视同仁地读它, * 不限于名字里带 `validator`/`triager` 的判词节点)——这次撞见的 `sdd-planning` 正是 * 普通生产节点自判失败的例子。只有图的入口节点没有同图内的上游可退,"没有 fail 边= * 该 blocked 等人介入"对它才是唯一自洽的选择;其余节点永远有"最近的上游"可以退。 * * ⚠️ **这次只收紧了 work 层**(本函数),org 层的 `org-graph-service.validateGraph` * 规则 5 仍是旧语义——org 层节点全部是 subgraph 类型,失败来源机制不同(子图收敛后回传, * 不是自己的 attempt-N.json),且套用同一条新规则会让 `dev.json`/`content.json`/`ops.json` * 立刻不合法(`org-coder`/`org-writer`/`org-monitor` 都是非入口、无 fail 边),这属于 * 27-p3b「验收 9」还没收口的那条链路,见 31 号文档决策 D,这里不动。 * * ⚠️ 两处规则不是同一份实现(见旧注释与 27-P3P4 §三.7 的风险记录):`validateWorkGraph` * 会先跑一遍 `validateGraph` 的旧版规则 5(可能报一次),再跑本函数的新版规则 5 * (可能对同一节点再报一次,且措辞不同)——这次不做去重,只保证本函数语义正确。 * * ⚠️ **`strict` 默认 `false`(保持旧语义)**:全仓 28 个测试文件在拿 `createWorkGraph` * 当测试夹具的通用建图工具(测预算/熔断/回退地址这些跟"图设计得完不完善"毫无关系的机制), * 图大多是"a→b→c 只给要测的那条边配 fail"的最小形状,非入口节点普遍没有自己的兜底边。 * 无条件收紧这条规则时实测**炸了 46 个用例、13 个文件**——那不是这些用例错了, * 是这条新规则的"结构完整性"诉求跟"内部机制测试用最小图"这个通用测试模式天然不兼容。 * 真正需要新语义的只有一个调用点:`graph-engine.createWorkGraphSafely` * (workgraph-planning 交回的图落库前的校验,也是它 Step 7 那张 errors 表实际读到的来源), * 那里显式传 `strict: true`。其余调用方(含测试)不传,拿到的是原来那条更宽松的规则, * 行为不变。 * * @param {Array<{nodeName, systemAgentId?, nodeKind?}>} nodes 用于算 forward 入度、找入口 * @param {Array<{from, to, condition, failReason?, edgeKind?}>} edges * @param {{strict?: boolean}} [opts] * @returns {Array<{rule: string, detail: string}>} */ function validateFailFallback(nodes, edges, { strict = false } = {}) { const errors = []; const byFrom = new Map(); // fromKey -> { hasFail: bool, hasFallback: bool } for (const e of edges || []) { if ((e.condition || 'success') !== 'fail') continue; const reason = e.failReason !== undefined ? e.failReason : e.fail_reason; const slot = byFrom.get(e.from) || { hasFail: false, hasFallback: false }; slot.hasFail = true; if (reason === null || reason === undefined || reason === '') slot.hasFallback = true; byFrom.set(e.from, slot); } if (!strict) { // 旧语义:只查已经画了 fail 边、但兜底缺失的节点。 for (const [from, slot] of byFrom) { if (slot.hasFail && !slot.hasFallback) { errors.push({ rule: 'fail-fallback-edge', detail: `节点 ${from} 有 fail 出边,但没有一条 fail_reason 为空的兜底边,归因结果不匹配任何边时会直接死路`, }); } } return errors; } // 新语义:除入口节点外,一条 fail 边都没有也算违规。 const names = (nodes || []).map((n) => (typeof n === 'string' ? n : n && n.nodeName)).filter(Boolean); if (!names.length) return errors; const indeg = new Map(names.map((n) => [n, 0])); for (const e of edges || []) { if ((e.edgeKind || 'forward') !== 'forward') continue; if (indeg.has(e.to)) indeg.set(e.to, indeg.get(e.to) + 1); } const entries = names.filter((n) => indeg.get(n) === 0); // 入口不唯一(0 个或多个)时不豁免任何节点——那本身已经是规则 2 single-entry 的违规, // 这里没有"唯一入口"可指,宁可多报一点噪声也不要悄悄放过该查的节点。 const exempt = entries.length === 1 ? new Set(entries) : new Set(); for (const name of names) { if (exempt.has(name)) continue; const slot = byFrom.get(name); if (!slot) { errors.push({ rule: 'fail-fallback-edge', detail: `节点 ${name} 没有任何 fail 出边——除图的唯一入口节点外,每个节点都必须至少留一条` + '不带 failReason 的兜底边,否则它自判失败时(VERIFY 阶段写出 attempt-N.json)无处可退,直接 blocked', }); } else if (!slot.hasFallback) { errors.push({ rule: 'fail-fallback-edge', detail: `节点 ${name} 有 fail 出边,但没有一条 fail_reason 为空的兜底边,归因结果不匹配任何边时会直接死路`, }); } } return errors; } /** * work graph 写入前的全部校验 = §三.7 四项(acyclic / single-entry / reachable / * backward-labeled,直接调 org-graph-service.validateGraph)+ 第五项 fail-fallback-edge。 * * @param {Array} nodes [{nodeName, systemAgentId?, nodeKind?}] 或 ['nodeName', ...] * @param {Array} edges [{from, to, condition?, failReason?, edgeKind?}],from/to 是 nodeName * @param {{strictFailFallback?: boolean}} [opts] 见 `validateFailFallback` 的 `strict` 参数说明 * @returns {{ok: boolean, errors: Array<{rule: string, detail: string}>}} */ function validateWorkGraph(nodes, edges, { strictFailFallback = false } = {}) { const ns = normalizeNodes(nodes); const es = normalizeEdges(edges); // 四项复用:org 侧用 roleName 表达节点,work 侧叫 nodeName,此处做一次名字映射 const base = validateGraph( ns.map((n) => ({ roleName: n.nodeName, systemAgentId: n.systemAgentId })), es ); const errors = base.errors.concat(validateFailFallback(ns, es, { strict: strictFailFallback })); return { ok: errors.length === 0, errors }; } const NODE_KINDS = ['agent_call', 'subgraph']; function normalizeNodes(nodes) { return (nodes || []).map((n) => { const nodeName = typeof n === 'string' ? n : n && (n.nodeName || n.node_name); const systemAgentId = (typeof n === 'object' && n && (n.systemAgentId || n.system_agent_id)) || nodeName; const nodeKind = (typeof n === 'object' && n && (n.nodeKind || n.node_kind)) || 'agent_call'; return { nodeName, systemAgentId, nodeKind }; }); } function normalizeEdges(edges) { return (edges || []).map((e) => ({ from: e.from, to: e.to, condition: e.condition || 'success', failReason: e.failReason !== undefined ? e.failReason : e.fail_reason ?? null, edgeKind: e.edgeKind || e.edge_kind || 'forward', // R-19:边上声明的数据契约(createWorkGraph 插边后写 graph_edge_contracts) contracts: normalizeContracts(e), })); } // ──────────────────────────────────────────────────────────────── // 定义层 CRUD // ──────────────────────────────────────────────────────────────── /** * 建一张 work graph(图定义一次性写完,之后不再改——work graph 生命周期是一次性的, * 要改就重新规划一张,所以这里没有 patch)。 * * @param {object} p * @param {string} p.companyId * @param {number} [p.ownerOrgNodeId] 这张图属于哪个 org node * @param {string} [p.plannedBy='workgraph-planning'] * @param {string} [p.goalId] @param {string} [p.taskId] @param {string} [p.okrNodeId] * 三者基数不同不可混用,见 27-NAMING §四 * @param {Array} p.nodes 见 validateWorkGraph * @param {Array} p.edges 见 validateWorkGraph,from/to 是 nodeName * @returns {{ok: true, id, nodeIds: Map} | {ok: false, errors: Array}} */ function createWorkGraph({ companyId, ownerOrgNodeId = null, plannedBy = 'workgraph-planning', goalId = null, taskId = null, okrNodeId = null, nodes = [], edges = [], strictFailFallback = false, } = {}) { const ns = normalizeNodes(nodes); const es = normalizeEdges(edges); const verdict = validateWorkGraph(ns, es, { strictFailFallback }); if (!verdict.ok) return { ok: false, errors: verdict.errors }; const bad = ns.find((n) => !NODE_KINDS.includes(n.nodeKind)); if (bad) { return { ok: false, errors: [{ rule: 'schema', detail: `节点 ${bad.nodeName} 的 nodeKind 非法: ${bad.nodeKind}` }], }; } const db = getDb(); const now = new Date().toISOString(); const graphId = newId('wg'); const nodeIds = new Map(); db.transaction(() => { db.prepare(` INSERT INTO work_graphs (id, company_id, owner_org_node_id, planned_by, goal_id, task_id, okr_node_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) `).run(graphId, companyId, ownerOrgNodeId, plannedBy, goalId, taskId, okrNodeId, now); const insNode = db.prepare(` INSERT INTO work_graph_nodes (id, work_graph_id, node_name, system_agent_id, node_kind, created_at) VALUES (?, ?, ?, ?, ?, ?) `); for (const n of ns) { const id = newId('wgn'); nodeIds.set(n.nodeName, id); insNode.run(id, graphId, n.nodeName, n.systemAgentId, n.nodeKind, now); } const insEdge = db.prepare(` INSERT INTO work_graph_edges (id, work_graph_id, from_node_id, to_node_id, condition, fail_reason, edge_kind, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) `); for (const e of es) { const edgeId = newId('wge'); insEdge.run( edgeId, graphId, nodeIds.get(e.from), nodeIds.get(e.to), e.condition, e.failReason, e.edgeKind, now ); // R-19 落库侧(work 层):declareEdgeContract 在这一层的**唯一调用方**。 // 没有它,inboundContracts 恒返回空 → describeInputs 拼不出任何上游路径 → // 下游节点重演第四次真跑那句「不是上游没产出,是它的产出结构上到不了我手上」。 for (const c of e.contracts) { declareEdgeContract({ edgeLayer: 'work', edgeId, kind: c.kind, required: c.required, role: c.role, }); } } })(); return { ok: true, id: graphId, nodeIds }; } /** 读一张 work graph 定义;不存在返回 null */ function getWorkGraph(workGraphId) { const db = getDb(); const graph = db.prepare(`SELECT * FROM work_graphs WHERE id = ?`).get(workGraphId); if (!graph) return null; return { graph, nodes: db .prepare(`SELECT * FROM work_graph_nodes WHERE work_graph_id = ? ORDER BY created_at, id`) .all(workGraphId), edges: db .prepare(`SELECT * FROM work_graph_edges WHERE work_graph_id = ? ORDER BY created_at, id`) .all(workGraphId), }; } // ──────────────────────────────────────────────────────────────── // 定义层的统一读取(org / work 两层同构,运行时层只认 graph_kind + graph_id) // ──────────────────────────────────────────────────────────────── /** * 按 graph_kind 读定义层的节点 + 边,归一成同一形状: * node: {defNodeId(TEXT), name, systemAgentId, nodeKind} * edge: {id, from, to, condition, failReason, edgeKind} from/to 是 defNodeId * * org 侧 id 是 INTEGER,这里统一 String() 成 TEXT(§四.A.2 那条 ⚠️ 的落地处)。 */ function readDefGraph(kind, graphId) { const db = getDb(); if (kind === 'work') { const g = getWorkGraph(graphId); if (!g) return { nodes: [], edges: [] }; return { nodes: g.nodes.map((n) => ({ defNodeId: n.id, name: n.node_name, systemAgentId: n.system_agent_id, nodeKind: n.node_kind, })), edges: g.edges.map((e) => ({ id: e.id, from: e.from_node_id, to: e.to_node_id, condition: e.condition, failReason: e.fail_reason, edgeKind: e.edge_kind, })), }; } // kind === 'org':graphId 就是 company_id(§四.A.2) const nodes = db .prepare(` SELECT id, role_name, system_agent_id FROM org_graph_nodes WHERE company_id = ? AND status = 'active' ORDER BY id `) .all(graphId); // fail_reason 列由 T-P3 追加,未加时按 NULL 读,避免本文件被 T-P3 的落地顺序卡住 const failCol = hasColumn(db, 'org_graph_edges', 'fail_reason') ? 'fail_reason' : 'NULL AS fail_reason'; const edges = db .prepare(` SELECT id, from_node_id, to_node_id, condition, ${failCol}, edge_kind FROM org_graph_edges WHERE company_id = ? AND status = 'active' ORDER BY id `) .all(graphId); const alive = new Set(nodes.map((n) => n.id)); return { nodes: nodes.map((n) => ({ defNodeId: String(n.id), name: n.role_name, systemAgentId: n.system_agent_id, // org graph 的节点全部是 subgraph executor(§四.D.0) nodeKind: 'subgraph', })), edges: edges .filter((e) => alive.has(e.from_node_id) && alive.has(e.to_node_id)) .map((e) => ({ id: String(e.id), from: String(e.from_node_id), to: String(e.to_node_id), condition: e.condition, failReason: e.fail_reason, edgeKind: e.edge_kind, })), }; } // ──────────────────────────────────────────────────────────────── // 运行时层 CRUD // ──────────────────────────────────────────────────────────────── /** * 建一次执行实例,并按定义层的节点批量铺出 graph_run_nodes(status 全 'pending')。 * * root run:不传 parentRunId,root_run_id = 自身 id,budget_json 只在这里非空(§四.A.3)。 * 子 run: 传 parentRunId + parentNodeRunId,root_run_id 从父 run 继承(冗余存储避免上溯)。 * * **建 run 本身不占坑,也不做额度检查**(批 4 起唯一的坑约束是 `updateRunStatus` 的准入事务)。 * 建出来的 run 只有走「进入活跃集」那一跳时才会被数一次坑 —— 计数守卫**故意不放在这里**: * 建 run 与进活跃集之间隔着排队,放在这里等于把「此刻有没有位置」当成「以后有没有位置」。 * * @returns {{ok: true, id, nodeRunIds: Map}} */ function createRun({ graphKind, graphId, companyId, parentRunId = null, parentNodeRunId = null, goalId = null, taskId = null, okrNodeId = null, 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(); const runId = newId('run'); const nodeRunIds = new Map(); db.transaction(() => { let rootRunId = runId; if (parentRunId) { const parent = db.prepare(`SELECT root_run_id FROM graph_runs WHERE id = ?`).get(parentRunId); if (!parent) throw new Error(`parentRunId 不存在: ${parentRunId}`); rootRunId = parent.root_run_id; } 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, 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 ); const ins = db.prepare(` INSERT INTO graph_run_nodes (id, run_id, def_node_id, status, created_at, updated_at) VALUES (?, ?, ?, 'pending', ?, ?) `); for (const n of readDefGraph(graphKind, graphId).nodes) { const id = newId('rn'); nodeRunIds.set(n.defNodeId, id); ins.run(id, runId, n.defNodeId, now, now); } })(); return { ok: true, id: runId, nodeRunIds }; } function getRun(runId) { return getDb().prepare(`SELECT * FROM graph_runs WHERE id = ?`).get(runId) || null; } /** * 这个节点行**一次都没被执行过**(可以安全删掉,删了不销毁任何审计轨迹)。 * `status='pending'` 单独一条不成立:回退重置会把跑过的节点写回 pending 并**保留 attempt**。 * 判据与 `syncRunNodesToDef` 里那条 DELETE 的 WHERE 逐字同源,改一处要改两处。 */ function isUntouchedNode(r) { return r.status === 'pending' && !r.attempt && !r.agent_run_id && !r.child_run_id && !r.output_ref; } /** * 把一条 run 的 `graph_run_nodes` 重新对齐**当前**的图定义(27-P3C §四.3 末)。 * * 为什么需要它:`createRun` 是在建行的那个事务里、按**当时**的 `readDefGraph` 把节点行全插好的, * 而 `graph-engine.defIndex` 每推进一步都重新读**最新**定义。两者一旦分叉,症状全是静默的: * · 定义里新增的节点没有 run 行 → `markReady` 第一句 `if (!rn …) return null` 直接吞掉, * `seedEntryNodes` / `activateSuccessTargets` 都只 `if (activated) push`,没有任何日志; * · 定义里删掉的节点留下孤儿 `pending` 行 → `settleRun` 把它们批量 skipped, * 图看起来跑完了,实际少跑了一半。 * 这个洞**今天就存在**(createRun 到 startRun 之间几行代码,窗口毫秒级);排队把窗口拉到小时级 * ——「排队期间有人点重新规划组织」是一等的用户操作 —— 于是从理论缺陷变成必然事件。 * * **只对 `run_status = 'queued'` 的 run 生效 —— 白名单,不是黑名单**(批 3 复核 F2)。 * 曾经写成「stopped / completed 之外都放行」,那等于把**正在跑的 run** 也放了进来。 * 多进程(pm2 cluster)下的形状:P1、P2 同时 tick 读到同一条 task → P2 先赢下出队 CAS, * run 转 running 开始执行 → P1 这时才走到本函数,run 是 running 不在黑名单里 → 它按 * **旧定义之外**的判据删掉一批 pending 行 → P1 随后的 CAS 返回 0,一切「正常」地 continue * → 这条 run 少了几个节点,`settleRun` 数不到活跃节点直接判 completed。 * 任务「成功」了,却少跑了一段图,全程零报错。白名单让这种交叠只可能删到还在队里的 run。 * * 非 queued 一律返回 null 且不报警:调用方(tick 5b / promoteQueuedTask)紧接着就是一次 * `queued → ready` 的 CAS,它同样会返回 null,「这轮没抢到」在调用方那里本来就是可见的。 * * 孤儿行只删**一点执行痕迹都没有**的(pending 且 attempt=0、没 agent_run_id / child_run_id / * output_ref)。以前的判据只有 `status='pending'`,并在注释里声称「带痕迹的都留着」—— * 那句话是假的:`resetRunNodes` / `resetForwardDownstream` / `reopenChildRun` 会把**跑过的** * 节点写回 pending(保留 attempt、清 fail_reason),于是 `status='pending'` ≠ 没跑过, * 那条 warn 也不会响。批 3 之后这更不是理论问题:`requeueBlockedRootRun` 会把一条跑过一半的 * run 退回 queued,它下一次出队就会走到这里。留下的行计入 kept 并告警。 * * ponytail: 这是「定义变了就补行」的最小实现,默认新节点从 pending 开始是安全的。 * 将来若允许对**跑到一半**的 run 做 replan,必须升级成带 contract_hash 判据的级联重置 * (loops-l3.md 那套四分类),这里的 kept 计数就是升级的触发信号。 * * @returns {{added: string[], removed: string[], kept: string[]}|null} null = run 不存在或不在队列里 */ function syncRunNodesToDef(runId) { const db = getDb(); const run = getRun(runId); if (!run) return null; if (run.run_status !== 'queued') return null; const defNodeIds = new Set(readDefGraph(run.graph_kind, run.graph_id).nodes.map((n) => n.defNodeId)); const rows = db .prepare(` SELECT id, def_node_id, status, attempt, agent_run_id, child_run_id, output_ref FROM graph_run_nodes WHERE run_id = ? `) .all(runId); const have = new Set(rows.map((r) => r.def_node_id)); const now = new Date().toISOString(); const added = []; const removed = []; const kept = []; db.transaction(() => { const ins = db.prepare(` INSERT INTO graph_run_nodes (id, run_id, def_node_id, status, created_at, updated_at) VALUES (?, ?, ?, 'pending', ?, ?) `); for (const defNodeId of defNodeIds) { if (have.has(defNodeId)) continue; ins.run(newId('rn'), runId, defNodeId, now, now); added.push(defNodeId); } // WHERE 里把「没跑过」的判据整条复述一遍,而不是只靠上面读到的快照:读和删之间 // 另一个连接仍可能刚给这一行写上 agent_run_id(这正是白名单挡不住的那半个窗口)。 const del = db.prepare(` DELETE FROM graph_run_nodes WHERE id = ? AND status = 'pending' AND attempt = 0 AND agent_run_id IS NULL AND child_run_id IS NULL AND output_ref IS NULL `); for (const r of rows) { if (defNodeIds.has(r.def_node_id)) continue; if (!isUntouchedNode(r) || !del.run(r.id).changes) { kept.push(r.def_node_id); continue; } removed.push(r.def_node_id); } })(); if (kept.length) { // 静默不可接受:这一条意味着「定义删掉的节点上有执行痕迹」,属于上面 ponytail 说的升级信号 console.warn(`[work-graph-db] syncRunNodesToDef run=${runId} 保留了带执行痕迹的孤儿节点: ${kept.join(', ')}`); } return { added, removed, kept }; } /** * 取一次 run 的全量状态:run 元信息 + 每个节点(定义 + 运行时合并)+ 边(定义层原样)。 * §四.B / T6 的 `GET /api/graph-runs/:id` 直接吐这个形状。 * * @returns {{run, nodes: Array, edges: Array} | null} */ function getRunState(runId) { const db = getDb(); const run = getRun(runId); if (!run) return null; const def = readDefGraph(run.graph_kind, run.graph_id); const defById = new Map(def.nodes.map((n) => [n.defNodeId, n])); const rows = db .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? ORDER BY created_at, id`) .all(runId); return { run, nodes: rows.map((r) => { const d = defById.get(r.def_node_id) || {}; return { id: r.id, defNodeId: r.def_node_id, name: d.name || null, systemAgentId: d.systemAgentId || null, nodeKind: d.nodeKind || null, status: r.status, attempt: r.attempt, agentRunId: r.agent_run_id, childRunId: r.child_run_id, outputRef: r.output_ref, failReason: r.fail_reason, // R-17.1 的两列同样要读回来(控制台 §四.B.3 用它们表达层级语义): // failNodeRef = 归因指向的跨层地址;childRunAttempt = 这张子 run 服务的是哪一轮 failNodeRef: r.fail_node_ref, childRunAttempt: r.child_run_attempt, // 落库了就必须读回来:字段在「产生 → 传递 → 落库 → 读回」四环里漏掉任何一环 // 都等于没有这个字段(P3 前几轮反复踩的就是这个) remediationNote: r.remediation_note, startedAt: r.started_at, completedAt: r.completed_at, }; }), edges: def.edges, }; } /** 某次 run 的运行时节点(按定义节点 id 索引),供引擎选边时用 */ function getRunNodeByDefId(runId, defNodeId) { return ( getDb() .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND def_node_id = ?`) .get(runId, String(defNodeId)) || null ); } function getRunNode(nodeRunId) { return getDb().prepare(`SELECT * FROM graph_run_nodes WHERE id = ?`).get(nodeRunId) || null; } /** 允许被条件更新一并写入的运行时列(白名单,避免拼 SQL 时把任意键塞进来) */ const NODE_PATCH_COLUMNS = { attempt: 'attempt', agentRunId: 'agent_run_id', childRunId: 'child_run_id', outputRef: 'output_ref', failReason: 'fail_reason', // 跟 fail_reason 走**同一次**条件更新:地址与根因是同一份归因产出的两半, // 分成两次写就会出现「有 fail_reason 没地址」的中间态,回退恰好读在那一刻就退化了 failNodeRef: 'fail_node_ref', childRunAttempt: 'child_run_attempt', // 跟着状态转换一起落库:写在这里而不是另开一个 setter,是为了让「改状态 + 写判词」 // 共用 updateNodeStatus 那一次条件更新,不出现「状态改了但判词没落」的第二条路径 remediationNote: 'remediation_note', startedAt: 'started_at', completedAt: 'completed_at', }; /** * 条件更新节点状态(§四.A.6 幂等的落地点)。 * * UPDATE ... WHERE id = ? AND status = ,返回受影响行数: * 1 → 本次转换是我做成的 * 0 → 已经被别人(自动驱动器 / 人工点击)拿走了,调用方原样返回当前状态即可,不报错 * 不加分布式锁——SQLite 行级事务 + 条件更新就够。 * * from 缺省时从当前行读,此时不具备幂等语义(仅用于确定单线程的场景)。 * 转换本身先过 NODE_TRANSITIONS,非法直接抛 InvalidTransitionError。 * * @param {string} nodeRunId * @param {string|null} from 期望的当前状态 * @param {string} to * @param {object} [patch] NODE_PATCH_COLUMNS 里的键 * @returns {number} 受影响行数(0 或 1) */ function updateNodeStatus(nodeRunId, from, to, patch = {}) { const db = getDb(); const cur = getRunNode(nodeRunId); if (!cur) throw new Error(`graph_run_nodes 不存在: ${nodeRunId}`); const expected = from === null || from === undefined ? cur.status : from; if (!canTransitionNode(expected, to)) { throw new InvalidTransitionError('node', expected, to, nodeRunId); } const sets = ['status = @to', 'updated_at = @now']; const params = { id: nodeRunId, to, from: expected, now: new Date().toISOString() }; for (const [key, col] of Object.entries(NODE_PATCH_COLUMNS)) { if (patch[key] !== undefined) { sets.push(`${col} = @${key}`); params[key] = patch[key]; } } return db .prepare(`UPDATE graph_run_nodes SET ${sets.join(', ')} WHERE id = @id AND status = @from`) .run(params).changes; } /** attempt + 1(回退时用,§四.D.2.1 第 3 步);返回受影响行数 */ function bumpNodeAttempt(nodeRunId) { return getDb() .prepare(`UPDATE graph_run_nodes SET attempt = attempt + 1, updated_at = ? WHERE id = ?`) .run(new Date().toISOString(), nodeRunId).changes; } const RUN_PATCH_COLUMNS = { pauseRequested: 'pause_requested', stopRequested: 'stop_requested', executionMode: 'execution_mode', budgetJson: 'budget_json', blockReason: 'block_reason', }; /** * 当前占着这家公司执行坑的 root run 数。 * * WHERE 从 ACTIVE_ROOT_RUN_STATUS 拼出来,与准入事务的判据同源。计数只有这一个持有者, * 调度器 / 路由都调它,不许自己再写一遍 SQL —— 批 4 DROP 掉唯一索引之后这条 SQL 就是 * 「坑满没满」的**全部**依据,多一个持有者就多一个能悄悄分叉的地方。 * * @param {string} companyId * @param {object} [db] 传进来是为了让准入的计数与随后的 UPDATE 落在**同一个事务**里 */ function countActiveRootRuns(companyId, db = getDb()) { return listActiveRootRunIds(companyId, db).length; } /** * 占着坑的是**谁**。`countActiveRootRuns` 由它派生(`.length`),所以那条 WHERE 全仓库 * 仍然只有这一个持有者 —— 为了给 `RunSlotBusyError` 填 heldBy 而另抄一条 SELECT, * 就是给「守卫放行、索引拒绝」那类分叉再开一个口子。活跃集最多几条,取 id 与取 COUNT 同价。 */ function listActiveRootRunIds(companyId, db = getDb()) { const marks = ACTIVE_ROOT_RUN_STATUS.map(() => '?').join(','); return db .prepare(` SELECT id FROM graph_runs WHERE parent_run_id IS NULL AND run_status IN (${marks}) AND company_id = ? ORDER BY created_at `) .all(...ACTIVE_ROOT_RUN_STATUS, companyId) .map((r) => r.id); } /** * 条件更新 run 状态,语义与 updateNodeStatus 完全一致(同样返回受影响行数供幂等判定)。 * * 外加**准入守卫**(27-P3C 批 3):见下方注释。坑满时抛 `RunSlotBusyError`。 * @returns {number} * @throws {InvalidTransitionError|RunSlotBusyError} */ function updateRunStatus(runId, from, to, patch = {}) { const db = getDb(); const cur = getRun(runId); if (!cur) throw new Error(`graph_runs 不存在: ${runId}`); const expected = from === null || from === undefined ? cur.run_status : from; if (!canTransitionRun(expected, to)) { throw new InvalidTransitionError('run', expected, to, runId); } const sets = ['run_status = @to', 'updated_at = @now']; const params = { id: runId, to, from: expected, now: new Date().toISOString() }; for (const [key, col] of Object.entries(RUN_PATCH_COLUMNS)) { if (patch[key] !== undefined) { sets.push(`${col} = @${key}`); params[key] = patch[key]; } } const write = () => db .prepare(`UPDATE graph_runs SET ${sets.join(', ')} WHERE id = @id AND run_status = @from`) .run(params).changes; // ── 准入守卫:数坑 + 占坑必须是一个原子步 ───────────────────────────── // // 判据是「**从活跃集外进入活跃集**」,不是「转成某个具体状态」: // · `ready→running` / `running→paused` / `paused→running` 都是**集内移动**,占的还是 // 原来那个坑。按 to 判就会把这条 run 自己算成竞争者(count 已经包含它), // max=1 时第一条 run 一起跑就把自己挡死 —— 而且是「起不来还不报错」那种挡死。 // · `queued→ready` / `planning→ready` / `blocked→running` / `completed→ready` / // `failed→ready` 才是真的进坑(这几个 from 都不在 ACTIVE_ROOT_RUN_STATUS 里)。 // 只管 root run:那条唯一索引也只约束 `parent_run_id IS NULL`,子 run 从不占公司级的坑。 // // ⚠️ **后三条同时是「恢复」路径**(人加完预算点启动 / 子 run 救回父 run / 对已收敛 run // 里的节点点重试),而批 3 让「坑一空就被队里下一条填上」成了必然事件 —— 于是恢复在 // 有队列的公司里几乎必然撞坑满。本函数**不为恢复开后门**(开了就绕过唯一索引, // 换成一句 SQLite 原文的 500):它只负责干净地拒绝(抛之前一个字都没写), // 「拒绝之后怎么办」由 `graph-engine.admitOrRequeue` 决定 —— 那里会把这条 root run // 退回队列(blocked → queued),让调度器在坑空时按位次把它接回来。 // // 为什么挂在 updateRunStatus 而不是各个调用点:实测有 6 条路径能把 root run 推进活跃集 // (markRunReady / 调度器 queued→ready / startRunSync 恢复 / retryNode / // reviveParentFromChild / nextStep),枚举一次漏一次(§五.1.1 那笔老账)。 // updateRunStatus 是它们共同的唯一漏斗,挂这里一处覆盖全部。 const admitting = !cur.parent_run_id && ACTIVE_ROOT_RUN_STATUS.includes(to) && !ACTIVE_ROOT_RUN_STATUS.includes(expected); if (!admitting) return write(); // 额度在**进事务之前**解析:它要读一次盘(company.json)+ JSON.parse,塞进 BEGIN IMMEDIATE // 就是拿着 SQLite 写锁去做 syscall,而且文件坏掉时会在事务里抛。 // 这一步不需要在事务里:需要原子的是「数坑 + 占坑」,不是「读配置」—— // 这一刻读到的额度即使下一毫秒被人改了,最多影响这一次调度,不影响任何不变量。 const max = maxActiveRootRuns(cur.company_id); // BEGIN IMMEDIATE:**读之前**就拿到写锁,于是「数出还有空位」到「自己占上」之间 // 不可能有第二个连接插进来各自数出还有空位(check-then-act 的经典超发)。 // 同进程内 better-sqlite3 是同步单写者本来就串行,这把锁防的是多进程 / 多连接。 // 批 4 起这是**唯一**的守卫:`idx_graph_runs_one_active_root` 那道 DB 兜底已经 DROP // (理由与四步论证见 getDb 迁移块),所以下面这四行今天就得是对的,没有第二道网接着。 return db.transaction(() => { // 事务内重读才是权威的。状态已经不是 expected → 这次调用本来就是幂等 no-op // (下面 UPDATE 的 WHERE 同样匹配不到),返回 0。**绝不能**当成坑满抛错: // 那是把「别人抢先做完了」误报成「容量不足」,两件事的处置完全不同。 const fresh = db.prepare(`SELECT run_status FROM graph_runs WHERE id = ?`).get(runId); if (!fresh || fresh.run_status !== expected) return 0; const heldBy = listActiveRootRunIds(cur.company_id, db); if (heldBy.length >= max) { throw new RunSlotBusyError(cur.company_id, runId, heldBy, max); } return write(); }).immediate(); } /** 只改软信号,不动 run_status(「暂停」/「停止」是下一个节点边界才生效的软信号) */ function setRunFlags(runId, { pauseRequested, stopRequested } = {}) { const sets = ['updated_at = @now']; const params = { id: runId, now: new Date().toISOString() }; if (pauseRequested !== undefined) { sets.push('pause_requested = @pause'); params.pause = pauseRequested ? 1 : 0; } if (stopRequested !== undefined) { sets.push('stop_requested = @stop'); params.stop = stopRequested ? 1 : 0; } return getDb() .prepare(`UPDATE graph_runs SET ${sets.join(', ')} WHERE id = @id`) .run(params).changes; } /** run 级回退总数 +1(§四.D.2.1 双层硬顶的外层),返回加完之后的值 */ function bumpBackwardCount(runId) { const db = getDb(); db.prepare(`UPDATE graph_runs SET backward_count = backward_count + 1, updated_at = ? WHERE id = ?`) .run(new Date().toISOString(), runId); const row = db.prepare(`SELECT backward_count FROM graph_runs WHERE id = ?`).get(runId); return row ? row.backward_count : null; } /** 某个 run 的直接子 run(级联暂停/停止用) */ function listChildRuns(runId) { return getDb().prepare(`SELECT * FROM graph_runs WHERE parent_run_id = ? ORDER BY created_at`).all(runId); } /** 同一棵 run 树的全部 run(按 root_run_id,预算扣减与三层级联都按这个口径) */ function listRunTree(rootRunId) { return getDb() .prepare(`SELECT * FROM graph_runs WHERE root_run_id = ? ORDER BY created_at`) .all(rootRunId); } /** 一家公司的 run 列表(T6 的 GET /api/company/:cid/graph-runs) */ function listCompanyRuns(companyId, { rootOnly = false } = {}) { return getDb() .prepare(` SELECT * FROM graph_runs WHERE company_id = ? ${rootOnly ? 'AND parent_run_id IS NULL' : ''} ORDER BY created_at DESC `) .all(companyId); } // ──────────────────────────────────────────────────────────────── // 27-P3C goal / task 两级模型(§四.1 / §四.2 / §四.4 / §四.5) // // 本节只管两张新表的存取。排队(run_status='queued' / 调度器 / depends_on 的出队判据) // 是另一批的事,这里一个字都不碰 —— depends_on 在本批只被**校验**,不被消费。 // ──────────────────────────────────────────────────────────────── /** * 改了已冻结的字段(§四.4)。路由层 `company.js:sendDomainError` 用 `instanceof` 翻成 409 + hint。 * * 为什么必须拒绝、而不是「顺手把 artifact 也更新一下」:task 的 description 在入队时就写进了 * runs/{runId}/artifacts/task-goal.json,describeTask 等 agent 侧链路读的是 artifact。 * 只改库不改 artifact,两份记录就对不上;两份都改则等于**篡改已被 agent 消费过的审计输入**。 * 要改就取消后新建 —— 这句话直接进 hint。 */ class FrozenFieldError extends Error { constructor(scope, id, fields) { super(`${scope} ${id} 已下达,${fields.join(' / ')} 不可再改(要改就取消后新建)`); this.name = 'FrozenFieldError'; this.scope = scope; // 'goal' | 'task' this.id = id; this.fields = fields; } } /** depends_on 不合法(§四.2)。路由层翻 400。校验在插入之前跑完,库里不留半条脏数据。 */ class DependsOnError extends Error { constructor(message, detail = {}) { super(message); this.name = 'DependsOnError'; Object.assign(this, detail); } } /** depends_on 列存的是 JSON 文本;解不出来当空数组 —— 一条脏行不该让整张 goal 的读取炸掉 */ function parseDependsOn(raw) { try { const v = JSON.parse(raw || '[]'); return Array.isArray(v) ? v.map(String) : []; } catch { return []; } } /** 允许被 patch 的 goal 列(白名单,照抄 RUN_PATCH_COLUMNS 那套,不让任意键拼进 SQL) */ const GOAL_PATCH_COLUMNS = { title: 'title', description: 'description', status: 'status', priority: 'priority', okrNodeId: 'okr_node_id', reportTo: 'report_to', planningTokens: 'planning_tokens', validationTokens: 'validation_tokens', verdict: 'verdict', planningError: 'planning_error', dispatchedAt: 'dispatched_at', archivedAt: 'archived_at', }; /** * 建一个 goal。产出一律是 draft —— 入队是另一个动作(§六.2 的 dispatch),建行本身没有副作用。 * * @returns {object} 刚插入的整行(snake_case,与 getRun 同口径) */ function createGoal({ companyId, title, description, priority = 'P1', okrNodeId = null, createdBy = null, reportTo = null, } = {}) { if (!companyId) throw new Error('createGoal: companyId 必填'); if (!title) throw new Error('createGoal: title 必填'); // description 就是验收边界:空描述等于没有可验收的东西。让它在建行时炸, // 好过让 goal 级 org-validator 拿着空字符串去判「达成了没有」——那一头判不出错,只会瞎判。 if (!description) throw new Error('createGoal: description 必填'); const now = new Date().toISOString(); const ins = getDb().prepare(` INSERT INTO goals (id, company_id, title, description, status, priority, okr_node_id, created_by, report_to, created_at, updated_at) VALUES (?, ?, ?, ?, 'draft', ?, ?, ?, ?, ?, ?) `); const id = insertWithFreshId('goals', genGoalId, (gid) => ins.run(gid, String(companyId), String(title), String(description), priority, okrNodeId, createdBy, reportTo, now, now)); return getGoal(id); } /** 读一个 goal;不存在返回 null */ function getGoal(goalId) { return getDb().prepare(`SELECT * FROM goals WHERE id = ?`).get(goalId) || null; } /** * 一家公司的 goal 列表。 * includeArchived 默认 false:归档只是隐藏(行还在,随时能查回来),默认列表不该看见它。 */ function listCompanyGoals(companyId, { status = null, includeArchived = false, limit = null } = {}) { const where = ['company_id = @companyId']; const params = { companyId }; if (status) { where.push('status = @status'); params.status = status; } if (!includeArchived) where.push('archived_at IS NULL'); const lim = Number.isInteger(limit) && limit > 0 ? ` LIMIT ${limit}` : ''; return getDb() .prepare(`SELECT * FROM goals WHERE ${where.join(' AND ')} ORDER BY created_at DESC${lim}`) .all(params); } /** * 改 goal(§四.4 描述冻结的落地点之一)。 * * dispatched_at 非空(首个 task 已入队)后带 description 一律抛 FrozenFieldError, * **只看键在不在、不看值一不一样**:「值没变就放行」会让调用方以为改描述有时候可以, * 那是最难排查的一种半生效。title / priority / okr_node_id 留着可改 —— 它们只是列表上的标签, * 不进任何 agent 的上下文。 * * @param {string} goalId * @param {object} patch GOAL_PATCH_COLUMNS 里的键 * @returns {object|null} 更新后的整行;goal 不存在返回 null(由调用方翻 404) * @throws {FrozenFieldError} 描述已冻结 */ function updateGoal(goalId, patch = {}) { const cur = getGoal(goalId); if (!cur) return null; if (patch.description !== undefined && cur.dispatched_at) { throw new FrozenFieldError('goal', goalId, ['description']); } const sets = ['updated_at = @now']; const params = { id: goalId, now: new Date().toISOString() }; for (const [key, col] of Object.entries(GOAL_PATCH_COLUMNS)) { if (patch[key] !== undefined) { sets.push(`${col} = @${key}`); params[key] = patch[key]; } } getDb().prepare(`UPDATE goals SET ${sets.join(', ')} WHERE id = @id`).run(params); return getGoal(goalId); } /** * 校验一条 task 的 depends_on(§四.2:只能引用**同 goal 内已存在的** task,有环报错)。 * * 为什么限定同 goal:depends_on 表达的是一个需求内部的先后。跨 goal 依赖会让出队判据 * 跨越另一个 goal 的生命周期(上游 goal 归档了,下游永远等不到),而「静默躺在队里、 * 界面上跟正常排队一模一样」正是这套模型要消灭的症状。 * * 环检测在**写入之前**跑完并抛错,库里不留半条脏数据。判据是「把这条 task 的依赖换成新值后, * 从它出发能不能走回自己」—— 只查这一条的可达性,不做全图 DFS:其余边在此之前每次写入 * 都过了这一关、已经是无环的,新引入的环必然经过被改的这条。 * * @param {string} goalId * @param {string|null} taskId 正在改的 task;新建时传 null * @param {any} dependsOn * @returns {string[]} 归一化后的依赖(去重、保序) * @throws {DependsOnError} */ function validateDependsOn(goalId, taskId, dependsOn) { if (dependsOn === undefined || dependsOn === null) return []; if (!Array.isArray(dependsOn)) { throw new DependsOnError('depends_on 必须是数组', { goalId, taskId, dependsOn }); } const deps = [...new Set(dependsOn.map(String))]; if (taskId && deps.includes(taskId)) { throw new DependsOnError(`depends_on 不能依赖自己: ${taskId}`, { goalId, taskId }); } const rows = getDb().prepare(`SELECT id, depends_on FROM tasks WHERE goal_id = ?`).all(goalId); const known = new Map(rows.map((r) => [r.id, parseDependsOn(r.depends_on)])); const missing = deps.filter((d) => !known.has(d)); if (missing.length) { throw new DependsOnError( `depends_on 只能引用同 goal 内已存在的 task,这些不是: ${missing.join(', ')}`, { goalId, taskId, missing } ); } // 新建时这条还不在库里,用一个不可能与 id 撞的占位名当起点 const start = taskId || '__pending__'; known.set(start, deps); const seen = new Set(); const stack = [...deps]; while (stack.length) { const cur = stack.pop(); if (cur === start) { throw new DependsOnError(`depends_on 形成环: ${start} → … → ${start}`, { goalId, taskId, dependsOn: deps }); } if (seen.has(cur)) continue; seen.add(cur); stack.push(...(known.get(cur) || [])); } return deps; } /** * 在一个 goal 下建一条 task。同样一律是 draft,入队是另一个动作。 * * @param {object} p * @param {string} p.goalId * @param {string} [p.companyId] 缺省从 goal 抄 * @param {number} [p.seq] 缺省取该 goal 内当前最大 seq + 1 * @param {string[]} [p.dependsOn=[]] 过 validateDependsOn * @param {object|string|null} [p.budgetJson] 这条 task 下达时要用的预算;对象会被序列化。 * **必须能在建行时就存下来**(缺陷 F3):草稿路径(`POST /:cid/tasks` 撞 409 → 存草稿 → * 之后 dispatch)上,请求体里的 budget 早就不在了。只有写侧没有读侧的列 = 用户填了、 * 接口回了 200、run 却拿着 DEFAULT_TASK_BUDGET 在跑,中途预算耗尽 blocked 且没人知道为什么。 * @returns {object} 刚插入的整行 */ function createTask({ goalId, companyId = null, seq = null, title, description, executionMode = 'manual', dependsOn = [], priority = 'P1', budgetJson = null, reportTo = null, createdBy = null, } = {}) { if (!goalId) throw new Error('createTask: goalId 必填'); if (!title) throw new Error('createTask: title 必填'); if (!description) throw new Error('createTask: description 必填'); const db = getDb(); const goal = getGoal(goalId); if (!goal) throw new Error(`createTask: goal 不存在: ${goalId}`); // company_id 这一列冗余存在的**唯一**理由是让调度器不必 JOIN;让调用方各传各的, // 就等于给这份冗余留了一条能对不上的路。默认从 goal 抄,传了也以 goal 为准的那种「以防万一」 // 反而会掩盖调用方的 bug,所以显式传值时照传值走,由调用方负责。 const cid = companyId || goal.company_id; const deps = validateDependsOn(goalId, null, dependsOn); const now = new Date().toISOString(); // ponytail: seq 用 MAX+1,并发建行时可能撞。撞了只是列表里两行的先后不定 // (seq 只用于排序,不参与任何判据),不值得为它上一把锁; // 真要严格连号就得把 seq 挪进 goals 上的计数器列,届时在同一个事务里自增。 const nextSeq = Number.isInteger(seq) ? seq : db.prepare(`SELECT COALESCE(MAX(seq), 0) + 1 AS n FROM tasks WHERE goal_id = ?`).get(goalId).n; // 调用方给对象还是给已序列化的文本都收:库里这一列是 TEXT,两种形态在这里归一, // 免得各调用方各 JSON.stringify 一遍(漏一个就是把 "[object Object]" 存进库)。 const budget = budgetJson == null || budgetJson === '' ? null : (typeof budgetJson === 'string' ? budgetJson : JSON.stringify(budgetJson)); const ins = db.prepare(` INSERT INTO tasks (id, goal_id, company_id, seq, title, description, status, execution_mode, depends_on, priority, budget_json, report_to, created_by, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, 'draft', ?, ?, ?, ?, ?, ?, ?, ?) `); const id = insertWithFreshId('tasks', genTaskId, (tid) => ins.run( tid, goalId, String(cid), nextSeq, String(title), String(description), executionMode, JSON.stringify(deps), priority, budget, // report_to 不落库就等于没有:请求作用域的闭包活不过一次排队(§四.7 禁令二)。 // goal 上有就继承下来,别让每个调用方各记一遍。 reportTo || goal.report_to || null, createdBy, now, now )); return getTask(id); } /** 读一条 task;不存在返回 null */ function getTask(taskId) { return getDb().prepare(`SELECT * FROM tasks WHERE id = ?`).get(taskId) || null; } /** 一个 goal 下的全部 task,按 seq 排(列表页的两级树就是这个顺序) */ function listGoalTasks(goalId, { includeArchived = false } = {}) { const where = ['goal_id = @goalId']; if (!includeArchived) where.push('archived_at IS NULL'); return getDb() .prepare(`SELECT * FROM tasks WHERE ${where.join(' AND ')} ORDER BY seq, created_at`) .all({ goalId }); } /** * 一家公司的 task 列表。 * * 排序**故意**只按 created_at:队列位次的口径(priority → seq → created_at)只留调度器一处, * 这里再写一份就是第二个持有者 —— canDispatch 的取值域被前端复述了一遍、还多写一个值, * 导致公司失败一次后再也下不了任务,就是这么来的。要位次找调度器要。 */ function listCompanyTasks(companyId, { status = null, goalId = null, includeArchived = false, limit = null } = {}) { const where = ['company_id = @companyId']; const params = { companyId }; if (status) { where.push('status = @status'); params.status = status; } if (goalId) { where.push('goal_id = @goalId'); params.goalId = goalId; } if (!includeArchived) where.push('archived_at IS NULL'); const lim = Number.isInteger(limit) && limit > 0 ? ` LIMIT ${limit}` : ''; return getDb() .prepare(`SELECT * FROM tasks WHERE ${where.join(' AND ')} ORDER BY created_at DESC${lim}`) .all(params); } /** 允许被 patch 的 task 列。depends_on 不在这里 —— 它要先过环检测,单独处理。 */ const TASK_PATCH_COLUMNS = { title: 'title', description: 'description', status: 'status', priority: 'priority', seq: 'seq', executionMode: 'execution_mode', budgetJson: 'budget_json', currentRunId: 'current_run_id', attempt: 'attempt', assigneeAgentId: 'assignee_agent_id', reportTo: 'report_to', blockedReason: 'blocked_reason', queuedAt: 'queued_at', startedAt: 'started_at', finishedAt: 'finished_at', archivedAt: 'archived_at', summary: 'summary', executionConvId: 'execution_conv_id', }; /** * 还能改 depends_on 的 task 状态(见 updateTask 的闸门 2)。 * `blocked` 在里面是「强制放行」那条自救路径的**全部依赖**:改完依赖 route 会立刻 tick 一次, * `requeueDerivedBlockedTask` 把它连同 blocked_reason 一起推回 queued。 */ const DEPENDS_ON_EDITABLE_STATUS = ['draft', 'queued', 'blocked']; /** queued_at 非空后就不许再改的字段(§四.4):它们都已经被 task-goal.json 或预算池消费过 */ const TASK_FROZEN_FIELDS = ['description', 'executionMode', 'budgetJson']; /** * 改 task。 * * 两道闸门: * 1. queued_at 非空 → description / execution_mode / budget 一律拒(理由见 FrozenFieldError); * 2. depends_on **在 draft / queued / blocked 都允许改**(§六.3 定案)。真实场景是「B 已经 * 排上了,才发现它得等 C」,此前只能取消重建、白丢排队位次。位次不受影响 —— 位次按 * priority → seq → created_at 排,depends_on 决定的是「够不够格出队」,两者正交。 * `blocked` 必须在里面:它是每轮 tick 从依赖**推导**出来的结果、不是不可逆状态,而 * 「强制放行」(删掉那条已经死了的上游)正是要在 blocked 上改 depends_on。挡掉它等于 * 界面上那个按钮必然 4xx —— 一条 blocked 任务再也没有自救的出路(批 3 对抗复核 F1)。 * 跑起来之后就不许改了:那时依赖判据已经用过,再改只会让 timeline 上的因果对不上。 * * @param {string} taskId * @param {object} patch TASK_PATCH_COLUMNS 里的键,外加 dependsOn * @returns {object|null} 更新后的整行;task 不存在返回 null * @throws {FrozenFieldError|DependsOnError} */ function updateTask(taskId, patch = {}) { const cur = getTask(taskId); if (!cur) return null; if (cur.queued_at) { const frozen = TASK_FROZEN_FIELDS.filter((f) => patch[f] !== undefined); if (frozen.length) throw new FrozenFieldError('task', taskId, frozen); } const sets = ['updated_at = @now']; const params = { id: taskId, now: new Date().toISOString() }; if (patch.dependsOn !== undefined) { if (!DEPENDS_ON_EDITABLE_STATUS.includes(cur.status)) { throw new FrozenFieldError('task', taskId, ['dependsOn']); } params.dependsOn = JSON.stringify(validateDependsOn(cur.goal_id, taskId, patch.dependsOn)); sets.push('depends_on = @dependsOn'); } for (const [key, col] of Object.entries(TASK_PATCH_COLUMNS)) { if (patch[key] !== undefined) { sets.push(`${col} = @${key}`); params[key] = patch[key]; } } getDb().prepare(`UPDATE tasks SET ${sets.join(', ')} WHERE id = @id`).run(params); return getTask(taskId); } /** * task 的终态:到了就不再被 run 收敛回写覆盖(重跑要走显式的 rerun,那会先把 status 推回 queued)。 * * **`blocked` 不在里面**(27-P3C 第二轮复核 缺陷 1)。它的语义是「挂起,等人介入」, * 和 completed / failed / cancelled 不同类:预算耗尽走 `graph-engine.suspendRunTree` * (那段注释自己写着「不是失败」),run 转 blocked 后 `RUN_TRANSITIONS.blocked = ['running','stopped']` * 允许人加完预算后 `POST /api/graph-runs/:id/start` 接着跑,`reviveParentFromChild` 甚至会**自动** * 把 blocked 的父 run 拉回 running。把它算成终态的后果是下面 `settleTaskFromRun` 的幂等闸门 * 反过来变成死锁:run 真跑完转 completed 时 UPDATE 命中 0 行,于是 run=completed 而 * task 永远停在 blocked —— 静默、且没有任何一条路径能再把它改回来。 */ const TASK_TERMINAL_STATUS = ['completed', 'failed', 'cancelled']; /** goal 的终态:同上,`archived` 也算 —— 归档过的 goal 不该被一条迟到的 run 收敛翻回 done/failed */ const GOAL_TERMINAL_STATUS = ['done', 'failed', 'cancelled', 'archived']; /** * 人类任务被 graph-engine 接管跑图时,`assignee_agent_id` 没有真实派单对象可写—— * 占位成这个哨兵值,让 COO 心跳巡检的「to 是否为 null」判据不再把「有图在跑」误判成「没人认领」 * (T-mt5dkvjm)。`enqueueTaskRun` 入队时写一次,这里(run 收敛)做兜底 COALESCE: * 只在该列仍是 NULL 时才补,不覆盖派单式任务(agent-chat 的 dispatch 流程)已经写好的真实值。 */ const GRAPH_TAKEOVER_ASSIGNEE = 'org-graph'; /** * run 收敛 → task 回写的**条件更新**(27-P3C §五.1.1 的钩子用,缺陷 F1)。 * * 两个 WHERE 条件各挡一类事故,缺一不可: * 1. `current_run_id = @runId` —— 挡「上一次 attempt 的迟到收敛」。task 重跑后 * current_run_id 已指向新 run,旧 run 此刻才转 stopped 的话,不加这条就会把 * 正在跑的第 2 次执行标成 cancelled,而界面上看不出任何异常。 * 2. `status NOT IN (终态)` —— 幂等闸门。钩子挂在状态转换的尾巴上, * 同一条 run 被重开(`completed → ready` 是合法转换)再收敛会二次触发。 * * @returns {number} 受影响行数;0 = 这次不该由我写(已终态 / 不是当前那次执行),调用方据此不做后续副作用 */ function settleTaskFromRun(taskId, runId, status) { const now = new Date().toISOString(); return getDb() .prepare(` UPDATE tasks SET status = @status, blocked_reason = NULL, finished_at = @now, updated_at = @now, assignee_agent_id = COALESCE(assignee_agent_id, '${GRAPH_TAKEOVER_ASSIGNEE}') WHERE id = @id AND current_run_id = @runId AND status NOT IN (${TASK_TERMINAL_STATUS.map((s) => `'${s}'`).join(', ')}) `) .run({ id: taskId, runId, status, now }).changes; } /** * 「上游已经不可能满足了」这个 blocked 的原因前缀(27-P3C §五.1.2)。 * * 它同时是**判据**:只有 `blocked_reason` 以它开头的 blocked task 才是调度器**推导**出来的, * 才允许被下一次 tick 推回 queued。别的来源的 blocked(最典型的是预算耗尽 —— run 转 blocked、 * task-settle 跟着把 task 写成 blocked)绝不能被自动重排:那条 run 还躺在 blocked 等人加预算, * 把 task 推回 queued 只会让调度器反复去 CAS 一条根本不是 queued 的 run,界面上任务在 * queued / blocked 之间抖动,而没有任何人真的在跑。 */ const UPSTREAM_BLOCK_PREFIX = 'upstream_failed:'; /** * queued → blocked(依赖已不可能满足)。条件更新:只动**还在排队**的行。 * 正在跑 / 已终态的任务不该被依赖推导碰到 —— 它们的状态由 run 收敛回写说了算。 * @returns {number} 1 = 本次由我标的 */ function blockQueuedTask(taskId, reason) { const now = new Date().toISOString(); return getDb() .prepare(` UPDATE tasks SET status = 'blocked', blocked_reason = @reason, updated_at = @now WHERE id = @id AND status = 'queued' `) .run({ id: taskId, reason, now }).changes; } /** * blocked → queued(上游恢复了)。**这是「blocked 不落成不可逆状态」的兑现处**: * §五.1.2 承诺「blocked 每次 tick 重算一遍,上游恢复了就自己回 queued」,没有这个函数 * 那句注释就是空头支票 —— A 重跑成功后 B 会永远躺在 blocked。 * * 两个 WHERE 条件缺一不可:`status='blocked'` 是幂等闸门,`blocked_reason LIKE 'upstream_failed:%'` * 把「调度器推导的」和「别的来源的 blocked」分开(见 UPSTREAM_BLOCK_PREFIX)。 * 写进去的 'queued' / NULL 都不在 WHERE 里。 * @returns {number} 1 = 本次由我放回队列的 */ function requeueDerivedBlockedTask(taskId) { const now = new Date().toISOString(); return getDb() .prepare(` UPDATE tasks SET status = 'queued', blocked_reason = NULL, updated_at = @now WHERE id = @id AND status = 'blocked' AND blocked_reason LIKE @prefix `) .run({ id: taskId, now, prefix: `${UPSTREAM_BLOCK_PREFIX}%` }).changes; } /** * 出队认领:queued → running(27-P3C §五.1 第 5 步)。 * * 必须是条件更新,不能用 `updateTask`(它的 WHERE 只有 `id`,是彻头彻尾的 check-then-act): * 两个 tick / 两个进程同时看到同一条 queued task 时,无条件 UPDATE 会让两边都以为自己赢了, * 各起一条 run。 * * 三个 WHERE 条件: * 1. `status = 'queued'` —— 真正的闸门。**写进去的 'running' 绝不能出现在这里**: * SQLite 的 changes 数的是「匹配到的行」,把目标值也写进 WHERE 会让已被别人认领的行 * 照样匹配,changes 恒为 1,闸门等于没有。 * 2. `current_run_id = @runId` —— 认的必须是**我刚出队的那条 run**。少了它, * 一次迟到的认领会把另一条 run 的 task 抢过来记账。 * 3. `archived_at IS NULL` —— 归档过的任务不该被调度器捡起来跑。 * * `attempt` 在这里 +1(而不是在调用方读一遍再写回去):读改写在并发下会丢代次, * 而 `attempt` 是 R-17.1 判「子 run 服务的是不是本轮」的依据,丢了会串轮。 * * @returns {number} 1 = 认领成功;0 = 不是我的(被抢走 / 已取消 / 已归档),调用方必须回滚那条 run */ function claimQueuedTask(taskId, runId) { const now = new Date().toISOString(); return getDb() .prepare(` UPDATE tasks SET status = 'running', attempt = attempt + 1, started_at = @now, blocked_reason = NULL, updated_at = @now WHERE id = @id AND current_run_id = @runId AND status = 'queued' AND archived_at IS NULL `) .run({ id: taskId, runId, now }).changes; } /** `requeueBlockedRootRun` 内部用来回滚事务的私有信号,不外泄 */ class RequeueRollback extends Error {} /** * 恢复不成 → **退回队列**:blocked 的 root run + 它的 blocked task 一起回到 queued。 * * 为什么需要它(27-P3C 批 3 复核 F1):`updateRunStatus` 的准入守卫会拒掉一切「从活跃集外 * 进活跃集」的转换,而恢复(人加完预算点启动 / 重试)恰恰全是这个形状。批 3 之后坑一空 * 就被队里下一条填上,于是恢复几乎必然撞坑满。只是干净地报 409 并不够 —— 那条 run 留在 * blocked(**非终态**),调度器只看 queued 的 task,人再点也是一样的赛跑,结果就是这条 * 任务永远回不来、它的 goal 永不收敛。放回队列才是真的把它交还给调度器。 * * **两行必须在同一个事务里**,顺序无所谓、原子性有所谓: * · 只有 run 回 queued → 队列按 task 选候选,没人会来出队它 → 换了个姿势永远卡住; * · 只有 task 回 queued → 调度器每轮都去 CAS 一条不是 queued 的 run,界面上任务在 * queued/blocked 之间抖动而没人真的在跑(`UPSTREAM_BLOCK_PREFIX` 那段注释描述的正是它)。 * * 三个前提缺一不可:root run(子 run 不占坑也不进队列)、run 与 task 都还在 blocked * (别的状态说明有人已经处置过了)、task 未归档。写进去的 'queued' 都不在 WHERE 里。 * * ⚠️ 它**不走** `sse.transitionRun`:那里的 apply 只能改 run 一张表,包不住 task 那一行。 * 所以调用方(graph-engine.admitOrRequeue)必须自己补一条 timeline/SSE 事件 —— * 状态变了却没有事件,正是这套代码一直在消灭的东西。 * * @returns {number} 1 = 本次由我放回队列;0 = 前提不成立(调用方据此照抛 RunSlotBusyError) */ function requeueBlockedRootRun(runId) { const db = getDb(); const run = getRun(runId); if (!run || run.parent_run_id || run.run_status !== 'blocked' || !run.task_id) return 0; const now = new Date().toISOString(); try { return db.transaction(() => { const t = db .prepare(` UPDATE tasks SET status = 'queued', blocked_reason = NULL, started_at = NULL, updated_at = @now WHERE id = @taskId AND current_run_id = @runId AND status = 'blocked' AND archived_at IS NULL `) .run({ taskId: run.task_id, runId, now }).changes; if (!t) return 0; const r = db .prepare(` UPDATE graph_runs SET run_status = 'queued', updated_at = @now WHERE id = @runId AND run_status = 'blocked' AND parent_run_id IS NULL `) .run({ runId, now }).changes; if (!r) throw new RequeueRollback(); // 别人抢先动了 run:整个事务回滚,task 不许独自留在 queued return 1; }).immediate(); } catch (err) { if (err instanceof RequeueRollback) return 0; throw err; } } /** * 把「为什么还卡着」写到 task 行上,供任务列表直接显示(27-P3C 批 3 复核 F1 场景 C)。 * * 用途是**留痕**,不是状态机:只动仍然 blocked、且 current_run_id 指着这条 run 的行, * 于是一条迟到的备注不会盖掉别人的处置结果。写进去的 reason 不在 WHERE 里。 * * ⚠️ reason **不许**带 `UPSTREAM_BLOCK_PREFIX`:带了就会被调度器当成「依赖推导出来的 * blocked」自动推回 queued,而这类 blocked(预算耗尽 / 救援失败)本来就等着人来处置。 * @returns {number} 1 = 记上了 */ function noteTaskBlockedReason(taskId, runId, reason) { if (!taskId) return 0; return getDb() .prepare(` UPDATE tasks SET blocked_reason = @reason, updated_at = @now WHERE id = @id AND current_run_id = @runId AND status = 'blocked' `) .run({ id: taskId, runId, reason, now: new Date().toISOString() }).changes; } /** * goal 收敛的 compare-and-swap 闸门(缺陷 F2)。 * * 为什么必须是条件更新而不是「读一下再写」:一个 goal 下的 N 个 task 可能几乎同时收敛, * 每一条都会算出「全部 task 都到终态了」并试图收敛 goal。goal 收敛带着两个**不可重放**的 * 副作用(写 OKR 节点、POST 唤醒上级 agent),做两遍就是同一个节点被写两次、上级被叫醒两次。 * 让 SQLite 的单写者语义来决出唯一的赢家:changes = 1 的那一个才有资格做副作用。 * * @returns {number} 1 = 本次抢到收敛权;0 = 别人已经收敛过(或 goal 不存在) */ function settleGoalStatus(goalId, status) { const now = new Date().toISOString(); return getDb() .prepare(` UPDATE goals SET status = @status, updated_at = @now WHERE id = @id AND status NOT IN (${GOAL_TERMINAL_STATUS.map((s) => `'${s}'`).join(', ')}) `) .run({ id: goalId, status, now }).changes; } /** `claimGoalSlot` / `settleGoalVerdict` 允许操作的列(白名单,绝不让调用方把列名拼进 SQL) */ const GOAL_SLOT_COLUMNS = ['verdict', 'reported_at']; /** * 认领 goal 上的一个「占位 → 唯一 token」闸门(第二轮复核:认领语句不能自己构成它要避开的形态)。 * * 这是 `verdict` / `reported_at` 两格共用的原语:**旧值必须精确匹配 `from`,新值 `to` 全局唯一**。 * 两个条件缺一不可 —— * · `from` 不精确(比如写成 `verdict IN (占位A, 占位B)`)而 `to` 恰好也是 `占位B` 时, * 已经被别人认领的行照样匹配,`changes` 恒为 1,**闸门等于没有**:两个进程各拿一个 1, * 各起一次 org-validator、各写一次 OKR、各叫醒一次上级。SQLite 的 changes 数的是 * 「匹配到的行」,不管值变没变,所以「相同值写回」永远是 1。 * · `to` 不唯一时,即使 WHERE 只认上一个态,两个进程也可能一前一后各命中一次 * (A 认领 → A 的值 = B 要找的值 → B 再认领一次)。唯一 token 让「谁认领的」本身可判定, * 后面的回写才能用 `WHERE 列 = 我的 token` 决出赢家。 * * @param {string} goalId * @param {'verdict'|'reported_at'} column * @param {{from: string|null, to: string}} p from=null 表示要求该列当前为 NULL * @returns {number} 1 = 认领成功;0 = 旧值已经不是 from(别人先到,或值已被改写) */ function claimGoalSlot(goalId, column, { from = null, to } = {}) { if (!GOAL_SLOT_COLUMNS.includes(column)) throw new Error(`claimGoalSlot: 不支持的列 ${column}`); if (!to) throw new Error('claimGoalSlot: to 必填(认领值必须唯一,不能为空)'); const cond = from === null ? `${column} IS NULL` : `${column} = @from`; return getDb() .prepare(` UPDATE goals SET ${column} = @to, updated_at = @now WHERE id = @id AND ${cond} `) .run({ id: goalId, from, to, now: new Date().toISOString() }).changes; } /** * goal 级验收结论的**条件回写**(第二轮复核:把 check-then-act 换成一条原子 UPDATE)。 * * 原来是「`getGoal` 读一下判词还是不是我认领的那个 → `updateGoal` 写回去」。这在单进程里也只是 * 侥幸成立,跨进程根本不成立:读和写之间隔着完整的一次 JS 事件循环,另一个实例完全可以在这中间 * 把这条定案掉,于是两边都判「还是我的」,判词互相覆盖、汇报做两次。 * 现在把判据放进 UPDATE 自己的 WHERE 里,靠 `changes` 决出唯一赢家 —— 只有赢家才有资格汇报。 * * `validation_tokens` 用 SQL 侧自增而不是「读出来 +tokens 再写回」:后者拿的是旧快照, * 拆分链路(planning_tokens)在同一行上并发写时会被整行回写抹掉。 * * @returns {number} 1 = 本次回写生效(可以去汇报);0 = 判词已被别处改写(绝不能再汇报) */ function settleGoalVerdict(goalId, { claim, status, verdict, addValidationTokens = 0 } = {}) { if (!claim) throw new Error('settleGoalVerdict: claim 必填(没有认领 token 就没有闸门)'); const add = Number.isFinite(addValidationTokens) ? addValidationTokens : 0; return getDb() .prepare(` UPDATE goals SET status = @status, verdict = @verdict, validation_tokens = validation_tokens + @add, updated_at = @now WHERE id = @id AND verdict = @claim `) .run({ id: goalId, claim, status, verdict, add, now: new Date().toISOString() }).changes; } /** * 拆分收工的 compare-and-swap 闸门(批 2 缺陷 1)。同 `settleGoalStatus` 那招, * 差别是这里的判据是「**还在** planning 且没被归档」而不是「还没到终态」。 * * 为什么不能用 `updateGoal(goalId, { status: 'draft' })`:那是一条无条件 UPDATE。 * 拆分要跑几十秒,这期间 goal 完全可能已经被别人推走 —— * · 用户点了表单上的「跳过,按单任务下达」→ `POST /:cid/tasks` 建 task + 起 run + * `markGoalDispatched` 写成 active; * · 用户在左栏归档了这个 goal(只盖 archived_at,status 仍是 planning)。 * 无条件写回 draft 就是把一个**正在烧钱跑 run 的 goal 标成草稿**,随后插进去的 N 条 draft task * 让 `settleGoalIfAllTasksDone` 永远看不到「全部终态」→ OKR 不写、report_to 不唤醒、 * verdict 恒空,且全程无日志无报错。归档那条同理:插进去的 task 归属一个不可见的 goal, * 左栏(按 goal 挂树)看不到、`GET /:cid/tasks` 里却在,永远不会被归档、下达或发现。 * * **没命中时仍然把 planning_tokens 记上**:那几轮 LLM 是真花了钱的,goal 层的花费本来就只有 * 这一个记账口,不记 = 这半边花费永远看不见。只有 status / planning_error 是被放弃的 —— * planning_error 的语义(界面上是「自动拆分失败,已按单任务处理」)在 goal 已经跑起来时是假话。 * * @param {string} goalId * @param {object} p * @param {string} [p.status='draft'] 命中时要写的状态 * @param {number} [p.planningTokens=0] 本次拆分烧掉的 token;命不命中都记 * @param {string|null} [p.planningError=null] 只在命中时写 * @returns {{settled: boolean}} settled=false 表示 goal 已被别人推离 planning(或归档 / 不存在) */ /** * 只补记验收 token,不碰状态与判词。 * * 用在「跨进程抢跑输了」那条路上:判词被别人定案了,但**这一次 org-validator 是真跑了、 * 钱是真烧了**。不记的话这半边花费永远看不见 —— 与 settleGoalPlanning 那条「没命中也 * 补记账」的分支同一个口径。tokens 为 0 时连 updated_at 都不碰。 */ function bumpValidationTokens(goalId, tokens) { if (!(tokens > 0)) return 0; return getDb() .prepare(`UPDATE goals SET validation_tokens = validation_tokens + @t, updated_at = @now WHERE id = @id`) .run({ id: goalId, t: tokens, now: new Date().toISOString() }).changes; } function settleGoalPlanning(goalId, { status = 'draft', planningTokens = 0, planningError = null } = {}) { const db = getDb(); const now = new Date().toISOString(); const tokens = Number.isFinite(planningTokens) ? planningTokens : 0; const changes = db .prepare(` UPDATE goals SET status = @status, planning_tokens = @tokens, planning_error = @err, updated_at = @now WHERE id = @id AND status = 'planning' AND archived_at IS NULL `) .run({ id: goalId, status, tokens, err: planningError, now }).changes; if (changes) return { settled: true }; // 只补记账。tokens 为 0 时连 updated_at 都不碰:拆分没花钱还去 touch 一个已经在跑的 goal, // 只会让「这行最后一次变动是什么时候」这个问题多一个假答案。 if (tokens > 0) { db.prepare(`UPDATE goals SET planning_tokens = @tokens, updated_at = @now WHERE id = @id`) .run({ id: goalId, tokens, now }); } return { settled: false }; } /** * 一个 goal 内当前最大的 seq(含已归档的 task),没有 task 时返回 0(批 2 缺陷 4)。 * * 拆分产出的 seq 必须以它为基准偏移:`idx_tasks_goal` 不是唯一索引,往非空 goal 里插 1..N * 不会报错,只会让同一个 goal 下出现两条 seq=1 —— 列表两行都显示 `#1`、依赖 chip 的 * 「依赖 #1」出现两个同名候选、dispatch 挑「第一条跑得动的 draft」顺序不定。**从此对不上号。** * * 归档的 task 也要算进来:它们的号还挂在界面上(`includeArchived=1` 随时查得回来), * 复用等于让历史行和新行重号。 */ function maxTaskSeq(goalId) { const row = getDb().prepare(`SELECT COALESCE(MAX(seq), 0) AS n FROM tasks WHERE goal_id = ?`).get(goalId); return (row && row.n) || 0; } /** * 一批 run 的节点完成度,**一条 SQL 出全部**(§五 / 验收 10)。 * * 为什么强制批量:列表页要给每行 task 画进度条,20 goal × 3 task 逐个查就是 60 次往返 —— * 那正是验收 10 点名要挡的 N+1。所以这个函数只接受数组、只返回 Map,不留单条版本的入口: * 留了就一定有人在循环里调它。 * * done 的口径 = success + skipped。skipped 是「按条件跳过」,它和成功一样**不会再动**; * 算进分母不算进分子,会让一张带跳过分支的图永远停在 80%。 * * ponytail: run 数超过 SQLite 的变量上限(默认 999)时要分批。列表页一屏几十条, * 目前够不着;真到那天在这里切 chunk 即可,调用方形态不变。 * * @param {string[]} runIds * @returns {Map} 没有节点行的 run 不出现在 Map 里 */ function taskProgress(runIds) { const ids = [...new Set((runIds || []).filter(Boolean).map(String))]; if (!ids.length) return new Map(); const rows = getDb() .prepare(` SELECT run_id, COUNT(*) AS total, SUM(CASE WHEN status IN ('success', 'skipped') THEN 1 ELSE 0 END) AS done FROM graph_run_nodes WHERE run_id IN (${ids.map(() => '?').join(', ')}) GROUP BY run_id `) .all(...ids); return new Map(rows.map((r) => [r.run_id, { total: r.total, done: r.done }])); } // ──────────────────────────────────────────────────────────────── // 回退记录(§四.D.2.1 的备案账本,判词由 T7 写) // ──────────────────────────────────────────────────────────────── /** 这条边在本 run 上已经退过几次(loop_count = 该值 + 1) */ function countLoopRecords(runId, edgeId) { return getDb() .prepare(`SELECT COUNT(*) AS n FROM graph_run_loop_records WHERE run_id = ? AND edge_id = ?`) .get(runId, String(edgeId)).n; } /** * 备案一次回退(§四.D.2.1 第 1 步)。loop_count 缺省按已有记录数 + 1 自动算。 * @returns {{id, loopCount}} */ function insertLoopRecord({ runId, edgeId, fromNodeRunId, toNodeRunId, loopCount, verdict = 'registered', remediationNote = null, targetNodeRef = null, } = {}) { const db = getDb(); const id = newId('lr'); const n = loopCount === undefined || loopCount === null ? countLoopRecords(runId, edgeId) + 1 : loopCount; db.prepare(` INSERT INTO graph_run_loop_records (id, run_id, edge_id, from_node_run_id, to_node_run_id, loop_count, verdict, remediation_note, target_node_ref, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `).run(id, runId, String(edgeId), fromNodeRunId, toNodeRunId, n, verdict, remediationNote, targetNodeRef, new Date().toISOString()); return { id, loopCount: n }; } /** 改判词(registered → fixed / not_fixed / force_terminated),返回受影响行数 */ function updateLoopVerdict(recordId, verdict, remediationNote) { const sets = ['verdict = @verdict']; const params = { id: recordId, verdict }; if (remediationNote !== undefined) { sets.push('remediation_note = @note'); params.note = remediationNote; } return getDb() .prepare(`UPDATE graph_run_loop_records SET ${sets.join(', ')} WHERE id = @id`) .run(params).changes; } /** 本 run 的回退记录,最新在前(控制台「循环」tab + 下一轮读 remediation_note 用) */ function listLoopRecords(runId, { edgeId = null } = {}) { const db = getDb(); return edgeId === null ? db.prepare(`SELECT * FROM graph_run_loop_records WHERE run_id = ? ORDER BY created_at DESC`).all(runId) : db .prepare(` SELECT * FROM graph_run_loop_records WHERE run_id = ? AND edge_id = ? ORDER BY created_at DESC `) .all(runId, String(edgeId)); } // ──────────────────────────────────────────────────────────────── // R-19 产物索引(27-P4 §4.6) // // 分工与 output_ref 不重复: // output_ref 节点级指针 —— 「我这次跑出了什么」,一对一挂在 graph_run_nodes 行上 // graph_files 公司级索引 —— 「这家公司有哪些产物、谁能用」,聚合视图,由前者派生 // 所以 graph_files 必须可从节点行重建(rebuildFilesFromNodes),索引坏了不至于丢产物。 // ──────────────────────────────────────────────────────────────── /** * 记一条产物实例。**只由引擎调用**,不给 agent 自己写 —— * 否则索引与实际文件必然漂移(本轮反复验证过这一点)。 * * 同一 (company, task, relPath) 重复产出时:把旧行标 superseded_by 指向新行, * 而不是堆一堆同路径的行。这样「最新的那份」是唯一 superseded_by IS NULL 的行, * 历史仍可追溯。 */ function recordFile({ companyId, taskId = null, relPath, kind, runId = null, nodeRunId = null }) { if (!companyId || !relPath || !kind) throw new Error('recordFile 需要 companyId / relPath / kind'); const db = getDb(); const now = new Date().toISOString(); const id = newId('gf'); const tx = db.transaction(() => { const prev = db .prepare(` SELECT id FROM graph_files WHERE company_id = ? AND COALESCE(task_id, '') = ? AND rel_path = ? AND superseded_by IS NULL `) .get(companyId, taskId || '', relPath); if (prev) { // 同路径再产出:先把旧行指向新行(腾出部分唯一索引的位置),再插新行。 // 顺序不能反 —— 反了会撞 idx_graph_files_current。 db.prepare(`UPDATE graph_files SET superseded_by = ? WHERE id = ?`).run(id, prev.id); } db.prepare(` INSERT INTO graph_files (id, company_id, task_id, rel_path, kind, produced_by_run_id, produced_by_node_run_id, created_at, superseded_by) VALUES (@id, @company_id, @task_id, @rel_path, @kind, @run_id, @node_run_id, @now, NULL) `).run({ id, company_id: companyId, task_id: taskId, rel_path: relPath, kind, run_id: runId, node_run_id: nodeRunId, now, }); }); tx(); return id; } /** 按 kind 查这家公司/这个 task 下最新的产物实例(superseded_by IS NULL) */ function findFiles({ companyId, taskId = null, kinds = null }) { const db = getDb(); const rows = db .prepare(` SELECT * FROM graph_files WHERE company_id = ? AND COALESCE(task_id, '') = ? AND superseded_by IS NULL ORDER BY created_at `) .all(companyId, taskId || ''); if (!kinds || !kinds.length) return rows; const want = new Set(kinds); return rows.filter((r) => want.has(r.kind)); } /** 在边上声明「这条边要传什么 kind」。规划时写,只有 kind 没有实例。 */ function declareEdgeContract({ edgeLayer, edgeId, kind, required = true, role = 'produces' }) { if (!['org', 'work'].includes(edgeLayer)) throw new Error(`edgeLayer 非法: ${edgeLayer}`); if (!edgeId || !kind) throw new Error('declareEdgeContract 需要 edgeId / kind'); // role 默认 produces:出边声明产出是压倒性多数,needs 必须显式写。 // 去重键仍是 (layer, edge, kind) 而不含 role —— 同一条边对同一个 kind 既声明 produces // 又声明 needs 是自相矛盾的输入(「我产它」和「我要别人产它」), // normalizeContracts 的遍历顺序让 produces 先落库、矛盾的那条被 DO NOTHING 吞掉, // 即保留更强的那个断言。 getDb() .prepare(` INSERT INTO graph_edge_contracts (id, edge_layer, edge_id, kind, required, role, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(edge_layer, edge_id, kind) DO NOTHING `) .run( newId('gec'), edgeLayer, String(edgeId), kind, required ? 1 : 0, role === 'needs' ? 'needs' : 'produces', new Date().toISOString() ); } /** 某条边上声明的契约 */ function edgeContracts(edgeLayer, edgeId) { return getDb() .prepare(`SELECT * FROM graph_edge_contracts WHERE edge_layer = ? AND edge_id = ?`) .all(edgeLayer, String(edgeId)); } /** * 某个节点的**入边**上声明了哪些 kind —— 这是「这个节点需要什么输入」的答案。 * 只看 forward 入边:backward 入边是回退路径,不是数据来源。 * * **produces 与 needs 都收**——两者对下游而言都是「要吃的输入」,只是来源不同: * produces 是紧邻上游自己的产出(R-19.3 ①「沿边传递」,如 `impl→test: produces:[artifact]`), * needs 是中继声明的非相邻依赖(R-19.3 ②,如 `test→validator: needs:[artifact]`, * 实际产出方是更早的 impl)。33 号文档最初以为这里该按 role 过滤只留 needs, * 但那会把单纯声明 produces(没有配对 needs)的绝大多数边(`impl→test`、`brd→prd`……) * 的下游需求判空,且会削弱 T-mt25ecbd 那次刚加固的产出闸门(`org-planner-sdd`/ * `org-planner-benchmark` 的 `produces:[sdd]`/`produces:[acceptance]` 就是靠这条链路 * 被 requiredInputsMissing 强制校验的)——按 role 过滤是错的,真正的根因和修法在 * `system/work-graph-templates/*.json` 那几条 `produces:["log"]` 声明本身 * (见 33 号文档修订版)。 */ function inboundContracts(edgeLayer, edges, defNodeId) { const inbound = (edges || []).filter( (e) => String(e.to) === String(defNodeId) && (e.edgeKind || 'forward') !== 'backward' ); const out = []; for (const e of inbound) { for (const c of edgeContracts(edgeLayer, e.id)) out.push({ ...c, fromNodeId: e.from }); } return out; } /** * 从 graph_run_nodes 重建索引(索引损坏时的兜底,类比 okr 的 _rebuildIndex)。 * 只能恢复「节点产出过什么目录」这一层,恢复不了 kind —— 所以 kind 统一记 'artifact', * 这是有损重建,够用来定位文件,不够还原语义。 */ function rebuildFilesFromNodes(companyId) { const db = getDb(); const rows = db .prepare(` SELECT n.id AS node_run_id, n.run_id, n.output_ref, r.task_id FROM graph_run_nodes n JOIN graph_runs r ON r.id = n.run_id WHERE r.company_id = ? AND n.output_ref IS NOT NULL `) .all(companyId); let n = 0; for (const r of rows) { recordFile({ companyId, taskId: r.task_id, relPath: r.output_ref, kind: 'artifact', runId: r.run_id, nodeRunId: r.node_run_id, }); n += 1; } return n; } module.exports = { getDb, newId, // 校验 validateWorkGraph, validateFailFallback, // 定义层 createWorkGraph, getWorkGraph, readDefGraph, // 运行时层 createRun, getRun, getRunState, syncRunNodesToDef, getRunNode, getRunNodeByDefId, updateNodeStatus, bumpNodeAttempt, updateRunStatus, // 「坑满没满」的唯一计数处(WHERE 与准入事务同源),调度器 / 路由复用它 countActiveRootRuns, listActiveRootRunIds, setRunFlags, bumpBackwardCount, listChildRuns, listRunTree, listCompanyRuns, // R-19 产物索引(27-P4 §4.6) recordFile, findFiles, declareEdgeContract, edgeContracts, inboundContracts, rebuildFilesFromNodes, // 回退记录 countLoopRecords, insertLoopRecord, updateLoopVerdict, listLoopRecords, // 27-P3C goal / task 两级模型 genGoalId, genTaskId, createGoal, getGoal, listCompanyGoals, updateGoal, createTask, getTask, listGoalTasks, listCompanyTasks, updateTask, // validateDependsOn 故意**不导出**:它只有 createTask / updateTask 两个内部调用方, // 导出一个零外部调用方的函数就是 §九.4quater 点名的那类「写了没人用」的存量(缺陷 F10)。 // 真有外部要单独校验时再导出,那时它会带着自己的调用方一起进来。 taskProgress, maxTaskSeq, settleTaskFromRun, GRAPH_TAKEOVER_ASSIGNEE, // 批 3 排队:三条都是条件更新,调用方靠 changes 判「是不是我做的」 claimQueuedTask, blockQueuedTask, requeueDerivedBlockedTask, requeueBlockedRootRun, noteTaskBlockedReason, UPSTREAM_BLOCK_PREFIX, settleGoalStatus, settleGoalPlanning, bumpValidationTokens, claimGoalSlot, settleGoalVerdict, TASK_TERMINAL_STATUS, GOAL_TERMINAL_STATUS, // 这两个类有真调用方:company.js:sendDomainError 用 instanceof 判定(不是比 err.name) FrozenFieldError, DependsOnError, // 状态机 NODE_TRANSITIONS, RUN_TRANSITIONS, // 准入事务与调度器的「坑满没满」判据共用这一个数组,别再抄字面量 ACTIVE_ROOT_RUN_STATUS, // 并行度上限:per-company,读 company.json.max_parallel_tasks。 // **故意不导出一个常量**:批 3 那个 `MAX_ACTIVE_ROOT_RUNS = 1` 一旦留下来, // 就会有人拿它当「上限」用,而它对一家 max=3 的公司是错的(且不报错)。 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, RunSlotBusyError, // 「这次没抢到坑」的唯一判据,调度器与路由都取它,不许各写一份(理由见函数注释) isSlotRaceError, };