#!/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 的正文把换行写成了
,标题要还原一下才不会糊成一长条 */ function tasklogTitle(task) { const plain = String(task || '').replace(//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 };