/** * 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')}`; } // ──────────────────────────────────────────────────────────────── // 建表(幂等,可反复调用) // ──────────────────────────────────────────────────────────────── 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); `); // 存量库迁移(写法照抄 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`); } 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); } // ──────────────────────────────────────────────────────────────── // 回退记录(§四.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, // 状态机 NODE_TRANSITIONS, RUN_TRANSITIONS, canTransitionNode, canTransitionRun, InvalidTransitionError, };