宜博 / 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,
      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,
      created_at       TEXT NOT NULL
    );
  `);

  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'],
  completed: [],
  failed: [],
  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 层（T14 的另一半）以同样的语义复用，不必再实现一遍。
 *
 * @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,
        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',
  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,
} = {}) {
  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, created_at)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
  `).run(id, runId, String(edgeId), fromNodeRunId, toNodeRunId, n, verdict, remediationNote, 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,
};