/** * Unit Tests — 恢复动作在 **HTTP 这一头**的结局(27-P3C 批 3 对抗复核 F1 / 409) * * 引擎侧的两种结局已经有用例钉着(graph-run-slot-recovery.test.js),但用户点的是按钮、 * 外部 agent 打的是 curl —— 真正决定「这次恢复有没有可执行的下一步」的是**响应**: * · 退得回队列的:必须是 2xx + 一个说得清「它回队列了」的体。回 500 的话前端只会当故障, * 而库里其实已经排好队了(状态和响应对不上,正是最难查的一类); * · 退不回队列的:必须是 409 + 点名占坑者。500 的含义是「服务端坏了」,调用方会去重试 * 或报故障,而这里唯一有用的动作是「等它收敛 / 去停掉它」。 * * 为什么非得走真 express + fetch:`sendError` 是私有函数,且 `/start` 是**同步** handler —— * `h()` 早先的 `Promise.resolve(fn()).catch()` 挂 catch 时 fn 已经抛完了,异常直接冒到 express * 默认处理器、回一页 HTML 的 500,前端 `await r.json()` 炸在 "Unexpected token '<'"。 * 只有从 HTTP 这一头才测得出这件事(已有的 409 用例走的是 `/nodes/:id/retry` 那条, * 这里换 `/start`:四个同步端点各自的 try/catch 是分别写的,一条路通不代表另一条通)。 * * 每个用例一家自己的公司 —— 「坑」是 per-company 的,各测各的互不干扰,也就不存在用例顺序依赖。 */ const fs = require('fs'); const os = require('os'); const path = require('path'); const express = require('express'); const TEST_DB_PATH = '/tmp/test-p3c-recovery-http.sqlite'; const DB_FILES = [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]; const MODULES = [ '../../services/graph-db', '../../services/org-graph-service', '../../services/work-graph-db', '../../services/company-paths', '../../services/graph-run-sse', '../../services/graph-engine', '../../services/task-settle', '../../services/task-scheduler', '../../routes/graph-run', ]; describe('恢复动作的 HTTP 结局:退得回队列 → 2xx,退不回 → 409(都不是 500)', () => { let wdb; let engine; let scheduler; let server; let origin; let runsDir; /** 挂起的 executor promise,afterAll 统一放掉 */ const pendingResolvers = []; beforeAll(async () => { for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 本来就没有 */ } } process.env.GRAPH_DB_PATH = TEST_DB_PATH; runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'p3c-http-runs-')); process.env.GRAPH_RUNS_DIR = runsDir; for (const m of MODULES) delete require.cache[require.resolve(m)]; wdb = require('../../services/work-graph-db'); engine = require('../../services/graph-engine'); scheduler = require('../../services/task-scheduler'); const app = express(); app.use(express.json()); app.use('/api/graph-runs', require('../../routes/graph-run')); await new Promise((resolve) => { server = app.listen(0, resolve); }); origin = `http://127.0.0.1:${server.address().port}`; }); afterAll(async () => { for (const r of pendingResolvers) r(); await new Promise((r) => setTimeout(r, 100)); if (server) server.close(); scheduler.stopSweeper(); cleanupLeakedCompanyDirs(); for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 已经删了 */ } } try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch { /* 同上 */ } delete process.env.GRAPH_DB_PATH; delete process.env.GRAPH_RUNS_DIR; for (const m of MODULES) delete require.cache[require.resolve(m)]; }); /** * advanceOne 的 ensureProjectRoot 认的是仓库里的 data/companies/,只删本文件造的那些。 * nextCompany() 现在会顺手建一份 company.json(38 号文档 Phase 2,ensureTestCompany 写的 * ownerId 恒为 -1,真实生产公司的 ownerId 是 users.id,不会撞上 -1)——判据从"有没有 * company.json"改成"ownerId 是不是 -1 这个测试专用标记",否则这批目录会被误判成真公司 * 而永远不清理。 */ function cleanupLeakedCompanyDirs() { const root = path.resolve(__dirname, '../../../data/companies'); if (!fs.existsSync(root)) return; for (const cid of fs.readdirSync(root)) { if (!cid.startsWith('p3chttp-co-')) continue; const dir = path.join(root, cid); const companyJsonPath = path.join(dir, 'company.json'); if (fs.existsSync(companyJsonPath)) { try { const { ownerId } = JSON.parse(fs.readFileSync(companyJsonPath, 'utf-8')); if (ownerId !== -1) continue; // 那是真公司 } catch { continue; } // 读不出来就保守跳过,不乱删 } try { fs.rmSync(dir, { recursive: true, force: true }); } catch { /* 删不掉就留着 */ } } } let seq = 0; // 38 号文档 Phase 2:graph-run.js 的 :id 路由现在会反查 run 的 company_id 是否是一家 // 真实存在的公司,这里造出来的 id 从没真的建过公司,补一份最小 company.json。 const nextCompany = () => { const cid = `p3chttp-co-${++seq}`; require('../helpers/ensure-test-company').ensureTestCompany(cid); return cid; }; function ensureAgents(ids) { const now = new Date().toISOString(); const ins = wdb.getDb().prepare(` INSERT OR IGNORE INTO system_agents (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at) VALUES (?, ?, 'public', NULL, ?, 'active', ?, ?) `); for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, now, now); } function buildGraph(companyId, node) { ensureAgents([node]); const g = wdb.createWorkGraph({ companyId, nodes: [node], edges: [] }); expect(g.errors || []).toEqual([]); return g.id; } /** 占坑的那条:一条**有 task** 的 running root run(409 的 holders 要带得出 taskId) */ function occupySlot(companyId, tag) { const goalId = wdb.createGoal({ companyId, title: 'g', description: 'g' }).id; const taskId = wdb.createTask({ goalId, companyId, title: `holder-${tag}`, description: '占坑的任务' }).id; const run = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, `h${tag}`), companyId, goalId, taskId, }); wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'running' WHERE id = ?`).run(run.id); wdb.updateTask(taskId, { status: 'running', currentRunId: run.id }); return { runId: run.id, taskId, goalId }; } /** 预算耗尽那种局面:一条 blocked 的 root run + 一条 blocked 的 task(人正要点「启动」) */ function blockedTaskRun(companyId, goalId, tag) { const taskId = wdb.createTask({ goalId, companyId, title: `t-${tag}`, description: '预算耗尽卡住的任务', executionMode: 'auto', }).id; const run = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, `n${tag}`), companyId, goalId, taskId, executionMode: 'auto', }); wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'blocked' WHERE id = ?`).run(run.id); wdb.updateTask(taskId, { status: 'blocked', currentRunId: run.id, blockedReason: 'budget_exhausted', queuedAt: new Date().toISOString(), }); return { runId: run.id, taskId }; } /** 永不 resolve 的 executor:被接回来的那条稳稳停在 running,断言不跟异步驱动赛跑 */ function hangingExecutor() { engine.setAgentExecutor(() => new Promise((resolve) => { pendingResolvers.push(() => resolve({ status: 'success' })); })); } /** 等一个条件成立(调度是 setImmediate + 异步驱动,不能只 flush 一次就断言) */ async function waitFor(fn, ms = 1000) { const deadline = Date.now() + ms; for (;;) { const v = fn(); if (v) return v; if (Date.now() > deadline) return v; await new Promise((r) => setTimeout(r, 10)); } } it('退不回队列的恢复 → 409 + 点名占坑者,绝不是 500(/start 是同步 handler,早先直接冒到 express)', async () => { const companyId = nextCompany(); const holder = occupySlot(companyId, 'a'); // 没有 task 的裸 root run:队列按 task 选候选,它退不回去 —— 这时唯一正确的答复是 409 const bare = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, 'bare'), companyId }); wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'blocked' WHERE id = ?`).run(bare.id); const r = await fetch(`${origin}/api/graph-runs/${bare.id}/start`, { method: 'POST' }); expect(r.headers.get('content-type')).toMatch(/application\/json/); // 500 那条路回的是一页 HTML const body = await r.json(); expect(r.status).toBe(409); expect(body.reason).toBe('run-slot-busy'); expect(body.companyId).toBe(companyId); expect(body.holders).toEqual([ expect.objectContaining({ runId: holder.runId, taskId: holder.taskId, runStatus: 'running' }), ]); // hint 必须指向此刻真的走得通的一步:占坑那条是 running,/stop 认它 expect(body.hint).toContain(`/api/graph-runs/${holder.runId}/stop`); // 拒绝是干净的:一个字都没写进去 expect(wdb.getRun(bare.id).run_status).toBe('blocked'); }); it('人对卡住的任务点「启动」撞上坑满 → 2xx + 明说它回了队列(不是 500,也不是原地卡死)', async () => { const companyId = nextCompany(); const goalId = wdb.createGoal({ companyId, title: 'g', description: 'g' }).id; occupySlot(companyId, 'b'); const t1 = blockedTaskRun(companyId, goalId, 'b'); const r = await fetch(`${origin}/api/graph-runs/${t1.runId}/start`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ mode: 'auto' }), }); const body = await r.json(); expect(r.status).toBe(200); expect(body).toMatchObject({ ok: true, requeued: true, runStatus: 'queued' }); // 响应说什么,库里就得是什么 —— run 与 task 两行一起回队列,少一半都是新的卡死形态 expect(wdb.getRun(t1.runId).run_status).toBe('queued'); expect(wdb.getTask(t1.taskId).status).toBe('queued'); expect(wdb.getTask(t1.taskId).blocked_reason).toBeNull(); }); it('坑一空,刚才那条自己被接回来 —— 人只点了一次「启动」,没有第二次操作', async () => { const companyId = nextCompany(); const goalId = wdb.createGoal({ companyId, title: 'g', description: 'g' }).id; const holder = occupySlot(companyId, 'c'); const t1 = blockedTaskRun(companyId, goalId, 'c'); hangingExecutor(); expect((await fetch(`${origin}/api/graph-runs/${t1.runId}/start`, { method: 'POST' })).status).toBe(200); expect(wdb.getRun(t1.runId).run_status).toBe('queued'); // 人去把占坑那条停掉(正是 409 的 hint 指的那条路),之后**不再点任何按钮** const stopped = await fetch(`${origin}/api/graph-runs/${holder.runId}/stop`, { method: 'POST' }); expect(stopped.status).toBe(200); await waitFor(() => wdb.getRun(t1.runId).run_status === 'running'); expect(wdb.getRun(t1.runId).run_status).toBe('running'); expect(wdb.getTask(t1.taskId).status).toBe('running'); }); });