/** * 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'); // ──────────────────────────────────────────────────────────────── // 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, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); -- v1 约束:同一 company 至多一个 run_status ∈ {ready,running,paused} 的 root run CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_runs_one_active_root ON graph_runs(company_id) WHERE parent_run_id IS NULL AND run_status IN ('ready','running','paused'); 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', 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)。 // 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。它们在旧代码下早就报过了(或永远不该报),一律按「已报」封口。 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(', ')}) `); } schemaEnsured = true; return db; } // ──────────────────────────────────────────────────────────────── // §四.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'], 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'], blocked: ['running', 'stopped'], // 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 末尾):任何有 fail 出边的节点,必须至少留一条 * fail_reason IS NULL 的兜底边。否则归因产出一个图上没有对应边的取值时会落到死路, * 表现为「莫名其妙 blocked」。 * * ⚠️ 现状:org 层**没有**复用本函数,而是在 org-graph-service.validateGraph 的「规则 5」里 * 内联实现了同一条规则——本文件 require org-graph-service(validateGraph),反向 require 会成环。 * 于是 validateWorkGraph 走的是「validateGraph 的规则 5 + 本函数」两份实现,同一条违规会报两行; * 两份逻辑目前语义一致,但改一处漏一处就会漂移。要收敛的话把规则抽到一个无依赖的小模块里, * 两边都 require 它(见 27-P3P4 §三.7 的风险记录)。 * * @param {Array<{from, to, condition, failReason?}>} edges from/to 用什么标识都行(只做分组) * @returns {Array<{rule: string, detail: string}>} */ function validateFailFallback(edges) { 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); } for (const [from, slot] of byFrom) { if (slot.hasFail && !slot.hasFallback) { errors.push({ rule: 'fail-fallback-edge', detail: `节点 ${from} 有 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 * @returns {{ok: boolean, errors: Array<{rule: string, detail: string}>}} */ function validateWorkGraph(nodes, edges) { 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(es)); 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 = [], } = {}) { const ns = normalizeNodes(nodes); const es = normalizeEdges(edges); const verdict = validateWorkGraph(ns, es); 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 继承(冗余存储避免上溯)。 * * 同一 company 只允许一个 active root run(idx_graph_runs_one_active_root); * 冲突时抛 SqliteError,调用方自己决定是拒绝还是复用已有 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, } = {}) { if (graphKind !== 'org' && graphKind !== 'work') { throw new Error(`graphKind 必须是 'org' 或 'work',收到: ${graphKind}`); } 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, 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), 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; } /** * 取一次 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', }; /** * 条件更新 run 状态,语义与 updateNodeStatus 完全一致(同样返回受影响行数供幂等判定)。 * @returns {number} */ 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]; } } return db .prepare(`UPDATE graph_runs SET ${sets.join(', ')} WHERE id = @id AND run_status = @from`) .run(params).changes; } /** 只改软信号,不动 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', reportTo: 'report_to', blockedReason: 'blocked_reason', queuedAt: 'queued_at', startedAt: 'started_at', finishedAt: 'finished_at', archivedAt: 'archived_at', }; /** 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 都允许改**(§六.3 定案)。真实场景是「B 已经排上了, * 才发现它得等 C」,此前只能取消重建、白丢排队位次。位次不受影响 —— 位次按 * priority → seq → created_at 排,depends_on 决定的是「够不够格出队」,两者正交。 * 跑起来之后就不许改了:那时依赖判据已经用过,再改只会让 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 (cur.status !== 'draft' && cur.status !== 'queued') { 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']; /** * 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, finished_at = @now, updated_at = @now 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; } /** * 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 入边是回退路径,不是数据来源。 */ 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, getRunNode, getRunNodeByDefId, updateNodeStatus, bumpNodeAttempt, updateRunStatus, 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, settleGoalStatus, settleGoalPlanning, bumpValidationTokens, claimGoalSlot, settleGoalVerdict, TASK_TERMINAL_STATUS, GOAL_TERMINAL_STATUS, // 这两个类有真调用方:company.js:sendDomainError 用 instanceof 判定(不是比 err.name) FrozenFieldError, DependsOnError, // 状态机 NODE_TRANSITIONS, RUN_TRANSITIONS, canTransitionNode, canTransitionRun, InvalidTransitionError, };