宜博 / lemonboss

#!/usr/bin/env node
/**
 * migrate-runs-to-goal-task.js — 把存量 root run 回填成 goal + task 两级行
 *
 * 依据: docs/todo/system-design/27-p3c-goal-task-model.md §四.8
 *
 * 为什么必须有这个脚本：批 1 建了 goals / tasks 两张表，左栏接下来要改成读 `tasks`。
 * 而**既有的 root run 在 tasks 里一行都没有** —— 那一改，所有历史任务会从界面上整个消失。
 * 本文档要修的第一个问题（A：历史任务看不见）会被批 1 自己反向放大一次。所以回填不是
 * 收尾的可选项，它跟建表是同一件事的两半。
 *
 * 一条 root run 拆回两级：
 *   goals  一条 —— 人当初提的那句需求（描述从 artifacts/task-goal.json 的 task 字段读回）
 *   tasks  一条 —— 那次执行单元，seq=1、depends_on=[]、current_run_id 指回这条 run
 *   graph_runs.goal_id / task_id 回填指向新行（这两列从 P3 起就在写，一直没有表接得住）
 *
 * 用法:
 *   node scripts/migrate-runs-to-goal-task.js --dry-run   # 预演，事务回滚，不留任何行
 *   node scripts/migrate-runs-to-goal-task.js             # 实际执行
 *
 * 【批 5 追加】同一个脚本还负责第二类回填：`system/tasklog/*.json`（agent 之间派单的旧存储）
 * → `tasks` 表的 `executor='agent'` 行。两类回填放同一个脚本是 §十二.5 的要求：同一次数据形态
 * 变更分两个脚本，只会让「跑了哪个没跑哪个」变成新问题。
 *
 * 幂等: 见 alreadyMigrated() / tasklogAlreadyMigrated()。重复跑不会产生第二份 goal/task。
 * 库位置: 走 work-graph-db.getDb()（即 data/graph.sqlite，可被 GRAPH_DB_PATH 覆盖）——
 *         必须走它而不是自己 new Database：goals / tasks 两张表的 DDL 就在它的 getDb() 里，
 *         别的入口连表都建不出来。
 */

const fs = require('fs');
const path = require('path');
const wdb = require('../server/services/work-graph-db');

const REPO_ROOT = path.resolve(__dirname, '..');
const DRY = process.argv.includes('--dry-run');

const log = (...a) => console.log(DRY ? '[dry-run]' : '[migrate]', ...a);
const stats = { roots: 0, goals: 0, tasks: 0, runsPatched: 0, skipped: 0, noArtifact: 0,
  tasklogFiles: 0, tasklogGoals: 0, tasklogTasks: 0, tasklogSkipped: 0, tasklogBad: 0 };

/**
 * root run 的产物目录 —— 形状必须与 graph-engine.runRootDir 逐字一致：
 * 覆盖态是扁平的 {GRAPH_RUNS_DIR}/{rootRunId}（**没有** _companies/{cid}/runs 这一段），
 * 默认态才是 data/companies/{cid}/runs/{rootRunId}。两边算不出同一个字符串的后果不是报错，
 * 是「读不到 task-goal.json」→ 所有历史任务都退化成 '(历史任务)' 占位标题，而且一声不吭。
 * 与 company-paths 一样**每次调用读 env**，不在模块加载时快照：脚本可能被先 require 再设 env。
 */
function runRootDir(run) {
  const runsRoot = process.env.GRAPH_RUNS_DIR || null;
  return runsRoot
    ? path.join(runsRoot, run.id)
    : path.join(REPO_ROOT, 'data', 'companies', run.company_id, 'runs', run.id);
}

/** 读回当初下达的那句话；读不到返回 null（目录被清理、或 run 老到还没有这个 artifact） */
function readTaskGoal(run) {
  try {
    const j = JSON.parse(fs.readFileSync(path.join(runRootDir(run), 'artifacts', 'task-goal.json'), 'utf-8'));
    const task = typeof j.task === 'string' ? j.task.trim() : '';
    return task || null;
  } catch {
    return null;
  }
}

/**
 * run 状态 → task 状态。task 的取值域里没有 paused / ready / planning，
 * 那三种都还挂着一条活的 run，对任务列表来说就是「在跑」。
 * stopped 是人按的停，映射成 cancelled 而不是 failed —— 后者会让历史列表凭空多出一批红条。
 */
const TASK_STATUS = {
  completed: 'completed',
  failed: 'failed',
  blocked: 'blocked',
  stopped: 'cancelled',
  planning: 'running',
  ready: 'running',
  running: 'running',
  paused: 'running',
};

/**
 * run 状态 → goal 状态（§四.8：completed → done，其余 → active / cancelled）。
 * 注意 **failed 的 run 对应的 goal 是 active 而不是 failed**：goals.failed 的语义是
 * 「goal 级验收没过」（§四.10 的 org-validator 判词），历史 run 从没跑过那道验收，
 * 硬标 failed 是替一个没发生过的判决下结论。active = 还没收工、可补 task 重下，
 * 正是这批遗留 run 的真实处境。
 */
const GOAL_STATUS = { completed: 'done', stopped: 'cancelled' };

/** 列表用的短标题：取描述第一行，截 60 字 */
function deriveTitle(desc) {
  const line = String(desc).split('\n')[0].trim();
  return line.length > 60 ? `${line.slice(0, 60)}…` : line;
}

/**
 * 幂等判据（两条，缺一条都会漏）：
 *   1. 已有 task 的 current_run_id 指着它 —— 本脚本上一次跑出来的行
 *   2. run.task_id 已经是 tasks 表里的真行 —— 批 2 之后由新版 POST /:cid/tasks 正常建出来的，
 *      那种 run 根本不是「存量」，回填只会给它凭空多一个 goal
 */
function alreadyMigrated(db, run) {
  const byRun = db.prepare(`SELECT id FROM tasks WHERE current_run_id = ?`).get(run.id);
  if (byRun) return byRun.id;
  if (run.task_id) {
    const byId = db.prepare(`SELECT id FROM tasks WHERE id = ?`).get(run.task_id);
    if (byId) return byId.id;
  }
  return null;
}

/** 生成一个当前表里没被占用的 id（§四.5 的 4 位 hex 只有 65536 种取值，同毫秒批量建会撞） */
function freshId(db, table, gen) {
  for (let i = 0; i < 5; i++) {
    const id = gen();
    if (!db.prepare(`SELECT 1 FROM ${table} WHERE id = ?`).get(id)) return id;
  }
  throw new Error(`${table} 连续 5 次 id 碰撞，异常，中止`);
}

function migrateRun(db, run) {
  const done = alreadyMigrated(db, run);
  if (done) {
    log(`跳过（已回填）: run=${run.id} → task=${done}`);
    stats.skipped++;
    return;
  }

  const spoken = readTaskGoal(run);
  if (!spoken) stats.noArtifact++;
  const description = spoken || `历史任务 ${run.task_id || run.id}（原始描述已丢失）`;
  const title = spoken ? deriveTitle(spoken) : '(历史任务)';

  const goalId = freshId(db, 'goals', wdb.genGoalId);
  db.prepare(`
    INSERT INTO goals (
      id, company_id, title, description, status, priority, okr_node_id,
      created_by, report_to, created_at, updated_at, dispatched_at
    ) VALUES (?, ?, ?, ?, ?, 'P1', ?, 'migration', NULL, ?, ?, ?)
  `).run(
    goalId, run.company_id, title, description,
    GOAL_STATUS[run.run_status] || 'active',
    // 禁令一（§四.7）：okr_node_id 只存在 goals 上。run 上那一份**不清** ——
    // 存量 run 的这列是历史事实，清空是不可逆的破坏性写，超出「回填」的授权范围。
    run.okr_node_id || null,
    run.created_at, run.updated_at,
    // 历史任务早就下达过了，dispatched_at 非空 = 描述立即冻结（§四.4）。本来也不该再改。
    run.created_at
  );
  stats.goals++;

  // task id 优先**沿用 run.task_id**：产物目录是 tasks/{taskId}/（company-paths.taskDirRel），
  // 换一个新 id 等于让已落盘的历史产物全部认不回来。只有 task_id 为空或已被占用才新生成。
  let taskId = run.task_id && !db.prepare(`SELECT 1 FROM tasks WHERE id = ?`).get(run.task_id)
    ? run.task_id
    : freshId(db, 'tasks', wdb.genTaskId);

  const terminal = ['completed', 'failed', 'blocked', 'stopped'].includes(run.run_status);
  db.prepare(`
    INSERT INTO tasks (
      id, goal_id, company_id, seq, title, description, status, execution_mode,
      depends_on, budget_json, current_run_id, attempt, report_to, created_by,
      blocked_reason, priority, created_at, updated_at, queued_at, started_at, finished_at
    ) VALUES (?, ?, ?, 1, ?, ?, ?, ?, '[]', ?, ?, 1, NULL, 'migration', NULL, 'P1', ?, ?, ?, ?, ?)
  `).run(
    taskId, goalId, run.company_id, title, description,
    TASK_STATUS[run.run_status] || 'running',
    run.execution_mode || 'manual',
    run.budget_json || null,
    run.id,
    run.created_at, run.updated_at,
    // queued_at / started_at 都取 run.created_at：这条 run 存在本身就是「已入队且已起跑」的证据，
    // 真实时刻无处可考，用建 run 的时间是唯一不会编造顺序的选择。
    run.created_at, run.created_at,
    terminal ? run.updated_at : null
  );
  stats.tasks++;

  db.prepare(`UPDATE graph_runs SET goal_id = ?, task_id = ? WHERE id = ?`).run(goalId, taskId, run.id);
  stats.runsPatched++;

  log(`回填: run=${run.id} [${run.run_status}] → goal=${goalId} task=${taskId} 「${title}」`);
}

function main() {
  log(DRY ? '=== DRY RUN（事务结束回滚，不留任何行）===' : '=== 开始回填 ===');

  const db = wdb.getDb();
  const roots = db.prepare(`
    SELECT * FROM graph_runs WHERE parent_run_id IS NULL ORDER BY created_at
  `).all();
  stats.roots = roots.length;
  log(`root run 共 ${roots.length} 条`);

  // 预演也走真事务再回滚，不是「if (!DRY) 跳过写」：后者撞不上外键与唯一约束，
  // 预演报「将建 N 条」而真跑第 3 条就炸，等于没预演。
  const ROLLBACK = new Error('dry-run rollback');
  try {
    db.transaction(() => {
      for (const run of roots) migrateRun(db, run);
      migrateAllTasklog(db);
      if (DRY) throw ROLLBACK;
    })();
  } catch (e) {
    if (e !== ROLLBACK) throw e;
  }

  console.log('\n' + (DRY ? '[dry-run] ' : '') + '汇总:', JSON.stringify(stats));
  if (stats.noArtifact) {
    console.log(`注意: ${stats.noArtifact} 条 run 读不到 artifacts/task-goal.json，描述用占位文本。`);
  }
  if (DRY) console.log('这是 dry-run，事务已回滚，未做任何改动。去掉 --dry-run 实际执行。');
  else console.log('完成。');
}


// ════════════════════════════════════════════════════════════════
// 批 5：system/tasklog/*.json → tasks 表（executor='agent'）
// ════════════════════════════════════════════════════════════════

const TASKLOG_DIR = path.join(REPO_ROOT, 'system', 'tasklog');

/**
 * 派单任务落在哪家公司名下。
 *
 * tasklog 的行**没有 company 这一维** —— 它记的是「谁派给谁」，而 from/to 都是本体 id。
 * 全局 agent 的会话本来就落在保留公司 id `_global` 下（28 §5.1），这里沿用同一个约定，
 * 不去猜某条派单「其实属于哪家公司」：猜错会让它出现在错误的公司任务列表里，
 * 而那比「都归到 _global」更难发现。
 */
const DISPATCH_COMPANY = '_global';

/** tasklog 的 status → tasks.status。dispatched = 派出去了还没回话，落 running。 */
const TASKLOG_STATUS = {
  dispatched: 'running',
  done: 'completed',
  failed: 'failed',
  blocked: 'blocked',
  cancelled: 'cancelled',
};

/** tasklog 的正文把换行写成了 <br>，标题要还原一下才不会糊成一长条 */
function tasklogTitle(task) {
  const plain = String(task || '').replace(/<br\s*\/?>/gi, '\n');
  return deriveTitle(plain) || '(无标题派单)';
}

/**
 * 幂等：这条 tasklog 是不是已经回填过。
 * 判据用 **tasks.id === taskId**（回填时刻意沿用原 taskId，见 migrateTasklog 的理由）。
 */
function tasklogAlreadyMigrated(db, taskId) {
  return Boolean(db.prepare('SELECT 1 AS ok FROM tasks WHERE id = ?').get(taskId));
}

function migrateTasklog(db, file) {
  let meta;
  try {
    meta = JSON.parse(fs.readFileSync(path.join(TASKLOG_DIR, file), 'utf-8'));
  } catch (e) {
    stats.tasklogBad += 1;
    log(`⚠️  ${file} 解析失败，跳过: ${e.message}`);
    return;
  }
  const taskId = meta.taskId || path.basename(file, '.json');
  if (tasklogAlreadyMigrated(db, taskId)) { stats.tasklogSkipped += 1; return; }

  const now = new Date().toISOString();
  const createdAt = meta.createdAt || now;
  const title = tasklogTitle(meta.task);

  // 每条派单建一个 goal：`tasks.goal_id` 是 NOT NULL 的真外键，而派单没有「人提的需求」那一层。
  // 与 §六.3 给 `POST /:cid/tasks` 设计的兼容路径同一条 —— 模型保持干净，不给 goal_id 开可空的口子。
  const goalId = wdb.genGoalId();
  db.prepare(`
    INSERT INTO goals (id, company_id, title, description, status, priority,
                       okr_node_id, created_by, report_to, created_at, updated_at, dispatched_at)
    VALUES (@id, @cid, @title, @desc, @status, @prio, @okr, @by, @to, @at, @at, @at)
  `).run({
    id: goalId, cid: DISPATCH_COMPANY, title: `来自 ${meta.from || '?'} 的派单`,
    desc: String(meta.task || ''), status: meta.status === 'done' ? 'done' : 'active',
    prio: meta.priority || 'P1', okr: meta.nodeId || null,
    by: meta.from || null, to: meta.from || null, at: createdAt,
  });
  stats.tasklogGoals += 1;

  // **沿用原 taskId 而不是新发一个**：conversations 目录名是 `task-{taskId}`、
  // 汇报链路按 taskId 找回这条派单、summary 里也互相引用。换 id 等于把这些引用全断掉，
  // 而回填的目的正是「别让历史消失」。
  db.prepare(`
    INSERT INTO tasks (id, goal_id, company_id, seq, title, description, status,
                       execution_mode, depends_on, attempt, priority, created_by,
                       executor, assignee_agent_id, parent_task_id,
                       created_at, updated_at, queued_at, started_at, finished_at)
    VALUES (@id, @gid, @cid, 1, @title, @desc, @status,
            'manual', '[]', 1, @prio, @by,
            'agent', @to, @parent,
            @at, @at, @at, @at, @done)
  `).run({
    id: taskId, gid: goalId, cid: DISPATCH_COMPANY, title,
    desc: String(meta.task || ''), status: TASKLOG_STATUS[meta.status] || 'running',
    prio: meta.priority || 'P1', by: meta.from || null,
    to: meta.to || null, parent: meta.parentTaskId || null,
    at: createdAt, done: meta.completedAt || null,
  });
  stats.tasklogTasks += 1;
}

function migrateAllTasklog(db) {
  if (!fs.existsSync(TASKLOG_DIR)) { log('system/tasklog/ 不存在，跳过派单回填'); return; }
  const files = fs.readdirSync(TASKLOG_DIR).filter((f) => f.endsWith('.json')).sort();
  stats.tasklogFiles = files.length;
  log(`system/tasklog/ 共 ${files.length} 条派单`);
  // ⚠️ parent_task_id 是**自引用外键**：父派单必须先入库。文件名按 taskId 排序 ≈ 按时间排序
  // （T-{base36 毫秒}），父一定早于子；真有乱序时下面那条 catch 会点名，不会静默丢。
  for (const f of files) {
    try { migrateTasklog(db, f); } catch (e) {
      stats.tasklogBad += 1;
      log(`⚠️  ${f} 落库失败（多半是 parent_task_id 还没入库）: ${e.message}`);
    }
  }
}

if (require.main === module) main();

module.exports = { runRootDir, readTaskGoal, deriveTitle, TASK_STATUS, GOAL_STATUS,
  migrateAllTasklog, tasklogTitle, TASKLOG_STATUS, DISPATCH_COMPANY };