diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index 742df3e..afd5def 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -2493,8 +2493,9 @@ router.get('/:agentId/uploads/:convId/:filename', (req, res) => { // 🔒 Agent 级并发互斥锁 — 同一 Agent 同时只允许一个活跃 chat session // 同一 convId 的多轮对话不受限制（ACP 进程复用） -const agentChatLocks = new Map(); // agentId → { convId, startedAt } -const AGENT_LOCK_TIMEOUT = 30 * 60 * 1000; // 30 分钟安全阀 +// 32 号文档方案 B：这把锁挪到了共享模块 agent-session-lock.js，company.js 的招聘装配 +// 重试逻辑也要用同一把——两条路径各自维护一份的话，各自都以为自己没有并发。 +const { agentChatLocks, AGENT_LOCK_TIMEOUT } = require('../src/core/agent-session-lock'); router.post('/:agentId/chat', async (req, res) => { const { message, conversationId: reqConvId } = req.body; diff --git a/server/routes/company.js b/server/routes/company.js index 9c48de5..eadb691 100644 --- a/server/routes/company.js +++ b/server/routes/company.js @@ -13,6 +13,10 @@ const { ScheduleService } = require('../services/schedule-service'); const orgGraphService = require('../services/org-graph-service'); // 28 批次 4：公司不再持有 agent 实体 —— 名册查 company_agents，目录一律经 agent-locator 推导 const locator = require('../services/agent-locator'); +// 32 号文档方案 B：招聘装配（agent-creator 填 agent.md/soul.md）改走同进程调用 + 互斥锁 + +// 重试，不再是裸 fetch 回环自己。 +const AgentManager = require('../src/core/agent-manager'); +const { fillAgentBodyReliably } = require('../services/agent-body-filler'); const companyService = new CompanyService(); const scheduleService = new ScheduleService(); @@ -1002,27 +1006,33 @@ router.post('/:companyId/hire', async (req, res) => { const bodyPath = locator.bodyDir(role); const PORT = config.port || 3000; - // 1. 发起 Agent-Creator 编写骨架定义 (异步不等待) - // ⚠️ 给的是**绝对路径**：agent-creator 的 cwd 现在是它自己的任务态目录 - // （system/agents/agent-creator/_global/{taskId}/），原来那个 `../{公司}/agents/{角色}` - // 相对路径从那里数出去指向的是一个不存在的位置 —— 它会照着写，且不报错。 - fetch(`http://127.0.0.1:${PORT}/api/agent/agent-creator/chat`, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - // 产出契约（28 批次 5-23）逐字点名三个必填文件：本体填不满就停在 - // status='draft'、别家公司雇不走。以前只写「补充完整的定义文件(…等)」， - // 「等」里那份 SKILL.md 谁也没保证过 —— org-coder 就是这么变成空定义的。 - // 契约文本（含运行期布局那半段）单点定义在 company-service，见其注释。 - message: buildAgentDefinitionBrief({ - agentId: role, - bodyPath, - description: description || name || role, - intro: `新员工 ${namespacedId} 已由脚手架创建空白目录在 \`${bodyPath}\`。请你按照标准模板，使用 Write 工具为该员工补充定义文件。写完后请答复。`, - }), - conversationId: `create-${req.params.companyId}-${role}-${Date.now().toString(36)}` - }) - }).catch(e => console.error('[Hire] Agent-Creator 请求错误:', e.message)); + // 1. 发起 Agent-Creator 编写骨架定义（32 号文档方案 B：同进程调用 + 互斥锁 + + // 有限重试，替掉原来那发裸 fetch。原注释里"异步不等待"已经不准确—— + // 这里 await 了，下面第 2/3 步因此天然拿到"agent-creator 真的写完了"这个 + // 保证，不再是"assume 5 秒内写完"（第 3 步 setTimeout(5000) 那条注释）。 + // ⚠️ 给的是**绝对路径**：agent-creator 的 cwd 是它自己的任务态目录 + // （system/agents/agent-creator/_global/{taskId}/），相对路径从那里数出去 + // 指向的是一个不存在的位置 —— 它会照着写，且不报错。 + const fillResult = await fillAgentBodyReliably({ + agentManager: new AgentManager({}, REPO_ROOT), + targetAgentId: role, + // 产出契约（28 批次 5-23）逐字点名三个必填文件：本体填不满就停在 + // status='draft'、别家公司雇不走。以前只写「补充完整的定义文件(…等)」， + // 「等」里那份 SKILL.md 谁也没保证过 —— org-coder 就是这么变成空定义的。 + // 契约文本（含运行期布局那半段）单点定义在 company-service，见其注释。 + message: buildAgentDefinitionBrief({ + agentId: role, + bodyPath, + description: description || name || role, + intro: `新员工 ${namespacedId} 已由脚手架创建空白目录在 \`${bodyPath}\`。请你按照标准模板，使用 Write 工具为该员工补充定义文件。写完后请答复。`, + }), + }); + if (!fillResult.ok) { + // 重试耗尽——不是把招聘流程一起判失败（目录/skill 注入照常做，参见 32 号文档 + // §三「不做保证自动跑到底」），只是把这个失败从"一行日志"变成可查询的哨兵文件 + // （agent-body-filler.js 已经写好），这里再喊一嗓子留个显眼的日志。 + console.error(`[Hire] ⚠️ agent-creator 装配 ${role} 失败（重试 ${fillResult.attempts} 次）：${fillResult.error}`); + } // 2. 去网上/Arena搜索并注入 3 个基础 Skill。 // 落 `.claude/skills/` 而不是 `skills/`：只有前者会被 CLI（和 agent-manager diff --git a/server/services/agent-body-filler.js b/server/services/agent-body-filler.js new file mode 100644 index 0000000..55e4a2d --- /dev/null +++ b/server/services/agent-body-filler.js @@ -0,0 +1,119 @@ +/** + * 招聘后同进程调 agent-creator 填自己的定义文件——带互斥锁 + 有限重试 + 失败落地。 + * + * 32 号文档方案 B 的落地。修的是 `company.js` hire 路由原来那发 + * `fetch('http://127.0.0.1:${PORT}/api/agent/agent-creator/chat', ...).catch(...)`： + * fire-and-forget，`.catch` 只接网络层错误，不检查响应状态码；agent-creator 是单会话， + * 撞车时 `/api/agent/:agentId/chat` 返回 429，被原样当成"成功"吞掉——新招的角色永久停在 + * base 空壳 + `status=training`，唯一痕迹是一行 console.error（撞车走的是 429 正常响应， + * 连这行都不会触发）。 + * + * 这里改成同进程直接调 `AgentManager.callAgent`，绕开那次多余的 HTTP 回环，换来： + * - 真实的调用结果/异常，而不是网络层 `.catch`； + * - 撞车时**等一等再试**（有限次数 + 退避），不是当场丢弃； + * - 重试耗尽后在本体目录写一个哨兵文件，而不是一行日志——不动 + * `system_agents.status`/`config.json.status`（那两个字段还有 draft→active 的 + * 毕业逻辑在读，改它们的取值域风险和收益不对等，见 32 号文档 §四.3）。 + * + * ⚠️ 互斥锁复用的是 `agent-session-lock.js`，跟 `routes/agent-chat.js` 的 + * `POST /:agentId/chat` 共享同一个 Map——不能各起一把。原因见该模块头注： + * `SessionPool.getOrCreate()` 不检查上一次调用有没有跑完，两条路径各自以为自己 + * 没有并发的话，会把第二轮 prompt 发进一个还在处理中的 CLI 会话，比现在的 + * "被拒绝丢弃"更危险（协议层内容串台，未定义行为）。 + */ +const fs = require('fs'); +const path = require('path'); +const locator = require('./agent-locator'); +const { agentChatLocks, AGENT_LOCK_TIMEOUT } = require('../src/core/agent-session-lock'); + +const MAX_ATTEMPTS = 3; +// 重试之间的等待——不是精算出来的数字，是"给撞车的那一方一点时间跑完"的量级； +// 真正的硬顶是 MAX_ATTEMPTS，不是靠拉长等待时间去覆盖更长的撞车窗口。 +const RETRY_DELAY_MS = [5_000, 15_000]; + +const FAILURE_MARKER_NAME = '.body_fill_failed.json'; + +function sleep(ms) { + return new Promise((resolve) => { setTimeout(resolve, ms); }); +} + +function failureMarkerPath(bodyDir) { + return path.join(bodyDir, FAILURE_MARKER_NAME); +} + +/** 重试耗尽后落地——**必须能被查询/扫描到**，这是 32 号文档方案 B 的唯一硬要求。 */ +function writeFailureMarker(bodyDir, info) { + try { + fs.writeFileSync(failureMarkerPath(bodyDir), JSON.stringify({ + ...info, + failedAt: new Date().toISOString(), + }, null, 2)); + } catch (e) { + // 连哨兵文件都写不出去（目录权限/磁盘满）——这一步再失败也不能让调用方炸， + // 招聘流程本身（scaffoldNewAgent 已经建好目录）不该被这个次要写入拖下水。 + console.error(`[BodyFill] ⚠️ 写失败标记本身也失败了: ${e.message}`); + } +} + +/** 成功之后必须清掉上一轮可能留下的失败标记——不然人工重试成功了，哨兵文件还在骗人。 */ +function clearFailureMarker(bodyDir) { + try { + fs.unlinkSync(failureMarkerPath(bodyDir)); + } catch { + // 本来就没有这个文件（多数情况）——不是错误 + } +} + +/** + * @param {object} a + * @param {object} a.agentManager `AgentManager` 实例（调用方传入，不在这里 new—— + * 测试时可以注入假实现，不用真起 Claude CLI） + * @param {string} a.targetAgentId 要填定义的那个新招角色 id（== 本体目录名） + * @param {string} a.message 发给 agent-creator 的完整 brief（调用方已经拼好， + * 这个函数不关心内容，只关心「发出去、等结果、失败重试」） + * @returns {Promise<{ok: boolean, attempts: number, error?: string}>} + */ +async function fillAgentBodyReliably({ agentManager, targetAgentId, message }) { + const bodyDir = locator.bodyDir(targetAgentId); + let lastError = null; + + for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + const existingLock = agentChatLocks.get('agent-creator'); + if (existingLock && Date.now() - existingLock.startedAt < AGENT_LOCK_TIMEOUT) { + // 忙——等一等，不占位、不硬抢、不当场放弃。这是原来那发 fetch 唯一缺的东西。 + console.log(`[BodyFill] 🔒 agent-creator 正忙（conv ${existingLock.convId}），` + + `第 ${attempt}/${MAX_ATTEMPTS} 次尝试改为等待`); + lastError = new Error(`agent-creator 正忙（活跃会话 ${existingLock.convId}）`); + if (attempt < MAX_ATTEMPTS) await sleep(RETRY_DELAY_MS[attempt - 1] ?? RETRY_DELAY_MS.at(-1)); + continue; + } + + const convId = `body-fill-${targetAgentId}-a${attempt}-${Date.now().toString(36)}`; + agentChatLocks.set('agent-creator', { convId, startedAt: Date.now() }); + try { + await agentManager.callAgent('agent-creator', message, { + metadata: { taskId: convId }, + }); + clearFailureMarker(bodyDir); + return { ok: true, attempts: attempt }; + } catch (err) { + lastError = err; + console.error(`[BodyFill] ⚠️ agent-creator 第 ${attempt}/${MAX_ATTEMPTS} 次调用失败: ${err.message}`); + if (attempt < MAX_ATTEMPTS) await sleep(RETRY_DELAY_MS[attempt - 1] ?? RETRY_DELAY_MS.at(-1)); + } finally { + // 只释放"自己那一把"——万一在途期间锁已经被别的机制清过又被第三方占了， + // 不能凭空把第三方的锁摘掉（同 agent-chat.js releaseLock 的判据）。 + const cur = agentChatLocks.get('agent-creator'); + if (cur && cur.convId === convId) agentChatLocks.delete('agent-creator'); + } + } + + writeFailureMarker(bodyDir, { + targetAgentId, + attempts: MAX_ATTEMPTS, + lastError: lastError ? lastError.message : null, + }); + return { ok: false, attempts: MAX_ATTEMPTS, error: lastError ? lastError.message : null }; +} + +module.exports = { fillAgentBodyReliably, FAILURE_MARKER_NAME }; diff --git a/server/src/core/agent-session-lock.js b/server/src/core/agent-session-lock.js new file mode 100644 index 0000000..e42d66c --- /dev/null +++ b/server/src/core/agent-session-lock.js @@ -0,0 +1,17 @@ +/** + * Agent 级并发互斥锁的共享存储（32 号文档方案 B）。 + * + * 从 `routes/agent-chat.js` 抽出来：那边的 `POST /:agentId/chat` 路由用一个进程内 + * `Map` 挡「同一个 agent 同时只允许一个活跃会话」，`company.js` 的招聘装配重试逻辑 + * （`services/agent-body-filler.js`）现在也要用到同一个判据——如果两条路径各自维护 + * 一份锁，各自都以为自己没有并发，实际上会同时往同一个 `SessionPool` 会话发两轮请求。 + * `SessionPool.getOrCreate()`（`session-pool.js:160`）只看连接在不在，不看上一次调用有没有 + * 跑完，两条路径不共享锁的话，这个坑照样在，只是换了个触发组合。 + * + * 纯数据结构，不含业务逻辑——acquire/release/超时判断仍然各自在调用方写，行为跟 + * agent-chat.js 原来内联声明时逐字一致，这次只是换了个文件。 + */ +const agentChatLocks = new Map(); // agentId → { convId, startedAt } +const AGENT_LOCK_TIMEOUT = 30 * 60 * 1000; // 30 分钟安全阀，与原实现一致 + +module.exports = { agentChatLocks, AGENT_LOCK_TIMEOUT }; diff --git a/server/tests/unit/agent-body-filler.test.js b/server/tests/unit/agent-body-filler.test.js new file mode 100644 index 0000000..6681bb8 --- /dev/null +++ b/server/tests/unit/agent-body-filler.test.js @@ -0,0 +1,160 @@ +/** + * Unit Tests — agent-body-filler.js（32 号文档方案 B 的落地） + * + * 锁的是：撞车时等一等重试（不是当场丢弃）、重试耗尽落地为可查询的哨兵文件（不是一行 + * 日志）、成功后清掉上一轮可能留下的失败标记、互斥锁只释放自己那一把（不误删别人的）、 + * 以及最要紧的一条——这把锁跟 `agent-chat.js` 的 `POST /:agentId/chat` 共享同一个 Map + * （用真实 `agent-session-lock.js` 模块，不 mock 它，直接验证两边操作的是同一个对象）。 + * + * 用 vi.useFakeTimers() 控制重试之间的等待——RETRY_DELAY_MS 是 5s/15s 量级，真等出来 + * 这个测试文件就要跑 20+ 秒。 + * + * locator.bodyDir() 没有 env 覆盖点，跟 identity-context.test.js / + * describe-role-definition-truncation.test.js 同款手法：起一个 zz- 前缀的真实临时本体 + * 目录，afterEach 清理不留痕迹。 + */ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import fs from 'fs'; +import path from 'path'; + +const REPO_ROOT = path.resolve(__dirname, '../../..'); +const AGENTS_ROOT = path.join(REPO_ROOT, 'system', 'agents'); + +describe('agent-body-filler.fillAgentBodyReliably', () => { + const targetAgentId = 'zz-test-body-filler-target'; + const bodyDir = path.join(AGENTS_ROOT, targetAgentId); + const markerPath = path.join(bodyDir, '.body_fill_failed.json'); + + let fillAgentBodyReliably; + let agentChatLocks; + + beforeEach(() => { + fs.mkdirSync(bodyDir, { recursive: true }); + vi.useFakeTimers(); + + delete require.cache[require.resolve('../../src/core/agent-session-lock.js')]; + delete require.cache[require.resolve('../../services/agent-body-filler.js')]; + ({ agentChatLocks } = require('../../src/core/agent-session-lock.js')); + ({ fillAgentBodyReliably } = require('../../services/agent-body-filler.js')); + agentChatLocks.clear(); + }); + + afterEach(() => { + vi.useRealTimers(); + agentChatLocks.clear(); + fs.rmSync(bodyDir, { recursive: true, force: true }); + }); + + /** 跑 fillAgentBodyReliably 的同时把假定时器一路推走，让内部的 sleep() 立刻"过去" */ + async function runWithTimers(promise) { + let settled = false; + promise.finally(() => { settled = true; }); + for (let i = 0; i < 10 && !settled; i++) { + await vi.advanceTimersByTimeAsync(20_000); + } + return promise; + } + + it('第一次就成功：attempts=1，不写失败标记', async () => { + const agentManager = { callAgent: vi.fn().mockResolvedValue({ output: 'ok' }) }; + const result = await runWithTimers(fillAgentBodyReliably({ + agentManager, targetAgentId, message: 'brief', + })); + expect(result).toEqual({ ok: true, attempts: 1 }); + expect(agentManager.callAgent).toHaveBeenCalledTimes(1); + expect(agentManager.callAgent).toHaveBeenCalledWith('agent-creator', 'brief', expect.any(Object)); + expect(fs.existsSync(markerPath)).toBe(false); + // 锁用完必须释放，不能一直占着 + expect(agentChatLocks.has('agent-creator')).toBe(false); + }); + + it('第一次调用抛错、第二次成功：attempts=2，不写失败标记', async () => { + const agentManager = { + callAgent: vi.fn() + .mockRejectedValueOnce(new Error('transient')) + .mockResolvedValueOnce({ output: 'ok' }), + }; + const result = await runWithTimers(fillAgentBodyReliably({ + agentManager, targetAgentId, message: 'brief', + })); + expect(result).toEqual({ ok: true, attempts: 2 }); + expect(agentManager.callAgent).toHaveBeenCalledTimes(2); + expect(fs.existsSync(markerPath)).toBe(false); + }); + + it('agent-creator 被占用时不当场放弃，等锁释放后重试成功', async () => { + // 模拟"agent-creator 正忙"：先手工占上锁，跟 agent-chat.js 的 POST /:agentId/chat + // 会做的事一模一样——这就是两边共享同一个 Map 才成立的场景。 + agentChatLocks.set('agent-creator', { convId: 'someone-elses-chat', startedAt: Date.now() }); + + const agentManager = { callAgent: vi.fn().mockResolvedValue({ output: 'ok' }) }; + const resultPromise = fillAgentBodyReliably({ agentManager, targetAgentId, message: 'brief' }); + + // 第一次尝试应该看到锁被占用，完全不调用 agentManager + await vi.advanceTimersByTimeAsync(1); + expect(agentManager.callAgent).not.toHaveBeenCalled(); + + // 锁释放（比如占用它的那次 chat 结束了），推进定时器让重试发生 + agentChatLocks.delete('agent-creator'); + const result = await runWithTimers(resultPromise); + + expect(result.ok).toBe(true); + expect(agentManager.callAgent).toHaveBeenCalledTimes(1); + }); + + it('重试耗尽：ok:false，写出可查询的失败标记，内容带上尝试次数与最后一次错误', async () => { + const agentManager = { callAgent: vi.fn().mockRejectedValue(new Error('agent-creator 挂了')) }; + const result = await runWithTimers(fillAgentBodyReliably({ + agentManager, targetAgentId, message: 'brief', + })); + + expect(result.ok).toBe(false); + expect(result.attempts).toBe(3); // MAX_ATTEMPTS + expect(agentManager.callAgent).toHaveBeenCalledTimes(3); + + expect(fs.existsSync(markerPath)).toBe(true); + const marker = JSON.parse(fs.readFileSync(markerPath, 'utf-8')); + expect(marker.targetAgentId).toBe(targetAgentId); + expect(marker.attempts).toBe(3); + expect(marker.lastError).toContain('agent-creator 挂了'); + expect(marker.failedAt).toBeTruthy(); + + // 锁不能因为失败就永久占着 + expect(agentChatLocks.has('agent-creator')).toBe(false); + }); + + it('上一轮留下失败标记后，这一轮成功了要清掉——不能让哨兵文件继续骗人', async () => { + fs.writeFileSync(markerPath, JSON.stringify({ stale: true })); + const agentManager = { callAgent: vi.fn().mockResolvedValue({ output: 'ok' }) }; + const result = await runWithTimers(fillAgentBodyReliably({ + agentManager, targetAgentId, message: 'brief', + })); + expect(result.ok).toBe(true); + expect(fs.existsSync(markerPath)).toBe(false); + }); + + it('只释放自己抢到的那把锁：调用期间锁被第三方抢先改写不会被这里误删', async () => { + // agentManager.callAgent 在途期间，模拟第三方（比如超时安全阀强制释放后又被 + // 另一次调用抢到）把锁换成了别的 convId——finally 里的判据必须是 + // "convId 匹配才删"，不能是"只要有锁就删"。 + let resolveCall; + const agentManager = { + callAgent: vi.fn(() => new Promise((r) => { resolveCall = r; })), + }; + const resultPromise = fillAgentBodyReliably({ agentManager, targetAgentId, message: 'brief' }); + await vi.advanceTimersByTimeAsync(1); + + const mine = agentChatLocks.get('agent-creator'); + expect(mine).toBeTruthy(); + // 第三方悄悄把锁换成了自己的 + agentChatLocks.set('agent-creator', { convId: 'third-party-conv', startedAt: Date.now() }); + + resolveCall({ output: 'ok' }); + await runWithTimers(resultPromise); + + // 第三方的锁必须还在——不能被这次调用的 finally 顺手清掉 + expect(agentChatLocks.get('agent-creator')).toEqual( + expect.objectContaining({ convId: 'third-party-conv' }) + ); + }); +});