fix: prevent concurrent Master sessions from racing on template files

Layer 1 (P0): Per-agent session mutex - heartbeat-service.js: skip heartbeat if previous session still active (30min safety valve) - agent-chat.js: reject concurrent /chat requests for same agentId with different convId (429) Multi-turn conversations on same convId still work normally Layer 2 (P1): Advisory file lock utility - server/utils/file-lock.js: O_EXCL-based .lock file mechanism Auto-cleanup stale locks, graceful degradation on timeout Root cause: HeartbeatService creates new convId per heartbeat, spawning independent Claude CLI processes that concurrently write config.json and SKILL.md, causing rollbacks and FRAUDULENT scores.

fix: prevent concurrent Master sessions from racing on template files
Layer 1 (P0): Per-agent session mutex - heartbeat-service.js: skip heartbeat if previous session still active (30min safety valve) - agent-chat.js: reject concurrent /chat requests for same agentId with different convId (429) Multi-turn conversations on same convId still work normally Layer 2 (P1): Advisory file lock utility - server/utils/file-lock.js: O_EXCL-based .lock file mechanism Auto-cleanup stale locks, graceful degradation on timeout Root cause: HeartbeatService creates new convId per heartbeat, spawning independent Claude CLI processes that concurrently write config.json and SKILL.md, causing rollbacks and FRAUDULENT scores.
yibo
1 parent b1a348b0
Showing 4 changed files with 129 additions and 3 deletions
.gitignore
server/routes/agent-chat.js
server/services/heartbeat-service.js
server/utils/file-lock.js
@@ -94,6 +94,7 @@ system/data/companies/
 # Temp/backup files
 *.bak
+*.lock
 EOF
 PYEOF
@@ -2178,6 +2178,11 @@ router.get(&#39;/:agentId/uploads/:convId/:filename&#39;, (req, res) =&gt; {
 //  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(&#39;/:agentId/chat&#39;, async (req, res) =&gt; {
   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(&#39;\n\n&#39;)}
     sendSSE('error', { message: err.message });
     console.log(`[SSE-DEBUG] 🚨 sendSSE('error') called:`, err.message.substring(0, 200));
   } finally {
+    agentChatLocks.delete(agentId);
     res.end();
   }
 });
@@ -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}`);
     });
+/**
+ * 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 };