From dae7e90a990bbe98e5565cbedd27ddd768154dea Mon Sep 17 00:00:00 2001 From: yibo Date: Mon, 13 Apr 2026 11:00:24 +0800 Subject: [PATCH] fix: prevent concurrent Master sessions from racing on template files --- .gitignore | 1 + server/routes/agent-chat.js | 25 +++++++++++++++++++++++++ server/services/heartbeat-service.js | 32 +++++++++++++++++++++++++++++--- server/utils/file-lock.js | 74 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 129 insertions(+), 3 deletions(-) create mode 100644 server/utils/file-lock.js diff --git a/.gitignore b/.gitignore index 78b28e5..162cb2e 100644 --- a/.gitignore +++ b/.gitignore @@ -94,6 +94,7 @@ system/data/companies/ # Temp/backup files *.bak +*.lock EOF PYEOF diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index c81baa6..0d5b4c2 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -2178,6 +2178,11 @@ router.get('/:agentId/uploads/:convId/:filename', (req, res) => { // POST /api/agent/:agentId/chat — SSE 流式对话 // ═══════════════════════════════════════ +// 🔒 Agent 级并发互斥锁 — 同一 Agent 同时只允许一个活跃 chat session +// 同一 convId 的多轮对话不受限制（ACP 进程复用） +const agentChatLocks = new Map(); // agentId → { convId, startedAt } +const AGENT_LOCK_TIMEOUT = 30 * 60 * 1000; // 30 分钟安全阀 + router.post('/:agentId/chat', async (req, res) => { const { message, conversationId: reqConvId } = req.body; if (!message) return res.status(400).json({ error: 'message is required' }); @@ -2187,6 +2192,25 @@ router.post('/:agentId/chat', async (req, res) => { if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const convId = reqConvId || `conv-${Date.now().toString(36)}`; + + // 🔒 并发互斥检查：同一 Agent 不同 convId 的并发请求被拒绝 + const existingLock = agentChatLocks.get(agentId); + if (existingLock && existingLock.convId !== convId) { + const elapsed = Date.now() - existingLock.startedAt; + if (elapsed < AGENT_LOCK_TIMEOUT) { + console.log(`[AgentChat] 🔒 拒绝 ${agentId} 并发请求 — 已有活跃会话 ${existingLock.convId} (${(elapsed / 1000).toFixed(0)}s)`); + return res.status(429).json({ + error: `Agent ${agentId} 正在执行另一个会话 (${existingLock.convId})，请等待完成`, + activeConvId: existingLock.convId, + elapsedMs: elapsed + }); + } + // 超时释放 + console.warn(`[AgentChat] ⚠️ 强制释放超时锁: ${agentId} / ${existingLock.convId}`); + agentChatLocks.delete(agentId); + } + agentChatLocks.set(agentId, { convId, startedAt: Date.now() }); + const msgId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`; const sendSSE = setupSSE(res); const identity = loadIdentity(agentDir); @@ -2492,6 +2516,7 @@ ${resultsText.join('\n\n')} sendSSE('error', { message: err.message }); console.log(`[SSE-DEBUG] 🚨 sendSSE('error') called:`, err.message.substring(0, 200)); } finally { + agentChatLocks.delete(agentId); res.end(); } }); diff --git a/server/services/heartbeat-service.js b/server/services/heartbeat-service.js index 105b9d6..7e4b4b3 100644 --- a/server/services/heartbeat-service.js +++ b/server/services/heartbeat-service.js @@ -14,7 +14,8 @@ function getAgentDir(id) { class HeartbeatService { constructor() { - this.timers = new Map(); // agentId -> NodeJS.Timeout + this.timers = new Map(); // agentId → NodeJS.Timeout + this.activeSessions = new Map(); // agentId → { convId, startedAt } — 防并发互斥 this.port = config.port || 3000; } @@ -97,17 +98,35 @@ class HeartbeatService { * 内部方法：向 Agent 的 Chat API 发送一次隐式的系统心跳 */ _triggerHeartbeat(agentId) { + // 🔒 互斥检查：如果该 Agent 已有活跃心跳会话，跳过本次 + const SESSION_TIMEOUT = 30 * 60 * 1000; // 30 分钟安全阀 + if (this.activeSessions.has(agentId)) { + const active = this.activeSessions.get(agentId); + const elapsed = Date.now() - active.startedAt; + if (elapsed < SESSION_TIMEOUT) { + console.log(`[Heartbeat] ⏭️ 跳过 ${agentId} — 已有活跃会话 ${active.convId} (${(elapsed / 1000).toFixed(0)}s)`); + return; + } + // 超时强制释放（防死锁） + console.warn(`[Heartbeat] ⚠️ 强制释放超时会话: ${agentId} / ${active.convId} (${(elapsed / 1000).toFixed(0)}s)`); + this.activeSessions.delete(agentId); + } + const timestamp = new Date().toISOString().replace('T', ' ').substring(0, 19); console.log(`[Heartbeat] 💓 触发 ${agentId} 主动心跳 (${timestamp})...`); const convId = `heartbeat-${Date.now().toString(36)}`; + + // 🔒 注册活跃会话 + this.activeSessions.set(agentId, { convId, startedAt: Date.now() }); + const heartbeatMessage = [ `[SYSTEM HEARTBEAT]`, `当前系统时间：${timestamp}`, `这是一个后台周期的主动心跳触发。`, `要求：请根据你的身份和 heartbeat.md 规则，扫描当前最新的全局 OKR 目标状态和节点。`, `1. 如果发现某个负责人超时、卡点或产出需要仲裁，请主动使用 dispatch 工具给他们发送通知或驳回。`, - `2. 如果一切正常，无事发生，请直接回复“[心跳正常]”并且不要使用任何工具。`, + `2. 如果一切正常，无事发生，请直接回复"[心跳正常]"并且不要使用任何工具。`, `无需等待人类确认，请完全自主决策。` ].join('\n'); @@ -128,12 +147,18 @@ class HeartbeatService { } }; + const releaseSession = () => { + this.activeSessions.delete(agentId); + console.log(`[Heartbeat] 🔓 释放 ${agentId} 会话锁 (${convId})`); + }; + const req = http.request(options, (res) => { let body = ''; res.on('data', (chunk) => body += chunk); res.on('end', () => { + releaseSession(); if (res.statusCode >= 200 && res.statusCode < 300) { - console.log(`[Heartbeat] 📨 ${agentId} 成功启动心跳推理会话 (${convId})`); + console.log(`[Heartbeat] 📨 ${agentId} 成功完成心跳推理会话 (${convId})`); } else { console.warn(`[Heartbeat] ⚠️ ${agentId} 心跳触发异常: ${res.statusCode} ${body}`); } @@ -141,6 +166,7 @@ class HeartbeatService { }); req.on('error', (err) => { + releaseSession(); console.error(`[Heartbeat] ❌ ${agentId} 心跳请求失败: ${err.message}`); }); diff --git a/server/utils/file-lock.js b/server/utils/file-lock.js new file mode 100644 index 0000000..4687b75 --- /dev/null +++ b/server/utils/file-lock.js @@ -0,0 +1,74 @@ +/** + * file-lock.js — 基于 .lock 文件的 Advisory Lock + * + * 用于保护模板关键文件（config.json, SKILL.md）的并发写入。 + * 使用 O_EXCL 原子操作创建锁文件，自动检测和清理过期锁。 + * + * 使用方式： + * const { withFileLock } = require('./utils/file-lock'); + * await withFileLock('/path/to/config.json', async () => { + * // 安全写入 config.json + * }); + */ + +const fs = require('fs'); + +const LOCK_TIMEOUT = 30_000; // 30s 锁超期 +const RETRY_INTERVAL = 200; // 200ms 重试间隔 +const MAX_WAIT = 10_000; // 最多等待 10s + +/** + * 在文件锁保护下执行 fn + * @param {string} filePath - 要保护的文件路径 + * @param {Function} fn - 在锁保护下执行的异步函数 + * @returns {Promise<*>} fn 的返回值 + */ +async function withFileLock(filePath, fn) { + const lockPath = filePath + '.lock'; + const deadline = Date.now() + MAX_WAIT; + + // 自旋等待获取锁 + while (Date.now() < deadline) { + try { + // O_EXCL: 如果文件已存在则失败（原子操作） + fs.writeFileSync(lockPath, JSON.stringify({ + pid: process.pid, + ts: Date.now(), + file: filePath, + }), { flag: 'wx' }); + + // 获取锁成功 + try { + return await fn(); + } finally { + try { fs.unlinkSync(lockPath); } catch {} + } + } catch (e) { + if (e.code === 'EEXIST') { + // 锁已存在 — 检查是否过期 + try { + const lockInfo = JSON.parse(fs.readFileSync(lockPath, 'utf-8')); + if (Date.now() - lockInfo.ts > LOCK_TIMEOUT) { + // 过期锁，强制移除 + console.warn(`[FileLock] ⚠️ 清理过期锁: ${lockPath} (age: ${((Date.now() - lockInfo.ts) / 1000).toFixed(0)}s)`); + try { fs.unlinkSync(lockPath); } catch {} + continue; + } + } catch { + // 锁文件被并发删除或损坏，重试 + } + + await new Promise(r => setTimeout(r, RETRY_INTERVAL)); + } else { + throw e; + } + } + } + + // 超时 — 强制移除锁文件并执行（降级策略，不死等） + console.error(`[FileLock] ❌ 获取文件锁超时 (${MAX_WAIT}ms): ${filePath}，强制执行`); + try { fs.unlinkSync(lockPath); } catch {} + return await fn(); +} + +module.exports = { withFileLock }; -- libgit2 0.21.4