From 12880b86207bf7b354cefe25a8707b6e23fac9cf Mon Sep 17 00:00:00 2001 From: yibo Date: Thu, 20 Aug 2026 01:32:09 +0800 Subject: [PATCH] feat(27-P3C): 批 5 第三步 —— /task/dispatch 的存储换成 tasks 表 --- server/routes/agent-chat.js | 196 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------------------------- server/services/work-graph-db.js | 13 +++++++++++++ server/tests/unit/task-dispatch-store.test.js | 102 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 261 insertions(+), 50 deletions(-) create mode 100644 server/tests/unit/task-dispatch-store.test.js diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index 04ab0b7..52cac1f 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -1659,17 +1659,110 @@ router.post('/scaffold', async (req, res) => { * autoExecute: true, // 是否自动异步触发执行 * } */ -// ── Task 元数据持久化目录 ── -const TASK_META_DIR = path.resolve(__dirname, '../../system/tasklog'); -if (!fs.existsSync(TASK_META_DIR)) fs.mkdirSync(TASK_META_DIR, { recursive: true }); +// ── Task 元数据：存储换成 tasks 表（27-P3C 批 5 §十二）───────────────── +// +// **本文件除了这两个函数的内脏，其余一行都没动。** `/task/dispatch` 与 `/task/report` 的 +// URL、请求体、响应体形态全部原样 —— 它们印在 packages/agent-core/base/common/tools-base.md +// 里，是**所有 agent 的工具说明**，改一个字等于同时改所有 agent 的提示词。 +// +// 原来的存储是 system/tasklog/{taskId}.json 一任务一文件。换掉它的三个理由： +// · 去重要「扫整个目录 + 逐个 JSON.parse」，任务一多就是 O(n) 次磁盘读，且没有索引可用； +// · 派单任务在库里看不见 —— 控制台的任务列表读的是 tasks 表，派单那套完全是平行世界； +// · 同一个 T- 前缀两套 id 体系，读日志时分不清。 +// +// **meta 的形状不变**（调用方按字段名取值），只是落点从文件换成表。 +const DISPATCH_COMPANY = '_global'; // 派单没有 company 这一维，全局 agent 的保留 id（28 §5.1） + +/** meta.status ↔ tasks.status。两边取值域不同，来回都要过一次翻译 */ +const META_TO_TASK_STATUS = { dispatched: 'running', done: 'completed', failed: 'failed', blocked: 'blocked', cancelled: 'cancelled' }; +const TASK_TO_META_STATUS = { running: 'dispatched', completed: 'done', queued: 'dispatched', draft: 'dispatched' }; + +/** 正文首行当标题；派单正文把换行写成
，先还原再取 */ +function dispatchTitle(task) { + const line = String(task || '').replace(//gi, '\n').split('\n')[0].trim(); + if (!line) return '(无标题派单)'; + return line.length > 60 ? `${line.slice(0, 60)}…` : line; +} function saveTaskMeta(taskId, meta) { - fs.writeFileSync(path.join(TASK_META_DIR, `${taskId}.json`), JSON.stringify(meta, null, 2), 'utf-8'); + const wdb = require('../services/work-graph-db'); + const db = wdb.getDb(); + const now = new Date().toISOString(); + const status = META_TO_TASK_STATUS[meta.status] || 'running'; + const existing = db.prepare('SELECT id, goal_id FROM tasks WHERE id = ?').get(taskId); + + if (existing) { + // 更新路径：/task/report 的写法是「loadTaskMeta → 改几个字段 → 整个对象写回」， + // 所以这里是全量 upsert 而不是增量 patch。 + db.prepare(` + UPDATE tasks SET status = @status, summary = @summary, finished_at = @done, + description = @desc, priority = @prio, updated_at = @now + WHERE id = @id + `).run({ + id: taskId, status, summary: meta.summary || null, done: meta.completedAt || null, + desc: String(meta.task || ''), prio: meta.priority || 'P1', now, + }); + return; + } + + // 新建路径：每条派单自带一个 goal —— tasks.goal_id 是 NOT NULL 真外键，而派单没有 + // 「人提的需求」那一层。与 §六.3 给 POST /:cid/tasks 设计的兼容路径同一条， + // 不给 goal_id 开可空的口子。 + const goalId = wdb.genGoalId(); + const createdAt = meta.createdAt || now; + 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, 'active', @prio, @okr, @by, @by, @at, @at, @at) + `).run({ + id: goalId, cid: DISPATCH_COMPANY, title: `来自 ${meta.from || '?'} 的派单`, + desc: String(meta.task || ''), prio: meta.priority || 'P1', + okr: meta.nodeId || null, by: meta.from || null, at: createdAt, + }); + 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, summary, execution_conv_id, + created_at, updated_at, queued_at, started_at) + VALUES (@id, @gid, @cid, 1, @title, @desc, @status, + 'manual', '[]', 1, @prio, @by, + 'agent', @to, @parent, @summary, @conv, + @at, @at, @at, @at) + `).run({ + id: taskId, gid: goalId, cid: DISPATCH_COMPANY, title: dispatchTitle(meta.task), + desc: String(meta.task || ''), status, prio: meta.priority || 'P1', + by: meta.from || null, to: meta.to || null, parent: meta.parentTaskId || null, + summary: meta.summary || null, conv: meta.executionConvId || null, at: createdAt, + }); } + function loadTaskMeta(taskId) { - const fp = path.join(TASK_META_DIR, `${taskId}.json`); - if (!fs.existsSync(fp)) return null; - try { return JSON.parse(fs.readFileSync(fp, 'utf-8')); } catch { return null; } + const wdb = require('../services/work-graph-db'); + const row = wdb.getDb().prepare(` + SELECT t.*, g.okr_node_id FROM tasks t JOIN goals g ON g.id = t.goal_id WHERE t.id = ? + `).get(taskId); + if (!row) return null; + // objectiveId 不单独存：它恒等于 okr node id 的首段（OkrService.parseNodeId 与 + // company.js:reportRunOutcome 都是这个口径），而且它只在与 nodeId 一起用时才有意义 + // （两处读取方都是 `if (objId && meta.nodeId)`）。 + const nodeId = row.okr_node_id || null; + const objectiveId = nodeId ? String(nodeId).split('.')[0] : null; + return { + taskId: row.id, + from: row.created_by, + to: row.assignee_agent_id, + task: row.description, + priority: row.priority, + objectiveId, + goalId: objectiveId, // 旧字段名，保留供存量读取方兼容（27-NAMING §六） + nodeId, + parentTaskId: row.parent_task_id, + executionConvId: row.execution_conv_id, + createdAt: row.created_at, + status: TASK_TO_META_STATUS[row.status] || row.status, + completedAt: row.finished_at, + summary: row.summary, + }; } router.post('/task/dispatch', async (req, res) => { @@ -1688,33 +1781,40 @@ router.post('/task/dispatch', async (req, res) => { const now = new Date().toISOString().slice(0, 16).replace('T', ' '); // ── 任务去重检查：相同接收方 30 分钟内是否有相似任务已分发 ── + // 批 5：从「readdirSync 整个 tasklog + 逐个 JSON.parse」换成一条带索引的 SQL + // （idx_tasks_assignee = company_id + assignee_agent_id + created_at）。 + // **判据本身一个字没改** —— 仍是「同一接收方 + 30 分钟内 + 仍在执行 + 正文互相包含」， + // 换的只是从哪儿把候选捞出来。响应体形态也原样（agent 的提示词里印着它）。 const DUPLICATE_WINDOW_MS = 30 * 60 * 1000; try { - const taskFiles = fs.readdirSync(TASK_META_DIR).filter(f => f.endsWith('.json')); - for (const tf of taskFiles) { - try { - const tm = JSON.parse(fs.readFileSync(path.join(TASK_META_DIR, tf), 'utf-8')); - if (tm.to === to && tm.status === 'dispatched') { - const age = Date.now() - new Date(tm.createdAt).getTime(); - if (age < DUPLICATE_WINDOW_MS) { - const normNew = task.toLowerCase().replace(/[\s\n\r]+/g, ' ').trim(); - const normExist = (tm.task || '').toLowerCase().replace(/
/g, ' ').replace(/[\s\n\r]+/g, ' ').trim(); - const isDuplicate = normNew === normExist || normNew.includes(normExist) || normExist.includes(normNew); - if (isDuplicate) { - console.log(`[TaskDispatch] 🚫 去重拦截: ${from} → ${to} 与 ${tm.taskId} 重复`); - return res.status(409).json({ - success: false, - error: 'Duplicate task detected', - existingTaskId: tm.taskId, - existingTask: tm.task, - message: `去重拦截：${to} 在 30 分钟内已有相似任务 ${tm.taskId} 处于进行中，请勿重复分发。`, - }); - } - } - } - } catch(e) {} + const wdb = require('../services/work-graph-db'); + const since = new Date(Date.now() - DUPLICATE_WINDOW_MS).toISOString(); + const rows = wdb.getDb().prepare(` + SELECT id, description FROM tasks + WHERE company_id = @cid AND assignee_agent_id = @to + AND executor = 'agent' AND status = 'running' + AND created_at > @since AND archived_at IS NULL + ORDER BY created_at DESC + `).all({ cid: DISPATCH_COMPANY, to, since }); + const normNew = task.toLowerCase().replace(/[\s\n\r]+/g, ' ').trim(); + for (const r of rows) { + const normExist = String(r.description || '').toLowerCase() + .replace(/
/g, ' ').replace(/[\s\n\r]+/g, ' ').trim(); + if (normNew === normExist || normNew.includes(normExist) || normExist.includes(normNew)) { + console.log(`[TaskDispatch] 🚫 去重拦截: ${from} → ${to} 与 ${r.id} 重复`); + return res.status(409).json({ + success: false, + error: 'Duplicate task detected', + existingTaskId: r.id, + existingTask: r.description, + message: `去重拦截：${to} 在 30 分钟内已有相似任务 ${r.id} 处于进行中，请勿重复分发。`, + }); + } } - } catch(e) {} + } catch (e) { + // 去重查不动不该让派单整个失败，但**必须留声** —— 静默跳过等于去重形同虚设 + console.error(`[TaskDispatch] ⚠️ 去重检查失败，本次放行: ${e.message}`); + } // 净化多行文本和竖线，防止破坏 Markdown 表格格式 const safeTask = task.replace(/\n/g, '
').replace(/\|/g, '|'); @@ -1723,28 +1823,25 @@ router.post('/task/dispatch', async (req, res) => { // 如果调用者没有显式传 parentTaskId，则尝试从 .tasks/ 目录下 // 查找 from agent 正在执行的（status=dispatched 且 to=from）的 task let parentTaskId = explicitParent || null; - let sourceConvId = null; if (!parentTaskId) { + // 批 5：同样从扫目录换成一条 SQL，判据不变 ——「派给 from 且仍在执行」里最近的一条， + // 那就是 from 此刻正在干的活，新派的这条挂在它下面。排序交给 SQL（原来是把全部候选 + // 读进内存再 sort，目录一大就是白读一遍）。 try { - const taskFiles = fs.readdirSync(TASK_META_DIR).filter(f => f.endsWith('.json')); - // 找到 "to === from" 且尚未完成的任务 → 这就是 from agent 正在执行的父任务 - let candidates = []; - for (const tf of taskFiles) { - try { - const tm = JSON.parse(fs.readFileSync(path.join(TASK_META_DIR, tf), 'utf-8')); - if (tm.to === from && tm.status === 'dispatched') { - candidates.push(tm); - } - } catch(e) {} - } - // 如果有多个，取最近创建的 - if (candidates.length > 0) { - candidates.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()); - parentTaskId = candidates[0].taskId; - sourceConvId = candidates[0].executionConvId || null; + const wdb = require('../services/work-graph-db'); + const parent = wdb.getDb().prepare(` + SELECT id FROM tasks + WHERE company_id = @cid AND assignee_agent_id = @from + AND executor = 'agent' AND status = 'running' AND archived_at IS NULL + ORDER BY created_at DESC LIMIT 1 + `).get({ cid: DISPATCH_COMPANY, from }); + if (parent) { + parentTaskId = parent.id; console.log(`[TaskDispatch] 🔗 Auto-detected parentTaskId: ${parentTaskId} (${from} is executing it)`); } - } catch(e) { /* ignore */ } + } catch (e) { + console.error(`[TaskDispatch] ⚠️ 父任务推断失败，本次不挂父: ${e.message}`); + } } // 1. 写入目标 Agent 的 todo.md（公司态：`co-alpha:ceo` 与 `co-beta:ceo` 各写各的） @@ -1788,7 +1885,6 @@ router.post('/task/dispatch', async (req, res) => { goalId: objectiveId || null, // 旧字段名，保留供存量读取方兼容 nodeId: nodeId || null, parentTaskId: parentTaskId || null, - sourceConvId: sourceConvId || null, executionConvId, createdAt: new Date().toISOString(), status: 'dispatched', diff --git a/server/services/work-graph-db.js b/server/services/work-graph-db.js index 04dc34e..0fe11f6 100644 --- a/server/services/work-graph-db.js +++ b/server/services/work-graph-db.js @@ -504,6 +504,11 @@ function getDb() { executor TEXT NOT NULL DEFAULT 'graph' CHECK (executor IN ('graph','agent')), assignee_agent_id TEXT, -- executor='agent' 时派给谁（dispatch 的 to） + -- 下面两列只对 executor='agent' 有意义（图任务用 current_run_id / task-goal.json）： + summary TEXT, -- 汇报摘要（/task/report 的 summary） + execution_conv_id TEXT, -- 派单起的那次对话 id（task-{taskId}） + -- ⚠️ 旧 meta 里还有个 sourceConvId，全仓**零读取方**，借这次统一删掉 —— + -- 「实现在、零调用方」是本仓库反复立案的惯犯模式，不再往新表里搬一份。 -- 派单链：A 让 B 干活、B 又拆给 C。与 depends_on 不是一回事 —— -- depends_on 是同级任务之间的先后，parent_task_id 是上下级的派生。 parent_task_id TEXT REFERENCES tasks(id), @@ -559,6 +564,12 @@ function getDb() { if (!hasColumn(db, 'tasks', 'parent_task_id')) { db.exec(`ALTER TABLE tasks ADD COLUMN parent_task_id TEXT`); } + if (!hasColumn(db, 'tasks', 'summary')) { + db.exec(`ALTER TABLE tasks ADD COLUMN summary TEXT`); + } + if (!hasColumn(db, 'tasks', 'execution_conv_id')) { + db.exec(`ALTER TABLE tasks ADD COLUMN execution_conv_id TEXT`); + } // ⚠️ 这条索引**必须建在 ALTER 之后**，不能放进上面的建表块： // 建表块整体先于迁移块执行，而存量库上 `CREATE TABLE IF NOT EXISTS tasks` 是 no-op —— // 表还是批 4 的形态、没有 assignee_agent_id，那一句 CREATE INDEX 会直接抛 @@ -1760,6 +1771,8 @@ const TASK_PATCH_COLUMNS = { startedAt: 'started_at', finishedAt: 'finished_at', archivedAt: 'archived_at', + summary: 'summary', + executionConvId: 'execution_conv_id', }; /** diff --git a/server/tests/unit/task-dispatch-store.test.js b/server/tests/unit/task-dispatch-store.test.js new file mode 100644 index 0000000..f319d5d --- /dev/null +++ b/server/tests/unit/task-dispatch-store.test.js @@ -0,0 +1,102 @@ +/** + * 批 5 第 3 步：`/task/dispatch` 的存储从 system/tasklog/*.json 换成 tasks 表。 + * + * **这一步唯一不能坏的是契约**：`/task/dispatch` 与 `/task/report` 的 URL、请求体、响应体 + * 印在 `packages/agent-core/base/common/tools-base.md` 里，是所有 agent 的工具说明 —— + * 改一个字等于同时改所有 agent 的提示词。所以这份用例盯的是「形状没变」，不是「实现对不对」。 + */ +import { describe, it, expect, beforeAll, afterAll } 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('派单存储换成 tasks 表', () => { + let wdb, app, srv, base; + + beforeAll(async () => { + process.env.GRAPH_DB_PATH = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'dispatch-store-')), 'graph.sqlite'); + for (const m of ['../../services/work-graph-db', '../../services/graph-db', '../../routes/agent-chat']) { + delete require_.cache[require_.resolve(m)]; + } + wdb = require_('../../services/work-graph-db'); + const express = require_('express'); + app = express(); + app.use(express.json()); + app.use('/api/agent', require_('../../routes/agent-chat')); + await new Promise((r) => { srv = app.listen(0, '127.0.0.1', r); }); + base = `http://127.0.0.1:${srv.address().port}`; + }); + afterAll(() => { try { srv?.close(); } catch {} }); + + const post = async (p, body) => { + const r = await fetch(base + p, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body), + }); + return { status: r.status, body: await r.json().catch(() => ({})) }; + }; + + /** 用真实存在的两个本体（端点会校验目录存在） */ + const FROM = 'ceo'; + const TO = 'qa-judge'; + + it('派单落进 tasks 表，且响应体是老形态（agent 提示词里印着它）', async () => { + const r = await post('/api/agent/task/dispatch', { + from: FROM, to: TO, task: '守卫用例：形态检查', priority: 'P1', + goalId: 'O-900', nodeId: 'O-900.1.1', autoExecute: false, + }); + expect(r.status).toBe(200); + // 响应体的字段一个都不能少 —— 少了就是改了所有 agent 的工具说明 + for (const k of ['success', 'taskId', 'from', 'to', 'task', 'priority', 'autoExecute', 'message']) { + expect(r.body).toHaveProperty(k); + } + expect(r.body.success).toBe(true); + + const row = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(r.body.taskId); + expect(row).toBeTruthy(); + expect(row.executor).toBe('agent'); // 调度器据此跳过它（图任务才排队） + expect(row.assignee_agent_id).toBe(TO); + expect(row.created_by).toBe(FROM); + expect(row.status).toBe('running'); // meta 的 dispatched ↔ 表的 running + + // 每条派单自带一个 goal：goal_id 是 NOT NULL 真外键，派单没有「人提的需求」那一层 + const goal = wdb.getDb().prepare('SELECT * FROM goals WHERE id = ?').get(row.goal_id); + expect(goal.title).toContain(FROM); + expect(goal.okr_node_id).toBe('O-900.1.1'); // objectiveId 由它的首段推导，不单独存 + }); + + it('去重仍然拦得住，且 409 的响应体形态没变', async () => { + const body = { from: FROM, to: TO, task: '守卫用例：去重', autoExecute: false }; + const first = await post('/api/agent/task/dispatch', body); + expect(first.status).toBe(200); + const dup = await post('/api/agent/task/dispatch', body); + expect(dup.status).toBe(409); + for (const k of ['success', 'error', 'existingTaskId', 'existingTask', 'message']) { + expect(dup.body).toHaveProperty(k); + } + expect(dup.body.existingTaskId).toBe(first.body.taskId); + }); + + it('汇报回路：状态 / summary / 完成时间都落到表上', async () => { + const d = await post('/api/agent/task/dispatch', { + from: FROM, to: TO, task: '守卫用例：汇报回路', autoExecute: false, + }); + const rep = await post('/api/agent/task/report', { + taskId: d.body.taskId, from: TO, to: FROM, status: 'done', summary: '干完了', + }); + expect(rep.status).toBe(200); + const row = wdb.getDb().prepare('SELECT * FROM tasks WHERE id = ?').get(d.body.taskId); + expect(row.status).toBe('completed'); // meta 的 done ↔ 表的 completed + expect(row.summary).toBe('干完了'); + expect(row.finished_at).toBeTruthy(); + }); + + it('不再往 system/tasklog/ 写新文件', async () => { + const dir = path.resolve(process.cwd(), 'system', 'tasklog'); + const before = fs.existsSync(dir) ? fs.readdirSync(dir).length : 0; + await post('/api/agent/task/dispatch', { from: FROM, to: TO, task: '守卫用例：不写文件', autoExecute: false }); + const after = fs.existsSync(dir) ? fs.readdirSync(dir).length : 0; + expect(after).toBe(before); + }); +}); -- libgit2 0.21.4