Restore edge agent shell polling

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 {