diff --git a/services/edge-agent/dist/agent.mjs b/services/edge-agent/dist/agent.mjs index cb3cd5cf..2ecece90 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, @@ -1764,19 +1765,18 @@ export async function startAgent({ const commandPollTimeoutSeconds = Number(config.commandPollTimeoutSeconds || 20); const commandPollRetryDelayMs = Number(config.commandPollRetryDelayMs || 1000); const brokerReconnectDelayMs = Number(config.brokerReconnectDelayMs || DEFAULT_BROKER_RECONNECT_DELAY_MS); + const shellActionPollTimeoutSeconds = Number(config.shellActionPollTimeoutSeconds || 20); + const shellActionPollRetryDelayMs = Number(config.shellActionPollRetryDelayMs || 1000); 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 shell = createShellBridgeImpl((message) => { + brokerBridge?.send(message); + shellEventPublisher.publish(message); + }); brokerBridge = createBrokerBridge({ config, shell, @@ -1841,8 +1841,32 @@ export async function startAgent({ } }; + const runShellActionPollLoop = async () => { + while (!stopped) { + try { + 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(() => {}); @@ -1853,8 +1877,9 @@ export async function startAgent({ stopped = true; clearInterval(timer); brokerBridge?.stop(); + await shellEventPublisher.drain(); shell.dispose(); - await Promise.allSettled([commandPollPromise]); + await Promise.allSettled([commandPollPromise, shellActionPollPromise]); }; return {