From 7c7895f776a2c7666fd38a2f51cc3ed61ff5e3e4 Mon Sep 17 00:00:00 2001 From: Jeppe Bundgaard Date: Tue, 21 Apr 2026 21:46:39 +0200 Subject: [PATCH] Restore edge agent shell polling --- services/edge-agent/dist/agent.mjs | 51 ++++++++++++++++++++++++------ 1 file changed, 41 insertions(+), 10 deletions(-) diff --git a/services/edge-agent/dist/agent.mjs b/services/edge-agent/dist/agent.mjs index cb3cd5cf..317c3059 100644 --- a/services/edge-agent/dist/agent.mjs +++ b/services/edge-agent/dist/agent.mjs @@ -137,6 +137,7 @@ function buildTransportHeartbeatState(brokerState = {}) { status: "ONLINE", metadata: { command_transport: brokerConnected ? "BROKER_FAST_PATH" : "API_POLLING", + shell_transport: brokerConnected ? "BROKER_FAST_PATH" : "API_POLLING", broker_connected: brokerConnected, broker_url: brokerState.url || null, broker_last_error: brokerState.lastError || null, @@ -1761,22 +1762,22 @@ export async function startAgent({ }); config.configPath = configPath; const intervalMs = Number(config.heartbeatIntervalSeconds || 15) * 1000; - const commandPollTimeoutSeconds = Number(config.commandPollTimeoutSeconds || 20); - const commandPollRetryDelayMs = Number(config.commandPollRetryDelayMs || 1000); + const commandPollTimeoutSeconds = Number(config.commandPollTimeoutSeconds ?? 20); + const commandPollRetryDelayMs = Number(config.commandPollRetryDelayMs ?? 1000); + const shellActionPollTimeoutSeconds = Number(config.shellActionPollTimeoutSeconds ?? 20); + const shellActionPollRetryDelayMs = Number(config.shellActionPollRetryDelayMs ?? 1000); const brokerReconnectDelayMs = Number(config.brokerReconnectDelayMs || DEFAULT_BROKER_RECONNECT_DELAY_MS); let stopped = false; let cpuSnapshot = null; let lastHeartbeatLatencyMs = null; - void createShellBridgeImpl; let brokerBridge = null; - const shell = { - open: async () => {}, - input: () => {}, - resize: () => {}, - close: () => {}, - dispose: () => {}, + const shellEventPublisher = createShellEventPublisher(config, fetchImpl); + const sendShellMessage = (message) => { + shellEventPublisher.publish(message); + brokerBridge?.send(message); }; + const shell = createShellBridgeImpl(sendShellMessage); brokerBridge = createBrokerBridge({ config, shell, @@ -1841,8 +1842,37 @@ export async function startAgent({ } }; + const runShellActionPollLoop = async () => { + while (!stopped) { + try { + if (brokerBridge?.state?.connected) { + await new Promise((resolve) => setTimeout(resolve, shellActionPollRetryDelayMs)); + continue; + } + + const action = await pollShellActionJob(config, fetchImpl, shellActionPollTimeoutSeconds); + if (stopped) { + break; + } + + if (!action) { + continue; + } + + await processPolledShellAction(config, action, shell, fetchImpl); + } catch { + if (stopped) { + break; + } + + await new Promise((resolve) => setTimeout(resolve, shellActionPollRetryDelayMs)); + } + } + }; + await sendTransportHeartbeat(); const commandPollPromise = runCommandPollLoop(); + const shellActionPollPromise = runShellActionPollLoop(); const timer = setInterval(() => { sendTransportHeartbeat().catch(() => {}); @@ -1854,7 +1884,8 @@ export async function startAgent({ clearInterval(timer); brokerBridge?.stop(); shell.dispose(); - await Promise.allSettled([commandPollPromise]); + await Promise.allSettled([commandPollPromise, shellActionPollPromise]); + await shellEventPublisher.drain(); }; return {