/** * 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 这一套,不另写一份 const { validateGraph } = 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 ); `); // 存量库迁移(写法照抄 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-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', })); } // ──────────────────────────────────────────────────────────────── // 定义层 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) { insEdge.run( newId('wge'), graphId, nodeIds.get(e.from), nodeIds.get(e.to), e.condition, e.failReason, e.edgeKind, now ); } })(); 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)); } module.exports = { getDb, newId, // 校验 validateWorkGraph, validateFailFallback, // 定义层 createWorkGraph, getWorkGraph, readDefGraph, // 运行时层 createRun, getRun, getRunState, getRunNode, getRunNodeByDefId, updateNodeStatus, bumpNodeAttempt, updateRunStatus, setRunFlags, bumpBackwardCount, listChildRuns, listRunTree, listCompanyRuns, // 回退记录 countLoopRecords, insertLoopRecord, updateLoopVerdict, listLoopRecords, // 状态机 NODE_TRANSITIONS, RUN_TRANSITIONS, canTransitionNode, canTransitionRun, InvalidTransitionError, };