"use strict"; /** * Inspection domain handlers for AI agent tools (RCA + natural-language ops). * * Each handler matches a capability id declared in capabilities/catalog/inspection.cjs * and is wired into the builtinRpcHandlerRegistry via registerInspectionHandlers(). * * Storage: * alertRecordStore and inspectionResultStore take (readFn, writeFn) — we resolve storage * lazily on first call so this module can be require()d before the app fully initializes. */ const alertRecordStore = require("../inspection/alertRecordStore.cjs"); const inspectionResultStore = require("../inspection/inspectionResultStore.cjs"); const inspectionAlertRuleStore = require("../inspection/alertRuleStore.cjs"); const inspectionDeviceStore = require("../inspection/deviceStore.cjs"); // Lazy storage adapter — resolved from Electron's global bridge on first call. let _storageBridge = null; let _vaultBridge = null; function getStorageBridge() { if (!_storageBridge) { try { // 和 inspectionStorage.cjs 保持一致的文件名 const { app } = require("electron"); const path = require("path"); const fs = require("fs"); const userData = app.getPath("userData"); const storeFile = path.join(userData, "inspection-data.json"); _storageBridge = { _storeFile: storeFile, read: (key) => { try { if (!fs.existsSync(storeFile)) return null; const raw = fs.readFileSync(storeFile, "utf-8"); const all = JSON.parse(raw || "{}"); return all[key] ?? null; } catch (_) { return null; } }, write: (key, value) => { try { let all = {}; if (fs.existsSync(storeFile)) { all = JSON.parse(fs.readFileSync(storeFile, "utf-8") || "{}"); } all[key] = value; fs.writeFileSync(storeFile, JSON.stringify(all, null, 2), "utf-8"); } catch (e) { console.error("[InspectionAgent] write failed:", e.message); } }, }; } catch (e) { console.warn("[InspectionAgent] storage bridge not available:", e.message); _storageBridge = { read: () => null, write: () => {} }; } console.log("[InspectionAgent] storage file:", _storageBridge._storeFile); } return _storageBridge; } function getVaultBridge() { if (!_vaultBridge) { try { const vaultStore = require("../../vault/hostStore.cjs"); _vaultBridge = vaultStore; } catch (e) { try { _vaultBridge = require("../vault/vaultStore.cjs"); } catch (_) { console.warn("[InspectionAgent] vault bridge not available"); _vaultBridge = null; } } } return _vaultBridge; } function readFn(key) { return getStorageBridge().read(key); } function writeFn(key, val) { getStorageBridge().write(key, val); } // ---------- Handlers ---------- function handleAlertDetail(params) { const id = params?.alertId || params?.id; if (!id) return { error: "alertId required" }; return alertRecordStore.getAlert(readFn, id); } function handleHostAlerts(params) { const deviceId = params?.deviceId || params?.hostId; if (!deviceId) return { error: "deviceId required" }; return alertRecordStore.listAlerts(readFn, { device_id: deviceId, limit: 50, }); } function handleHostMetrics(params) { const deviceId = params?.deviceId || params?.hostId; if (!deviceId) return { error: "deviceId required" }; const records = inspectionResultStore.getLatestByDevice(readFn, deviceId, 1); if (!records || records.length === 0) return { device_id: deviceId, metrics: null, status: "no_data" }; const latest = records[0]; // catalogId 兜底:优先 result(新巡检存的),没有则从 device extension 拿 let catalogId = latest.catalog_id || undefined; try { const ext = inspectionDeviceStore.getExtension(readFn, deviceId); console.log("[InspectionAgent] getExtension for", deviceId, "→", ext ? JSON.stringify({ host_id: ext.host_id, catalog_id: ext.catalog_id, hostname: ext.hostname }) : "null"); if (!catalogId) catalogId = ext?.catalog_id || undefined; } catch (e) { console.warn("[InspectionAgent] getExtension error:", e?.message); } console.log("[InspectionAgent] handleHostMetrics → result.catalog_id:", latest.catalog_id, "| final catalogId:", catalogId); return { device_id: deviceId, device_name: latest.device_name, catalogId, timestamp: latest.timestamp, metrics: latest.metrics || {}, online: latest.metrics?.online !== false, }; } function handleHostRecentResults(params) { const deviceId = params?.deviceId || params?.hostId; const limit = Math.min(params?.limit || 10, 50); if (!deviceId) return { error: "deviceId required" }; return inspectionResultStore.getLatestByDevice(readFn, deviceId, limit); } function handleAllActiveAlerts() { return alertRecordStore.getActiveAlerts(readFn); } function handleHostInfo(params) { const deviceId = params?.deviceId || params?.hostId; if (!deviceId) return { error: "deviceId required" }; // 先从 inspection device extension 拿 catalogId(这个不依赖 vault,始终可用) let catalogId = undefined; let hostname = undefined; let username = undefined; let port = undefined; try { const ext = inspectionDeviceStore.getExtension(readFn, deviceId); console.log("[InspectionAgent] hostInfo getExtension for", deviceId, "→", ext ? JSON.stringify({ catalog_id: ext.catalog_id, hostname: ext.hostname }) : "null"); catalogId = ext?.catalog_id || undefined; hostname = ext?.hostname || undefined; } catch (_) {} // 尝试 vault 拿更多设备信息 const vault = getVaultBridge(); if (vault) { let host = null; if (typeof vault.getHost === "function") host = vault.getHost(deviceId); if (!host && typeof vault.getById === "function") host = vault.getById(deviceId); if (!host && typeof vault.listHosts === "function") { const hosts = vault.listHosts(); host = hosts.find((h) => h.id === deviceId) || null; } if (host) { hostname = hostname || host.hostname || host.address; username = host.username; port = host.port || 22; if (!catalogId) catalogId = host.catalog_id || host.catalogId || undefined; } } console.log("[InspectionAgent] handleHostInfo → catalogId:", catalogId, "hostname:", hostname); // 至少返回 catalogId(从 inspection extension 兜底) return { device_id: deviceId, hostname, username, port, catalogId, }; } function handleHostTopology(params) { const deviceId = params?.deviceId || params?.hostId; if (!deviceId) return { error: "deviceId required" }; const vault = getVaultBridge(); if (!vault) return { error: "vault not available" }; // Get host list + offline status for topology context let hosts = []; if (typeof vault.listHosts === "function") hosts = vault.listHosts(); else if (typeof vault.getAll === "function") hosts = vault.getAll(); const alerts = alertRecordStore.getActiveAlerts(readFn); const offlineIds = new Set( alerts.filter((a) => a.rule_id === "rule_offline" && a.status === "active").map((a) => a.device_id) ); const criticalIds = new Set( alerts.filter((a) => a.level === "critical" && a.status === "active" && a.rule_id !== "rule_offline").map((a) => a.device_id) ); const peers = hosts .filter((h) => h.id !== deviceId) .map((h) => ({ id: h.id, name: h.label || h.hostname, offline: offlineIds.has(h.id), critical: criticalIds.has(h.id), })); return { device_id: deviceId, peers_total: peers.length, peers_offline: peers.filter((p) => p.offline).length, peers_critical: peers.filter((p) => p.critical).length, peers, }; } /** * inspection.runCommand — execute a single shell command on a device via independent SSH connection. * Used by RCA agent to run diagnosis commands on the target device. * * Params: { deviceId, command, timeoutMs? } * Returns: { ok, stdout, stderr, exitCode, durationMs, error? } */ async function handleRunCommand(params) { const deviceId = params?.deviceId; const command = params?.command; if (!deviceId) return { ok: false, error: "deviceId required" }; if (!command) return { ok: false, error: "command required" }; // 1. 优先从 sessions 按 hostname 复用已有 SSH 连接(最快、不用拿 vault) const sessions = getSessions(); let targetHostname = null; // 尝试从 inspection metrics 拿 hostname(通过 metrics 存储里的 catalog 扩展) try { const ext = inspectionDeviceStore.getExtension(readFn, deviceId); targetHostname = ext?.hostname || null; } catch (_) {} // 尝试 vault(如果碰巧能拿到) if (!targetHostname) { const vault = getVaultBridge(); if (vault) { let host = null; if (typeof vault.getHost === "function") host = vault.getHost(deviceId); if (!host && typeof vault.listHosts === "function") { host = vault.listHosts().find((h) => h.id === deviceId); } if (host) targetHostname = host.hostname || host.address; } } // 在 sessions Map 里按 hostname 匹配 let matchedSession = null; if (sessions && targetHostname) { for (const session of sessions.values()) { if (!session || !session.conn) continue; const sessionHost = session.hostname || session.host || ""; if (sessionHost === targetHostname) { matchedSession = session; break; } } } const start = Date.now(); if (matchedSession) { // —— 复用已有 SSH 连接发 exec 命令 —— return new Promise((resolve) => { const timeoutMs = params.timeoutMs || 15000; let timedOut = false; const timer = setTimeout(() => { timedOut = true; resolve({ ok: false, stdout: "", stderr: "", exitCode: -1, durationMs: Date.now() - start, error: `command timed out after ${timeoutMs}ms`, }); }, timeoutMs); try { matchedSession.conn.exec(command, { pty: false }, (err, channel) => { if (timedOut) return; if (err) { clearTimeout(timer); resolve({ ok: false, stdout: "", stderr: err.message || String(err), exitCode: -1, durationMs: Date.now() - start, error: err.message || String(err), }); return; } let stdout = ""; let stderr = ""; channel.on("data", (d) => { stdout += d.toString(); }); channel.stderr?.on("data", (d) => { stderr += d.toString(); }); channel.on("close", (code) => { if (timedOut) return; clearTimeout(timer); resolve({ ok: code === 0, stdout, stderr, exitCode: code ?? 0, durationMs: Date.now() - start, error: code !== 0 ? `exit code ${code}` : undefined, }); }); // 安全兜底 channel.on("error", (e) => { if (timedOut) return; clearTimeout(timer); resolve({ ok: false, stdout, stderr: e.message || String(e), exitCode: -1, durationMs: Date.now() - start, error: e.message || String(e), }); }); }); } catch (e) { clearTimeout(timer); resolve({ ok: false, stdout: "", stderr: e?.message || String(e), exitCode: -1, durationMs: Date.now() - start, error: e?.message || String(e), }); } }); } // 2. 没有匹配的已有 SSH session — 无法独立连接(vault 凭据不可得) // 告诉 RCA 在设备上打开一个终端 tab 后重试 return { ok: false, error: "目标设备没有活跃的 SSH 终端会话。请先在 Vault 里打开该设备的终端,让 RCA 复用连接执行诊断命令。", }; } /** * inspection.runNow — synchronous immediate inspection. * Requires the device to be reachable; uses the same SSH adapter as the scheduler. * We can't re-use runInspection from index.cjs directly (it lives in Electron ipcMain scope * and has side effects via broadcast). Instead we construct a fresh runner on demand. */ async function handleRunNow(params) { const deviceId = params?.deviceId || params?.hostId; if (!deviceId) return { error: "deviceId required" }; const vault = getVaultBridge(); if (!vault) return { error: "vault not available" }; let host = null; if (typeof vault.getHost === "function") host = vault.getHost(deviceId); if (!host && typeof vault.getById === "function") host = vault.getById(deviceId); if (!host && typeof vault.listHosts === "function") { host = vault.listHosts().find((h) => h.id === deviceId); } if (!host) return { error: `host ${deviceId} not found in Vault` }; const hostConfig = { id: host.id, label: host.label, hostname: host.hostname || host.address, port: host.port || 22, username: host.username, password: host.password, privateKey: host.privateKey, passphrase: host.passphrase, }; let catalogId = "linux_server"; try { const ext = inspectionDeviceStore.getExtension(readFn, deviceId); if (ext?.catalog_id) catalogId = ext.catalog_id; } catch (_) {} try { const { execSshCommand, execSshCommands } = require("../inspection/inspectionExec.cjs"); const { createAdapter } = require("../inspection/adapterRegistry.cjs"); const deviceStoreLocal = require("../inspection/deviceStore.cjs"); const sshConfig = { host: hostConfig.hostname, port: hostConfig.port, username: hostConfig.username, password: hostConfig.password, privateKey: hostConfig.privateKey, passphrase: hostConfig.passphrase, }; const execFn = async (command, options = {}) => { const timeoutMs = options.timeoutMs || 15000; return await execSshCommand(sshConfig, command, { connectTimeoutMs: 20000, execTimeoutMs: timeoutMs, }); }; const execCommandsFn = async (commands, options = {}) => { return await execSshCommands(sshConfig, commands, options); }; const catalogs = deviceStoreLocal.getCatalog(); const catalog = catalogs.find((c) => c.id === catalogId); const adapterClassName = catalog?.adapter_class || "LinuxServerAdapter"; const adapter = createAdapter(adapterClassName, sshConfig, { execFn, execCommands: execCommandsFn, timeoutMs: 20000, }); if (!adapter) { return { device_id: deviceId, status: "failed", error: "adapter " + adapterClassName + " not found" }; } const start = Date.now(); const collectResult = await adapter.collect(); const durationMs = Date.now() - start; // Save to result store try { inspectionResultStore.addResult(readFn, writeFn, { id: `ins_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`, device_id: deviceId, device_name: host.label || host.hostname || deviceId, catalog_id: catalogId, start_time: start, end_time: Date.now(), duration_ms: durationMs, status: collectResult.online ? "success" : "failed", online: collectResult.online, metrics: collectResult.metrics || {}, raw: collectResult.raw || {}, error_message: collectResult.error || null, }); } catch (_) {} return { device_id: deviceId, device_name: host.label || host.hostname, status: collectResult.online ? "success" : "failed", durationMs, metrics: collectResult.metrics || null, error: collectResult.error || null, }; } catch (e) { return { device_id: deviceId, status: "failed", error: e.message, metrics: null, }; } } // Lazy storage adapter — resolved from Electron's global bridge on first call. // (模块级变量已在文件顶部声明:_storageBridge / _vaultBridge) let _sessions = null; /** * Context injector — called by main process bridge registration to supply * shared dependencies. `sessions` is the main-process SSH session Map * (sessionId → session with .conn ssh2.Client). */ function configure(ctx = {}) { if (ctx.sessions) _sessions = ctx.sessions; } function getSessions() { return _sessions; } /** * Map capability id → handler function. Spread this into buildBuiltinRpcHandlerRegistry(). */ function registerInspectionHandlers() { return { "inspection.alertDetail": handleAlertDetail, "inspection.hostAlerts": handleHostAlerts, "inspection.hostMetrics": handleHostMetrics, "inspection.hostRecentResults": handleHostRecentResults, "inspection.hostTopology": handleHostTopology, "inspection.hostInfo": handleHostInfo, "inspection.runNow": handleRunNow, "inspection.runCommand": handleRunCommand, "inspection.allActiveAlerts": handleAllActiveAlerts, }; } module.exports = { registerInspectionHandlers, configure, getSessions, // exported for testing / direct invocation handleAlertDetail, handleHostAlerts, handleHostMetrics, handleHostRecentResults, handleHostTopology, handleHostInfo, handleAllActiveAlerts, };