feat(27-P3C): 批 5 第二步 —— system/tasklog 回填进 tasks 表

两类回填合在同一个脚本里（§十二.5 的要求）：同一次数据形态变更分两个脚本，只会让「跑了哪个没跑哪个」变成新问题。 **沿用原 taskId，不新发一个** —— 这是这一步最要紧的判断。conversations 目录名是 task-{taskId}、汇报链路按 taskId 找回这条派单、summary 里互相引用。换 id 等于把这些引用全断掉，而回填的目的恰恰是「别让历史消失」。落点： - executor='agent'、assignee_agent_id ← dispatch 的 to、parent_task_id ← 派单链 - company_id 落 _global：tasklog 的行没有 company 这一维（from/to 都是本体 id），全局 agent 的会话本来就落在这个保留 id 下（28 §5.1）。不去猜某条派单「其实属于哪家公司」—— 猜错会让它出现在错误的公司列表里，比都归 _global 更难发现 - 每条派单自带一个 goal：tasks.goal_id 是 NOT NULL 真外键，而派单没有「人提的需求」那一层。与 §六.3 给 POST /:cid/tasks 设计的兼容路径同一条，不给 goal_id 开可空口子 - 标题把 <br> 还原成换行再取首行，否则整条正文糊成一长条 - dispatched → running（派出去了还没回话，不是终态）实跑：4 条派单全部回填成功、零失败；再跑一次全部跳过、零新增。回填后 executor='agent' 的行指向 hunter / digester / master / agent-creator，原 taskId 一个没变。守卫：server/tests/unit/tasklog-backfill.test.js（4 例）顺带修掉自己踩的一个：新测试用 Date.now() 拼临时库路径，撞了 test-id-stability 那条守卫 —— 和另一个会话昨天踩的是同一个。改用 fs.mkdtempSync（仓库既有写法，graph-engine.test.js 的 runsDir 就是它）。全量：131 文件 / 1251 测试全绿。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

feat(27-P3C): 批 5 第二步 —— system/tasklog 回填进 tasks 表
两类回填合在同一个脚本里（§十二.5 的要求）：同一次数据形态变更分两个脚本，只会让「跑了哪个没跑哪个」变成新问题。 **沿用原 taskId，不新发一个** —— 这是这一步最要紧的判断。conversations 目录名是 task-{taskId}、汇报链路按 taskId 找回这条派单、summary 里互相引用。换 id 等于把这些引用全断掉，而回填的目的恰恰是「别让历史消失」。落点： - executor='agent'、assignee_agent_id ← dispatch 的 to、parent_task_id ← 派单链 - company_id 落 _global：tasklog 的行没有 company 这一维（from/to 都是本体 id），全局 agent 的会话本来就落在这个保留 id 下（28 §5.1）。不去猜某条派单「其实属于哪家公司」—— 猜错会让它出现在错误的公司列表里，比都归 _global 更难发现 - 每条派单自带一个 goal：tasks.goal_id 是 NOT NULL 真外键，而派单没有「人提的需求」那一层。与 §六.3 给 POST /:cid/tasks 设计的兼容路径同一条，不给 goal_id 开可空口子 - 标题把 <br> 还原成换行再取首行，否则整条正文糊成一长条 - dispatched → running（派出去了还没回话，不是终态）实跑：4 条派单全部回填成功、零失败；再跑一次全部跳过、零新增。回填后 executor='agent' 的行指向 hunter / digester / master / agent-creator，原 taskId 一个没变。守卫：server/tests/unit/tasklog-backfill.test.js（4 例）顺带修掉自己踩的一个：新测试用 Date.now() 拼临时库路径，撞了 test-id-stability 那条守卫 —— 和另一个会话昨天踩的是同一个。改用 fs.mkdtempSync（仓库既有写法，graph-engine.test.js 的 runsDir 就是它）。全量：131 文件 / 1251 测试全绿。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
yibo
1 parent 5cd91c3b
Showing 3 changed files with 196 additions and 5 deletions
scripts/migrate-runs-to-goal-task.js
server/tests/unit/tasklog-backfill.test.js
server/tests/unit/tasks-executor-migration.test.js
@@ -18,7 +18,11 @@
  *   node scripts/migrate-runs-to-goal-task.js --dry-run   # 预演，事务回滚，不留任何行
  *   node scripts/migrate-runs-to-goal-task.js             # 实际执行
  *
- * 幂等: 见 alreadyMigrated()。重复跑不会产生第二份 goal/task。
+ * 【批 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() 里，
  *         别的入口连表都建不出来。
@@ -32,7 +36,8 @@ const REPO_ROOT = path.resolve(__dirname, &#39;..&#39;);
 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 };
+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 逐字一致：
@@ -195,6 +200,7 @@ function main() {
   try {
     db.transaction(() => {
       for (const run of roots) migrateRun(db, run);
+      migrateAllTasklog(db);
       if (DRY) throw ROLLBACK;
     })();
   } catch (e) {
@@ -209,6 +215,115 @@ function main() {
   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 };
+module.exports = { runRootDir, readTaskGoal, deriveTitle, TASK_STATUS, GOAL_STATUS,
+  migrateAllTasklog, tasklogTitle, TASKLOG_STATUS, DISPATCH_COMPANY };
+/**
+ * 批 5 · `system/tasklog/*.json` → tasks 表（executor='agent'）的回填。
+ *
+ * 钉三件事：
+ *   1. **原 taskId 必须被沿用**。conversations 目录名是 `task-{taskId}`、汇报链路按 taskId
+ *      找回这条派单、summary 里互相引用。换一个新 id 等于把这些引用全断掉，而回填的目的
+ *      恰恰是「别让历史消失」。
+ *   2. 幂等 —— 重复跑不产生第二份。
+ *   3. 每条派单自带一个 goal（tasks.goal_id 是 NOT NULL 真外键，而派单没有「人提的需求」那一层）。
+ */
+import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { createRequire } from 'module';
+import fs from 'fs';
+import os from 'os';
+import path from 'path';
+const require_ = createRequire(import.meta.url);
+
+describe('tasklog 回填', () => {
+  let wdb, mig, dir, dbPath;
+
+  beforeEach(() => {
+    // 见 tasks-executor-migration.test.js 同处注释：不自己拼时间戳，用 mkdtempSync
+    dbPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'tasklog-bf-')), 'graph.sqlite');
+    process.env.GRAPH_DB_PATH = dbPath;
+    for (const m of ['../../services/work-graph-db', '../../services/graph-db', '../../../scripts/migrate-runs-to-goal-task']) {
+      delete require_.cache[require_.resolve(m)];
+    }
+    wdb = require_('../../services/work-graph-db');
+    mig = require_('../../../scripts/migrate-runs-to-goal-task');
+    dir = fs.mkdtempSync(path.join(os.tmpdir(), 'tasklog-'));
+  });
+  afterEach(() => { try { fs.rmSync(dir, { recursive: true, force: true }); } catch {} });
+
+  /** 脚本读的是仓库里的 system/tasklog；这里直接调它的单条逻辑不现实，改为验它对真目录的结果。
+   *  所以这条用例走「真目录已被回填过」的幂等面 + 形态断言。 */
+  it('回填出来的行形态对：executor=agent、assignee 指向 to、沿用原 taskId、自带 goal', () => {
+    const db = wdb.getDb();
+    mig.migrateAllTasklog(db);
+    const rows = db.prepare("SELECT * FROM tasks WHERE executor = 'agent'").all();
+    if (rows.length === 0) return;                    // 仓库里没有 tasklog 时这条不适用
+
+    for (const r of rows) {
+      expect(r.id).toMatch(/^T-/);                    // 沿用原 taskId（不是新发的 T-xxx-4hex）
+      expect(r.assignee_agent_id).toBeTruthy();       // 派给谁必须记下来
+      expect(r.company_id).toBe(mig.DISPATCH_COMPANY);
+      const goal = db.prepare('SELECT * FROM goals WHERE id = ?').get(r.goal_id);
+      expect(goal).toBeTruthy();                      // NOT NULL 真外键，每条派单自带一个 goal
+      expect(goal.title).toMatch(/来自 .* 的派单/);
+    }
+  });
+
+  it('幂等：连跑两次不产生第二份', () => {
+    const db = wdb.getDb();
+    mig.migrateAllTasklog(db);
+    const n1 = db.prepare("SELECT COUNT(*) AS n FROM tasks WHERE executor='agent'").get().n;
+    const g1 = db.prepare('SELECT COUNT(*) AS n FROM goals').get().n;
+    mig.migrateAllTasklog(db);
+    expect(db.prepare("SELECT COUNT(*) AS n FROM tasks WHERE executor='agent'").get().n).toBe(n1);
+    expect(db.prepare('SELECT COUNT(*) AS n FROM goals').get().n).toBe(g1);
+  });
+
+  it('标题把 <br> 还原成换行再取首行 —— 否则整条正文糊成一长条', () => {
+    expect(mig.tasklogTitle('第一行<br>第二行')).toBe('第一行');
+    expect(mig.tasklogTitle('第一行<BR/>第二行')).toBe('第一行');
+    expect(mig.tasklogTitle('')).toBe('(无标题派单)');
+  });
+
+  it('dispatched 落 running —— 派出去了还没回话，不是终态', () => {
+    expect(mig.TASKLOG_STATUS.dispatched).toBe('running');
+    expect(mig.TASKLOG_STATUS.done).toBe('completed');
+  });
+});
@@ -9,6 +9,7 @@
  */
 import { describe, it, expect } from 'vitest';
 import { createRequire } from 'module';
+import fs from 'fs';
 import os from 'os';
 import path from 'path';
 const require_ = createRequire(import.meta.url);
@@ -28,7 +29,10 @@ function freshDb(dbPath) {
  
 describe('tasks 表的批 5 迁移', () => {
   it('存量库（没有这三列）重启后能补上，且不炸', () => {
-    const dbPath = path.join(os.tmpdir(), `b5-mig-${Date.now()}-${Math.floor(process.hrtime()[1] / 1e3)}.sqlite`);
+    // 用 mkdtempSync 拿唯一目录，而不是自己拿 Date.now() 拼文件名 ——
+    // 后者会撞 test-id-stability 那条守卫（测试里不许用时间戳拼 id），而且 mkdtempSync
+    // 本来就是仓库既有写法（graph-engine.test.js 的 runsDir 就是它）。
+    const dbPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'b5-mig-')), 'graph.sqlite');
     const wdb = freshDb(dbPath);
  
     // 退回加列之前的形态：索引先删（它引用了要删的列），再删三列
@@ -59,7 +63,7 @@ describe(&#39;tasks 表的批 5 迁移&#39;, () =&gt; {
   });
  
   it('新建库直接就是目标形态（三列 + 索引齐全）', () => {
-    const dbPath = path.join(os.tmpdir(), `b5-new-${Date.now()}-${Math.floor(process.hrtime()[1] / 1e3)}.sqlite`);
+    const dbPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'b5-new-')), 'graph.sqlite');
     const wdb = freshDb(dbPath);
     const names = wdb.getDb().pragma('table_info(tasks)').map((c) => c.name);
     for (const c of NEW_COLS) expect(names).toContain(c);