宜博 / lemonboss

/**
 * 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
    );

    -- ── 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,
      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-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<string,string>} | {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<defNodeId,string>}}
 */
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 = <from>，返回受影响行数：
 *   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 }) {
  if (!['org', 'work'].includes(edgeLayer)) throw new Error(`edgeLayer 非法: ${edgeLayer}`);
  if (!edgeId || !kind) throw new Error('declareEdgeContract 需要 edgeId / kind');
  getDb()
    .prepare(`
      INSERT INTO graph_edge_contracts (id, edge_layer, edge_id, kind, required, created_at)
      VALUES (?, ?, ?, ?, ?, ?)
      ON CONFLICT(edge_layer, edge_id, kind) DO NOTHING
    `)
    .run(newId('gec'), edgeLayer, String(edgeId), kind, required ? 1 : 0, 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,
};