Усилить heartbeat AI Workspace Hub

services/ai-workspace-hub/src/server.mjs

@@ -6,6 +6,8 @@ import { WebSocket, WebSocketServer } from "ws";
 const DEFAULT_WS_PATH = "/api/ai-workspace/hub";
 const MAX_EVENTS_PER_AGENT = 5000;
 const MAX_REQUEST_META_PER_AGENT = 1000;
+const HEARTBEAT_INTERVAL_MS = 30000;
+const AGENT_STALE_MS = 75000;
 
 const config = readConfig();
 const app = express();
@@ -80,12 +82,20 @@ httpServer.on("upgrade", (req, socket, head) => {
 wss.on("connection", attachAgent);
 
 const heartbeat = setInterval(() => {
-  for (const agent of agentsByCode.values()) {
+  for (const agent of Array.from(agentsByCode.values())) {
+    if (!isSocketOpen(agent) || isAgentStale(agent)) {
+      removeAgent(agent, "bridge_agent_stale");
+      continue;
+    }
     try {
+      agent.awaitingPong = true;
+      agent.lastPingAt = new Date().toISOString();
       agent.socket.ping();
-    } catch {}
+    } catch {
+      removeAgent(agent, "bridge_agent_ping_failed");
+    }
   }
-}, 30000);
+}, HEARTBEAT_INTERVAL_MS);
 
 httpServer.listen(config.port, "0.0.0.0", () => {
   console.log(`NODE.DC AI Workspace Hub listening on http://0.0.0.0:${config.port}`);
@@ -105,8 +115,7 @@ function attachAgent(socket, req) {
 
   const previous = agentsByCode.get(pairingCode);
   if (previous) {
-    settleAll(previous, Object.assign(new Error("agent_replaced"), { status: 503 }));
-    previous.socket.close(4000, "agent_replaced");
+    removeAgent(previous, "agent_replaced");
   }
 
   const now = new Date().toISOString();
@@ -115,8 +124,14 @@ function attachAgent(socket, req) {
     pairingCode,
     machineName: cleanString(url.searchParams.get("machineName"), 120),
     host: "",
+    runtimeStatus: "",
+    runtimeError: "",
+    runtimeCheckedAt: "",
     connectedAt: now,
     lastSeenAt: now,
+    lastPingAt: "",
+    awaitingPong: false,
+    closed: false,
     pending: new Map(),
     requestMeta: previous?.requestMeta || new Map(),
     events: previous?.events || [],
@@ -124,12 +139,12 @@ function attachAgent(socket, req) {
   agentsByCode.set(pairingCode, agent);
   pushAgentEvent(agent, "", { kind: "connected", message: "Bridge agent connected." });
 
-  socket.on("message", (raw) => handleAgentMessage(agent, raw));
-  socket.on("close", () => {
-    pushAgentEvent(agent, "", { kind: "disconnected", message: "Bridge agent disconnected." });
-    if (agentsByCode.get(pairingCode) === agent) agentsByCode.delete(pairingCode);
-    settleAll(agent, Object.assign(new Error("bridge_agent_offline"), { status: 503 }));
+  socket.on("pong", () => {
+    agent.lastSeenAt = new Date().toISOString();
+    agent.awaitingPong = false;
   });
+  socket.on("message", (raw) => handleAgentMessage(agent, raw));
+  socket.on("close", () => removeAgent(agent, "bridge_agent_offline"));
   socket.on("error", () => {});
 
   sendJson(socket, {
@@ -141,6 +156,7 @@ function attachAgent(socket, req) {
 
 function handleAgentMessage(agent, raw) {
   agent.lastSeenAt = new Date().toISOString();
+  agent.awaitingPong = false;
   let message = null;
   try {
     message = JSON.parse(String(raw));
@@ -151,6 +167,9 @@ function handleAgentMessage(agent, raw) {
   if (message?.type === "hello") {
     agent.machineName = cleanString(message.machineName || agent.machineName, 120);
     agent.host = cleanString(message.host, 120);
+    agent.runtimeStatus = cleanString(message.runtimeStatus, 40);
+    agent.runtimeError = cleanString(message.runtimeError, 1000);
+    agent.runtimeCheckedAt = cleanString(message.runtimeCheckedAt, 80);
     pushAgentEvent(agent, "", {
       kind: "hello",
       message: "Bridge agent hello received.",
@@ -269,6 +288,7 @@ function pushAgentEvent(agent, requestId, rawEvent = {}) {
     pairingCode: agent.pairingCode,
     requestId: cleanRequestId,
     threadId: rawEvent.threadId == null ? cleanString(meta?.threadId, 160) : cleanString(rawEvent.threadId, 160),
+    sessionId: rawEvent.sessionId == null ? "" : cleanString(rawEvent.sessionId, 160),
     threadTitle: rawEvent.threadTitle == null ? cleanString(meta?.threadTitle, 240) : cleanString(rawEvent.threadTitle, 240),
     userMessage: rawEvent.userMessage == null ? cleanString(meta?.userMessage, 12000) : cleanString(rawEvent.userMessage, 12000),
     remoteControlMode: rawEvent.remoteControlMode == null ? cleanString(meta?.remoteControlMode, 40) : cleanString(rawEvent.remoteControlMode, 40),
@@ -296,15 +316,47 @@ function publicAgent(agent) {
     pairingCode: agent.pairingCode,
     machineName: agent.machineName,
     host: agent.host,
+    runtimeStatus: agent.runtimeStatus,
+    runtimeError: agent.runtimeError,
+    runtimeCheckedAt: agent.runtimeCheckedAt,
     connectedAt: agent.connectedAt,
     lastSeenAt: agent.lastSeenAt,
   };
 }
 
-function isAgentOnline(agent) {
+function isSocketOpen(agent) {
   return Boolean(agent?.socket && agent.socket.readyState === WebSocket.OPEN);
 }
 
+function isAgentStale(agent) {
+  const lastSeenMs = Date.parse(agent?.lastSeenAt || agent?.connectedAt || "");
+  if (!Number.isFinite(lastSeenMs)) return true;
+  return Date.now() - lastSeenMs > AGENT_STALE_MS;
+}
+
+function isAgentOnline(agent) {
+  return isSocketOpen(agent) && !isAgentStale(agent);
+}
+
+function removeAgent(agent, reason = "bridge_agent_offline") {
+  if (!agent || agent.closed) return;
+  agent.closed = true;
+  const error = Object.assign(new Error(reason), { status: reason === "agent_replaced" ? 409 : 503 });
+  pushAgentEvent(agent, "", {
+    kind: "disconnected",
+    message: reason === "agent_replaced"
+      ? "Bridge agent connection replaced."
+      : reason === "bridge_agent_stale"
+        ? "Bridge agent heartbeat timed out."
+        : "Bridge agent disconnected.",
+  });
+  if (agentsByCode.get(agent.pairingCode) === agent) agentsByCode.delete(agent.pairingCode);
+  settleAll(agent, error);
+  try {
+    if (isSocketOpen(agent)) agent.socket.terminate();
+  } catch {}
+}
+
 function settleAll(agent, error) {
   for (const pending of agent.pending.values()) {
     clearTimeout(pending.timeout);