From 4e57287b5ad592f60794d52d88e4c6f8d0c2b5b0 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 31 May 2026 12:50:51 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A3=D1=81=D0=B8=D0=BB=D0=B8=D1=82=D1=8C=20he?= =?UTF-8?q?artbeat=20AI=20Workspace=20Hub?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- services/ai-workspace-hub/src/server.mjs | 74 ++++++++++++++++++++---- 1 file changed, 63 insertions(+), 11 deletions(-) diff --git a/services/ai-workspace-hub/src/server.mjs b/services/ai-workspace-hub/src/server.mjs index aecd910..71867c2 100644 --- a/services/ai-workspace-hub/src/server.mjs +++ b/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);