Report stalled agent lanes and bound health lock waits

agent/ec_agent.py

@@ -12,6 +12,7 @@ import contextlib
 from datetime import datetime, timezone
 import math
 import hashlib
+import http.client
 import fcntl
 import json
 import os
@@ -32,7 +33,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.5.1"
+VERSION = "1.5.3"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -40,6 +41,7 @@ MANIFEST_MAX_BYTES = 16 * 1024 * 1024
 MANIFEST_MAX_SECONDS = 15
 MANIFEST_MAX_DEPTH = 64
 LIBRARY_DOWNLOAD_WORKERS = 4
+HEALTH_LOCK_TIMEOUT_SECONDS = 5
 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
 AGENT_CONFIG = Path("/etc/electric-cherry/agent.json")
 AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py")
@@ -922,6 +924,28 @@ class Agent:
         self.inventory_manifest_lock = threading.Lock()
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
         self.executing_id = None
+        self.lane_lock = threading.Lock()
+        self.lanes = {name: {"thread": None, "lastPollAt": None, "polled": time.monotonic(), "busy": False}
+                      for name in ("health-1", "health-2", "read", "write")}
+
+    def lane_started(self, name):
+        with self.lane_lock:
+            self.lanes[name]["thread"] = threading.current_thread()
+
+    def lane_polled(self, name):
+        with self.lane_lock:
+            self.lanes[name].update(lastPollAt=datetime.now(timezone.utc).isoformat(timespec="milliseconds"), polled=time.monotonic())
+
+    def lane_busy(self, name, busy):
+        with self.lane_lock:
+            self.lanes[name]["busy"] = busy
+
+    def lane_inventory(self):
+        with self.lane_lock:
+            return [{"name": name, "alive": lane["thread"] is None or lane["thread"].is_alive(),
+                     "busy": lane["busy"], "lastPollAt": lane["lastPollAt"],
+                     "pollAgeSeconds": max(0, time.monotonic() - lane["polled"])}
+                    for name, lane in self.lanes.items()]
 
     def can_uninstall(self):
         try:
@@ -1099,7 +1123,7 @@ class Agent:
                     raise AgentError("Controller redirects are not allowed")
             with urllib.request.build_opener(NoRedirect()).open(req, timeout=35) as response:
                 return json.load(response)
-        except (OSError, ValueError):
+        except (OSError, ValueError, http.client.HTTPException):
             raise AgentError("Controller request failed") from None
 
     def cpu_usage_percent(self):
@@ -1333,7 +1357,7 @@ class Agent:
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
         capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
-        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
+        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
     def safe_to_disrupt(self, instance_id, generation):
         health = self.inspect(instance_id, generation)
@@ -2354,6 +2378,8 @@ class Agent:
                 self.heartbeat()
             except (AgentError, OSError, ValueError):
                 pass
+            except Exception as err:
+                print(f"Heartbeat recovered from {type(err).__name__}", file=sys.stderr)
 
     @staticmethod
     def read_operation(operation):
@@ -2363,13 +2389,29 @@ class Agent:
         key = operation.get("instanceId")
         return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext()
 
+    @contextlib.contextmanager
+    def observation_lock(self, operation, health=False):
+        lock = self.instance_lock(operation)
+        if not health or not operation.get("instanceId"):
+            with lock:
+                yield
+            return
+        if not lock.acquire(timeout=HEALTH_LOCK_TIMEOUT_SECONDS):
+            raise AgentError("Health check timed out waiting for the instance operation")
+        try:
+            yield
+        finally:
+            lock.release()
+
     @staticmethod
     def health_operation(operation):
         return operation.get("kind") in ("instance.inspect", "rcon.status-batch") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
 
-    def read_loop(self, lane="read"):
+    def read_loop(self, lane="read", lane_name=None):
         # Each lane owns its SQLite connection and never writes executing_id,
         # checkpoints, or current_operation used by the mutation state machine.
+        lane_name = lane_name or ("health-1" if lane == "health" else "read")
+        self.lane_started(lane_name)
         journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
         next_prune = time.monotonic()
         try:
@@ -2385,7 +2427,9 @@ class Agent:
                                 if journal.prune_observations(self.observation_keep_days) == 500:
                                     next_prune = time.monotonic() + 15
                     response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20")
+                    self.lane_polled(lane_name)
                     handled = False
+                    self.lane_busy(lane_name, bool(response.get("operations")))
                     for operation in response.get("operations", []):
                         if self.stopping.is_set() or not self.read_operation(operation):
                             continue
@@ -2402,8 +2446,8 @@ class Agent:
                                 row, _ = journal.accept(operation)
                                 if row["state"] in ("accepted", "unknown", "running"):
                                     self.report(journal.update(operation["id"], "running"), journal=journal)
-                                    with self.read_gate.read(), self.instance_lock(operation):
                                     try:
+                                        with self.read_gate.read(), self.observation_lock(operation, health=lane == "health"):
                                             result = self.dispatch(operation)
                                             row = journal.update(operation["id"], "completed", result)
                                     except Exception as err:
@@ -2421,6 +2465,11 @@ class Agent:
                         self.stopping.wait(0.25)
                 except (AgentError, OSError, ValueError, KeyError, sqlite3.Error):
                     self.stopping.wait(3)
+                except Exception as err:
+                    print(f"{lane} lane recovered from {type(err).__name__}", file=sys.stderr)
+                    self.stopping.wait(3)
+                finally:
+                    self.lane_busy(lane_name, False)
         finally:
             journal.close()
 
@@ -2434,6 +2483,8 @@ class Agent:
                 return
         # Never hold the updater lock while waiting for delivery.
         response = self.request("GET", "/api/agent/operations?lane=write&wait=20")
+        self.lane_polled("write")
+        self.lane_busy("write", bool(response.get("operations")))
         with self.activity() as acquired:
             if not acquired or self.stopping.is_set() or self.update_pending():
                 return
@@ -2461,6 +2512,7 @@ class Agent:
             return completed
 
     def run(self):
+        self.lane_started("write")
         # Obtain the configured host port range before accepting any deployment.
         # Enrollment's default values must never race the first heartbeat.
         while not self.stopping.is_set():
@@ -2475,10 +2527,14 @@ class Agent:
                 break
             except (AgentError, OSError, ValueError):
                 self.stopping.wait(3)
+            except Exception as err:
+                print(f"Initial heartbeat recovered from {type(err).__name__}", file=sys.stderr)
+                self.stopping.wait(3)
         heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True)
         heartbeats.start()
         # Preserve two concurrent health checks even while files are being read.
-        readers = [threading.Thread(target=self.read_loop, args=(lane,), daemon=True) for lane in ("health", "health", "read")]
+        readers = [threading.Thread(target=self.read_loop, args=(lane, name), daemon=True)
+                   for name, lane in (("health-1", "health"), ("health-2", "health"), ("read", "read"))]
         for reader in readers:
             reader.start()
         while not self.stopping.is_set():
@@ -2488,6 +2544,10 @@ class Agent:
             except (AgentError, OSError, ValueError, KeyError):
                 # Logs deliberately omit URLs, tokens, payloads and raw errors.
                 print("Host controller contact unavailable; existing containers continue running", file=sys.stderr)
+            except Exception as err:
+                print(f"Write lane recovered from {type(err).__name__}", file=sys.stderr)
+            finally:
+                self.lane_busy("write", False)
             self.stopping.wait(0.25 if completed else 3)
         for reader in readers:
             reader.join(timeout=2)

agent/tests/test_agent.py

@@ -1,5 +1,6 @@
 import base64
 import importlib.util
+import http.client
 import io
 import json
 from pathlib import Path
@@ -392,6 +393,120 @@ class AgentTests(unittest.TestCase):
         self.agent.journal.close()
         self.temp.cleanup()
 
+    def test_controller_truncated_response_is_a_sanitized_retryable_error(self):
+        # Exercise Python's actual HTTP body reader against a truncated stream.
+        sock = MagicMock()
+        sock.makefile.return_value = io.BytesIO(b'HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n{"secret":')
+        response = http.client.HTTPResponse(sock)
+        response.begin()
+        with patch.object(module.urllib.request, "build_opener") as opener:
+            opener.return_value.open.return_value = response
+            with self.assertRaisesRegex(AgentError, "^Controller request failed$"):
+                self.agent.request("GET", "/api/agent/operations")
+
+    def test_read_and_health_lanes_retry_unexpected_failures_without_logging_secrets(self):
+        for lane in ("read", "health"):
+            with self.subTest(lane=lane):
+                self.agent.stopping.clear()
+                polls = []
+
+                def request(*args):
+                    polls.append(args)
+                    if len(polls) == 1:
+                        raise RuntimeError("secret controller response")
+                    self.agent.stopping.set()
+                    return {"operations": []}
+
+                with patch.object(self.agent, "request", side_effect=request), patch.object(self.agent.stopping, "wait", return_value=False), patch.object(module.sys, "stderr", new_callable=io.StringIO) as logs:
+                    self.agent.read_loop(lane)
+                self.assertEqual(len(polls), 2)
+                self.assertIn("RuntimeError", logs.getvalue())
+                self.assertNotIn("secret", logs.getvalue())
+
+    def test_heartbeat_loop_survives_unexpected_failures(self):
+        with patch.object(self.agent.stopping, "wait", side_effect=[False, False, True]), patch.object(self.agent, "heartbeat", side_effect=[RuntimeError("secret"), None]) as heartbeat, patch.object(module.sys, "stderr", new_callable=io.StringIO) as logs:
+            self.agent.heartbeat_loop()
+        self.assertEqual(heartbeat.call_count, 2)
+        self.assertIn("RuntimeError", logs.getvalue())
+        self.assertNotIn("secret", logs.getvalue())
+
+    def test_startup_and_write_loop_survive_unexpected_failures(self):
+        polls = []
+
+        def poll():
+            polls.append(True)
+            if len(polls) == 1:
+                raise RuntimeError("secret")
+            self.agent.stopping.set()
+
+        with patch.object(module.threading, "Thread"), patch.object(self.agent.stopping, "wait", return_value=False), patch.object(self.agent, "heartbeat", side_effect=[RuntimeError("secret"), None]) as heartbeat, patch.object(self.agent, "poll_operations", side_effect=poll), patch.object(module.sys, "stderr", new_callable=io.StringIO) as logs:
+            self.agent.run()
+        self.assertEqual(heartbeat.call_count, 2)
+        self.assertEqual(len(polls), 2)
+        self.assertEqual(logs.getvalue().count("RuntimeError"), 2)
+        self.assertNotIn("secret", logs.getvalue())
+
+    def test_lane_inventory_tracks_each_health_worker_and_detects_dead_threads(self):
+        def poll():
+            self.agent.lane_started("health-1")
+            self.agent.lane_polled("health-1")
+
+        with patch.object(module.time, "monotonic", return_value=100):
+            thread = threading.Thread(target=poll)
+            thread.start()
+            thread.join(timeout=2)
+        self.assertFalse(thread.is_alive())
+        with patch.object(module.time, "monotonic", return_value=150):
+            self.agent.lane_started("health-2")
+            self.agent.lane_polled("health-2")
+            self.agent.lane_busy("health-2", True)
+        with patch.object(module.time, "monotonic", return_value=200):
+            lanes = {lane["name"]: lane for lane in self.agent.lane_inventory()}
+        self.assertEqual(set(lanes), {"health-1", "health-2", "read", "write"})
+        self.assertFalse(lanes["health-1"]["alive"])
+        self.assertEqual(lanes["health-1"]["pollAgeSeconds"], 100)
+        self.assertTrue(lanes["health-2"]["alive"])
+        self.assertTrue(lanes["health-2"]["busy"])
+        self.assertEqual(lanes["health-2"]["pollAgeSeconds"], 50)
+        self.assertIsNotNone(lanes["health-2"]["lastPollAt"])
+        self.assertEqual(len(self.agent.inventory()["agentLanes"]), 4)
+
+    def test_health_lane_times_out_on_instance_lock_and_processes_next_check(self):
+        blocked = self.operation("instance.inspect", op_id="blocked-health")
+        fresh = {**self.operation("instance.inspect", op_id="fresh-health"), "instanceId": "other-instance"}
+        pending = iter([blocked, fresh])
+        failures = []
+
+        def request(method, path, payload=None):
+            if method == "POST":
+                return {"operation": {"id": path.rsplit("/", 1)[1], "state": payload["state"]}}
+            operation = next(pending, None)
+            if operation is None:
+                self.agent.stopping.set()
+            return {"operations": [operation] if operation else []}
+
+        def read():
+            try:
+                self.agent.read_loop("health")
+            except BaseException as error:
+                failures.append(error)
+
+        with patch.object(module, "HEALTH_LOCK_TIMEOUT_SECONDS", 0.01), patch.object(self.agent, "request", side_effect=request), patch.object(self.agent, "dispatch", return_value={"healthy": True}) as dispatch:
+            with self.agent.instance_lock(blocked):
+                thread = threading.Thread(target=read)
+                thread.start()
+                thread.join(timeout=2)
+                finished_while_locked = not thread.is_alive()
+            self.agent.stopping.set()
+            thread.join(timeout=2)
+        self.assertTrue(finished_while_locked)
+        self.assertEqual(failures, [])
+        self.assertEqual(self.agent.journal.get(blocked["id"])["state"], "failed")
+        self.assertIn("waiting for the instance", self.agent.journal.get(blocked["id"])["error"])
+        self.assertEqual(self.agent.journal.get(fresh["id"])["state"], "completed")
+        dispatch.assert_called_once_with(fresh)
+        self.assertEqual(self.agent.read_claims, set())
+
     def test_inventory_reuses_manifests_until_mtime_changes(self):
         marker = self.library / "manifest.json"
         saved = json.loads(marker.read_text())

src/lib/hosts.ts

@@ -35,6 +35,21 @@ export interface HostAgentUpdate {
     checkedAt?: string;
 }
 
+export interface HostAgentLane {
+    name: 'health-1' | 'health-2' | 'read' | 'write';
+    alive: boolean; busy: boolean; lastPollAt: string | null; pollAgeSeconds: number;
+}
+export function hostAgentLaneLabel(name: HostAgentLane['name']): string {
+    return { 'health-1': 'Health check 1', 'health-2': 'Health check 2', read: 'File reads', write: 'Actions' }[name];
+}
+export function hostAgentLaneState(lane: HostAgentLane): 'stopped' | 'stalled' | 'busy' | 'starting' | 'polling' {
+    if (!lane.alive) return 'stopped';
+    // Reads and mutations can legitimately take minutes. Health work has a
+    // bounded budget; an old poll there indicates a blocked worker even if busy.
+    if (lane.pollAgeSeconds > 90 && (!lane.busy || lane.name.startsWith('health-'))) return 'stalled';
+    return lane.busy ? 'busy' : lane.lastPollAt === null ? 'starting' : 'polling';
+}
+
 export function hostAgentUpdatesEnabled(inventory: HostInventory): boolean {
     return inventory.agentUpdate?.enabled === true && inventory.capabilities?.includes('agent.update') === true;
 }
@@ -68,6 +83,7 @@ export interface HostInventory {
     currentOperation?: string | null; [key: string]: unknown;
     capabilities?: string[];
     agentUpdate?: HostAgentUpdate;
+    agentLanes?: HostAgentLane[];
     instanceResources?: InstanceResources[];
     instanceResourcesError?: string;
     instanceStates?: InstanceState[];

src/lib/server/hosts/index.ts

@@ -11,6 +11,7 @@ import { parseHostInventory } from './request-context.js';
 import { fleetSettings } from '../settings.js';
 import { recordInstanceResources, validateInstanceResources } from './calibration.js';
 import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js';
+import { hostAgentLaneLabel, hostAgentLaneState } from '../../hosts.js';
 import type { Host, HostDecommission, HostDeletion, HostInventory, HostMode, HostOperation, HostOperationFilter, HostOperationHistoryPage, HostOperationScope, HostOperationState } from '../../hosts.js';
 
 export const AGENT_PROTOCOL_VERSION = 1;
@@ -46,6 +47,10 @@ function hostFromRow(row: HostRow): Host {
     if (!row.enrolled_at) reasons.push('Agent has not enrolled.');
     else if (!row.credential_active) reasons.push('Agent credential has been revoked.');
     if (!connected) reasons.push('Agent heartbeat is unavailable or stale.');
+    if (connected) for (const lane of inventory.agentLanes ?? []) {
+        const state = hostAgentLaneState(lane);
+        if (state === 'stopped' || state === 'stalled') reasons.push(`Agent ${hostAgentLaneLabel(lane.name).toLowerCase()} worker is ${state}. Check the host agent service.`);
+    }
     if (inventory.protocolVersion !== AGENT_PROTOCOL_VERSION) reasons.push('A compatible agent protocol is required.');
     if (!inventory.dockerRunning) reasons.push('Docker is not ready.');
     if (!inventory.directoriesWritable) reasons.push('Managed directories are not writable.');
@@ -287,6 +292,20 @@ export function validateInventory(value: unknown): HostInventory {
             ...(update.error === undefined ? {} : { error: update.error }),
             ...(update.checkedAt === undefined ? {} : { checkedAt: update.checkedAt }) };
     }
+    if (inventory.agentLanes !== undefined) {
+        if (!Array.isArray(inventory.agentLanes) || inventory.agentLanes.length > 4) throw new Error('Invalid agent lane inventory.');
+        // Keep valid lane diagnostics without discarding a host heartbeat over
+        // one malformed sample. Do not persist thread details or raw errors.
+        const names = new Set<string>();
+        inventory.agentLanes = inventory.agentLanes.filter(lane => {
+            if (!lane || !['health-1','health-2','read','write'].includes(lane.name) || names.has(lane.name)
+                || typeof lane.alive !== 'boolean' || typeof lane.busy !== 'boolean'
+                || typeof lane.pollAgeSeconds !== 'number' || !Number.isFinite(lane.pollAgeSeconds) || lane.pollAgeSeconds < 0
+                || (lane.lastPollAt !== null && (typeof lane.lastPollAt !== 'string' || lane.lastPollAt.length > 40 || !Number.isFinite(Date.parse(lane.lastPollAt))))) return false;
+            names.add(lane.name);
+            return true;
+        }).map(({name,alive,busy,lastPollAt,pollAgeSeconds}) => ({name,alive,busy,lastPollAt,pollAgeSeconds}));
+    }
     if (inventory.libraries !== undefined && (!Array.isArray(inventory.libraries) || inventory.libraries.length > 2000 || inventory.libraries.some(library =>
         !library || typeof library !== 'object' || typeof library.id !== 'string' || !library.id || library.id.length > 200 || !['missing','downloading','verifying','ready','failed','outdated'].includes(library.state)))) throw new Error('Invalid game library inventory.');
     return inventory;
@@ -522,11 +541,14 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = '
             .all(hostId, Number(host.readiness === 'ready' || host.mode === 'draining'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false),
                 ...(lane === 'health' ? [new Date(Date.now() - 30_000).toISOString()] : [])) as OperationDeliveryRow[];
         const preparationLimit = fleetSettings().preparationConcurrency;
-        const preparing = new Set((database().prepare(`SELECT DISTINCT host_id FROM ec_host_operations
+        const libraryKinds = ['library.download','library.import'];
+        const preparing = preparationLimit && rows.some(operation => libraryKinds.includes(operation.kind))
+            ? new Set((database().prepare(`SELECT DISTINCT host_id FROM ec_host_operations
                 WHERE kind IN ('library.download','library.import') AND last_delivered_at IS NOT NULL
-            AND state IN ('pending','accepted','running','unknown')`).all() as { host_id: string }[]).map(row => row.host_id));
+                AND state IN ('pending','accepted','running','unknown')`).all() as { host_id: string }[]).map(row => row.host_id))
+            : new Set<string>();
         const admitted = rows.filter(operation => {
-            if (!['library.download','library.import'].includes(operation.kind) || preparing.has(hostId) || !preparationLimit) return true;
+            if (!libraryKinds.includes(operation.kind) || preparing.has(hostId) || !preparationLimit) return true;
             if (preparing.size >= preparationLimit) return false;
             preparing.add(hostId); return true;
         });
@@ -584,7 +606,25 @@ export async function waitHostOperationStatus(id: string, timeoutMs = 30_000): P
         const operation = getHostOperationStatus(id);
         if (!operation) throw new Error('Host operation was not found.');
         if (['completed','failed','unknown'].includes(operation.state)) return operation;
-        if (Date.now() >= deadline) throw new Error(`Host operation ${id} is awaiting acknowledgement; reconcile this operation ID before retrying.`);
+        if (Date.now() >= deadline) {
+            // Delivery and expiry both claim the pending row transactionally.
+            // Never cancel a delivered check or a command with side effects.
+            const expired = database().transaction(() => {
+                const timestamp = now();
+                const error = 'Expired before delivery.';
+                const result = database().prepare(`UPDATE ec_host_operations SET state='failed',error=?,updated_at=?
+                    WHERE id=? AND state='pending' AND last_delivered_at IS NULL
+                    AND kind IN ('instance.inspect','rcon.status-batch')`).run(error, timestamp, id);
+                if (result.changes) database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,?,?,?)')
+                    .run(id, 'failed', timestamp, error);
+                return result.changes > 0;
+            })();
+            if (expired) {
+                queueMicrotask(() => workEvents.emit(`operation:${id}`));
+                throw new Error(`Host operation ${id} expired before delivery.`);
+            }
+            throw new Error(`Host operation ${id} is awaiting acknowledgement; reconcile this operation ID before retrying.`);
+        }
         await waitForHostWork(`operation:${id}`, new AbortController().signal, Math.min(1000, Math.max(1, deadline - Date.now())));
     }
 }

src/lib/server/hosts/schema.ts

@@ -43,6 +43,7 @@ export function ensureHostSchema(handle: Database.Database): void {
         -- history. The queue index cannot narrow those lookups beyond host_id,
         -- causing a history scan for every queued command on Node's main thread.
         CREATE INDEX IF NOT EXISTS ec_host_operations_kind ON ec_host_operations(host_id, kind, created_at);
+        CREATE INDEX IF NOT EXISTS ec_host_operations_kind_state ON ec_host_operations(kind, state);
         CREATE TABLE IF NOT EXISTS ec_host_operation_payloads (
             operation_id TEXT PRIMARY KEY REFERENCES ec_host_operations(id) ON DELETE CASCADE,
             payload TEXT NOT NULL

src/routes/hosts/[id]/+page.svelte

@@ -9,7 +9,7 @@
     import { page } from '$app/stores';
     import { goto } from '$app/navigation';
     import type { Host, CapacityCalibration } from '$lib/hosts.js';
-    import { hostAgentUpdateLabel, hostAgentUpdateMessage, hostAgentUpdatesEnabled, hostDeletionMessage, hostReadinessLabel } from '$lib/hosts.js';
+    import { hostAgentLaneLabel, hostAgentLaneState, hostAgentUpdateLabel, hostAgentUpdateMessage, hostAgentUpdatesEnabled, hostDeletionMessage, hostReadinessLabel } from '$lib/hosts.js';
     import SectionHeader from '$ui/components/primitives/SectionHeader.svelte';
     import Button from '$ui/components/primitives/Button.svelte';
     import DeleteHostDialog from '$ui/components/DeleteHostDialog.svelte';
@@ -85,6 +85,11 @@
         <div class="host-panel" id="host-panel" role="tabpanel" aria-labelledby={`host-tabs-${tab}`}>
         {#if tab === 'overview'}
             <div class="overview-actions"><Button onclick={() => changeTab('servers')}>View {instances.length} servers</Button></div>
+        <section class="panel"><h2>Agent workers</h2>
+            {#if host.inventory.agentLanes?.length}<dl>{#each host.inventory.agentLanes as lane (lane.name)}
+                <dt>{hostAgentLaneLabel(lane.name)}</dt><dd>{hostAgentLaneState(lane)}{#if !host.connected} (last reported){/if}<br />Last successful poll: {lane.lastPollAt ? new Date(lane.lastPollAt).toLocaleString() : 'Not yet reported'}</dd>
+            {/each}</dl>{:else}<p>This agent has not reported worker status yet.</p>{/if}
+        </section>
         <section class="panel"><h2>Inventory</h2><dl>
             <dt>Hostname</dt><dd>{host.hostname || 'Pending'}</dd><dt>Agent</dt><dd>{host.inventory.agentVersion ?? 'Unknown'}</dd><dt>Platform</dt><dd>{host.inventory.os ?? 'Unknown'} · {host.inventory.architecture ?? 'Unknown'}</dd><dt>CPU</dt><dd>{host.inventory.cpu ?? 'Unknown'} ({host.inventory.cpuCount ?? '?'} cores)</dd><dt>CPU usage</dt><dd title="Average across all cores since the previous inventory sample">{host.inventory.cpuUsagePercent === undefined ? 'Not reported' : `${host.inventory.cpuUsagePercent.toFixed(1)}%`}{#if !host.connected && host.inventory.cpuUsagePercent !== undefined} (last reported){/if}</dd><dt>Memory available / total</dt><dd>{bytes(host.inventory.memoryAvailable)} / {bytes(host.inventory.memoryTotal)}</dd><dt>Game storage available / total</dt><dd>{bytes(host.inventory.storageAvailable)} / {bytes(host.inventory.storageTotal)}</dd><dt>Docker</dt><dd>{host.inventory.dockerVersion ?? 'Unknown'} · {host.inventory.dockerRunning ? 'Running' : 'Unavailable'}</dd><dt>Current operation</dt><dd>{host.inventory.currentOperation ?? 'None reported'}</dd>
         </dl></section>        {#if calibration}<section class="panel"><h2>Learned server capacity</h2>

static/agent/ec-agent.py

@@ -12,6 +12,7 @@ import contextlib
 from datetime import datetime, timezone
 import math
 import hashlib
+import http.client
 import fcntl
 import json
 import os
@@ -32,7 +33,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.5.1"
+VERSION = "1.5.3"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -40,6 +41,7 @@ MANIFEST_MAX_BYTES = 16 * 1024 * 1024
 MANIFEST_MAX_SECONDS = 15
 MANIFEST_MAX_DEPTH = 64
 LIBRARY_DOWNLOAD_WORKERS = 4
+HEALTH_LOCK_TIMEOUT_SECONDS = 5
 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
 AGENT_CONFIG = Path("/etc/electric-cherry/agent.json")
 AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py")
@@ -922,6 +924,28 @@ class Agent:
         self.inventory_manifest_lock = threading.Lock()
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
         self.executing_id = None
+        self.lane_lock = threading.Lock()
+        self.lanes = {name: {"thread": None, "lastPollAt": None, "polled": time.monotonic(), "busy": False}
+                      for name in ("health-1", "health-2", "read", "write")}
+
+    def lane_started(self, name):
+        with self.lane_lock:
+            self.lanes[name]["thread"] = threading.current_thread()
+
+    def lane_polled(self, name):
+        with self.lane_lock:
+            self.lanes[name].update(lastPollAt=datetime.now(timezone.utc).isoformat(timespec="milliseconds"), polled=time.monotonic())
+
+    def lane_busy(self, name, busy):
+        with self.lane_lock:
+            self.lanes[name]["busy"] = busy
+
+    def lane_inventory(self):
+        with self.lane_lock:
+            return [{"name": name, "alive": lane["thread"] is None or lane["thread"].is_alive(),
+                     "busy": lane["busy"], "lastPollAt": lane["lastPollAt"],
+                     "pollAgeSeconds": max(0, time.monotonic() - lane["polled"])}
+                    for name, lane in self.lanes.items()]
 
     def can_uninstall(self):
         try:
@@ -1099,7 +1123,7 @@ class Agent:
                     raise AgentError("Controller redirects are not allowed")
             with urllib.request.build_opener(NoRedirect()).open(req, timeout=35) as response:
                 return json.load(response)
-        except (OSError, ValueError):
+        except (OSError, ValueError, http.client.HTTPException):
             raise AgentError("Controller request failed") from None
 
     def cpu_usage_percent(self):
@@ -1333,7 +1357,7 @@ class Agent:
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
         capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
-        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
+        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
     def safe_to_disrupt(self, instance_id, generation):
         health = self.inspect(instance_id, generation)
@@ -2354,6 +2378,8 @@ class Agent:
                 self.heartbeat()
             except (AgentError, OSError, ValueError):
                 pass
+            except Exception as err:
+                print(f"Heartbeat recovered from {type(err).__name__}", file=sys.stderr)
 
     @staticmethod
     def read_operation(operation):
@@ -2363,13 +2389,29 @@ class Agent:
         key = operation.get("instanceId")
         return self.instance_locks.setdefault(key, threading.RLock()) if key else contextlib.nullcontext()
 
+    @contextlib.contextmanager
+    def observation_lock(self, operation, health=False):
+        lock = self.instance_lock(operation)
+        if not health or not operation.get("instanceId"):
+            with lock:
+                yield
+            return
+        if not lock.acquire(timeout=HEALTH_LOCK_TIMEOUT_SECONDS):
+            raise AgentError("Health check timed out waiting for the instance operation")
+        try:
+            yield
+        finally:
+            lock.release()
+
     @staticmethod
     def health_operation(operation):
         return operation.get("kind") in ("instance.inspect", "rcon.status-batch") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
 
-    def read_loop(self, lane="read"):
+    def read_loop(self, lane="read", lane_name=None):
         # Each lane owns its SQLite connection and never writes executing_id,
         # checkpoints, or current_operation used by the mutation state machine.
+        lane_name = lane_name or ("health-1" if lane == "health" else "read")
+        self.lane_started(lane_name)
         journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
         next_prune = time.monotonic()
         try:
@@ -2385,7 +2427,9 @@ class Agent:
                                 if journal.prune_observations(self.observation_keep_days) == 500:
                                     next_prune = time.monotonic() + 15
                     response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20")
+                    self.lane_polled(lane_name)
                     handled = False
+                    self.lane_busy(lane_name, bool(response.get("operations")))
                     for operation in response.get("operations", []):
                         if self.stopping.is_set() or not self.read_operation(operation):
                             continue
@@ -2402,8 +2446,8 @@ class Agent:
                                 row, _ = journal.accept(operation)
                                 if row["state"] in ("accepted", "unknown", "running"):
                                     self.report(journal.update(operation["id"], "running"), journal=journal)
-                                    with self.read_gate.read(), self.instance_lock(operation):
                                     try:
+                                        with self.read_gate.read(), self.observation_lock(operation, health=lane == "health"):
                                             result = self.dispatch(operation)
                                             row = journal.update(operation["id"], "completed", result)
                                     except Exception as err:
@@ -2421,6 +2465,11 @@ class Agent:
                         self.stopping.wait(0.25)
                 except (AgentError, OSError, ValueError, KeyError, sqlite3.Error):
                     self.stopping.wait(3)
+                except Exception as err:
+                    print(f"{lane} lane recovered from {type(err).__name__}", file=sys.stderr)
+                    self.stopping.wait(3)
+                finally:
+                    self.lane_busy(lane_name, False)
         finally:
             journal.close()
 
@@ -2434,6 +2483,8 @@ class Agent:
                 return
         # Never hold the updater lock while waiting for delivery.
         response = self.request("GET", "/api/agent/operations?lane=write&wait=20")
+        self.lane_polled("write")
+        self.lane_busy("write", bool(response.get("operations")))
         with self.activity() as acquired:
             if not acquired or self.stopping.is_set() or self.update_pending():
                 return
@@ -2461,6 +2512,7 @@ class Agent:
             return completed
 
     def run(self):
+        self.lane_started("write")
         # Obtain the configured host port range before accepting any deployment.
         # Enrollment's default values must never race the first heartbeat.
         while not self.stopping.is_set():
@@ -2475,10 +2527,14 @@ class Agent:
                 break
             except (AgentError, OSError, ValueError):
                 self.stopping.wait(3)
+            except Exception as err:
+                print(f"Initial heartbeat recovered from {type(err).__name__}", file=sys.stderr)
+                self.stopping.wait(3)
         heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True)
         heartbeats.start()
         # Preserve two concurrent health checks even while files are being read.
-        readers = [threading.Thread(target=self.read_loop, args=(lane,), daemon=True) for lane in ("health", "health", "read")]
+        readers = [threading.Thread(target=self.read_loop, args=(lane, name), daemon=True)
+                   for name, lane in (("health-1", "health"), ("health-2", "health"), ("read", "read"))]
         for reader in readers:
             reader.start()
         while not self.stopping.is_set():
@@ -2488,6 +2544,10 @@ class Agent:
             except (AgentError, OSError, ValueError, KeyError):
                 # Logs deliberately omit URLs, tokens, payloads and raw errors.
                 print("Host controller contact unavailable; existing containers continue running", file=sys.stderr)
+            except Exception as err:
+                print(f"Write lane recovered from {type(err).__name__}", file=sys.stderr)
+            finally:
+                self.lane_busy("write", False)
             self.stopping.wait(0.25 if completed else 3)
         for reader in readers:
             reader.join(timeout=2)

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.5.1",
+  "version": "1.5.3",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "67003980127327d07144041d0c17b8adaa0d466b30b0ebf7508a80efd5cf6481",
+  "sha256": "ebd6f67a44fdb4bd0683597d4abd54999311473561204de368864b04a5436256",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }

tests/unit/host-agent-lanes.test.ts

@@ -0,0 +1,45 @@
+import { afterEach, beforeEach, expect, it, vi } from 'vitest';
+import { hostAgentLaneState, type HostAgentLane } from '../../src/lib/hosts.js';
+import { newFixture, type Fixture } from '../helpers/fixture.js';
+
+let fixture: Fixture;
+const inventory = { protocolVersion:1, dockerRunning:true, directoriesWritable:true, storageAvailable:100 * 1024 ** 3,
+    runtimeImages:['image'], libraries:[{id:'library',state:'ready'}] };
+const lane: HostAgentLane = {name:'health-1',alive:true,busy:false,lastPollAt:'2026-10-06T12:00:00Z',pollAgeSeconds:5};
+beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); });
+afterEach(async () => { await fixture.cleanup(); });
+
+it('flags a stopped or stalled health worker despite a live heartbeat and recovers on fresh polls', async () => {
+    const api = await import('../../src/lib/server/hosts/index.js');
+    const host = api.createHost({name:'Worker host'});
+    api.enrollHost(api.createEnrollment(host.id).token,inventory);
+    for (const broken of [{...lane,alive:false},{...lane,busy:true,pollAgeSeconds:91}]) {
+        const updated = api.heartbeat(host.id,{...inventory,agentLanes:[broken,{...lane,name:'health-2'}]});
+        expect(updated).toMatchObject({connected:true,connectionState:'online',readiness:'degraded'});
+        expect(updated.readinessReasons[0]).toContain('health check 1 worker');
+        // Diagnostics must not prevent health polls from recovering the worker.
+        const check = api.queueHostOperation(host.id,'instance.inspect',{});
+        expect(api.pendingHostOperations(host.id,'health').map(op=>op.id)).toContain(check.id);
+    }
+    expect(api.heartbeat(host.id,{...inventory,agentLanes:[lane]}).readiness).toBe('ready');
+    // Older agents remain compatible without diagnostics.
+    expect(api.heartbeat(host.id,inventory).readiness).toBe('ready');
+});
+
+it('allows long reads and mutations while detecting stalled idle workers and missing initial polls', () => {
+    expect(hostAgentLaneState({...lane,lastPollAt:null})).toBe('starting');
+    expect(hostAgentLaneState({...lane,lastPollAt:null,pollAgeSeconds:91})).toBe('stalled');
+    expect(hostAgentLaneState({...lane,busy:true,pollAgeSeconds:91})).toBe('stalled');
+    for (const name of ['read','write'] as const) {
+        expect(hostAgentLaneState({...lane,name,busy:true,pollAgeSeconds:3600})).toBe('busy');
+        expect(hostAgentLaneState({...lane,name,busy:false,pollAgeSeconds:91})).toBe('stalled');
+        expect(hostAgentLaneState({...lane,name,alive:false,busy:true})).toBe('stopped');
+    }
+});
+
+it('sanitizes lane diagnostics without rejecting a heartbeat for one malformed sample', async () => {
+    const { validateInventory } = await import('../../src/lib/server/hosts/index.js');
+    expect(validateInventory({...inventory,agentLanes:[null,{...lane,secret:'omit'},{...lane},{...lane,name:'write',pollAgeSeconds:-1}]}).agentLanes)
+        .toEqual([lane]);
+    expect(validateInventory({...inventory,agentLanes:[{...lane,pollAgeSeconds:'stale'},{...lane,lastPollAt:'invalid'}]}).agentLanes).toEqual([]);
+});

tests/unit/host-request-cost.test.ts

@@ -23,6 +23,47 @@ async function enrolled() {
     return { ...hosts, ...context, hostId: host.id, credential };
 }
 
+describe('agent preparation admission', () => {
+    it('skips fleet preparation lookups for empty, health, read and unlimited library polls', async () => {
+        const api = await enrolled();
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js');
+        saveFleetSettings({...fleetSettings(),preparationConcurrency:1});
+        const prepare = vi.spyOn(db(),'prepare');
+        expect(api.pendingHostOperations(api.hostId)).toEqual([]);
+        api.queueHostOperation(api.hostId,'instance.inspect',{});
+        api.pendingHostOperations(api.hostId,'health');
+        api.queueHostOperation(api.hostId,'files.read',{path:'server.cfg'});
+        api.pendingHostOperations(api.hostId,'read');
+        saveFleetSettings({...fleetSettings(),preparationConcurrency:0});
+        const download = api.queueHostOperation(api.hostId,'library.download',{});
+        expect(api.pendingHostOperations(api.hostId,'write').map(op=>op.id)).toContain(download.id);
+        expect(prepare.mock.calls.some(([sql])=>sql.includes('SELECT DISTINCT host_id'))).toBe(false);
+    });
+
+    it('uses an indexed lookup and counts delivered preparation by host until all its jobs finish', async () => {
+        const api = await enrolled();
+        const other = api.createHost({name:'Other host'});
+        api.enrollHost(api.createEnrollment(other.id).token,inventory);
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js');
+        saveFleetSettings({...fleetSettings(),preparationConcurrency:1});
+        const download = api.queueHostOperation(api.hostId,'library.download',{});
+        const imported = api.queueHostOperation(api.hostId,'library.import',{});
+        const waiting = api.queueHostOperation(other.id,'library.download',{});
+        const prepare = vi.spyOn(db(),'prepare');
+        expect(api.pendingHostOperations(api.hostId,'write').map(op=>op.id)).toEqual([download.id,imported.id]);
+        const sql = prepare.mock.calls.find(([sql])=>sql.includes('SELECT DISTINCT host_id'))![0];
+        const plan = db().prepare(`EXPLAIN QUERY PLAN ${sql}`).all() as {detail:string}[];
+        expect(plan.map(row=>row.detail).join('\n')).toContain('USING INDEX ec_host_operations_kind_state (kind=? AND state=?)');
+        expect(api.pendingHostOperations(other.id,'write')).toEqual([]);
+        api.reportHostOperation(api.hostId,download.id,{state:'completed'});
+        expect(api.pendingHostOperations(other.id,'write')).toEqual([]);
+        api.reportHostOperation(api.hostId,imported.id,{state:'failed'});
+        expect(api.pendingHostOperations(other.id,'write').map(op=>op.id)).toEqual([waiting.id]);
+    });
+});
+
 describe('agent credential usage writes', () => {
     it('writes on first use and at most once per minute, including after reopening the database', async () => {
         const api = await enrolled();

tests/unit/native-hosts.test.ts

@@ -15,6 +15,38 @@ async function enrolled() {
 }
 
 describe('native host enrollment and scoped authentication', () => {
+    it.each(['instance.inspect','rcon.status-batch'])('expires undelivered %s on timeout and delivers the next fresh check', async kind => {
+        const api = await enrolled();
+        const { db } = await import('../../src/lib/server/db/index.js');
+        api.heartbeat(api.host.id,{...inventory,capabilities:['operations.health-lane']});
+        const stale = api.queueHostOperation(api.host.id,kind,{});
+        await expect(api.waitHostOperationStatus(stale.id,0)).rejects.toThrow('expired before delivery');
+        expect(api.getHostOperationStatus(stale.id)).toMatchObject({state:'failed',error:'Expired before delivery.'});
+        expect(await api.waitHostOperationStatus(stale.id,0)).toMatchObject({state:'failed'});
+        expect(db().prepare("SELECT state,error FROM ec_host_operation_events WHERE operation_id=? AND state='failed'").all(stale.id))
+            .toEqual([{state:'failed',error:'Expired before delivery.'}]);
+        const fresh = api.queueHostOperation(api.host.id,kind,{});
+        expect(api.pendingHostOperations(api.host.id,'health').map(op=>op.id)).toEqual([fresh.id]);
+    });
+
+    it.each(['pending','accepted','running'])('preserves delivered health checks in %s state after the caller times out', async state => {
+        const api = await enrolled();
+        const operation = api.queueHostOperation(api.host.id,'instance.inspect',{});
+        api.pendingHostOperations(api.host.id,'health');
+        if (state !== 'pending') api.reportHostOperationStatus(api.host.id,operation.id,{state});
+        await expect(api.waitHostOperationStatus(operation.id,0)).rejects.toThrow('awaiting acknowledgement');
+        expect(api.getHostOperationStatus(operation.id)?.state).toBe(state);
+        api.reportHostOperationStatus(api.host.id,operation.id,{state:'completed',result:{healthy:true}});
+        expect(await api.waitHostOperationStatus(operation.id)).toMatchObject({state:'completed',result:{healthy:true}});
+    });
+
+    it.each(['instance.restart','files.stage','rcon.execute'])('preserves undelivered %s when its caller times out', async kind => {
+        const api = await enrolled();
+        const operation = api.queueHostOperation(api.host.id,kind,{command:'status'});
+        await expect(api.waitHostOperationStatus(operation.id,0)).rejects.toThrow('awaiting acknowledgement');
+        expect(api.getHostOperationStatus(operation.id)?.state).toBe('pending');
+    });
+
     it('keeps the host online and delivers work when individual state observations are invalid', async () => {
         const api = await enrolled();
         const good = { instanceId: 'healthy-server', generation: 1, state: 'running', healthy: true, observedAt: new Date().toISOString() };