宜博 / 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 这一套，不另写一份
// R-19：normalizeContracts 也从那边拿 —— 边上的 produces / needs 两层通用，
// 两处各写一份归一逻辑就是「差一个字母静默失效」的温床
const { validateGraph, normalizeContracts } = require('./org-graph-service');

// ────────────────────────────────────────────────────────────────
// 活跃 root run 取值域（27-P3C §五.1）
// ────────────────────────────────────────────────────────────────

/**
 * 「这条 root run 正占着公司的执行坑」的取值域。
 *
 * **它同时是 `idx_graph_runs_one_active_root` 那个唯一索引的 WHERE 子句**（下面 getDb 里直接
 * 拼进去），所以这里是全仓库唯一的定义处，调度器 / 路由复用它而不是再抄一份字面量。
 * 为什么强调这件事：`canDispatch` 那次事故就是前端把这三个值复述了一遍、还多写了一个
 * `blocked`，于是公司失败一次之后再也下不了任务 —— 取值域一旦有第二个持有者就会分叉。
 *
 * ⚠️ **`queued` 故意不在里面**：排队中的 run 不占坑，一家公司可以同时躺着任意多条。
 * 这正是「出队原子性」的**全部依赖** —— 出队那一跳 `queued → ready` 会撞上这个唯一索引，
 * 超发的那一条被 DB 直接拒（SqliteError），而不是靠应用层「先查后插」的乐观检查。
 * 改这个数组之前先读 27-P3C §四.3 与 §五.2；批 4 放开并行时要连索引一起动。
 */
const ACTIVE_ROOT_RUN_STATUS = ['ready', 'running', 'paused'];

/**
 * 一家公司同时能占几个执行坑。
 *
 * ponytail: 批 3 硬编码为 1，与 `idx_graph_runs_one_active_root`（UNIQUE(company_id)）
 * 表达的是同一件事。批 4 放开并行时这里改成读 `company.json.max_parallel_tasks`
 * 并 DROP 那个索引 —— 到那时 `updateRunStatus` 里的准入事务就是**唯一**的守卫，
 * 所以它今天就得是对的，而不是等索引没了再回头补。
 */
const MAX_ACTIVE_ROOT_RUNS = 1;

/**
 * 坑满：一条 root run 想从活跃集外进入活跃集，但公司的执行坑已经占满。
 *
 * 单独一个类型，不复用别的：
 *   · 不是 `InvalidTransitionError` —— 转换本身完全合法（queued→ready 就是出队），
 *     错的是「这一刻没位置」，换个时间同一条转换会成功；
 *   · 不是那条 UNIQUE 的 SqliteError —— 那是**兜底**那一层的产物，语义正确但报文是
 *     一句 SQLite 原文，调用方要靠正则去猜。
 * 调度器用它判「这轮没抢到」（正常结局），路由层用它出 409。
 *
 * ⚠️ 这里以前挂着一条「message 里的 `constraint` 一词不许清掉，因为 `routes/company.js`
 * 有两处（dispatch / POST tasks）用 `/UNIQUE|constraint/i` 嗅探错误文本来翻 409」的告示。
 * **批 3 把那两处 catch 整段删了**（下达不再因坑满失败，改成入队），告示描述的调用方
 * 已经不存在，却一直留着 —— 一条注释在替一段不存在的代码索要保证，是本文档反复点名的
 * 那个形状（§十四.3 第 3 条：注释是对的、代码没兑现；这次反过来）。
 * 现在文本不再是任何人的判据：唯一的判据是 `isSlotRaceError`（下面那个函数），按类型判。
 * message 里那句 `active-root-run constraint` 只当人读的说明留着，改它不会弄坏任何分支。
 */
class RunSlotBusyError extends Error {
  /**
   * @param {string} companyId
   * @param {string} runId    想进坑而没进去的那条 run
   * @param {string[]} heldBy 此刻占着坑的 root run id —— **必须带上**：路由层要把它翻成
   *   一句人能看懂的 409（「被 run X 占着」），少了它界面上只剩「坑满」，人无从判断该等谁。
   * @param {number} max
   */
  constructor(companyId, runId, heldBy, max) {
    const held = heldBy.length ? heldBy.join(', ') : '(未知)';
    super(`company ${companyId} 的执行坑已满（${heldBy.length}/${max}，占用者 ${held}）：`
      + `run ${runId} 无法进入活跃集（active-root-run constraint）`);
    this.name = 'RunSlotBusyError';
    this.companyId = companyId;
    this.runId = runId;
    this.heldBy = heldBy;
    this.active = heldBy.length;
    this.max = max;
  }
}

/**
 * 「这次没抢到坑」—— 竞争的正常结局，不是故障。**全仓库唯一的判据持有者。**
 *
 * 为什么必须只有一个：批 3 落地时它有两份，而且两份不一样 ——
 * `task-scheduler.isRaceError` 按 `/UNIQUE constraint failed/i` 嗅报错文本，
 * `routes/company.js#isSlotRace` 按 `err.code`，后者的注释还写着「口径同 task-scheduler」。
 * 两个持有者迟早分叉，这一份从落地那天起就已经分叉了：改一句报错文案，调度器那边会把
 * 一次正常竞争当成故障去 `abandonRun`（run 收成 stopped、task 收成 cancelled = 丢任务），
 * 而路由那边照常 409。这正是 `canDispatch` 那次事故的形状，只是换了个取值域。
 *
 * 三种形态，三道防线各一种：
 *   · `RunSlotBusyError`         主守卫：准入事务在 BEGIN IMMEDIATE 里数出坑已满；
 *   · `SQLITE_CONSTRAINT_UNIQUE` 兜底：真有两条同时写进活跃集时唯一索引直接拒（批 4 DROP 它后消失）。
 *     判 `code` 而不是判 message：better-sqlite3 的 SqliteError 自带这个字段，文案换了它不变；
 *   · `InvalidTransitionError`   这条 run 已经被别人从 queued 推走了。
 */
function isSlotRaceError(err) {
  if (!err) return false;
  return err instanceof InvalidTransitionError
    || err instanceof RunSlotBusyError
    || err.code === 'SQLITE_CONSTRAINT_UNIQUE';
}

// ────────────────────────────────────────────────────────────────
// ID 生成
// ────────────────────────────────────────────────────────────────

/** 前缀 + 12 位 hex。不引入 nanoid 依赖，crypto 够用（碰撞概率 2^-48 量级）。 */
function newId(prefix) {
  return `${prefix}_${crypto.randomBytes(6).toString('hex')}`;
}

/**
 * goal / task 的 id：`G-{base36 毫秒}-{4 位 hex}`（27-P3C §四.5）。
 *
 * 为什么不复用上面的 newId：`T-` 这个前缀与短横线分隔在 company.js / agent-chat.js 已经用开了，
 * 换成 `T_xxx` 会让两套 id 看上去不是同一类东西。
 *
 * 为什么不能沿用现有的 `T-${Date.now().toString(36)}`：那个格式**同一毫秒内生成的两个 id
 * 完全相同**。今天没炸只是因为一次只建一条；一个 goal 一次性拆出 N 条 task 是**循环建行**，
 * 同毫秒撞是必然事件 —— 而 tasks.id 是主键，撞了就是 500。
 * 随机后缀是仓库既有约定（_spec/CONVENTIONS.md §2：时间戳部分不保证全局唯一，靠后缀随机位避碰）。
 */
function genId(prefix) {
  return `${prefix}-${Date.now().toString(36)}-${crypto.randomBytes(2).toString('hex')}`;
}

const genGoalId = () => genId('G');
const genTaskId = () => genId('T');

/**
 * 用新生成的 id 插一行，主键撞了就换一个再来（最多 5 次）。
 *
 * 4 位 hex 只有 65536 种：同毫秒批量建行时，生日碰撞是**百分之几**量级，不是「绝不会」。
 * id 格式按 §四.5 定死不动（它已经写进文档、也已经是对外可见的形状），
 * 所以补的是重试而不是加长随机位 —— 撞一次的代价是再跑一次 INSERT，
 * 而不补的代价是「一个 goal 拆 20 条 task，偶尔有一次整体 500」这种最难复现的故障。
 *
 * 只吞这张表主键的 UNIQUE 冲突，别的约束照常抛出去。
 */
function insertWithFreshId(table, gen, insert) {
  for (let attempt = 0; ; attempt += 1) {
    const id = gen();
    try {
      insert(id);
      return id;
    } catch (err) {
      if (attempt >= 4 || !new RegExp(`UNIQUE constraint failed: ${table}\\.id`).test(err.message)) throw err;
    }
  }
}

// ────────────────────────────────────────────────────────────────
// 建表（幂等，可反复调用）
// ────────────────────────────────────────────────────────────────

let schemaEnsured = false;

/** 表是否已有某列（org_graph_edges.fail_reason 由 T-P3 加，本文件读时要能容忍它还没加） */
function hasColumn(db, table, column) {
  return db.pragma(`table_info(${table})`).some((c) => c.name === column);
}

function getDb() {
  const db = getGraphDb(); // 复用 graph.sqlite 的同一连接（WAL + foreign_keys=ON 已由它设好）
  if (schemaEnsured) return db;

  db.exec(`
    -- ── 定义层：work graph ──────────────────────────────────────
    CREATE TABLE IF NOT EXISTS work_graphs (
      id                TEXT PRIMARY KEY,
      company_id        TEXT NOT NULL,
      owner_org_node_id INTEGER REFERENCES org_graph_nodes(id),
      planned_by        TEXT NOT NULL DEFAULT 'workgraph-planning',
      goal_id           TEXT,
      task_id           TEXT,
      okr_node_id       TEXT,
      created_at        TEXT NOT NULL
    );

    CREATE TABLE IF NOT EXISTS work_graph_nodes (
      id              TEXT PRIMARY KEY,
      work_graph_id   TEXT NOT NULL REFERENCES work_graphs(id),
      node_name       TEXT NOT NULL,
      system_agent_id TEXT NOT NULL REFERENCES system_agents(id),
      node_kind       TEXT NOT NULL DEFAULT 'agent_call'
                        CHECK (node_kind IN ('agent_call','subgraph')),
      created_at      TEXT NOT NULL
    );

    CREATE TABLE IF NOT EXISTS work_graph_edges (
      id            TEXT PRIMARY KEY,
      work_graph_id TEXT NOT NULL REFERENCES work_graphs(id),
      from_node_id  TEXT NOT NULL REFERENCES work_graph_nodes(id),
      to_node_id    TEXT NOT NULL REFERENCES work_graph_nodes(id),
      condition     TEXT NOT NULL CHECK (condition IN ('success','fail')),
      fail_reason   TEXT,
      edge_kind     TEXT NOT NULL DEFAULT 'forward'
                      CHECK (edge_kind IN ('forward','backward')),
      created_at    TEXT NOT NULL
    );

    -- ── 运行时层：org / work 两层共用 ───────────────────────────
    CREATE TABLE IF NOT EXISTS graph_runs (
      id                 TEXT PRIMARY KEY,
      graph_kind         TEXT NOT NULL CHECK (graph_kind IN ('org','work')),
      graph_id           TEXT NOT NULL,
      company_id         TEXT NOT NULL,
      root_run_id        TEXT NOT NULL,
      parent_run_id      TEXT REFERENCES graph_runs(id),
      parent_node_run_id TEXT REFERENCES graph_run_nodes(id),
      goal_id            TEXT,
      task_id            TEXT,
      okr_node_id        TEXT,
      execution_mode     TEXT NOT NULL DEFAULT 'manual'
                           CHECK (execution_mode IN ('manual','auto')),
      run_status         TEXT NOT NULL DEFAULT 'planning',
      pause_requested    INTEGER NOT NULL DEFAULT 0,
      stop_requested     INTEGER NOT NULL DEFAULT 0,
      backward_count     INTEGER NOT NULL DEFAULT 0,
      budget_json        TEXT,
      created_at         TEXT NOT NULL,
      updated_at         TEXT NOT NULL
    );
    -- v1 约束：同一 company 至多一个「活跃」root run。取值域从 ACTIVE_ROOT_RUN_STATUS 拼进来，
    -- 不再手抄一份 —— 索引的 WHERE 和调度器判「坑满没满」的判据必须是同一个数组（§五.1）。
    -- 注意 queued 不在其中：排队的 run 不占坑，这是出队 CAS 的依据（见该常量注释）。
    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 (${ACTIVE_ROOT_RUN_STATUS.map((s) => `'${s}'`).join(',')});

    CREATE TABLE IF NOT EXISTS graph_run_nodes (
      id            TEXT PRIMARY KEY,
      run_id        TEXT NOT NULL REFERENCES graph_runs(id),
      def_node_id   TEXT NOT NULL,
      status        TEXT NOT NULL DEFAULT 'pending',
      attempt       INTEGER NOT NULL DEFAULT 0,
      agent_run_id  TEXT,
      child_run_id  TEXT REFERENCES graph_runs(id),
      output_ref    TEXT,
      fail_reason   TEXT,
      -- R-17.1 跨层节点地址：归因**指向**的那个 work graph node，形如
      --   {"orgNode":"org-coder","workNode":"impl-a"}
      -- 语义是「这一行的失败，根因在 orgNode 那张 work graph 的 workNode 上」。
      -- 与 fail_reason 同源同轨：同一份 attempt-{n}.json 产出、同一次条件更新落库、
      -- 同一条 deriveChildOutcome 透传上去。**不复用 fail_reason 承载**（它是选边的匹配键，
      -- 掺进地址会让所有精确 fail 边失配），也**不放在边上**（边是定义层，
      -- 而 work graph 是运行时才由 workgraph-planning 规划出来的，定义层根本写不出这个 id）。
      fail_node_ref TEXT,
      -- child_run_id 那张子 run 是为**本节点的哪一次 attempt** 起/复用的（R-17.1 幂等判据）。
      -- 判据从「子 run 是不是还活着」改成「它服务的是不是本轮」：R-17.1 要求复用旧子 run，
      -- 只看终态就会让「上一轮留下的终态」和「这一轮刚重置过」长得一样——那正是第二轮
      -- 那个永久死锁的成因。存代次是唯一能把两者分开的信息。
      child_run_attempt INTEGER,
      -- 「留给这个节点下一轮的修复建议」。载体必须是节点行而不是 graph_run_loop_records：
      -- 后者只在命中 backward 边时才有记录，走 forward fail 边 / 死路 blocked / infra_flake
      -- 原地重跑这三条路径上归因节点写下的 remediation_note 就地蒸发（R4-F1）。
      -- 且 loop record 是熔断账本（loop_count 按 (run_id, edge_id) 记录数算），
      -- 借它存判词会直接污染熔断计数。
      remediation_note TEXT,
      started_at    TEXT,
      completed_at  TEXT,
      created_at    TEXT NOT NULL,
      updated_at    TEXT NOT NULL
    );
    CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_run_nodes_unique
      ON graph_run_nodes(run_id, def_node_id);

    CREATE TABLE IF NOT EXISTS graph_run_loop_records (
      id               TEXT PRIMARY KEY,
      run_id           TEXT NOT NULL REFERENCES graph_runs(id),
      edge_id          TEXT NOT NULL,
      from_node_run_id TEXT NOT NULL REFERENCES graph_run_nodes(id),
      to_node_run_id   TEXT NOT NULL REFERENCES graph_run_nodes(id),
      loop_count       INTEGER NOT NULL,
      verdict          TEXT NOT NULL
                         CHECK (verdict IN ('registered','fixed','not_fixed','force_terminated')),
      remediation_note TEXT,
      -- R-17.1：这次回退**实际落到**的跨层地址（解析后的结果，含 childRunId / nodeRunId）。
      -- 与 graph_run_nodes.fail_node_ref 的分工：那边是归因的**主张**，这边是引擎的**动作**。
      -- 主张可能解析不了（图换过、节点名不存在、与选中的边不是同一个 org node），
      -- 那时这一列为空 = 「退化成整张 work graph 重跑」，控制台据此显示回退粒度。
      target_node_ref  TEXT,
      created_at       TEXT NOT NULL
    );

    -- ── R-19：产物索引（27-P4 §4.6）────────────────────────────────────
    -- 补的是「同一张 work graph 内部上游到下游没有数据通路」这个断点：边一直只承载
    -- 控制流（condition / edge_kind / fail_reason），而「边的输入输出」此前只是
    -- 从 output_ref **推断**出来的 UI 展示、没有实体，推断的那一端是节点私有目录，
    -- 下游读不到。第四次真跑就死在这：prd 节点的归因原话是
    -- 「不是上游没产出，是它的产出结构上到不了我手上」。

    -- 产物**实例**：运行时由**引擎**追加。不靠 agent 自觉写 —— 否则索引与实际文件必然漂移
    -- （本轮反复验证：凡「靠 agent 自觉」的环节几乎全部落空过）。
    -- rel_path 是**相对公司根**的路径，根由代码推导不入库：存绝对路径会烂，
    -- W13 清掉的那 10 条 legacy_path 悬空行就是这个病。
    CREATE TABLE IF NOT EXISTS graph_files (
      id                      TEXT PRIMARY KEY,
      company_id              TEXT NOT NULL,
      task_id                 TEXT,
      rel_path                TEXT NOT NULL,
      kind                    TEXT NOT NULL,
      produced_by_run_id      TEXT,
      produced_by_node_run_id TEXT,
      created_at              TEXT NOT NULL,
      superseded_by           TEXT
    );
    -- **部分**唯一：只约束「当前版本」（superseded_by IS NULL），历史行可以同路径共存。
    -- 与 idx_org_graph_nodes_company_role_active 同一个模式 —— 想要「同一键只有一个活的」
    -- 又要留历史，就得把条件写进索引，而不是靠删行。
    CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_files_current
      ON graph_files(company_id, COALESCE(task_id, ''), rel_path) WHERE superseded_by IS NULL;
    CREATE INDEX IF NOT EXISTS idx_graph_files_lookup
      ON graph_files(company_id, COALESCE(task_id, ''), kind);

    -- 边上声明的数据**契约**：规划时写，只有 kind 没有实例。
    -- edge_layer 区分两层边表 —— org_graph_edges.id 是 INTEGER、work_graph_edges.id 是 TEXT，
    -- 与 graph_run_nodes.def_node_id 同一处类型不整齐（见 27-P3/P4 §四.A.2 的注），
    -- 所以这里存字符串形式、不建外键，引用完整性靠服务层。
    CREATE TABLE IF NOT EXISTS graph_edge_contracts (
      id          TEXT PRIMARY KEY,
      edge_layer  TEXT NOT NULL CHECK (edge_layer IN ('org','work')),
      edge_id     TEXT NOT NULL,
      kind        TEXT NOT NULL,
      required    INTEGER NOT NULL DEFAULT 1,
      -- produces = 产出方在自己的出边上声明「我产这个 kind 交给下游」
      -- needs    = 消费方在自己的入边上声明「我要吃这个 kind」，产出方不必是紧邻上游
      -- 这个非对称必须落库：规则 6（data-contract）靠「有人 needs 但没人 produces」判错，
      -- 只存 kind 的话读回来就分不清谁该产、谁只是要吃，校验在库这一侧永远算不出来。
      role        TEXT NOT NULL DEFAULT 'produces' CHECK (role IN ('produces','needs')),
      created_at  TEXT NOT NULL
    );
    CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_edge_contracts_dedupe
      ON graph_edge_contracts(edge_layer, edge_id, kind);

    -- ── 27-P3C：goal / task 两级模型（27-p3c-goal-task-model.md §四.1 / §四.2）──
    -- 这不是新概念，是给一对**已经存在的悬空外键**补上它们的表：graph_runs.goal_id /
    -- task_id 从 P3 起就一直在写，却没有任何表接得住，于是「一个需求拆成几件活」
    -- 在库里根本不存在，界面上也就画不出来。
    -- 三级：goal（人提的一个需求，description 即验收边界）→ task（一件可独立跑的活）
    --      → run（task 的一次执行，已有）。

    -- **没有 budget_json**：预算闸门是按 root run 算的（DEFAULT_TASK_BUDGET 只挂 root run，
    -- 子 run 一律上溯 root_run_id 扣同一个池），goal 级预算没有任何执行机制 ——
    -- 加一个没人消费的额度列，就是本仓库反复被咬的「实现在、零调用方」。
    -- 但 task-planning 的拆分调用跑在任何 run 存在之前，那是一笔真花钱、又没有池子可扣的
    -- 开销（goal 层「隐形的一半」），所以它只**记账**（planning_tokens / validation_tokens），
    -- 不设闸门；总额控制另排期。
    CREATE TABLE IF NOT EXISTS goals (
      id             TEXT PRIMARY KEY,          -- G-{base36}-{4hex}，见 §四.5
      company_id     TEXT NOT NULL,
      title          TEXT NOT NULL,             -- 短标题，列表显示用；可改
      description    TEXT NOT NULL,             -- 人类需求原文 = 验收边界；下达后不可改
      status         TEXT NOT NULL DEFAULT 'draft',
                     -- draft | planning | active | done | failed | cancelled | archived
                     -- planning = task-planning 正在拆分，是个真会持续几十秒的状态
                     -- failed   = goal 级验收没过，或有 task 未能 completed；可回 active 补 task 重下
      priority       TEXT NOT NULL DEFAULT 'P1',
      okr_node_id    TEXT,                      -- 可选，挂到 OKR 度量树的哪个节点（O-001.1.1）
      created_by     TEXT,                      -- 'human' | agent id（Master/CEO 代下单）
      report_to      TEXT,                      -- 收敛后要唤醒的上级 agent，见 §四.7 禁令二
      planning_tokens INTEGER NOT NULL DEFAULT 0, -- task-planning 拆分烧掉的 token
      validation_tokens INTEGER NOT NULL DEFAULT 0, -- goal 级 org-validator 烧掉的 token
      verdict        TEXT,                      -- goal 级验收判词（org-validator 的原话）
      planning_error TEXT,                      -- 非空 = task-planning 拆分失败、已降级成单 task（§四.9）
      -- 「这个 goal 的收敛已经对外汇报过了」这件事的**可查证据**（第二轮复核：汇报那一格）。
      -- 判词落库和汇报（写 OKR / 唤醒上级）是两件事，中间那几十毫秒到几秒里进程被杀、
      -- 或端点回 5xx，库里这条 goal 会长得**和一次完美收敛一模一样** —— 判词漂亮、状态终态，
      -- 而 OKR 永远没写、上级永远没醒。没有这一列就没有任何查询能把两者分开，也就补不了。
      -- 三态（与 verdict 同一套形状）：
      --   NULL           → 从没汇报过
      --   '⏳ …#{token}' → 认领了但没确认成功（正在报 / 报到一半丢了），启动重推会补
      --   ISO 时间戳     → 真报成功了
      reported_at    TEXT,
      created_at     TEXT NOT NULL,
      updated_at     TEXT NOT NULL,
      dispatched_at  TEXT,                      -- 首次下达时间；非空 = 描述已冻结（§四.4）
      archived_at    TEXT
    );
    CREATE INDEX IF NOT EXISTS idx_goals_company ON goals(company_id, status);

    -- goal_id 是**真外键**：tasks 是新表，可以直接写上（graph_runs 那两列只能是逻辑外键 ——
    -- SQLite 不支持给已有列追加外键，重建 graph_runs 要连带 graph_run_nodes 的真外键一起动，
    -- 代价远大于收益，见 §四.6）。
    CREATE TABLE IF NOT EXISTS tasks (
      id             TEXT PRIMARY KEY,          -- T-{base36}-{4hex}
      goal_id        TEXT NOT NULL REFERENCES goals(id),
      company_id     TEXT NOT NULL,             -- 冗余一份：调度器按公司选任务，避免每次 JOIN
      seq            INTEGER NOT NULL,          -- goal 内序号，列表排序用（1,2,3…）
      title          TEXT NOT NULL,
      description    TEXT NOT NULL,             -- 落进 task-goal.json 的那句；下达后不可改
      status         TEXT NOT NULL DEFAULT 'draft',
                     -- draft | queued | running | completed | failed | blocked | cancelled
                     -- queued 现在就在取值域里，但排队逻辑（调度器 / run_status='queued'）另批实现：
                     -- 取值域是数据形状的一部分，先钉死它，免得批 3 再来改一次列注释
      execution_mode TEXT NOT NULL DEFAULT 'manual' CHECK (execution_mode IN ('manual','auto')),
      depends_on     TEXT NOT NULL DEFAULT '[]', -- JSON 数组，同 goal 内的 task id
      budget_json    TEXT,
      current_run_id TEXT,                       -- 最近一次 root run（逻辑外键，见 §四.6）
      attempt        INTEGER NOT NULL DEFAULT 0, -- 起过几次 run（任务级重跑，≠ 节点级 retry）
      report_to      TEXT,                       -- 从 goal 继承或调用方指定；见 §四.7 禁令二
      created_by     TEXT,                       -- 'task-planning' | 'human'：拆出来的还是人加的
      blocked_reason TEXT,                       -- status='blocked' 时的原因，如 upstream_failed:T-xxx
      priority       TEXT NOT NULL DEFAULT 'P1',
      created_at     TEXT NOT NULL,
      updated_at     TEXT NOT NULL,
      queued_at      TEXT,                       -- 入队时间；非空 = 描述已冻结（§四.4）
      started_at     TEXT,
      finished_at    TEXT,
      archived_at    TEXT
    );
    CREATE INDEX IF NOT EXISTS idx_tasks_goal    ON tasks(goal_id, seq);
    -- 列序与调度器的选取口径（company → status → priority → created_at）对齐
    CREATE INDEX IF NOT EXISTS idx_tasks_queue   ON tasks(company_id, status, priority, created_at);
  `);

  // 存量库迁移（写法照抄 graph-db.js migrate()：hasColumn 判断 + ALTER TABLE ADD COLUMN，
  // 新建库走上面的 CREATE TABLE 就已经是目标形态，这里是 no-op）。
  // R4-F1：graph_run_nodes 追加 remediation_note，可空，旧行 NULL = 「没有留给下一轮的话」。
  if (!hasColumn(db, 'graph_run_nodes', 'remediation_note')) {
    db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN remediation_note TEXT`);
  }
  // R-19.A2：graph_edge_contracts 追加 role。旧行一律回填 'produces' 而不是留空：
  // 规则 6 里 NULL 既不算产出也不算需求，一条本该是 produces 的旧行会让下游的 needs
  // 突然找不到产出方 —— 那是**误拒一张合法的图**。校验闸门宁可漏判不可错拦，
  // 且 produces 是压倒性多数（forward 边基本都写它），所以回填它是更接近事实的一侧。
  if (!hasColumn(db, 'graph_edge_contracts', 'role')) {
    db.exec(`ALTER TABLE graph_edge_contracts ADD COLUMN role TEXT`);
    db.exec(`UPDATE graph_edge_contracts SET role = 'produces' WHERE role IS NULL`);
  }
  // R-17.1：跨层地址 + 子 run 代次。旧行两列都是 NULL——
  //   fail_node_ref = NULL     → 没有跨层归因，回退退化成整张图重跑（与返工前同语义）
  //   child_run_attempt = NULL → 代次未知，executeSubgraph 按「不是本轮的」处理（复用并重置），
  //                              最坏情况多重跑一轮，绝不会返回 pending 卡死。
  if (!hasColumn(db, 'graph_run_nodes', 'fail_node_ref')) {
    db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN fail_node_ref TEXT`);
  }
  if (!hasColumn(db, 'graph_run_nodes', 'child_run_attempt')) {
    db.exec(`ALTER TABLE graph_run_nodes ADD COLUMN child_run_attempt INTEGER`);
  }
  if (!hasColumn(db, 'graph_run_loop_records', 'target_node_ref')) {
    db.exec(`ALTER TABLE graph_run_loop_records ADD COLUMN target_node_ref TEXT`);
  }
  // 27-P3C §四.9（批 2）：自动拆分失败时的降级留痕。
  // 降级后的 goal 与「本来就只有一条 task」的 goal 在库里长得**一模一样**（1 条 task、
  // created_by='human'、depends_on=[]），没有这一列就分不出「这是拆出来的」和
  // 「这是拆失败后兜底的」——界面也就没法按验收 20 标注「自动拆分失败，已按单任务处理」。
  // 可空：NULL = 没有失败过（含从没拆过的存量 goal 与同步路径建的 goal）。
  if (!hasColumn(db, 'goals', 'planning_error')) {
    db.exec(`ALTER TABLE goals ADD COLUMN planning_error TEXT`);
  }
  // 27-P3C 第二轮复核：汇报的可查性（语义见 CREATE TABLE 里那段）。
  // **回填是这次迁移的重点，不是附赠品**：不回填的话，加完列之后第一次启动，
  // `resumePendingGoalValidations` 会把库里**所有**历史终态 goal 都看成「汇报丢了」，
  // 于是把它们统统重报一遍 —— OKR 节点被历史结论覆盖、上级 agent 被几百条陈年旧事叫醒，
  // 每一条都真起一轮 LLM。它们在旧代码下早就报过了（或永远不该报），一律按「已报」封口。
  if (!hasColumn(db, 'goals', 'reported_at')) {
    db.exec(`ALTER TABLE goals ADD COLUMN reported_at TEXT`);
    db.exec(`
      UPDATE goals SET reported_at = updated_at
       WHERE reported_at IS NULL
         AND status IN (${GOAL_TERMINAL_STATUS.map((s) => `'${s}'`).join(', ')})
    `);
  }

  schemaEnsured = true;
  return db;
}

// ────────────────────────────────────────────────────────────────
// §四.A.4 状态机（模式照抄 devloop/unit-registry.js：显式转换表 + 非法转换抛错）
// ────────────────────────────────────────────────────────────────

class InvalidTransitionError extends Error {
  constructor(scope, from, to, id) {
    super(`Invalid ${scope} transition for ${id}: ${from} → ${to}`);
    this.name = 'InvalidTransitionError';
    this.scope = scope; // 'node' | 'run'
    this.from = from;
    this.to = to;
    this.id = id;
  }
}

/** graph_run_nodes.status 转换表 */
const NODE_TRANSITIONS = {
  pending: ['ready', 'skipped'],
  ready: ['running', 'skipped'],
  running: ['success', 'fail'],
  // fail → ready 是命中 backward 边或人工「重试」；fail → blocked 是死路 / 硬顶耗尽
  fail: ['ready', 'blocked'],
  // success 是终态，唯一出路是人工「重试」强制打回 ready
  success: ['ready'],
  // blocked 唯一走出方式就是人工「重试」
  blocked: ['ready'],
  skipped: [],
};

/** graph_runs.run_status 转换表 */
const RUN_TRANSITIONS = {
  planning: ['ready'],
  // 27-P3C §四.3：排队态。入队时 createRun 直接建成 queued（不经过 planning），
  // 出队就是 `updateRunStatus(runId,'queued','ready')` 这一条 CAS，**不走 markRunReady**
  // （它把 from 写死成 'planning'，对 queued run 是空转：WHERE run_status='planning' 匹配不到）。
  //
  // 出口只有两条，两条都必不可少：
  //   · ready   —— 出队。之后照走既有的 ready → running，起跑路径全仓库仍然只有一条。
  //   · stopped —— 撤出队列 / 出队后又发现抢不到坑要收回。**漏了这条的后果是静默的**：
  //     `graph-engine.stopRun` 对每条 run 都先问一句 `canTransitionRun(cur,'stopped')`，
  //     为 false 就跳过 —— 于是「停止」一个排队任务只会写个 stop_requested=1，
  //     run 永远躺在队里，界面上和正常排队一模一样。
  queued: ['ready', 'stopped'],
  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'],
  // queued 这条出口是**恢复路径的退路**（27-P3C 批 3 复核 F1）：人加完预算点「启动」时坑
  // 可能刚被队里下一条占走，那时唯一能让这条 run 还有将来的动作就是把它放回队列，
  // 由调度器在坑空时按位次接回来 —— 否则它永远躺在 blocked（非终态），goal 也就永不收敛。
  // 唯一实现处是 `requeueBlockedRootRun`（run 行与 task 行必须同一个事务，见那里）。
  blocked: ['running', 'stopped', 'queued'],
  // 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<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) {
      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<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;
}

/**
 * 这个节点行**一次都没被执行过**（可以安全删掉，删了不销毁任何审计轨迹）。
 * `status='pending'` 单独一条不成立：回退重置会把跑过的节点写回 pending 并**保留 attempt**。
 * 判据与 `syncRunNodesToDef` 里那条 DELETE 的 WHERE 逐字同源，改一处要改两处。
 */
function isUntouchedNode(r) {
  return r.status === 'pending' && !r.attempt && !r.agent_run_id && !r.child_run_id && !r.output_ref;
}

/**
 * 把一条 run 的 `graph_run_nodes` 重新对齐**当前**的图定义（27-P3C §四.3 末）。
 *
 * 为什么需要它：`createRun` 是在建行的那个事务里、按**当时**的 `readDefGraph` 把节点行全插好的，
 * 而 `graph-engine.defIndex` 每推进一步都重新读**最新**定义。两者一旦分叉，症状全是静默的：
 *   · 定义里新增的节点没有 run 行 → `markReady` 第一句 `if (!rn …) return null` 直接吞掉，
 *     `seedEntryNodes` / `activateSuccessTargets` 都只 `if (activated) push`，没有任何日志；
 *   · 定义里删掉的节点留下孤儿 `pending` 行 → `settleRun` 把它们批量 skipped，
 *     图看起来跑完了，实际少跑了一半。
 * 这个洞**今天就存在**（createRun 到 startRun 之间几行代码，窗口毫秒级）；排队把窗口拉到小时级
 * ——「排队期间有人点重新规划组织」是一等的用户操作 —— 于是从理论缺陷变成必然事件。
 *
 * **只对 `run_status = 'queued'` 的 run 生效 —— 白名单，不是黑名单**（批 3 复核 F2）。
 * 曾经写成「stopped / completed 之外都放行」，那等于把**正在跑的 run** 也放了进来。
 * 多进程（pm2 cluster）下的形状：P1、P2 同时 tick 读到同一条 task → P2 先赢下出队 CAS，
 * run 转 running 开始执行 → P1 这时才走到本函数，run 是 running 不在黑名单里 → 它按
 * **旧定义之外**的判据删掉一批 pending 行 → P1 随后的 CAS 返回 0，一切「正常」地 continue
 * → 这条 run 少了几个节点，`settleRun` 数不到活跃节点直接判 completed。
 * 任务「成功」了，却少跑了一段图，全程零报错。白名单让这种交叠只可能删到还在队里的 run。
 *
 * 非 queued 一律返回 null 且不报警：调用方（tick 5b / promoteQueuedTask）紧接着就是一次
 * `queued → ready` 的 CAS，它同样会返回 null，「这轮没抢到」在调用方那里本来就是可见的。
 *
 * 孤儿行只删**一点执行痕迹都没有**的（pending 且 attempt=0、没 agent_run_id / child_run_id /
 * output_ref）。以前的判据只有 `status='pending'`，并在注释里声称「带痕迹的都留着」——
 * 那句话是假的：`resetRunNodes` / `resetForwardDownstream` / `reopenChildRun` 会把**跑过的**
 * 节点写回 pending（保留 attempt、清 fail_reason），于是 `status='pending'` ≠ 没跑过，
 * 那条 warn 也不会响。批 3 之后这更不是理论问题：`requeueBlockedRootRun` 会把一条跑过一半的
 * run 退回 queued，它下一次出队就会走到这里。留下的行计入 kept 并告警。
 *
 * ponytail: 这是「定义变了就补行」的最小实现，默认新节点从 pending 开始是安全的。
 * 将来若允许对**跑到一半**的 run 做 replan，必须升级成带 contract_hash 判据的级联重置
 * （loops-l3.md 那套四分类），这里的 kept 计数就是升级的触发信号。
 *
 * @returns {{added: string[], removed: string[], kept: string[]}|null} null = run 不存在或不在队列里
 */
function syncRunNodesToDef(runId) {
  const db = getDb();
  const run = getRun(runId);
  if (!run) return null;
  if (run.run_status !== 'queued') return null;

  const defNodeIds = new Set(readDefGraph(run.graph_kind, run.graph_id).nodes.map((n) => n.defNodeId));
  const rows = db
    .prepare(`
      SELECT id, def_node_id, status, attempt, agent_run_id, child_run_id, output_ref
        FROM graph_run_nodes WHERE run_id = ?
    `)
    .all(runId);
  const have = new Set(rows.map((r) => r.def_node_id));

  const now = new Date().toISOString();
  const added = [];
  const removed = [];
  const kept = [];

  db.transaction(() => {
    const ins = db.prepare(`
      INSERT INTO graph_run_nodes (id, run_id, def_node_id, status, created_at, updated_at)
      VALUES (?, ?, ?, 'pending', ?, ?)
    `);
    for (const defNodeId of defNodeIds) {
      if (have.has(defNodeId)) continue;
      ins.run(newId('rn'), runId, defNodeId, now, now);
      added.push(defNodeId);
    }
    // WHERE 里把「没跑过」的判据整条复述一遍，而不是只靠上面读到的快照：读和删之间
    // 另一个连接仍可能刚给这一行写上 agent_run_id（这正是白名单挡不住的那半个窗口）。
    const del = db.prepare(`
      DELETE FROM graph_run_nodes
       WHERE id = ? AND status = 'pending' AND attempt = 0
         AND agent_run_id IS NULL AND child_run_id IS NULL AND output_ref IS NULL
    `);
    for (const r of rows) {
      if (defNodeIds.has(r.def_node_id)) continue;
      if (!isUntouchedNode(r) || !del.run(r.id).changes) { kept.push(r.def_node_id); continue; }
      removed.push(r.def_node_id);
    }
  })();

  if (kept.length) {
    // 静默不可接受：这一条意味着「定义删掉的节点上有执行痕迹」，属于上面 ponytail 说的升级信号
    console.warn(`[work-graph-db] syncRunNodesToDef run=${runId} 保留了带执行痕迹的孤儿节点: ${kept.join(', ')}`);
  }
  return { added, removed, kept };
}

/**
 * 取一次 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',
};

/**
 * 当前占着这家公司执行坑的 root run 数。
 *
 * WHERE 与 `idx_graph_runs_one_active_root` **逐字同源**（同一个 ACTIVE_ROOT_RUN_STATUS 拼出来）。
 * 一旦两边分叉，症状是「事务守卫放行、索引拒绝」的 500 —— 一个把关的说行、另一个说不行，
 * 而报出来的只有一句 SQLite 原文。所以计数只有这一个持有者，调度器也调它，不许自己再写一遍 SQL。
 *
 * @param {string} companyId
 * @param {object} [db] 传进来是为了让准入的计数与随后的 UPDATE 落在**同一个事务**里
 */
function countActiveRootRuns(companyId, db = getDb()) {
  return listActiveRootRunIds(companyId, db).length;
}

/**
 * 占着坑的是**谁**。`countActiveRootRuns` 由它派生（`.length`），所以那条 WHERE 全仓库
 * 仍然只有这一个持有者 —— 为了给 `RunSlotBusyError` 填 heldBy 而另抄一条 SELECT，
 * 就是给「守卫放行、索引拒绝」那类分叉再开一个口子。活跃集最多几条，取 id 与取 COUNT 同价。
 */
function listActiveRootRunIds(companyId, db = getDb()) {
  const marks = ACTIVE_ROOT_RUN_STATUS.map(() => '?').join(',');
  return db
    .prepare(`
      SELECT id FROM graph_runs
       WHERE parent_run_id IS NULL
         AND run_status IN (${marks})
         AND company_id = ?
       ORDER BY created_at
    `)
    .all(...ACTIVE_ROOT_RUN_STATUS, companyId)
    .map((r) => r.id);
}

/**
 * 条件更新 run 状态，语义与 updateNodeStatus 完全一致（同样返回受影响行数供幂等判定）。
 *
 * 外加**准入守卫**（27-P3C 批 3）：见下方注释。坑满时抛 `RunSlotBusyError`。
 * @returns {number}
 * @throws {InvalidTransitionError|RunSlotBusyError}
 */
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];
    }
  }
  const write = () => db
    .prepare(`UPDATE graph_runs SET ${sets.join(', ')} WHERE id = @id AND run_status = @from`)
    .run(params).changes;

  // ── 准入守卫：数坑 + 占坑必须是一个原子步 ─────────────────────────────
  //
  // 判据是「**从活跃集外进入活跃集**」，不是「转成某个具体状态」：
  //   · `ready→running` / `running→paused` / `paused→running` 都是**集内移动**，占的还是
  //     原来那个坑。按 to 判就会把这条 run 自己算成竞争者（count 已经包含它），
  //     max=1 时第一条 run 一起跑就把自己挡死 —— 而且是「起不来还不报错」那种挡死。
  //   · `queued→ready` / `planning→ready` / `blocked→running` / `completed→ready` /
  //     `failed→ready` 才是真的进坑（这几个 from 都不在 ACTIVE_ROOT_RUN_STATUS 里）。
  // 只管 root run：那条唯一索引也只约束 `parent_run_id IS NULL`，子 run 从不占公司级的坑。
  //
  // ⚠️ **后三条同时是「恢复」路径**（人加完预算点启动 / 子 run 救回父 run / 对已收敛 run
  // 里的节点点重试），而批 3 让「坑一空就被队里下一条填上」成了必然事件 —— 于是恢复在
  // 有队列的公司里几乎必然撞坑满。本函数**不为恢复开后门**（开了就绕过唯一索引，
  // 换成一句 SQLite 原文的 500）：它只负责干净地拒绝（抛之前一个字都没写），
  // 「拒绝之后怎么办」由 `graph-engine.admitOrRequeue` 决定 —— 那里会把这条 root run
  // 退回队列（blocked → queued），让调度器在坑空时按位次把它接回来。
  //
  // 为什么挂在 updateRunStatus 而不是各个调用点：实测有 6 条路径能把 root run 推进活跃集
  // （markRunReady / 调度器 queued→ready / startRunSync 恢复 / retryNode /
  // reviveParentFromChild / nextStep），枚举一次漏一次（§五.1.1 那笔老账）。
  // updateRunStatus 是它们共同的唯一漏斗，挂这里一处覆盖全部。
  const admitting = !cur.parent_run_id
    && ACTIVE_ROOT_RUN_STATUS.includes(to)
    && !ACTIVE_ROOT_RUN_STATUS.includes(expected);
  if (!admitting) return write();

  // BEGIN IMMEDIATE：**读之前**就拿到写锁，于是「数出 0 个活跃」到「自己占上」之间
  // 不可能有第二个连接插进来各自数出 0（check-then-act 的经典超发）。
  // 同进程内 better-sqlite3 是同步单写者本来就串行，这把锁防的是多进程 / 多连接。
  // 唯一索引仍在，是兜底那一层：主守卫（本事务）与兜底（DB 约束）并存，批 4 只 DROP 后者。
  return db.transaction(() => {
    // 事务内重读才是权威的。状态已经不是 expected → 这次调用本来就是幂等 no-op
    // （下面 UPDATE 的 WHERE 同样匹配不到），返回 0。**绝不能**当成坑满抛错：
    // 那是把「别人抢先做完了」误报成「容量不足」，两件事的处置完全不同。
    const fresh = db.prepare(`SELECT run_status FROM graph_runs WHERE id = ?`).get(runId);
    if (!fresh || fresh.run_status !== expected) return 0;

    const heldBy = listActiveRootRunIds(cur.company_id, db);
    if (heldBy.length >= MAX_ACTIVE_ROOT_RUNS) {
      throw new RunSlotBusyError(cur.company_id, runId, heldBy, MAX_ACTIVE_ROOT_RUNS);
    }
    return write();
  }).immediate();
}

/** 只改软信号，不动 run_status（「暂停」/「停止」是下一个节点边界才生效的软信号） */
function setRunFlags(runId, { pauseRequested, stopRequested } = {}) {
  const sets = ['updated_at = @now'];
  const params = { id: runId, now: new Date().toISOString() };
  if (pauseRequested !== undefined) {
    sets.push('pause_requested = @pause');
    params.pause = pauseRequested ? 1 : 0;
  }
  if (stopRequested !== undefined) {
    sets.push('stop_requested = @stop');
    params.stop = stopRequested ? 1 : 0;
  }
  return getDb()
    .prepare(`UPDATE graph_runs SET ${sets.join(', ')} WHERE id = @id`)
    .run(params).changes;
}

/** run 级回退总数 +1（§四.D.2.1 双层硬顶的外层），返回加完之后的值 */
function bumpBackwardCount(runId) {
  const db = getDb();
  db.prepare(`UPDATE graph_runs SET backward_count = backward_count + 1, updated_at = ? WHERE id = ?`)
    .run(new Date().toISOString(), runId);
  const row = db.prepare(`SELECT backward_count FROM graph_runs WHERE id = ?`).get(runId);
  return row ? row.backward_count : null;
}

/** 某个 run 的直接子 run（级联暂停/停止用） */
function listChildRuns(runId) {
  return getDb().prepare(`SELECT * FROM graph_runs WHERE parent_run_id = ? ORDER BY created_at`).all(runId);
}

/** 同一棵 run 树的全部 run（按 root_run_id，预算扣减与三层级联都按这个口径） */
function listRunTree(rootRunId) {
  return getDb()
    .prepare(`SELECT * FROM graph_runs WHERE root_run_id = ? ORDER BY created_at`)
    .all(rootRunId);
}

/** 一家公司的 run 列表（T6 的 GET /api/company/:cid/graph-runs） */
function listCompanyRuns(companyId, { rootOnly = false } = {}) {
  return getDb()
    .prepare(`
      SELECT * FROM graph_runs
      WHERE company_id = ? ${rootOnly ? 'AND parent_run_id IS NULL' : ''}
      ORDER BY created_at DESC
    `)
    .all(companyId);
}

// ────────────────────────────────────────────────────────────────
// 27-P3C goal / task 两级模型（§四.1 / §四.2 / §四.4 / §四.5）
//
// 本节只管两张新表的存取。排队（run_status='queued' / 调度器 / depends_on 的出队判据）
// 是另一批的事，这里一个字都不碰 —— depends_on 在本批只被**校验**，不被消费。
// ────────────────────────────────────────────────────────────────

/**
 * 改了已冻结的字段（§四.4）。路由层 `company.js:sendDomainError` 用 `instanceof` 翻成 409 + hint。
 *
 * 为什么必须拒绝、而不是「顺手把 artifact 也更新一下」：task 的 description 在入队时就写进了
 * runs/{runId}/artifacts/task-goal.json，describeTask 等 agent 侧链路读的是 artifact。
 * 只改库不改 artifact，两份记录就对不上；两份都改则等于**篡改已被 agent 消费过的审计输入**。
 * 要改就取消后新建 —— 这句话直接进 hint。
 */
class FrozenFieldError extends Error {
  constructor(scope, id, fields) {
    super(`${scope} ${id} 已下达，${fields.join(' / ')} 不可再改（要改就取消后新建）`);
    this.name = 'FrozenFieldError';
    this.scope = scope; // 'goal' | 'task'
    this.id = id;
    this.fields = fields;
  }
}

/** depends_on 不合法（§四.2）。路由层翻 400。校验在插入之前跑完，库里不留半条脏数据。 */
class DependsOnError extends Error {
  constructor(message, detail = {}) {
    super(message);
    this.name = 'DependsOnError';
    Object.assign(this, detail);
  }
}

/** depends_on 列存的是 JSON 文本；解不出来当空数组 —— 一条脏行不该让整张 goal 的读取炸掉 */
function parseDependsOn(raw) {
  try {
    const v = JSON.parse(raw || '[]');
    return Array.isArray(v) ? v.map(String) : [];
  } catch {
    return [];
  }
}

/** 允许被 patch 的 goal 列（白名单，照抄 RUN_PATCH_COLUMNS 那套，不让任意键拼进 SQL） */
const GOAL_PATCH_COLUMNS = {
  title: 'title',
  description: 'description',
  status: 'status',
  priority: 'priority',
  okrNodeId: 'okr_node_id',
  reportTo: 'report_to',
  planningTokens: 'planning_tokens',
  validationTokens: 'validation_tokens',
  verdict: 'verdict',
  planningError: 'planning_error',
  dispatchedAt: 'dispatched_at',
  archivedAt: 'archived_at',
};

/**
 * 建一个 goal。产出一律是 draft —— 入队是另一个动作（§六.2 的 dispatch），建行本身没有副作用。
 *
 * @returns {object} 刚插入的整行（snake_case，与 getRun 同口径）
 */
function createGoal({
  companyId,
  title,
  description,
  priority = 'P1',
  okrNodeId = null,
  createdBy = null,
  reportTo = null,
} = {}) {
  if (!companyId) throw new Error('createGoal: companyId 必填');
  if (!title) throw new Error('createGoal: title 必填');
  // description 就是验收边界：空描述等于没有可验收的东西。让它在建行时炸，
  // 好过让 goal 级 org-validator 拿着空字符串去判「达成了没有」——那一头判不出错，只会瞎判。
  if (!description) throw new Error('createGoal: description 必填');

  const now = new Date().toISOString();
  const ins = getDb().prepare(`
    INSERT INTO goals (id, company_id, title, description, status, priority,
                       okr_node_id, created_by, report_to, created_at, updated_at)
    VALUES (?, ?, ?, ?, 'draft', ?, ?, ?, ?, ?, ?)
  `);
  const id = insertWithFreshId('goals', genGoalId, (gid) =>
    ins.run(gid, String(companyId), String(title), String(description), priority,
            okrNodeId, createdBy, reportTo, now, now));
  return getGoal(id);
}

/** 读一个 goal；不存在返回 null */
function getGoal(goalId) {
  return getDb().prepare(`SELECT * FROM goals WHERE id = ?`).get(goalId) || null;
}

/**
 * 一家公司的 goal 列表。
 * includeArchived 默认 false：归档只是隐藏（行还在，随时能查回来），默认列表不该看见它。
 */
function listCompanyGoals(companyId, { status = null, includeArchived = false, limit = null } = {}) {
  const where = ['company_id = @companyId'];
  const params = { companyId };
  if (status) {
    where.push('status = @status');
    params.status = status;
  }
  if (!includeArchived) where.push('archived_at IS NULL');
  const lim = Number.isInteger(limit) && limit > 0 ? ` LIMIT ${limit}` : '';
  return getDb()
    .prepare(`SELECT * FROM goals WHERE ${where.join(' AND ')} ORDER BY created_at DESC${lim}`)
    .all(params);
}

/**
 * 改 goal（§四.4 描述冻结的落地点之一）。
 *
 * dispatched_at 非空（首个 task 已入队）后带 description 一律抛 FrozenFieldError，
 * **只看键在不在、不看值一不一样**：「值没变就放行」会让调用方以为改描述有时候可以，
 * 那是最难排查的一种半生效。title / priority / okr_node_id 留着可改 —— 它们只是列表上的标签，
 * 不进任何 agent 的上下文。
 *
 * @param {string} goalId
 * @param {object} patch GOAL_PATCH_COLUMNS 里的键
 * @returns {object|null} 更新后的整行；goal 不存在返回 null（由调用方翻 404）
 * @throws {FrozenFieldError} 描述已冻结
 */
function updateGoal(goalId, patch = {}) {
  const cur = getGoal(goalId);
  if (!cur) return null;
  if (patch.description !== undefined && cur.dispatched_at) {
    throw new FrozenFieldError('goal', goalId, ['description']);
  }

  const sets = ['updated_at = @now'];
  const params = { id: goalId, now: new Date().toISOString() };
  for (const [key, col] of Object.entries(GOAL_PATCH_COLUMNS)) {
    if (patch[key] !== undefined) {
      sets.push(`${col} = @${key}`);
      params[key] = patch[key];
    }
  }
  getDb().prepare(`UPDATE goals SET ${sets.join(', ')} WHERE id = @id`).run(params);
  return getGoal(goalId);
}

/**
 * 校验一条 task 的 depends_on（§四.2：只能引用**同 goal 内已存在的** task，有环报错）。
 *
 * 为什么限定同 goal：depends_on 表达的是一个需求内部的先后。跨 goal 依赖会让出队判据
 * 跨越另一个 goal 的生命周期（上游 goal 归档了，下游永远等不到），而「静默躺在队里、
 * 界面上跟正常排队一模一样」正是这套模型要消灭的症状。
 *
 * 环检测在**写入之前**跑完并抛错，库里不留半条脏数据。判据是「把这条 task 的依赖换成新值后，
 * 从它出发能不能走回自己」—— 只查这一条的可达性，不做全图 DFS：其余边在此之前每次写入
 * 都过了这一关、已经是无环的，新引入的环必然经过被改的这条。
 *
 * @param {string} goalId
 * @param {string|null} taskId 正在改的 task；新建时传 null
 * @param {any} dependsOn
 * @returns {string[]} 归一化后的依赖（去重、保序）
 * @throws {DependsOnError}
 */
function validateDependsOn(goalId, taskId, dependsOn) {
  if (dependsOn === undefined || dependsOn === null) return [];
  if (!Array.isArray(dependsOn)) {
    throw new DependsOnError('depends_on 必须是数组', { goalId, taskId, dependsOn });
  }

  const deps = [...new Set(dependsOn.map(String))];
  if (taskId && deps.includes(taskId)) {
    throw new DependsOnError(`depends_on 不能依赖自己: ${taskId}`, { goalId, taskId });
  }

  const rows = getDb().prepare(`SELECT id, depends_on FROM tasks WHERE goal_id = ?`).all(goalId);
  const known = new Map(rows.map((r) => [r.id, parseDependsOn(r.depends_on)]));
  const missing = deps.filter((d) => !known.has(d));
  if (missing.length) {
    throw new DependsOnError(
      `depends_on 只能引用同 goal 内已存在的 task，这些不是: ${missing.join(', ')}`,
      { goalId, taskId, missing }
    );
  }

  // 新建时这条还不在库里，用一个不可能与 id 撞的占位名当起点
  const start = taskId || '__pending__';
  known.set(start, deps);
  const seen = new Set();
  const stack = [...deps];
  while (stack.length) {
    const cur = stack.pop();
    if (cur === start) {
      throw new DependsOnError(`depends_on 形成环: ${start} → … → ${start}`, { goalId, taskId, dependsOn: deps });
    }
    if (seen.has(cur)) continue;
    seen.add(cur);
    stack.push(...(known.get(cur) || []));
  }
  return deps;
}

/**
 * 在一个 goal 下建一条 task。同样一律是 draft，入队是另一个动作。
 *
 * @param {object} p
 * @param {string} p.goalId
 * @param {string} [p.companyId] 缺省从 goal 抄
 * @param {number} [p.seq] 缺省取该 goal 内当前最大 seq + 1
 * @param {string[]} [p.dependsOn=[]] 过 validateDependsOn
 * @param {object|string|null} [p.budgetJson] 这条 task 下达时要用的预算；对象会被序列化。
 *   **必须能在建行时就存下来**（缺陷 F3）：草稿路径（`POST /:cid/tasks` 撞 409 → 存草稿 →
 *   之后 dispatch）上，请求体里的 budget 早就不在了。只有写侧没有读侧的列 = 用户填了、
 *   接口回了 200、run 却拿着 DEFAULT_TASK_BUDGET 在跑，中途预算耗尽 blocked 且没人知道为什么。
 * @returns {object} 刚插入的整行
 */
function createTask({
  goalId,
  companyId = null,
  seq = null,
  title,
  description,
  executionMode = 'manual',
  dependsOn = [],
  priority = 'P1',
  budgetJson = null,
  reportTo = null,
  createdBy = null,
} = {}) {
  if (!goalId) throw new Error('createTask: goalId 必填');
  if (!title) throw new Error('createTask: title 必填');
  if (!description) throw new Error('createTask: description 必填');

  const db = getDb();
  const goal = getGoal(goalId);
  if (!goal) throw new Error(`createTask: goal 不存在: ${goalId}`);

  // company_id 这一列冗余存在的**唯一**理由是让调度器不必 JOIN；让调用方各传各的，
  // 就等于给这份冗余留了一条能对不上的路。默认从 goal 抄，传了也以 goal 为准的那种「以防万一」
  // 反而会掩盖调用方的 bug，所以显式传值时照传值走，由调用方负责。
  const cid = companyId || goal.company_id;
  const deps = validateDependsOn(goalId, null, dependsOn);
  const now = new Date().toISOString();

  // ponytail: seq 用 MAX+1，并发建行时可能撞。撞了只是列表里两行的先后不定
  // （seq 只用于排序，不参与任何判据），不值得为它上一把锁；
  // 真要严格连号就得把 seq 挪进 goals 上的计数器列，届时在同一个事务里自增。
  const nextSeq = Number.isInteger(seq)
    ? seq
    : db.prepare(`SELECT COALESCE(MAX(seq), 0) + 1 AS n FROM tasks WHERE goal_id = ?`).get(goalId).n;

  // 调用方给对象还是给已序列化的文本都收：库里这一列是 TEXT，两种形态在这里归一，
  // 免得各调用方各 JSON.stringify 一遍（漏一个就是把 "[object Object]" 存进库）。
  const budget = budgetJson == null || budgetJson === ''
    ? null
    : (typeof budgetJson === 'string' ? budgetJson : JSON.stringify(budgetJson));

  const ins = db.prepare(`
    INSERT INTO tasks (id, goal_id, company_id, seq, title, description, status,
                       execution_mode, depends_on, priority, budget_json, report_to, created_by,
                       created_at, updated_at)
    VALUES (?, ?, ?, ?, ?, ?, 'draft', ?, ?, ?, ?, ?, ?, ?, ?)
  `);
  const id = insertWithFreshId('tasks', genTaskId, (tid) => ins.run(
    tid, goalId, String(cid), nextSeq, String(title), String(description),
    executionMode, JSON.stringify(deps), priority, budget,
    // report_to 不落库就等于没有：请求作用域的闭包活不过一次排队（§四.7 禁令二）。
    // goal 上有就继承下来，别让每个调用方各记一遍。
    reportTo || goal.report_to || null,
    createdBy, now, now
  ));
  return getTask(id);
}

/** 读一条 task；不存在返回 null */
function getTask(taskId) {
  return getDb().prepare(`SELECT * FROM tasks WHERE id = ?`).get(taskId) || null;
}

/** 一个 goal 下的全部 task，按 seq 排（列表页的两级树就是这个顺序） */
function listGoalTasks(goalId, { includeArchived = false } = {}) {
  const where = ['goal_id = @goalId'];
  if (!includeArchived) where.push('archived_at IS NULL');
  return getDb()
    .prepare(`SELECT * FROM tasks WHERE ${where.join(' AND ')} ORDER BY seq, created_at`)
    .all({ goalId });
}

/**
 * 一家公司的 task 列表。
 *
 * 排序**故意**只按 created_at：队列位次的口径（priority → seq → created_at）只留调度器一处，
 * 这里再写一份就是第二个持有者 —— canDispatch 的取值域被前端复述了一遍、还多写一个值，
 * 导致公司失败一次后再也下不了任务，就是这么来的。要位次找调度器要。
 */
function listCompanyTasks(companyId, { status = null, goalId = null, includeArchived = false, limit = null } = {}) {
  const where = ['company_id = @companyId'];
  const params = { companyId };
  if (status) {
    where.push('status = @status');
    params.status = status;
  }
  if (goalId) {
    where.push('goal_id = @goalId');
    params.goalId = goalId;
  }
  if (!includeArchived) where.push('archived_at IS NULL');
  const lim = Number.isInteger(limit) && limit > 0 ? ` LIMIT ${limit}` : '';
  return getDb()
    .prepare(`SELECT * FROM tasks WHERE ${where.join(' AND ')} ORDER BY created_at DESC${lim}`)
    .all(params);
}

/** 允许被 patch 的 task 列。depends_on 不在这里 —— 它要先过环检测，单独处理。 */
const TASK_PATCH_COLUMNS = {
  title: 'title',
  description: 'description',
  status: 'status',
  priority: 'priority',
  seq: 'seq',
  executionMode: 'execution_mode',
  budgetJson: 'budget_json',
  currentRunId: 'current_run_id',
  attempt: 'attempt',
  reportTo: 'report_to',
  blockedReason: 'blocked_reason',
  queuedAt: 'queued_at',
  startedAt: 'started_at',
  finishedAt: 'finished_at',
  archivedAt: 'archived_at',
};

/**
 * 还能改 depends_on 的 task 状态（见 updateTask 的闸门 2）。
 * `blocked` 在里面是「强制放行」那条自救路径的**全部依赖**：改完依赖 route 会立刻 tick 一次，
 * `requeueDerivedBlockedTask` 把它连同 blocked_reason 一起推回 queued。
 */
const DEPENDS_ON_EDITABLE_STATUS = ['draft', 'queued', 'blocked'];

/** queued_at 非空后就不许再改的字段（§四.4）：它们都已经被 task-goal.json 或预算池消费过 */
const TASK_FROZEN_FIELDS = ['description', 'executionMode', 'budgetJson'];

/**
 * 改 task。
 *
 * 两道闸门：
 *   1. queued_at 非空 → description / execution_mode / budget 一律拒（理由见 FrozenFieldError）；
 *   2. depends_on **在 draft / queued / blocked 都允许改**（§六.3 定案）。真实场景是「B 已经
 *      排上了，才发现它得等 C」，此前只能取消重建、白丢排队位次。位次不受影响 —— 位次按
 *      priority → seq → created_at 排，depends_on 决定的是「够不够格出队」，两者正交。
 *      `blocked` 必须在里面：它是每轮 tick 从依赖**推导**出来的结果、不是不可逆状态，而
 *      「强制放行」（删掉那条已经死了的上游）正是要在 blocked 上改 depends_on。挡掉它等于
 *      界面上那个按钮必然 4xx —— 一条 blocked 任务再也没有自救的出路（批 3 对抗复核 F1）。
 *      跑起来之后就不许改了：那时依赖判据已经用过，再改只会让 timeline 上的因果对不上。
 *
 * @param {string} taskId
 * @param {object} patch TASK_PATCH_COLUMNS 里的键，外加 dependsOn
 * @returns {object|null} 更新后的整行；task 不存在返回 null
 * @throws {FrozenFieldError|DependsOnError}
 */
function updateTask(taskId, patch = {}) {
  const cur = getTask(taskId);
  if (!cur) return null;

  if (cur.queued_at) {
    const frozen = TASK_FROZEN_FIELDS.filter((f) => patch[f] !== undefined);
    if (frozen.length) throw new FrozenFieldError('task', taskId, frozen);
  }

  const sets = ['updated_at = @now'];
  const params = { id: taskId, now: new Date().toISOString() };

  if (patch.dependsOn !== undefined) {
    if (!DEPENDS_ON_EDITABLE_STATUS.includes(cur.status)) {
      throw new FrozenFieldError('task', taskId, ['dependsOn']);
    }
    params.dependsOn = JSON.stringify(validateDependsOn(cur.goal_id, taskId, patch.dependsOn));
    sets.push('depends_on = @dependsOn');
  }
  for (const [key, col] of Object.entries(TASK_PATCH_COLUMNS)) {
    if (patch[key] !== undefined) {
      sets.push(`${col} = @${key}`);
      params[key] = patch[key];
    }
  }

  getDb().prepare(`UPDATE tasks SET ${sets.join(', ')} WHERE id = @id`).run(params);
  return getTask(taskId);
}

/**
 * task 的终态：到了就不再被 run 收敛回写覆盖（重跑要走显式的 rerun，那会先把 status 推回 queued）。
 *
 * **`blocked` 不在里面**（27-P3C 第二轮复核 缺陷 1）。它的语义是「挂起，等人介入」，
 * 和 completed / failed / cancelled 不同类：预算耗尽走 `graph-engine.suspendRunTree`
 * （那段注释自己写着「不是失败」），run 转 blocked 后 `RUN_TRANSITIONS.blocked = ['running','stopped']`
 * 允许人加完预算后 `POST /api/graph-runs/:id/start` 接着跑，`reviveParentFromChild` 甚至会**自动**
 * 把 blocked 的父 run 拉回 running。把它算成终态的后果是下面 `settleTaskFromRun` 的幂等闸门
 * 反过来变成死锁：run 真跑完转 completed 时 UPDATE 命中 0 行，于是 run=completed 而
 * task 永远停在 blocked —— 静默、且没有任何一条路径能再把它改回来。
 */
const TASK_TERMINAL_STATUS = ['completed', 'failed', 'cancelled'];
/** goal 的终态：同上，`archived` 也算 —— 归档过的 goal 不该被一条迟到的 run 收敛翻回 done/failed */
const GOAL_TERMINAL_STATUS = ['done', 'failed', 'cancelled', 'archived'];

/**
 * run 收敛 → task 回写的**条件更新**（27-P3C §五.1.1 的钩子用，缺陷 F1）。
 *
 * 两个 WHERE 条件各挡一类事故，缺一不可：
 *   1. `current_run_id = @runId` —— 挡「上一次 attempt 的迟到收敛」。task 重跑后
 *      current_run_id 已指向新 run，旧 run 此刻才转 stopped 的话，不加这条就会把
 *      正在跑的第 2 次执行标成 cancelled，而界面上看不出任何异常。
 *   2. `status NOT IN (终态)` —— 幂等闸门。钩子挂在状态转换的尾巴上，
 *      同一条 run 被重开（`completed → ready` 是合法转换）再收敛会二次触发。
 *
 * @returns {number} 受影响行数；0 = 这次不该由我写（已终态 / 不是当前那次执行），调用方据此不做后续副作用
 */
function settleTaskFromRun(taskId, runId, status) {
  const now = new Date().toISOString();
  return getDb()
    .prepare(`
      UPDATE tasks SET status = @status, finished_at = @now, updated_at = @now
       WHERE id = @id
         AND current_run_id = @runId
         AND status NOT IN (${TASK_TERMINAL_STATUS.map((s) => `'${s}'`).join(', ')})
    `)
    .run({ id: taskId, runId, status, now }).changes;
}

/**
 * 「上游已经不可能满足了」这个 blocked 的原因前缀（27-P3C §五.1.2）。
 *
 * 它同时是**判据**：只有 `blocked_reason` 以它开头的 blocked task 才是调度器**推导**出来的，
 * 才允许被下一次 tick 推回 queued。别的来源的 blocked（最典型的是预算耗尽 —— run 转 blocked、
 * task-settle 跟着把 task 写成 blocked）绝不能被自动重排：那条 run 还躺在 blocked 等人加预算，
 * 把 task 推回 queued 只会让调度器反复去 CAS 一条根本不是 queued 的 run，界面上任务在
 * queued / blocked 之间抖动，而没有任何人真的在跑。
 */
const UPSTREAM_BLOCK_PREFIX = 'upstream_failed:';

/**
 * queued → blocked（依赖已不可能满足）。条件更新：只动**还在排队**的行。
 * 正在跑 / 已终态的任务不该被依赖推导碰到 —— 它们的状态由 run 收敛回写说了算。
 * @returns {number} 1 = 本次由我标的
 */
function blockQueuedTask(taskId, reason) {
  const now = new Date().toISOString();
  return getDb()
    .prepare(`
      UPDATE tasks SET status = 'blocked', blocked_reason = @reason, updated_at = @now
       WHERE id = @id AND status = 'queued'
    `)
    .run({ id: taskId, reason, now }).changes;
}

/**
 * blocked → queued（上游恢复了）。**这是「blocked 不落成不可逆状态」的兑现处**：
 * §五.1.2 承诺「blocked 每次 tick 重算一遍，上游恢复了就自己回 queued」，没有这个函数
 * 那句注释就是空头支票 —— A 重跑成功后 B 会永远躺在 blocked。
 *
 * 两个 WHERE 条件缺一不可：`status='blocked'` 是幂等闸门，`blocked_reason LIKE 'upstream_failed:%'`
 * 把「调度器推导的」和「别的来源的 blocked」分开（见 UPSTREAM_BLOCK_PREFIX）。
 * 写进去的 'queued' / NULL 都不在 WHERE 里。
 * @returns {number} 1 = 本次由我放回队列的
 */
function requeueDerivedBlockedTask(taskId) {
  const now = new Date().toISOString();
  return getDb()
    .prepare(`
      UPDATE tasks SET status = 'queued', blocked_reason = NULL, updated_at = @now
       WHERE id = @id AND status = 'blocked' AND blocked_reason LIKE @prefix
    `)
    .run({ id: taskId, now, prefix: `${UPSTREAM_BLOCK_PREFIX}%` }).changes;
}

/**
 * 出队认领：queued → running（27-P3C §五.1 第 5 步）。
 *
 * 必须是条件更新，不能用 `updateTask`（它的 WHERE 只有 `id`，是彻头彻尾的 check-then-act）：
 * 两个 tick / 两个进程同时看到同一条 queued task 时，无条件 UPDATE 会让两边都以为自己赢了，
 * 各起一条 run。
 *
 * 三个 WHERE 条件：
 *   1. `status = 'queued'` —— 真正的闸门。**写进去的 'running' 绝不能出现在这里**：
 *      SQLite 的 changes 数的是「匹配到的行」，把目标值也写进 WHERE 会让已被别人认领的行
 *      照样匹配，changes 恒为 1，闸门等于没有。
 *   2. `current_run_id = @runId` —— 认的必须是**我刚出队的那条 run**。少了它，
 *      一次迟到的认领会把另一条 run 的 task 抢过来记账。
 *   3. `archived_at IS NULL` —— 归档过的任务不该被调度器捡起来跑。
 *
 * `attempt` 在这里 +1（而不是在调用方读一遍再写回去）：读改写在并发下会丢代次，
 * 而 `attempt` 是 R-17.1 判「子 run 服务的是不是本轮」的依据，丢了会串轮。
 *
 * @returns {number} 1 = 认领成功；0 = 不是我的（被抢走 / 已取消 / 已归档），调用方必须回滚那条 run
 */
function claimQueuedTask(taskId, runId) {
  const now = new Date().toISOString();
  return getDb()
    .prepare(`
      UPDATE tasks
         SET status = 'running', attempt = attempt + 1, started_at = @now,
             blocked_reason = NULL, updated_at = @now
       WHERE id = @id AND current_run_id = @runId AND status = 'queued' AND archived_at IS NULL
    `)
    .run({ id: taskId, runId, now }).changes;
}

/** `requeueBlockedRootRun` 内部用来回滚事务的私有信号，不外泄 */
class RequeueRollback extends Error {}

/**
 * 恢复不成 → **退回队列**：blocked 的 root run + 它的 blocked task 一起回到 queued。
 *
 * 为什么需要它（27-P3C 批 3 复核 F1）：`updateRunStatus` 的准入守卫会拒掉一切「从活跃集外
 * 进活跃集」的转换，而恢复（人加完预算点启动 / 重试）恰恰全是这个形状。批 3 之后坑一空
 * 就被队里下一条填上，于是恢复几乎必然撞坑满。只是干净地报 409 并不够 —— 那条 run 留在
 * blocked（**非终态**），调度器只看 queued 的 task，人再点也是一样的赛跑，结果就是这条
 * 任务永远回不来、它的 goal 永不收敛。放回队列才是真的把它交还给调度器。
 *
 * **两行必须在同一个事务里**，顺序无所谓、原子性有所谓：
 *   · 只有 run 回 queued → 队列按 task 选候选，没人会来出队它 → 换了个姿势永远卡住；
 *   · 只有 task 回 queued → 调度器每轮都去 CAS 一条不是 queued 的 run，界面上任务在
 *     queued/blocked 之间抖动而没人真的在跑（`UPSTREAM_BLOCK_PREFIX` 那段注释描述的正是它）。
 *
 * 三个前提缺一不可：root run（子 run 不占坑也不进队列）、run 与 task 都还在 blocked
 * （别的状态说明有人已经处置过了）、task 未归档。写进去的 'queued' 都不在 WHERE 里。
 *
 * ⚠️ 它**不走** `sse.transitionRun`：那里的 apply 只能改 run 一张表，包不住 task 那一行。
 * 所以调用方（graph-engine.admitOrRequeue）必须自己补一条 timeline/SSE 事件 ——
 * 状态变了却没有事件，正是这套代码一直在消灭的东西。
 *
 * @returns {number} 1 = 本次由我放回队列；0 = 前提不成立（调用方据此照抛 RunSlotBusyError）
 */
function requeueBlockedRootRun(runId) {
  const db = getDb();
  const run = getRun(runId);
  if (!run || run.parent_run_id || run.run_status !== 'blocked' || !run.task_id) return 0;
  const now = new Date().toISOString();
  try {
    return db.transaction(() => {
      const t = db
        .prepare(`
          UPDATE tasks SET status = 'queued', blocked_reason = NULL, started_at = NULL, updated_at = @now
           WHERE id = @taskId AND current_run_id = @runId AND status = 'blocked' AND archived_at IS NULL
        `)
        .run({ taskId: run.task_id, runId, now }).changes;
      if (!t) return 0;
      const r = db
        .prepare(`
          UPDATE graph_runs SET run_status = 'queued', updated_at = @now
           WHERE id = @runId AND run_status = 'blocked' AND parent_run_id IS NULL
        `)
        .run({ runId, now }).changes;
      if (!r) throw new RequeueRollback(); // 别人抢先动了 run：整个事务回滚，task 不许独自留在 queued
      return 1;
    }).immediate();
  } catch (err) {
    if (err instanceof RequeueRollback) return 0;
    throw err;
  }
}

/**
 * 把「为什么还卡着」写到 task 行上，供任务列表直接显示（27-P3C 批 3 复核 F1 场景 C）。
 *
 * 用途是**留痕**，不是状态机：只动仍然 blocked、且 current_run_id 指着这条 run 的行，
 * 于是一条迟到的备注不会盖掉别人的处置结果。写进去的 reason 不在 WHERE 里。
 *
 * ⚠️ reason **不许**带 `UPSTREAM_BLOCK_PREFIX`：带了就会被调度器当成「依赖推导出来的
 * blocked」自动推回 queued，而这类 blocked（预算耗尽 / 救援失败）本来就等着人来处置。
 * @returns {number} 1 = 记上了
 */
function noteTaskBlockedReason(taskId, runId, reason) {
  if (!taskId) return 0;
  return getDb()
    .prepare(`
      UPDATE tasks SET blocked_reason = @reason, updated_at = @now
       WHERE id = @id AND current_run_id = @runId AND status = 'blocked'
    `)
    .run({ id: taskId, runId, reason, now: new Date().toISOString() }).changes;
}

/**
 * goal 收敛的 compare-and-swap 闸门（缺陷 F2）。
 *
 * 为什么必须是条件更新而不是「读一下再写」：一个 goal 下的 N 个 task 可能几乎同时收敛，
 * 每一条都会算出「全部 task 都到终态了」并试图收敛 goal。goal 收敛带着两个**不可重放**的
 * 副作用（写 OKR 节点、POST 唤醒上级 agent），做两遍就是同一个节点被写两次、上级被叫醒两次。
 * 让 SQLite 的单写者语义来决出唯一的赢家：changes = 1 的那一个才有资格做副作用。
 *
 * @returns {number} 1 = 本次抢到收敛权；0 = 别人已经收敛过（或 goal 不存在）
 */
function settleGoalStatus(goalId, status) {
  const now = new Date().toISOString();
  return getDb()
    .prepare(`
      UPDATE goals SET status = @status, updated_at = @now
       WHERE id = @id AND status NOT IN (${GOAL_TERMINAL_STATUS.map((s) => `'${s}'`).join(', ')})
    `)
    .run({ id: goalId, status, now }).changes;
}

/** `claimGoalSlot` / `settleGoalVerdict` 允许操作的列（白名单，绝不让调用方把列名拼进 SQL） */
const GOAL_SLOT_COLUMNS = ['verdict', 'reported_at'];

/**
 * 认领 goal 上的一个「占位 → 唯一 token」闸门（第二轮复核：认领语句不能自己构成它要避开的形态）。
 *
 * 这是 `verdict` / `reported_at` 两格共用的原语：**旧值必须精确匹配 `from`，新值 `to` 全局唯一**。
 * 两个条件缺一不可 ——
 *   · `from` 不精确（比如写成 `verdict IN (占位A, 占位B)`）而 `to` 恰好也是 `占位B` 时，
 *     已经被别人认领的行照样匹配，`changes` 恒为 1，**闸门等于没有**：两个进程各拿一个 1，
 *     各起一次 org-validator、各写一次 OKR、各叫醒一次上级。SQLite 的 changes 数的是
 *     「匹配到的行」，不管值变没变，所以「相同值写回」永远是 1。
 *   · `to` 不唯一时，即使 WHERE 只认上一个态，两个进程也可能一前一后各命中一次
 *     （A 认领 → A 的值 = B 要找的值 → B 再认领一次）。唯一 token 让「谁认领的」本身可判定，
 *     后面的回写才能用 `WHERE 列 = 我的 token` 决出赢家。
 *
 * @param {string} goalId
 * @param {'verdict'|'reported_at'} column
 * @param {{from: string|null, to: string}} p from=null 表示要求该列当前为 NULL
 * @returns {number} 1 = 认领成功；0 = 旧值已经不是 from（别人先到，或值已被改写）
 */
function claimGoalSlot(goalId, column, { from = null, to } = {}) {
  if (!GOAL_SLOT_COLUMNS.includes(column)) throw new Error(`claimGoalSlot: 不支持的列 ${column}`);
  if (!to) throw new Error('claimGoalSlot: to 必填（认领值必须唯一，不能为空）');
  const cond = from === null ? `${column} IS NULL` : `${column} = @from`;
  return getDb()
    .prepare(`
      UPDATE goals SET ${column} = @to, updated_at = @now
       WHERE id = @id AND ${cond}
    `)
    .run({ id: goalId, from, to, now: new Date().toISOString() }).changes;
}

/**
 * goal 级验收结论的**条件回写**（第二轮复核：把 check-then-act 换成一条原子 UPDATE）。
 *
 * 原来是「`getGoal` 读一下判词还是不是我认领的那个 → `updateGoal` 写回去」。这在单进程里也只是
 * 侥幸成立，跨进程根本不成立：读和写之间隔着完整的一次 JS 事件循环，另一个实例完全可以在这中间
 * 把这条定案掉，于是两边都判「还是我的」，判词互相覆盖、汇报做两次。
 * 现在把判据放进 UPDATE 自己的 WHERE 里，靠 `changes` 决出唯一赢家 —— 只有赢家才有资格汇报。
 *
 * `validation_tokens` 用 SQL 侧自增而不是「读出来 +tokens 再写回」：后者拿的是旧快照，
 * 拆分链路（planning_tokens）在同一行上并发写时会被整行回写抹掉。
 *
 * @returns {number} 1 = 本次回写生效（可以去汇报）；0 = 判词已被别处改写（绝不能再汇报）
 */
function settleGoalVerdict(goalId, { claim, status, verdict, addValidationTokens = 0 } = {}) {
  if (!claim) throw new Error('settleGoalVerdict: claim 必填（没有认领 token 就没有闸门）');
  const add = Number.isFinite(addValidationTokens) ? addValidationTokens : 0;
  return getDb()
    .prepare(`
      UPDATE goals
         SET status = @status, verdict = @verdict,
             validation_tokens = validation_tokens + @add,
             updated_at = @now
       WHERE id = @id AND verdict = @claim
    `)
    .run({ id: goalId, claim, status, verdict, add, now: new Date().toISOString() }).changes;
}

/**
 * 拆分收工的 compare-and-swap 闸门（批 2 缺陷 1）。同 `settleGoalStatus` 那招，
 * 差别是这里的判据是「**还在** planning 且没被归档」而不是「还没到终态」。
 *
 * 为什么不能用 `updateGoal(goalId, { status: 'draft' })`：那是一条无条件 UPDATE。
 * 拆分要跑几十秒，这期间 goal 完全可能已经被别人推走 ——
 *   · 用户点了表单上的「跳过，按单任务下达」→ `POST /:cid/tasks` 建 task + 起 run +
 *     `markGoalDispatched` 写成 active；
 *   · 用户在左栏归档了这个 goal（只盖 archived_at，status 仍是 planning）。
 * 无条件写回 draft 就是把一个**正在烧钱跑 run 的 goal 标成草稿**，随后插进去的 N 条 draft task
 * 让 `settleGoalIfAllTasksDone` 永远看不到「全部终态」→ OKR 不写、report_to 不唤醒、
 * verdict 恒空，且全程无日志无报错。归档那条同理：插进去的 task 归属一个不可见的 goal，
 * 左栏（按 goal 挂树）看不到、`GET /:cid/tasks` 里却在，永远不会被归档、下达或发现。
 *
 * **没命中时仍然把 planning_tokens 记上**：那几轮 LLM 是真花了钱的，goal 层的花费本来就只有
 * 这一个记账口，不记 = 这半边花费永远看不见。只有 status / planning_error 是被放弃的 ——
 * planning_error 的语义（界面上是「自动拆分失败，已按单任务处理」）在 goal 已经跑起来时是假话。
 *
 * @param {string} goalId
 * @param {object} p
 * @param {string} [p.status='draft'] 命中时要写的状态
 * @param {number} [p.planningTokens=0] 本次拆分烧掉的 token；命不命中都记
 * @param {string|null} [p.planningError=null] 只在命中时写
 * @returns {{settled: boolean}} settled=false 表示 goal 已被别人推离 planning（或归档 / 不存在）
 */
/**
 * 只补记验收 token，不碰状态与判词。
 *
 * 用在「跨进程抢跑输了」那条路上：判词被别人定案了，但**这一次 org-validator 是真跑了、
 * 钱是真烧了**。不记的话这半边花费永远看不见 —— 与 settleGoalPlanning 那条「没命中也
 * 补记账」的分支同一个口径。tokens 为 0 时连 updated_at 都不碰。
 */
function bumpValidationTokens(goalId, tokens) {
  if (!(tokens > 0)) return 0;
  return getDb()
    .prepare(`UPDATE goals SET validation_tokens = validation_tokens + @t, updated_at = @now WHERE id = @id`)
    .run({ id: goalId, t: tokens, now: new Date().toISOString() }).changes;
}

function settleGoalPlanning(goalId, { status = 'draft', planningTokens = 0, planningError = null } = {}) {
  const db = getDb();
  const now = new Date().toISOString();
  const tokens = Number.isFinite(planningTokens) ? planningTokens : 0;

  const changes = db
    .prepare(`
      UPDATE goals
         SET status = @status, planning_tokens = @tokens, planning_error = @err, updated_at = @now
       WHERE id = @id AND status = 'planning' AND archived_at IS NULL
    `)
    .run({ id: goalId, status, tokens, err: planningError, now }).changes;
  if (changes) return { settled: true };

  // 只补记账。tokens 为 0 时连 updated_at 都不碰：拆分没花钱还去 touch 一个已经在跑的 goal，
  // 只会让「这行最后一次变动是什么时候」这个问题多一个假答案。
  if (tokens > 0) {
    db.prepare(`UPDATE goals SET planning_tokens = @tokens, updated_at = @now WHERE id = @id`)
      .run({ id: goalId, tokens, now });
  }
  return { settled: false };
}

/**
 * 一个 goal 内当前最大的 seq（含已归档的 task），没有 task 时返回 0（批 2 缺陷 4）。
 *
 * 拆分产出的 seq 必须以它为基准偏移：`idx_tasks_goal` 不是唯一索引，往非空 goal 里插 1..N
 * 不会报错，只会让同一个 goal 下出现两条 seq=1 —— 列表两行都显示 `#1`、依赖 chip 的
 * 「依赖 #1」出现两个同名候选、dispatch 挑「第一条跑得动的 draft」顺序不定。**从此对不上号。**
 *
 * 归档的 task 也要算进来：它们的号还挂在界面上（`includeArchived=1` 随时查得回来），
 * 复用等于让历史行和新行重号。
 */
function maxTaskSeq(goalId) {
  const row = getDb().prepare(`SELECT COALESCE(MAX(seq), 0) AS n FROM tasks WHERE goal_id = ?`).get(goalId);
  return (row && row.n) || 0;
}

/**
 * 一批 run 的节点完成度，**一条 SQL 出全部**（§五 / 验收 10）。
 *
 * 为什么强制批量：列表页要给每行 task 画进度条，20 goal × 3 task 逐个查就是 60 次往返 ——
 * 那正是验收 10 点名要挡的 N+1。所以这个函数只接受数组、只返回 Map，不留单条版本的入口：
 * 留了就一定有人在循环里调它。
 *
 * done 的口径 = success + skipped。skipped 是「按条件跳过」，它和成功一样**不会再动**；
 * 算进分母不算进分子，会让一张带跳过分支的图永远停在 80%。
 *
 * ponytail: run 数超过 SQLite 的变量上限（默认 999）时要分批。列表页一屏几十条，
 * 目前够不着；真到那天在这里切 chunk 即可，调用方形态不变。
 *
 * @param {string[]} runIds
 * @returns {Map<string,{total:number, done:number}>} 没有节点行的 run 不出现在 Map 里
 */
function taskProgress(runIds) {
  const ids = [...new Set((runIds || []).filter(Boolean).map(String))];
  if (!ids.length) return new Map();

  const rows = getDb()
    .prepare(`
      SELECT run_id,
             COUNT(*) AS total,
             SUM(CASE WHEN status IN ('success', 'skipped') THEN 1 ELSE 0 END) AS done
      FROM graph_run_nodes
      WHERE run_id IN (${ids.map(() => '?').join(', ')})
      GROUP BY run_id
    `)
    .all(...ids);

  return new Map(rows.map((r) => [r.run_id, { total: r.total, done: r.done }]));
}

// ────────────────────────────────────────────────────────────────
// 回退记录（§四.D.2.1 的备案账本，判词由 T7 写）
// ────────────────────────────────────────────────────────────────

/** 这条边在本 run 上已经退过几次（loop_count = 该值 + 1） */
function countLoopRecords(runId, edgeId) {
  return getDb()
    .prepare(`SELECT COUNT(*) AS n FROM graph_run_loop_records WHERE run_id = ? AND edge_id = ?`)
    .get(runId, String(edgeId)).n;
}

/**
 * 备案一次回退（§四.D.2.1 第 1 步）。loop_count 缺省按已有记录数 + 1 自动算。
 * @returns {{id, loopCount}}
 */
function insertLoopRecord({
  runId,
  edgeId,
  fromNodeRunId,
  toNodeRunId,
  loopCount,
  verdict = 'registered',
  remediationNote = null,
  targetNodeRef = null,
} = {}) {
  const db = getDb();
  const id = newId('lr');
  const n = loopCount === undefined || loopCount === null ? countLoopRecords(runId, edgeId) + 1 : loopCount;
  db.prepare(`
    INSERT INTO graph_run_loop_records
      (id, run_id, edge_id, from_node_run_id, to_node_run_id, loop_count, verdict,
       remediation_note, target_node_ref, created_at)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
  `).run(id, runId, String(edgeId), fromNodeRunId, toNodeRunId, n, verdict, remediationNote,
    targetNodeRef, new Date().toISOString());
  return { id, loopCount: n };
}

/** 改判词（registered → fixed / not_fixed / force_terminated），返回受影响行数 */
function updateLoopVerdict(recordId, verdict, remediationNote) {
  const sets = ['verdict = @verdict'];
  const params = { id: recordId, verdict };
  if (remediationNote !== undefined) {
    sets.push('remediation_note = @note');
    params.note = remediationNote;
  }
  return getDb()
    .prepare(`UPDATE graph_run_loop_records SET ${sets.join(', ')} WHERE id = @id`)
    .run(params).changes;
}

/** 本 run 的回退记录，最新在前（控制台「循环」tab + 下一轮读 remediation_note 用） */
function listLoopRecords(runId, { edgeId = null } = {}) {
  const db = getDb();
  return edgeId === null
    ? db.prepare(`SELECT * FROM graph_run_loop_records WHERE run_id = ? ORDER BY created_at DESC`).all(runId)
    : db
        .prepare(`
          SELECT * FROM graph_run_loop_records WHERE run_id = ? AND edge_id = ?
          ORDER BY created_at DESC
        `)
        .all(runId, String(edgeId));
}


// ────────────────────────────────────────────────────────────────
// R-19 产物索引（27-P4 §4.6）
//
// 分工与 output_ref 不重复：
//   output_ref  节点级指针 —— 「我这次跑出了什么」，一对一挂在 graph_run_nodes 行上
//   graph_files 公司级索引 —— 「这家公司有哪些产物、谁能用」，聚合视图，由前者派生
// 所以 graph_files 必须可从节点行重建（rebuildFilesFromNodes），索引坏了不至于丢产物。
// ────────────────────────────────────────────────────────────────

/**
 * 记一条产物实例。**只由引擎调用**，不给 agent 自己写 ——
 * 否则索引与实际文件必然漂移（本轮反复验证过这一点）。
 *
 * 同一 (company, task, relPath) 重复产出时：把旧行标 superseded_by 指向新行，
 * 而不是堆一堆同路径的行。这样「最新的那份」是唯一 superseded_by IS NULL 的行，
 * 历史仍可追溯。
 */
function recordFile({ companyId, taskId = null, relPath, kind, runId = null, nodeRunId = null }) {
  if (!companyId || !relPath || !kind) throw new Error('recordFile 需要 companyId / relPath / kind');
  const db = getDb();
  const now = new Date().toISOString();
  const id = newId('gf');

  const tx = db.transaction(() => {
    const prev = db
      .prepare(`
        SELECT id FROM graph_files
         WHERE company_id = ? AND COALESCE(task_id, '') = ? AND rel_path = ?
           AND superseded_by IS NULL
      `)
      .get(companyId, taskId || '', relPath);

    if (prev) {
      // 同路径再产出：先把旧行指向新行（腾出部分唯一索引的位置），再插新行。
      // 顺序不能反 —— 反了会撞 idx_graph_files_current。
      db.prepare(`UPDATE graph_files SET superseded_by = ? WHERE id = ?`).run(id, prev.id);
    }
    db.prepare(`
      INSERT INTO graph_files
        (id, company_id, task_id, rel_path, kind,
         produced_by_run_id, produced_by_node_run_id, created_at, superseded_by)
      VALUES (@id, @company_id, @task_id, @rel_path, @kind, @run_id, @node_run_id, @now, NULL)
    `).run({
      id, company_id: companyId, task_id: taskId, rel_path: relPath, kind,
      run_id: runId, node_run_id: nodeRunId, now,
    });
  });
  tx();
  return id;
}

/** 按 kind 查这家公司/这个 task 下最新的产物实例（superseded_by IS NULL） */
function findFiles({ companyId, taskId = null, kinds = null }) {
  const db = getDb();
  const rows = db
    .prepare(`
      SELECT * FROM graph_files
       WHERE company_id = ? AND COALESCE(task_id, '') = ? AND superseded_by IS NULL
       ORDER BY created_at
    `)
    .all(companyId, taskId || '');
  if (!kinds || !kinds.length) return rows;
  const want = new Set(kinds);
  return rows.filter((r) => want.has(r.kind));
}

/** 在边上声明「这条边要传什么 kind」。规划时写，只有 kind 没有实例。 */
function declareEdgeContract({ edgeLayer, edgeId, kind, required = true, role = 'produces' }) {
  if (!['org', 'work'].includes(edgeLayer)) throw new Error(`edgeLayer 非法: ${edgeLayer}`);
  if (!edgeId || !kind) throw new Error('declareEdgeContract 需要 edgeId / kind');
  // role 默认 produces：出边声明产出是压倒性多数，needs 必须显式写。
  // 去重键仍是 (layer, edge, kind) 而不含 role —— 同一条边对同一个 kind 既声明 produces
  // 又声明 needs 是自相矛盾的输入（「我产它」和「我要别人产它」），
  // normalizeContracts 的遍历顺序让 produces 先落库、矛盾的那条被 DO NOTHING 吞掉，
  // 即保留更强的那个断言。
  getDb()
    .prepare(`
      INSERT INTO graph_edge_contracts (id, edge_layer, edge_id, kind, required, role, created_at)
      VALUES (?, ?, ?, ?, ?, ?, ?)
      ON CONFLICT(edge_layer, edge_id, kind) DO NOTHING
    `)
    .run(
      newId('gec'), edgeLayer, String(edgeId), kind, required ? 1 : 0,
      role === 'needs' ? 'needs' : 'produces', new Date().toISOString()
    );
}

/** 某条边上声明的契约 */
function edgeContracts(edgeLayer, edgeId) {
  return getDb()
    .prepare(`SELECT * FROM graph_edge_contracts WHERE edge_layer = ? AND edge_id = ?`)
    .all(edgeLayer, String(edgeId));
}

/**
 * 某个节点的**入边**上声明了哪些 kind —— 这是「这个节点需要什么输入」的答案。
 * 只看 forward 入边：backward 入边是回退路径，不是数据来源。
 */
function inboundContracts(edgeLayer, edges, defNodeId) {
  const inbound = (edges || []).filter(
    (e) => String(e.to) === String(defNodeId) && (e.edgeKind || 'forward') !== 'backward'
  );
  const out = [];
  for (const e of inbound) {
    for (const c of edgeContracts(edgeLayer, e.id)) out.push({ ...c, fromNodeId: e.from });
  }
  return out;
}

/**
 * 从 graph_run_nodes 重建索引（索引损坏时的兜底，类比 okr 的 _rebuildIndex）。
 * 只能恢复「节点产出过什么目录」这一层，恢复不了 kind —— 所以 kind 统一记 'artifact'，
 * 这是有损重建，够用来定位文件，不够还原语义。
 */
function rebuildFilesFromNodes(companyId) {
  const db = getDb();
  const rows = db
    .prepare(`
      SELECT n.id AS node_run_id, n.run_id, n.output_ref, r.task_id
        FROM graph_run_nodes n JOIN graph_runs r ON r.id = n.run_id
       WHERE r.company_id = ? AND n.output_ref IS NOT NULL
    `)
    .all(companyId);
  let n = 0;
  for (const r of rows) {
    recordFile({
      companyId, taskId: r.task_id, relPath: r.output_ref, kind: 'artifact',
      runId: r.run_id, nodeRunId: r.node_run_id,
    });
    n += 1;
  }
  return n;
}

module.exports = {
  getDb,
  newId,

  // 校验
  validateWorkGraph,
  validateFailFallback,

  // 定义层
  createWorkGraph,
  getWorkGraph,
  readDefGraph,

  // 运行时层
  createRun,
  getRun,
  getRunState,
  syncRunNodesToDef,
  getRunNode,
  getRunNodeByDefId,
  updateNodeStatus,
  bumpNodeAttempt,
  updateRunStatus,
  // 「坑满没满」的唯一计数处（WHERE 与唯一索引同源），调度器复用它
  countActiveRootRuns,
  listActiveRootRunIds,
  setRunFlags,
  bumpBackwardCount,
  listChildRuns,
  listRunTree,
  listCompanyRuns,

  // R-19 产物索引（27-P4 §4.6）
  recordFile,
  findFiles,
  declareEdgeContract,
  edgeContracts,
  inboundContracts,
  rebuildFilesFromNodes,

  // 回退记录
  countLoopRecords,
  insertLoopRecord,
  updateLoopVerdict,
  listLoopRecords,

  // 27-P3C goal / task 两级模型
  genGoalId,
  genTaskId,
  createGoal,
  getGoal,
  listCompanyGoals,
  updateGoal,
  createTask,
  getTask,
  listGoalTasks,
  listCompanyTasks,
  updateTask,
  // validateDependsOn 故意**不导出**：它只有 createTask / updateTask 两个内部调用方，
  // 导出一个零外部调用方的函数就是 §九.4quater 点名的那类「写了没人用」的存量（缺陷 F10）。
  // 真有外部要单独校验时再导出，那时它会带着自己的调用方一起进来。
  taskProgress,
  maxTaskSeq,
  settleTaskFromRun,
  // 批 3 排队：三条都是条件更新，调用方靠 changes 判「是不是我做的」
  claimQueuedTask,
  blockQueuedTask,
  requeueDerivedBlockedTask,
  requeueBlockedRootRun,
  noteTaskBlockedReason,
  UPSTREAM_BLOCK_PREFIX,
  settleGoalStatus,
  settleGoalPlanning,
  bumpValidationTokens,
  claimGoalSlot,
  settleGoalVerdict,
  TASK_TERMINAL_STATUS,
  GOAL_TERMINAL_STATUS,
  // 这两个类有真调用方：company.js:sendDomainError 用 instanceof 判定（不是比 err.name）
  FrozenFieldError,
  DependsOnError,

  // 状态机
  NODE_TRANSITIONS,
  RUN_TRANSITIONS,
  // 唯一索引的 WHERE 与调度器的「坑满没满」判据共用这一个数组，别再抄字面量
  ACTIVE_ROOT_RUN_STATUS,
  // 并行度上限：批 3 恒为 1，批 4 改成读 company.json。调度器取这里，不许再写一个 1
  MAX_ACTIVE_ROOT_RUNS,
  canTransitionNode,
  canTransitionRun,
  InvalidTransitionError,
  RunSlotBusyError,
  // 「这次没抢到坑」的唯一判据，调度器与路由都取它，不许各写一份（理由见函数注释）
  isSlotRaceError,
};