Reuse heartbeat container state and batch host RCON status

agent/ec_agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.9"
+VERSION = "1.4.10"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -519,7 +519,7 @@ class ReadWriteGate:
 def read_operation(operation):
     if not isinstance(operation, dict):
         return False
-    if operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
+    if operation.get("kind") in ("host.inspect", "instance.inspect", "rcon.status-batch", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
         return True
     payload = operation.get("payload")
     return operation.get("kind") == "rcon.execute" and isinstance(payload, dict) and payload.get("command") in ("status", "getstatus", "version")
@@ -913,6 +913,7 @@ class Agent:
         self.read_gate = ReadWriteGate()
         self.read_claim_lock = threading.Lock()
         self.read_claims = set()
+        self.health_probe_gate = threading.BoundedSemaphore(2)
         self.cancelled_creations = frozenset()
         self.observation_keep_days = 0  # Older controllers do not authorize cleanup.
         self.library_progress = {}
@@ -1127,18 +1128,19 @@ class Agent:
                 return None
             return round(100 * (total - deltas[3] - deltas[4]) / total, 1)
 
-    def instance_resources(self):
+    def instance_resources(self, observations=None):
         """One bounded Docker stats batch; unavailable samples are never zero load."""
-        ids = self.docker.call("container", "ls", "--filter", "label=" + LABEL + ".managed=true",
+        ids = self.docker.call("container", "ls", "--all", "--filter", "label=" + LABEL + ".managed=true",
                                "--no-trunc", "--format", "{{.ID}}", timeout=10).splitlines()[:1000]
         if not ids:
             return []
+        observed_at = datetime.now(timezone.utc).isoformat(timespec="milliseconds")
         containers = json.loads(self.docker.call("container", "inspect", *ids, timeout=10))
         owned = {}
         for info in containers:
             try:
                 labels = info.get("Config", {}).get("Labels", {}) or {}
-                if labels.get(LABEL + ".managed") != "true" or not info.get("State", {}).get("Running"):
+                if labels.get(LABEL + ".managed") != "true":
                     continue
                 instance_id, generation = instance_identity(labels.get(LABEL + ".instance"), int(labels.get(LABEL + ".generation", "")))
                 name = self.container_name(instance_id, generation)
@@ -1147,6 +1149,16 @@ class Agent:
                 expected = str(self.instance_dir(instance_id, generation) / "app")
                 if not any(mount.get("Type") == "bind" and mount.get("Source") == expected and mount.get("Destination") == "/home/plutainer/app" for mount in info.get("Mounts", [])):
                     continue
+                state = info.get("State", {})
+                healthy = state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
+                status = state.get("Status", "unknown")
+                if status == "running" and not healthy:
+                    status = state.get("Health", {}).get("Status", "unknown")
+                if observations is not None:
+                    observations.append({"instanceId": instance_id, "generation": generation,
+                                         "state": status, "healthy": healthy, "observedAt": observed_at})
+                if not state.get("Running"):
+                    continue
                 started = datetime.fromisoformat(info["State"]["StartedAt"].replace("Z", "+00:00"))
                 uptime = max(0, int((datetime.now(timezone.utc) - started).total_seconds()))
                 owned[name] = {"instanceId": instance_id, "generation": generation, "uptimeSeconds": uptime}
@@ -1280,17 +1292,21 @@ class Agent:
                         images.extend(alias for alias, image_id in json.loads(aliases.read_text()).items() if image_id in image_ids)
         resources = {}
         if docker_running:
+            observations = []
             try:
-                resources["instanceResources"] = self.instance_resources()
+                resources["instanceResources"] = self.instance_resources(observations)
             except (AgentError, ValueError, TypeError, KeyError):
                 resources["instanceResourcesError"] = "Container resource sampling failed. Check Docker access and stats availability on this host."
+            # A failed stats call does not discard a successful inspection.
+            # Missing entries remain unknown; they are never inferred stopped.
+            resources["instanceStates"] = observations
         os_name = platform.system()
         with contextlib.suppress(OSError):
             for line in Path("/etc/os-release").read_text().splitlines():
                 if line.startswith("PRETTY_NAME="):
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
-        capabilities = ["library.remove", "instance.cancel-create", "operations.read-lane", "operations.health-lane", "files.manifest"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
+        capabilities = ["library.remove", "instance.cancel-create", "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})}
 
     def safe_to_disrupt(self, instance_id, generation):
@@ -1314,7 +1330,46 @@ class Agent:
         if result["players"] != 0:
             raise Held("Players are present; disruption held")
 
-    def rcon(self, instance_id, generation, command):
+    def rcon_status_batch(self, operation):
+        targets = operation.get("payload", {}).get("instances")
+        if not isinstance(targets, list) or not 1 <= len(targets) <= 1000 or any(not isinstance(target, dict) for target in targets):
+            raise AgentError("Invalid status batch")
+        identities = [instance_identity(target.get("instanceId"), target.get("generation")) for target in targets]
+        if len(set(identities)) != len(identities):
+            raise AgentError("Duplicate status batch target")
+        deadline = time.monotonic() + 20
+
+        def probe(identity):
+            instance_id, generation = identity
+            target = {"instanceId": instance_id, "generation": generation}
+            reply = {"ok": False, "reason": "timeout", "text": ""}
+            # A batch does not hold every instance lock while one server waits.
+            # Bound concurrency across overlapping batches as well as within one.
+            lock = self.instance_lock(target)
+            # Use the same lock order as individual health-lane operations.
+            if lock.acquire(timeout=max(0, deadline - time.monotonic())):
+                try:
+                    if self.health_probe_gate.acquire(timeout=max(0, deadline - time.monotonic())):
+                        try:
+                            marker = self.instance_dir(instance_id, generation).parent / "current.json"
+                            current = json.loads(marker.read_text()) if marker.exists() else {}
+                            if current.get("generation", generation) != generation or current.get("removed"):
+                                reply = {"ok": False, "reason": "superseded", "text": ""}
+                            elif time.monotonic() < deadline:
+                                reply = self.rcon(instance_id, generation, "status", deadline=deadline)
+                        except (AgentError, OSError, ValueError, KeyError):
+                            reply = {"ok": False, "reason": "unreachable", "text": ""}
+                        finally:
+                            self.health_probe_gate.release()
+                finally:
+                    lock.release()
+            return {**target, "reply": reply}
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            results = list(executor.map(probe, identities))
+        return {"ok": all(row["reply"].get("ok") for row in results), "results": results}
+
+    def rcon(self, instance_id, generation, command, deadline=None):
         if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):
             raise AgentError("Invalid RCON command")
         saved = self.spec(instance_id, generation)
@@ -1327,21 +1382,25 @@ class Agent:
         # Retry only the read-only occupancy query. A missing UDP reply must
         # never cause commands with effects to be replayed.
         for attempt in range(3 if command == "status" else 1):
-            result = self.rcon_attempt(family, host, int(saved["port"]), password, command)
+            if deadline is not None and time.monotonic() >= deadline:
+                return {"ok": False, "reason": "timeout", "players": None, "text": ""}
+            result = self.rcon_attempt(family, host, int(saved["port"]), password, command, **({"deadline": deadline} if deadline is not None else {}))
             if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == 2 or command != "status":
                 return result
-            time.sleep(0.25 * 2 ** attempt)
+            time.sleep(min(0.25 * 2 ** attempt, max(0, deadline - time.monotonic())) if deadline is not None else 0.25 * 2 ** attempt)
 
-    def rcon_attempt(self, family, host, port, password, command):
+    def rcon_attempt(self, family, host, port, password, command, deadline=None):
         chunks = []
         with socket.socket(family, socket.SOCK_DGRAM) as sock:
-            sock.settimeout(3)
+            sock.settimeout(max(0.001, min(3, deadline - time.monotonic())) if deadline is not None else 3)
             try:
                 sock.connect((host, port))
                 sock.send(b"\xff\xff\xff\xffrcon " + password.encode() + b" " + command.encode() + b"\n")
                 while len(chunks) < 64 and sum(map(len, chunks)) < 1024 * 1024:
+                    if deadline is not None and time.monotonic() >= deadline:
+                        break
                     chunks.append(sock.recv(65535))
-                    sock.settimeout(0.15)
+                    sock.settimeout(max(0.001, min(0.15, deadline - time.monotonic())) if deadline is not None else 0.15)
             except socket.timeout:
                 pass
             except OSError:
@@ -2028,6 +2087,8 @@ class Agent:
             return {"image": image, "ready": True}
         if kind == "host.inspect":
             return self.inventory()
+        if kind == "rcon.status-batch":
+            return self.rcon_status_batch(operation)
         if kind == "library.download":
             return self.library_download(payload)
         if kind == "library.remove":
@@ -2049,6 +2110,9 @@ class Agent:
         if kind == "rcon.execute":
             if payload["command"] not in ("status", "getstatus", "version"):
                 self.checkpoint({"phase": "rcon-command"})
+            else:
+                with self.health_probe_gate:
+                    return self.rcon(instance_id, generation, payload["command"])
             return self.rcon(instance_id, generation, payload["command"])
         raise AgentError("Unsupported operation kind")
 
@@ -2266,7 +2330,7 @@ class Agent:
 
     @staticmethod
     def health_operation(operation):
-        return operation.get("kind") == "instance.inspect" or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
+        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"):
         # Each lane owns its SQLite connection and never writes executing_id,

agent/tests/test_agent.py

@@ -472,7 +472,7 @@ class AgentTests(unittest.TestCase):
     def resource_container(self):
         return {"Name": self.agent.container_name(INSTANCE, 1),
                 "Config": {"Labels": {module.LABEL + ".managed": "true", module.LABEL + ".instance": INSTANCE, module.LABEL + ".generation": "1"}},
-                "State": {"Running": True, "StartedAt": "2000-01-01T00:00:00.000000000Z"},
+                "State": {"Running": True, "Status": "running", "StartedAt": "2000-01-01T00:00:00.000000000Z"},
                 "Mounts": [{"Type": "bind", "Source": str(self.agent.instance_dir(INSTANCE, 1) / "app"), "Destination": "/home/plutainer/app"}]}
 
     def test_instance_resources_batches_owned_containers_and_normalizes_core_cpu_and_memory(self):
@@ -486,7 +486,7 @@ class AgentTests(unittest.TestCase):
         self.assertEqual(samples[0]["cpuCores"], 1.5)
         self.assertEqual(samples[0]["memoryBytes"], int(384.5 * 1024 ** 2))
         self.assertGreater(samples[0]["uptimeSeconds"], 300)
-        self.assertEqual(call.call_args_list[0].args, ("container", "ls", "--filter", "label=io.electric-cherry.managed=true", "--no-trunc", "--format", "{{.ID}}"))
+        self.assertEqual(call.call_args_list[0].args, ("container", "ls", "--all", "--filter", "label=io.electric-cherry.managed=true", "--no-trunc", "--format", "{{.ID}}"))
         self.assertEqual(call.call_args_list[-1].args, ("stats", "--no-stream", "--no-trunc", "--format", "{{json .}}", name))
 
     def test_instance_resources_ignores_stopped_foreign_and_mismatched_containers(self):
@@ -520,6 +520,86 @@ class AgentTests(unittest.TestCase):
             self.assertEqual(inventory["instanceResources"], [])
             self.assertNotIn("instanceResourcesError", inventory)
 
+    def test_heartbeat_states_share_inspection_and_include_stopped_and_unhealthy_containers(self):
+        for status, running, health in (("exited", False, "healthy"), ("running", True, "unhealthy"), ("running", True, "healthy")):
+            info = self.resource_container()
+            info["State"].update({"Status": status, "Running": running, "Health": {"Status": health}})
+            observations = []
+            outputs = ["container-id\n", json.dumps([info])] + ([""] if running else [])
+            with patch.object(self.docker, "call", side_effect=outputs) as docker:
+                self.agent.instance_resources(observations)
+            self.assertEqual(observations[0]["instanceId"], INSTANCE)
+            self.assertEqual(observations[0]["generation"], 1)
+            self.assertEqual(observations[0]["state"], health if running and health != "healthy" else status)
+            self.assertEqual(observations[0]["healthy"], running and health == "healthy")
+            self.assertIn("observedAt", observations[0])
+            self.assertEqual(sum(args.args[:2] == ("container", "inspect") for args in docker.call_args_list), 1)
+        info["Mounts"][0]["Source"] = "/foreign/app"
+        observations = []
+        with patch.object(self.docker, "call", side_effect=["container-id", json.dumps([info])]):
+            self.agent.instance_resources(observations)
+        self.assertEqual(observations, [])
+
+    def test_status_batch_limits_concurrency_and_isolates_target_failures(self):
+        operation = {"kind": "rcon.status-batch", "payload": {"instances": [
+            {"instanceId": f"00000000-0000-4000-8000-{i:012d}", "generation": 1} for i in range(5)]}}
+        self.assertTrue(module.read_operation(operation))
+        self.assertTrue(self.agent.health_operation(operation))
+        ready, release = threading.Event(), threading.Event()
+        lock = threading.Lock()
+        active, peak = 0, 0
+
+        def rcon(instance_id, generation, command, deadline):
+            nonlocal active, peak
+            with lock:
+                active += 1
+                peak = max(peak, active)
+                if active == 2:
+                    ready.set()
+            try:
+                self.assertEqual(command, "status")
+                self.assertTrue(release.wait(2))
+                if instance_id == "00000000-0000-4000-8000-000000000002":
+                    raise AgentError("local failure")
+                return {"ok": True, "text": "status"}
+            finally:
+                with lock:
+                    active -= 1
+
+        results = []
+        with patch.object(self.agent, "rcon", side_effect=rcon) as probe:
+            thread = threading.Thread(target=lambda: results.append(self.agent.dispatch(operation)))
+            thread.start()
+            try:
+                self.assertTrue(ready.wait(2))
+                self.assertEqual(probe.call_count, 2)
+            finally:
+                release.set()
+                thread.join(timeout=3)
+        self.assertFalse(thread.is_alive())
+        self.assertEqual(peak, 2)
+        self.assertEqual(len(results[0]["results"]), 5)
+        self.assertFalse(results[0]["ok"])
+        self.assertEqual(results[0]["results"][2]["reply"]["reason"], "unreachable")
+        self.assertTrue(results[0]["results"][4]["reply"]["ok"])
+
+    def test_status_batch_rejects_old_generations_and_bounds_expired_work(self):
+        marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json"
+        module.atomic_json(marker, {"generation": 2})
+        operation = {"kind": "rcon.status-batch", "payload": {"instances": [{"instanceId": INSTANCE, "generation": 1}]}}
+        with patch.object(self.agent, "rcon") as probe:
+            result = self.agent.dispatch(operation)
+            self.assertEqual(result["results"][0]["reply"]["reason"], "superseded")
+            probe.assert_not_called()
+            marker.unlink()
+            with patch.object(module.time, "monotonic", side_effect=[0] + [21] * 20):
+                result = self.agent.dispatch(operation)
+            self.assertEqual(result["results"][0]["reply"]["reason"], "timeout")
+            probe.assert_not_called()
+        for targets in ([], [None], [{"instanceId": INSTANCE, "generation": 1}] * 2):
+            with self.assertRaises(AgentError):
+                self.agent.dispatch({"kind": "rcon.status-batch", "payload": {"instances": targets}})
+
     def operation(self, kind, payload=None, op_id="operation-1", generation=1):
         return {"id": op_id, "kind": kind, "instanceId": INSTANCE, "generation": generation, "payload": payload or {}}
 

docs/native-hosts.md

@@ -584,6 +584,17 @@ and maintenance additionally share successful health results for up to 30 second
 from the request start, preserving the original observation time. Changed server
 identity or lifecycle intent invalidates reuse. Manual checks and disruption
 safety checks continue to request fresh evidence.
+Status scans allow two checks per host, so a slow host cannot stall the fleet.
+Accepted or running health operations are withheld from the other health lane
+for 30 seconds; an expired replay lease allows recovery with the original ID.
+Agent 1.4.10 includes generation-tagged container states in its 15-second
+heartbeat, using the same Docker inspection as resource sampling (including
+stopped containers). Routine monitoring uses samples younger than 30 seconds,
+preserving their sample time and falling back to inspection when evidence is
+missing, stale, or predates lifecycle changes. RCON status queries share one
+operation per host, with two concurrent local probes and a 20-second batch
+deadline. Individual failures remain visible per server. Both additions are
+capability-gated; older agents retain individual inspection and RCON operations.
 Agent 1.4.5 reserves two observation workers for instance health and read-only RCON,
 and one for files, logs, and host inventory. Background reads arrive in batches of
 up to four. File receipt uploads release the instance lock before contacting the

src/lib/hosts.ts

@@ -70,6 +70,10 @@ export interface HostInventory {
     agentUpdate?: HostAgentUpdate;
     instanceResources?: InstanceResources[];
     instanceResourcesError?: string;
+    instanceStates?: InstanceState[];
+}
+export interface InstanceState {
+    instanceId: string; generation: number; state: string; healthy: boolean; observedAt: string;
 }
 export interface InstanceResources {
     instanceId: string; generation: number; uptimeSeconds: number; cpuCores: number; memoryBytes: number;

src/lib/server/db/index.ts

@@ -380,7 +380,11 @@ export function db(): Database.Database {
             CREATE INDEX IF NOT EXISTS idx_server_deployments_state
             ON server_deployments(state, action, server_deleted_at);
             CREATE INDEX IF NOT EXISTS idx_server_deployments_started_at
-            ON server_deployments(started_at);`);
+            ON server_deployments(started_at);
+            CREATE INDEX IF NOT EXISTS idx_server_deployments_submission
+            ON server_deployments(server, server_deleted_at, action, submitted_at);
+            CREATE INDEX IF NOT EXISTS idx_server_deployments_finished_at
+            ON server_deployments(finished_at) WHERE server_deleted_at IS NULL;`);
         installReportRevisions(handle);
     } catch (err) {
         handle.close();

src/lib/server/fleet/health-observations.ts

@@ -1,9 +1,10 @@
 import { config } from '../config.js';
 import { db } from '../db/index.js';
-import { getHost } from '../hosts/index.js';
+import { getHost, queueHostOperation, waitHostOperationStatus } from '../hosts/index.js';
-import { nativeInstance } from '../backends/native.js';
+import { nativeInstance, type NativeInstance } from '../backends/native.js';
 import { komodoInspectHealthy } from '../integrations/komodo.js';
-import { rconStatusResult } from '../integrations/rcon.js';
+import { rconStatusResult, agentRconStatusResult, type AgentRconReply } from '../integrations/rcon.js';
+import { withOperationSource } from '../hosts/operation-context.js';
 import { getServer } from './servers.js';
 import { latestGameSubmission } from './lifecycle-evidence.js';
 
@@ -99,8 +100,83 @@ async function observe<T extends { ok: boolean }>(server: string, fresh: boolean
 const inspections = new Map<string, Entry<Awaited<ReturnType<typeof komodoInspectHealthy>>>>();
 const rcon = new Map<string, Entry<Awaited<ReturnType<typeof rconStatusResult>>>>();
 
+/** Keep the actual sample time: a later heartbeat must not renew old evidence. */
+export function heartbeatInspection(server: string): Observation<Awaited<ReturnType<typeof komodoInspectHealthy>>> | null {
+    const row = getServer(server);
+    if (!row?.native_instance_id || !row.host_id) return null;
+    const host = getHost(row.host_id);
+    if (!host?.connected || !host.inventory.capabilities?.includes('instance.heartbeat-state')) return null;
+    const state = host.inventory.instanceStates?.find(state => state.instanceId === row.native_instance_id && state.generation === row.instance_generation);
+    if (!state) return null;
+    const at = Date.parse(state.observedAt);
+    const lifecycle = db().prepare(`SELECT MAX(MAX(submitted_at,COALESCE(finished_at,submitted_at))) AS at
+        FROM server_deployments WHERE server=? AND server_deleted_at IS NULL`).get(server) as { at: string | null };
+    if (!reusable(at) || at < Date.parse(row.created_at) || at < Date.parse(lifecycle.at ?? '')
+        || at < Date.parse(latestGameSubmission(server) ?? '') || at < Date.parse(row.container_checked_at ?? '')) return null;
+    return { observedAt: new Date(at).toISOString(), result: state.healthy
+        ? { ok: true, state: state.state } : { ok: false, state: state.state, reason: `Container ${state.state}` } };
+}
+
+type RconObservation = Observation<Awaited<ReturnType<typeof rconStatusResult>>>;
+
+/** Prime a whole host before the per-host queue admits just its first two servers. */
+export function prepareMonitoringRcon(servers: string[]): Map<string, () => Promise<RconObservation>> {
+    const probes = new Map<string, () => Promise<RconObservation>>();
+    const batches = new Map<string, { instances: NativeInstance[]; result: Promise<{ results: Array<{ instanceId: string; generation: number; reply: AgentRconReply }> }> }>();
+    for (const server of servers) {
+        const instance = nativeInstance(server);
+        if (!instance) continue;
+        const host = getHost(instance.host_id);
+        if (!host?.connected || !host.inventory.capabilities?.includes('rcon.status-batch')) continue;
+        const container = heartbeatInspection(server);
+        if (container && !container.result.ok && container.result.state !== 'unknown') continue;
+        const before = identity(server, 'rcon');
+        const request = async () => {
+            let batch = batches.get(instance.host_id);
+            if (!batch) {
+                const instances: NativeInstance[] = [];
+                // observe() invokes each request in the same microtask turn;
+                // enqueue afterward so cached or shared probes need no batch slot.
+                const result = Promise.resolve().then(async () => {
+                    const operation = withOperationSource('Status monitor', () => queueHostOperation(instance.host_id, 'rcon.status-batch', {
+                        instances: instances.map(value => ({ instanceId: value.id, generation: value.generation }))
+                    }));
+                    const receipt = await waitHostOperationStatus(operation.id, 25_000);
+                    if (receipt.state !== 'completed') throw new Error(receipt.error ?? 'Host status batch did not complete.');
+                    return receipt.result as { results: Array<{ instanceId: string; generation: number; reply: AgentRconReply }> };
+                });
+                batch = { instances, result }; batches.set(instance.host_id, batch);
+            }
+            batch.instances.push(instance);
+            const endpoint = { host: `agent:${instance.host_id}`, port: instance.port };
+            try {
+                const receipt = await batch.result;
+                const replies = receipt.results.filter(value => value.instanceId === instance.id && value.generation === instance.generation);
+                if (replies.length !== 1 || typeof replies[0].reply?.text !== 'string') throw new Error('Host status batch omitted a valid instance result.');
+                return agentRconStatusResult(replies[0].reply, endpoint);
+            } catch (error) { return { ok: false as const, endpoint, reason: 'timeout' as const, detail: (error as Error).message }; }
+        };
+        // Observe before dispatch, including servers still waiting for their
+        // queue slot. Consume failures too, so a failed batch is never retried
+        // once per server or left as an unhandled rejected promise.
+        const pending = observe(server, false, 'rcon', rcon, request).then(value => ({ value }), error => ({ error }));
+        probes.set(server, async () => {
+            const result = await pending;
+            if ('error' in result) throw result.error;
+            const current = identity(server, 'rcon');
+            if (!sameIdentity(before, current) || newerEvidence(current, Date.parse(result.value.observedAt))) {
+                throw new MonitoringChangedError('Server changed before consuming the host status batch.');
+            }
+            return result.value;
+        });
+    }
+    return probes;
+}
+
 /** Routine monitoring only. Manual diagnostics and disruption gates use fresh probes directly. */
 export function inspectForMonitoring(server: string, fresh = false) {
+    const heartbeat = !fresh && heartbeatInspection(server);
+    if (heartbeat) return Promise.resolve(heartbeat);
     return observe(server, fresh, 'inspect', inspections, komodoInspectHealthy);
 }
 export function rconForMonitoring(server: string, fresh = false) {

src/lib/server/fleet/lifecycle-evidence.ts

@@ -3,6 +3,17 @@ import { db } from '../db/index.js';
 /** Reports from before the latest lifecycle submission describe the old process. */
 export function latestGameSubmission(server: string): string | null {
     return (db().prepare(`SELECT MAX(at) AS at FROM (
-        SELECT submitted_at AS at FROM server_deployments WHERE server=? AND server_deleted_at IS NULL AND action IN ('create','start','restart','rebuild')
+        SELECT MAX(submitted_at) AS at FROM server_deployments WHERE server=? AND server_deleted_at IS NULL AND action IN ('create','start','restart','rebuild')
         UNION ALL SELECT requested_at AS at FROM restart_rollout WHERE server=?)`).get(server, server) as { at: string | null }).at;
 }
+
+/** Look up each registered server through the covering submission index. */
+export function latestGameSubmissions(): Map<string, string | null> {
+    const rows = db().prepare(`SELECT server, MAX(at) AS at FROM (
+        SELECT s.name AS server, (SELECT MAX(submitted_at) FROM server_deployments
+            WHERE server=s.name AND server_deleted_at IS NULL AND action IN ('create','start','restart','rebuild')) AS at
+        FROM servers s
+        UNION ALL SELECT server, requested_at AS at FROM restart_rollout) GROUP BY server`).all() as
+        { server: string; at: string | null }[];
+    return new Map(rows.map(row => [row.server, row.at]));
+}

src/lib/server/fleet/overview.ts

@@ -34,11 +34,18 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
             href: `/campaigns/${encodeURIComponent(row.id)}` });
     }
     // Only the newest lifecycle operation can still require action on a server.
-    const lifecycle = handle.prepare(`SELECT d.* FROM server_deployments d
+    // Separate branches let SQLite use state and finish indexes; UNION deduplicates
-        WHERE d.server_deleted_at IS NULL AND (d.state IN ('queued', 'running', 'waiting', 'awaiting-health')
+    // recent warnings that are also the newest operation. Unary + keeps terminal
-        OR (d.state IN ('done', 'warning', 'cancelled') AND d.finished_at >= ?)
+    // branches on finished_at and rowid instead of searching historical states.
-        OR (d.state IN ('failed', 'interrupted', 'warning') AND NOT EXISTS
+    const lifecycle = handle.prepare(`SELECT id, server, action, state, stage, error, started_at FROM server_deployments
-            (SELECT 1 FROM server_deployments n WHERE n.server = d.server AND n.rowid > d.rowid)))`)
+        WHERE server_deleted_at IS NULL AND state IN ('queued', 'running', 'waiting', 'awaiting-health')
+        UNION
+        SELECT id, server, action, state, stage, error, started_at FROM server_deployments
+        WHERE server_deleted_at IS NULL AND +state IN ('done', 'warning', 'cancelled') AND finished_at >= ?
+        UNION
+        SELECT id, server, action, state, stage, error, started_at FROM server_deployments
+        WHERE server_deleted_at IS NULL AND +state IN ('failed', 'interrupted', 'warning')
+            AND rowid IN (SELECT MAX(rowid) FROM server_deployments GROUP BY server)`)
         .all(since) as { id: string; server: string; action: string; state: string; stage: string;
             error: string | null; started_at: string }[];
     for (const row of lifecycle) {

src/lib/server/fleet/status.ts

@@ -4,6 +4,7 @@ import { isProvisioning } from './provisioning.js';
 import { restartSettling } from '../provisioning/restarts.js';
 import { type ServerRow } from './servers.js';
 import { telemetryReport, latestGameSubmission } from './status-evidence.js';
+import { latestGameSubmissions } from './lifecycle-evidence.js';
 import { db } from '../db/index.js';
 import { parseGameEvent, type GameTelemetry } from '../../telemetry.js';
 import { restartRequestInFlight } from '../provisioning/restarts.js';
@@ -18,10 +19,7 @@ export function observationReader() {
         { server: string; action: string; state: string }[]).map(row => [row.server, row]));
     const restarts = new Map((handle.prepare('SELECT server, requested_at, healthy_at FROM restart_rollout').all() as
         { server: string; requested_at: string; healthy_at: string | null }[]).map(row => [row.server, row]));
-    const submissions = new Map((handle.prepare(`SELECT server, MAX(at) AS at FROM (
+    const submissions = latestGameSubmissions();
-        SELECT server, submitted_at AS at FROM server_deployments WHERE server_deleted_at IS NULL AND action IN ('create','start','restart','rebuild')
-        UNION ALL SELECT server, requested_at AS at FROM restart_rollout) GROUP BY server`).all() as
-        { server: string; at: string | null }[]).map(row => [row.server, row.at]));
     const reports = new Map<string, GameTelemetry>();
     for (const row of handle.prepare('SELECT server, received_at, payload FROM server_telemetry WHERE payload IS NOT NULL').all() as
         { server: string; received_at: string; payload: string }[]) {

src/lib/server/hosts/index.ts

@@ -254,6 +254,15 @@ export function validateInventory(value: unknown): HostInventory {
     }
     if (inventory.cpuUsagePercent !== undefined && (typeof inventory.cpuUsagePercent !== 'number' || !Number.isFinite(inventory.cpuUsagePercent) || inventory.cpuUsagePercent < 0 || inventory.cpuUsagePercent > 100)) throw new Error('Invalid inventory cpuUsagePercent.');
     validateInstanceResources(inventory);
+    if (inventory.instanceStates !== undefined && (!Array.isArray(inventory.instanceStates) || inventory.instanceStates.length > 1000
+        || inventory.instanceStates.some(state => !state || typeof state !== 'object'
+            || typeof state.instanceId !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/.test(state.instanceId)
+            || !Number.isSafeInteger(state.generation) || state.generation < 1
+            || typeof state.state !== 'string' || !/^[a-z-]{1,40}$/.test(state.state)
+            || typeof state.healthy !== 'boolean' || state.healthy && state.state !== 'running'
+            || typeof state.observedAt !== 'string' || state.observedAt.length > 40 || !Number.isFinite(Date.parse(state.observedAt))))) {
+        throw new Error('Invalid instance state inventory.');
+    }
     for (const key of ['dockerRunning','directoriesWritable']) if (inventory[key] !== undefined && typeof inventory[key] !== 'boolean') throw new Error(`Invalid inventory ${key}.`);
     if (inventory.runtimeImages !== undefined && (!Array.isArray(inventory.runtimeImages) || inventory.runtimeImages.some(image => typeof image !== 'string'))) throw new Error('Invalid runtime image inventory.');
     if (inventory.capabilities !== undefined && (!Array.isArray(inventory.capabilities) || inventory.capabilities.length > 32 || inventory.capabilities.some(value => typeof value !== 'string' || value.length > 80))) throw new Error('Invalid agent capability inventory.');
@@ -486,9 +495,9 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = '
         WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown') LIMIT 1`).get(hostId)) return [];
     // Accepted/running/unknown commands keep the same immutable ID. The agent
     // consults its durable journal before acknowledging or executing a replay.
-    const readOnly = `(kind IN ('host.inspect','instance.inspect','files.read','files.list','files.stat','files.manifest','logs.read')
+    const readOnly = `(kind IN ('host.inspect','instance.inspect','rcon.status-batch','files.read','files.list','files.stat','files.manifest','logs.read')
         OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`;
-    const healthOnly = `(kind='instance.inspect' OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`;
+    const healthOnly = `(kind IN ('instance.inspect','rcon.status-batch') OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`;
     const healthLane = host.inventory.capabilities?.includes('operations.health-lane');
     const laneFilter = lane === 'all' ? '' : lane === 'health' ? ` AND ${healthOnly}`
         : ` AND ${lane === 'write' ? 'NOT ' : ''}${readOnly}${lane === 'read' && healthLane ? ` AND NOT ${healthOnly}` : ''}`;
@@ -504,8 +513,12 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = '
             AND (kind!='instance.cancel-create' OR ?)
             AND NOT EXISTS (SELECT 1 FROM ec_host_operations cancellation WHERE cancellation.host_id=o.host_id
                 AND cancellation.kind='instance.cancel-create' AND json_extract(cancellation.payload,'$.createOperationId')=o.id)
+            ${lane === 'health' ? "AND (state NOT IN ('accepted','running') OR MAX(updated_at,COALESCE(last_delivered_at,'')) <= ?)" : ''}
             ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 ELSE 1 END,COALESCE(last_delivered_at,''),created_at,rowid LIMIT ${limit}`)
-            .all(hostId, Number(host.readiness === 'ready' || host.mode === 'draining'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false)) as OperationDeliveryRow[];
+            // A short replay lease avoids handing the idle health lane its busy
+            // sibling's work, but still recovers a lost receipt or agent restart.
+            .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 deliveredAt = now();
         for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id);
         return rows;

src/lib/server/hosts/observations.ts

@@ -1,4 +1,4 @@
-export const observationKinds = ['host.inspect', 'instance.inspect', 'files.read', 'files.list', 'files.stat', 'files.manifest', 'logs.read'];
+export const observationKinds = ['host.inspect', 'instance.inspect', 'rcon.status-batch', 'files.read', 'files.list', 'files.stat', 'files.manifest', 'logs.read'];
 const commands = ['status', 'getstatus', 'version'];
 
 export function safeOperationTarget(kind: string, value: string | null): string | null {

src/lib/server/integrations/plutonium.ts

@@ -3,6 +3,7 @@ import { config } from '../config.js';
 import { getServer, listServers } from '../fleet/servers.js';
 import { hasActiveProvisioning } from '../fleet/provisioning.js';
 import { latestGameSubmission } from '../fleet/status-evidence.js';
+import { latestGameSubmissions } from '../fleet/lifecycle-evidence.js';
 import { rconPlutoniumVersion } from './rcon.js';
 import { komodoInspectHealthy } from './komodo.js';
 import { restartRecoveryPending } from '../provisioning/index.js';
@@ -23,9 +24,7 @@ export function runningVersions(): PlutoniumVersion[] {
     const versions = db().prepare('SELECT * FROM plutonium_versions').all() as PlutoniumVersion[];
     const target = freshRelease();
     const byServer = new Map(versions.map(version => [version.server, version]));
-    const epochs = new Map((db().prepare(`SELECT server, MAX(at) AS at FROM (
+    const epochs = latestGameSubmissions();
-        SELECT server, submitted_at AS at FROM server_deployments WHERE server_deleted_at IS NULL AND action IN ('create','start','restart','rebuild')
-        UNION ALL SELECT server, requested_at AS at FROM restart_rollout) GROUP BY server`).all() as { server: string; at: string | null }[]).map(row => [row.server, row.at]));
     const recovering = new Set((db().prepare('SELECT server FROM restart_rollout WHERE healthy_at IS NULL').all() as { server: string }[]).map(row => row.server));
     return listServers().map(server => {
         const value = byServer.get(server.name)

src/lib/server/integrations/rcon.ts

@@ -367,22 +367,9 @@ export function formatRawReply(buf: Buffer): string {
     return lines.join('\n');
 }
 
-/**
+export interface AgentRconReply { ok?: boolean; reason?: string; text: string }
- * Full RCON status probe with structured diagnostics. Callers get either
+
- * (ok, endpoint, status) or (ok:false, endpoint?, reason, detail). This is
+export function agentRconStatusResult(reply: AgentRconReply, endpoint: { host: string; port: number }): RconStatusResult {
- * the shape the fleet UI consumes to explain *why* a server is unreachable
- * rather than just "unreachable".
- */
-export async function rconStatusResult(server: string): Promise<RconStatusResult> {
-    const native = getServer(server)?.native_instance_id ? nativeInstance(server) : null;
-    if (native) {
-        const key = JSON.stringify(['agent', native.id, native.generation, latestGameSubmission(server)]);
-        const existing = statusInFlight.get(key);
-        if (existing) return existing;
-        const endpoint = { host: `agent:${native.host_id}`, port: native.port };
-        const request = (async (): Promise<RconStatusResult> => {
-            try {
-                const reply = await nativeRequest<{ ok?: boolean; reason?: string; text: string }>(native, 'rcon.execute', { command: 'status' });
     if (reply.ok === false) {
         const failures: Record<string, { reason: RconFailure; detail: string }> = {
             unconfigured: { reason: 'no-config', detail: 'Agent RCON password is not configured.' },
@@ -402,6 +389,25 @@ export async function rconStatusResult(server: string): Promise<RconStatusResult
         capacity: positiveCapacity(info.sv_maxclients ?? header.sv_maxclients),
         map: info.mapname ?? info.sv_mapname ?? header.map ?? header.mapname ?? header.sv_mapname ?? null,
         gametype: info.g_gametype ?? header.g_gametype ?? header.gametype ?? null, roster: parsePlayerRoster(reply.text) }, rawReply: reply.text.slice(0, 4000) };
+}
+
+/**
+ * Full RCON status probe with structured diagnostics. Callers get either
+ * (ok, endpoint, status) or (ok:false, endpoint?, reason, detail). This is
+ * the shape the fleet UI consumes to explain *why* a server is unreachable
+ * rather than just "unreachable".
+ */
+export async function rconStatusResult(server: string): Promise<RconStatusResult> {
+    const native = getServer(server)?.native_instance_id ? nativeInstance(server) : null;
+    if (native) {
+        const key = JSON.stringify(['agent', native.id, native.generation, latestGameSubmission(server)]);
+        const existing = statusInFlight.get(key);
+        if (existing) return existing;
+        const endpoint = { host: `agent:${native.host_id}`, port: native.port };
+        const request = (async (): Promise<RconStatusResult> => {
+            try {
+                const reply = await nativeRequest<{ ok?: boolean; reason?: string; text: string }>(native, 'rcon.execute', { command: 'status' });
+                return agentRconStatusResult(reply, endpoint);
             } catch (error) { return { ok: false, endpoint, reason: 'timeout', detail: (error as Error).message }; }
         })().finally(() => statusInFlight.delete(key));
         statusInFlight.set(key, request); return request;

src/lib/server/pipeline/worker.ts

@@ -6,7 +6,7 @@ import { emit } from './bus.js';
 import type { ServerStatusEntry } from './bus.js';
 import type { RconFailure } from '../integrations/rcon.js';
 import { getServer, listServers, setServerFields, type ServerRow } from '../fleet/servers.js';
-import { inspectForMonitoring, rconForMonitoring, MonitoringChangedError } from '../fleet/health-observations.js';
+import { inspectForMonitoring, rconForMonitoring, prepareMonitoringRcon, MonitoringChangedError } from '../fleet/health-observations.js';
 import { restartSettling } from '../provisioning/index.js';
 import { serverObservation, observationReader } from '../fleet/status.js';
 import { telemetryEvidence, latestGameSubmission } from '../fleet/status-evidence.js';
@@ -20,9 +20,8 @@ import { trackActivity } from './activity.js';
 import { withOperationSource } from '../hosts/operation-context.js';
 
 const TICK_MS = 15_000;
-// Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the
+// Match each agent's two health lanes without making unrelated hosts wait.
-// host all at once.
+const STATUS_CONCURRENCY_PER_HOST = 2;
-const STATUS_CONCURRENCY = 4;
 
 let started = false;
 let tickTimer: NodeJS.Timeout | null = null;
@@ -76,8 +75,8 @@ function encodeError(reason: RconFailure, detail: string): string {
 // Deduplicate queued/running checks per server. A slow server must not prevent
 // completed servers from being polled again on the next 30-second tick.
 const pendingStatus = new Map<string, Promise<number | null>>();
-const statusQueue: Array<() => Promise<void>> = [];
+interface StatusQueue { active: number; queued: Array<() => Promise<void>> }
-let activeStatus = 0;
+const statusQueues = new Map<string, StatusQueue>();
 const changedStatus = new Map<string, ServerStatusEntry>();
 function publishStatus(entry: ServerStatusEntry) {
     const scheduled = changedStatus.size > 0;
@@ -88,38 +87,46 @@ function publishStatus(entry: ServerStatusEntry) {
     });
 }
 
-function drainStatusQueue(): void {
+function drainStatusQueue(host: string, queue: StatusQueue): void {
-    while (activeStatus < STATUS_CONCURRENCY && statusQueue.length) {
+    while (queue.active < STATUS_CONCURRENCY_PER_HOST && queue.queued.length) {
-        const next = statusQueue.shift()!;
+        const next = queue.queued.shift()!;
-        activeStatus++;
+        queue.active++;
-        void next().finally(() => { activeStatus--; drainStatusQueue(); });
+        void next().finally(() => { queue.active--; drainStatusQueue(host, queue); });
     }
+    if (!queue.active && !queue.queued.length) statusQueues.delete(host);
 }
 
-function queueStatus(server: string): Promise<number | null> {
+function queueStatus(srv: ServerRow, probe?: () => ReturnType<typeof rconForMonitoring>): Promise<number | null> {
+    const server = srv.name;
     const pending = pendingStatus.get(server);
     if (pending) return pending;
+    // Legacy servers without a recorded host share the configured host's pool.
+    const host = srv.host_id ? `native:${srv.host_id}` : `komodo:${srv.komodo_server || config.komodo.targetServer}`;
+    let queue = statusQueues.get(host);
+    if (!queue) { queue = { active: 0, queued: [] }; statusQueues.set(host, queue); }
     let finish!: (capacity: number | null) => void;
     const work = new Promise<number | null>(resolve => { finish = resolve; });
     pendingStatus.set(server, work);
-    statusQueue.push(async () => {
+    queue.queued.push(async () => {
         let capacity: number | null = null;
         try {
-            capacity = await withOperationSource('Status monitor', () => refreshServerStatus(server));
+            capacity = await withOperationSource('Status monitor', () => refreshServerStatus(server, probe));
             // Publish this server's latest evidence without waiting for the fleet.
             const current = getServer(server);
             if (current) publishStatus(statusEntry(current));
         } catch (err) { console.error(`[worker] status error for ${server}:`, err); }
         finally { pendingStatus.delete(server); finish(capacity); }
     });
-    drainStatusQueue();
+    drainStatusQueue(host, queue);
     return work;
 }
 
 export async function fanoutStatus(): Promise<void> {
     const servers = listServers();
     if (!servers.length) return;
-    const results = await Promise.all(servers.map(server => queueStatus(server.name)));
+    const probes = prepareMonitoringRcon(servers.filter(server => !pendingStatus.has(server.name)
+        && !isProvisioning(server.name) && !restartSettling(server.name) && !telemetryEvidence(server.name)).map(server => server.name));
+    const results = await Promise.all(servers.map(server => queueStatus(server, probes.get(server.name))));
     const capacities = new Map(servers.map((server, i) => [server.name, results[i]]));
     try {
         // Metrics remain complete fleet collections, using freshness at capture
@@ -135,7 +142,7 @@ export async function fanoutStatus(): Promise<void> {
     } catch (err) { console.error('[worker] metrics error:', err); }
 }
 
-async function refreshServerStatus(server: string): Promise<number | null> {
+async function refreshServerStatus(server: string, probeStatus?: () => ReturnType<typeof rconForMonitoring>): Promise<number | null> {
     const srv = getServer(server);
     if (!srv) return null;
     let capacity: number | null = null;
@@ -170,7 +177,7 @@ async function refreshServerStatus(server: string): Promise<number | null> {
             } else if (unavailable || !telemetryEvidence(srv.name)) {
                 flush();
                 const probeStartedAt = new Date().toISOString();
-                const probe = await rconForMonitoring(srv.name).catch(err => err instanceof MonitoringChangedError ? null
+                const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name)).catch(err => err instanceof MonitoringChangedError ? null
                     : { result: { ok: false as const, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }, observedAt: probeStartedAt });
                 if (!probe) return null;
                 const result = probe.result;

static/agent/ec-agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.9"
+VERSION = "1.4.10"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -519,7 +519,7 @@ class ReadWriteGate:
 def read_operation(operation):
     if not isinstance(operation, dict):
         return False
-    if operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
+    if operation.get("kind") in ("host.inspect", "instance.inspect", "rcon.status-batch", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
         return True
     payload = operation.get("payload")
     return operation.get("kind") == "rcon.execute" and isinstance(payload, dict) and payload.get("command") in ("status", "getstatus", "version")
@@ -913,6 +913,7 @@ class Agent:
         self.read_gate = ReadWriteGate()
         self.read_claim_lock = threading.Lock()
         self.read_claims = set()
+        self.health_probe_gate = threading.BoundedSemaphore(2)
         self.cancelled_creations = frozenset()
         self.observation_keep_days = 0  # Older controllers do not authorize cleanup.
         self.library_progress = {}
@@ -1127,18 +1128,19 @@ class Agent:
                 return None
             return round(100 * (total - deltas[3] - deltas[4]) / total, 1)
 
-    def instance_resources(self):
+    def instance_resources(self, observations=None):
         """One bounded Docker stats batch; unavailable samples are never zero load."""
-        ids = self.docker.call("container", "ls", "--filter", "label=" + LABEL + ".managed=true",
+        ids = self.docker.call("container", "ls", "--all", "--filter", "label=" + LABEL + ".managed=true",
                                "--no-trunc", "--format", "{{.ID}}", timeout=10).splitlines()[:1000]
         if not ids:
             return []
+        observed_at = datetime.now(timezone.utc).isoformat(timespec="milliseconds")
         containers = json.loads(self.docker.call("container", "inspect", *ids, timeout=10))
         owned = {}
         for info in containers:
             try:
                 labels = info.get("Config", {}).get("Labels", {}) or {}
-                if labels.get(LABEL + ".managed") != "true" or not info.get("State", {}).get("Running"):
+                if labels.get(LABEL + ".managed") != "true":
                     continue
                 instance_id, generation = instance_identity(labels.get(LABEL + ".instance"), int(labels.get(LABEL + ".generation", "")))
                 name = self.container_name(instance_id, generation)
@@ -1147,6 +1149,16 @@ class Agent:
                 expected = str(self.instance_dir(instance_id, generation) / "app")
                 if not any(mount.get("Type") == "bind" and mount.get("Source") == expected and mount.get("Destination") == "/home/plutainer/app" for mount in info.get("Mounts", [])):
                     continue
+                state = info.get("State", {})
+                healthy = state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
+                status = state.get("Status", "unknown")
+                if status == "running" and not healthy:
+                    status = state.get("Health", {}).get("Status", "unknown")
+                if observations is not None:
+                    observations.append({"instanceId": instance_id, "generation": generation,
+                                         "state": status, "healthy": healthy, "observedAt": observed_at})
+                if not state.get("Running"):
+                    continue
                 started = datetime.fromisoformat(info["State"]["StartedAt"].replace("Z", "+00:00"))
                 uptime = max(0, int((datetime.now(timezone.utc) - started).total_seconds()))
                 owned[name] = {"instanceId": instance_id, "generation": generation, "uptimeSeconds": uptime}
@@ -1280,17 +1292,21 @@ class Agent:
                         images.extend(alias for alias, image_id in json.loads(aliases.read_text()).items() if image_id in image_ids)
         resources = {}
         if docker_running:
+            observations = []
             try:
-                resources["instanceResources"] = self.instance_resources()
+                resources["instanceResources"] = self.instance_resources(observations)
             except (AgentError, ValueError, TypeError, KeyError):
                 resources["instanceResourcesError"] = "Container resource sampling failed. Check Docker access and stats availability on this host."
+            # A failed stats call does not discard a successful inspection.
+            # Missing entries remain unknown; they are never inferred stopped.
+            resources["instanceStates"] = observations
         os_name = platform.system()
         with contextlib.suppress(OSError):
             for line in Path("/etc/os-release").read_text().splitlines():
                 if line.startswith("PRETTY_NAME="):
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
-        capabilities = ["library.remove", "instance.cancel-create", "operations.read-lane", "operations.health-lane", "files.manifest"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
+        capabilities = ["library.remove", "instance.cancel-create", "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})}
 
     def safe_to_disrupt(self, instance_id, generation):
@@ -1314,7 +1330,46 @@ class Agent:
         if result["players"] != 0:
             raise Held("Players are present; disruption held")
 
-    def rcon(self, instance_id, generation, command):
+    def rcon_status_batch(self, operation):
+        targets = operation.get("payload", {}).get("instances")
+        if not isinstance(targets, list) or not 1 <= len(targets) <= 1000 or any(not isinstance(target, dict) for target in targets):
+            raise AgentError("Invalid status batch")
+        identities = [instance_identity(target.get("instanceId"), target.get("generation")) for target in targets]
+        if len(set(identities)) != len(identities):
+            raise AgentError("Duplicate status batch target")
+        deadline = time.monotonic() + 20
+
+        def probe(identity):
+            instance_id, generation = identity
+            target = {"instanceId": instance_id, "generation": generation}
+            reply = {"ok": False, "reason": "timeout", "text": ""}
+            # A batch does not hold every instance lock while one server waits.
+            # Bound concurrency across overlapping batches as well as within one.
+            lock = self.instance_lock(target)
+            # Use the same lock order as individual health-lane operations.
+            if lock.acquire(timeout=max(0, deadline - time.monotonic())):
+                try:
+                    if self.health_probe_gate.acquire(timeout=max(0, deadline - time.monotonic())):
+                        try:
+                            marker = self.instance_dir(instance_id, generation).parent / "current.json"
+                            current = json.loads(marker.read_text()) if marker.exists() else {}
+                            if current.get("generation", generation) != generation or current.get("removed"):
+                                reply = {"ok": False, "reason": "superseded", "text": ""}
+                            elif time.monotonic() < deadline:
+                                reply = self.rcon(instance_id, generation, "status", deadline=deadline)
+                        except (AgentError, OSError, ValueError, KeyError):
+                            reply = {"ok": False, "reason": "unreachable", "text": ""}
+                        finally:
+                            self.health_probe_gate.release()
+                finally:
+                    lock.release()
+            return {**target, "reply": reply}
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            results = list(executor.map(probe, identities))
+        return {"ok": all(row["reply"].get("ok") for row in results), "results": results}
+
+    def rcon(self, instance_id, generation, command, deadline=None):
         if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):
             raise AgentError("Invalid RCON command")
         saved = self.spec(instance_id, generation)
@@ -1327,21 +1382,25 @@ class Agent:
         # Retry only the read-only occupancy query. A missing UDP reply must
         # never cause commands with effects to be replayed.
         for attempt in range(3 if command == "status" else 1):
-            result = self.rcon_attempt(family, host, int(saved["port"]), password, command)
+            if deadline is not None and time.monotonic() >= deadline:
+                return {"ok": False, "reason": "timeout", "players": None, "text": ""}
+            result = self.rcon_attempt(family, host, int(saved["port"]), password, command, **({"deadline": deadline} if deadline is not None else {}))
             if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == 2 or command != "status":
                 return result
-            time.sleep(0.25 * 2 ** attempt)
+            time.sleep(min(0.25 * 2 ** attempt, max(0, deadline - time.monotonic())) if deadline is not None else 0.25 * 2 ** attempt)
 
-    def rcon_attempt(self, family, host, port, password, command):
+    def rcon_attempt(self, family, host, port, password, command, deadline=None):
         chunks = []
         with socket.socket(family, socket.SOCK_DGRAM) as sock:
-            sock.settimeout(3)
+            sock.settimeout(max(0.001, min(3, deadline - time.monotonic())) if deadline is not None else 3)
             try:
                 sock.connect((host, port))
                 sock.send(b"\xff\xff\xff\xffrcon " + password.encode() + b" " + command.encode() + b"\n")
                 while len(chunks) < 64 and sum(map(len, chunks)) < 1024 * 1024:
+                    if deadline is not None and time.monotonic() >= deadline:
+                        break
                     chunks.append(sock.recv(65535))
-                    sock.settimeout(0.15)
+                    sock.settimeout(max(0.001, min(0.15, deadline - time.monotonic())) if deadline is not None else 0.15)
             except socket.timeout:
                 pass
             except OSError:
@@ -2028,6 +2087,8 @@ class Agent:
             return {"image": image, "ready": True}
         if kind == "host.inspect":
             return self.inventory()
+        if kind == "rcon.status-batch":
+            return self.rcon_status_batch(operation)
         if kind == "library.download":
             return self.library_download(payload)
         if kind == "library.remove":
@@ -2049,6 +2110,9 @@ class Agent:
         if kind == "rcon.execute":
             if payload["command"] not in ("status", "getstatus", "version"):
                 self.checkpoint({"phase": "rcon-command"})
+            else:
+                with self.health_probe_gate:
+                    return self.rcon(instance_id, generation, payload["command"])
             return self.rcon(instance_id, generation, payload["command"])
         raise AgentError("Unsupported operation kind")
 
@@ -2266,7 +2330,7 @@ class Agent:
 
     @staticmethod
     def health_operation(operation):
-        return operation.get("kind") == "instance.inspect" or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
+        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"):
         # Each lane owns its SQLite connection and never writes executing_id,

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.4.9",
+  "version": "1.4.10",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "c0a02509a8307b2abb75e5450bf3f39596252f26e041ed938d662bc7127bec62",
+  "sha256": "c518a216ad4e9dd86ac82743b27bae54cdf1c51ef7ae3bf8d252dcc9863be1fb",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }

tests/unit/fleet-overview.test.ts

@@ -116,4 +116,36 @@ describe('persisted Fleet operations', () => {
         expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'beta', label: 'Observing a match', href: '/betas/1' }));
         expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'recovery', state: 'attention' }));
     });
+
+    it('preserves lifecycle selection across cutoff, deletion, supersession, and overlapping warning branches', async () => {
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const handle = db();
+        const since = iso(-86_400_000);
+        handle.transaction(() => {
+            const insert = handle.prepare(`INSERT INTO server_deployments
+                (id, server, action, state, stage, started_at, finished_at, server_deleted_at, triggered_by)
+                VALUES (?, ?, 'restart', ?, 'Stage', ?, ?, ?, 'test')`);
+            let i = 0;
+            for (const state of ['queued', 'running', 'waiting', 'awaiting-health', 'done', 'warning', 'cancelled', 'failed', 'interrupted']) {
+                for (const finished of [null, iso(-86_400_001), since, iso()]) {
+                    for (const deleted of [null, iso()]) {
+                        for (const successor of ['none', 'live', 'deleted']) {
+                            const id = `operation-${i++}`;
+                            insert.run(id, id, state, iso(-i), finished, deleted);
+                            if (successor !== 'none') insert.run(`${id}-next`, id, 'done', iso(-i), null, successor === 'deleted' ? iso() : null);
+                        }
+                    }
+                }
+            }
+        })();
+        // The former OR query is the compatibility reference for all three branches.
+        const expected = handle.prepare(`SELECT d.id FROM server_deployments d
+            WHERE d.server_deleted_at IS NULL AND (d.state IN ('queued', 'running', 'waiting', 'awaiting-health')
+            OR (d.state IN ('done', 'warning', 'cancelled') AND d.finished_at >= ?)
+            OR (d.state IN ('failed', 'interrupted', 'warning') AND NOT EXISTS
+                (SELECT 1 FROM server_deployments n WHERE n.server = d.server AND n.rowid > d.rowid)))`)
+            .all(since) as { id: string }[];
+        const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
+        expect(fleetOperations(now).map(row => row.key).sort()).toEqual(expected.map(row => `deployment:${row.id}`).sort());
+    });
 });

tests/unit/host-status-batch.test.ts

@@ -0,0 +1,138 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { newFixture, type Fixture } from '../helpers/fixture.js';
+
+vi.mock('../../src/lib/server/integrations/komodo.js', () => ({ komodoInspectHealthy: vi.fn() }));
+let fixture: Fixture;
+let at: number;
+beforeEach(async () => {
+    vi.resetModules(); vi.resetAllMocks();
+    fixture = await newFixture([]);
+    at = Date.now();
+    vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(at);
+});
+afterEach(async () => { vi.useRealTimers(); await fixture.cleanup(); });
+
+async function setup(count = 5, hostCount = 1) {
+    const hosts = await import('../../src/lib/server/hosts/index.js');
+    const { db } = await import('../../src/lib/server/db/index.js');
+    const { komodoInspectHealthy } = await import('../../src/lib/server/integrations/komodo.js');
+    vi.mocked(komodoInspectHealthy).mockResolvedValue({ ok: true, state: 'running' });
+    const hostIds: string[] = [];
+    for (let h = 0; h < hostCount; h++) {
+        const host = hosts.createHost({ name: `Host ${h}` }); hostIds.push(host.id);
+        const inventory = { protocolVersion: 1, capabilities: ['instance.heartbeat-state', 'rcon.status-batch', 'operations.health-lane'],
+            instanceStates: Array.from({ length: count }, (_, i) => ({ instanceId: `instance-${h}-${i}`, generation: 1,
+                healthy: true, state: 'running', observedAt: new Date(at).toISOString() })) };
+        hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
+        for (let i = 0; i < count; i++) {
+            const name = `server-${h}-${i}`;
+            db().prepare('INSERT INTO servers(name,komodo_name,created_at) VALUES(?,?,?)').run(name, '', new Date(at - 1000).toISOString());
+            db().prepare(`INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at)
+                VALUES(?,?,?,'library','image',?,?)`).run(`instance-${h}-${i}`, name, host.id, 28960 + i, new Date(at - 1000).toISOString());
+        }
+    }
+    const health = await import('../../src/lib/server/fleet/health-observations.js');
+    const worker = await import('../../src/lib/server/pipeline/worker.js');
+    const { getServer } = await import('../../src/lib/server/fleet/servers.js');
+    const operations = () => hostIds.flatMap(id => hosts.listHostOperations(id));
+    const complete = () => operations().filter(op => !['completed', 'failed'].includes(op.state)).forEach(op => {
+        const targets = op.payload.instances as { instanceId: string; generation: number }[];
+        hosts.reportHostOperation(op.hostId, op.id, { state: 'completed', result: op.kind === 'rcon.status-batch'
+            ? { ok: true, results: targets.map(target => ({ ...target, reply: { ok: true, text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n' } })) }
+            : { ok: true, text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n' } });
+    });
+    return { ...hosts, ...health, ...worker, getServer, db: db(), inspect: vi.mocked(komodoInspectHealthy), hostIds, operations, complete };
+}
+
+describe('host status observations', () => {
+    it('uses one batch per host for a fifteen-server pass and shares overlapping scans', async () => {
+        const api = await setup(5, 3);
+        const passes = [api.fanoutStatus(), api.fanoutStatus()];
+        try {
+            await vi.waitFor(() => expect(api.operations()).toHaveLength(3));
+            for (const operation of api.operations()) {
+                expect(operation.kind).toBe('rcon.status-batch');
+                expect(operation.payload.instances).toHaveLength(5);
+                expect(api.pendingHostOperations(operation.hostId, 'health').map(op => op.id)).toEqual([operation.id]);
+                expect(api.pendingHostOperations(operation.hostId, 'write')).toEqual([]);
+            }
+        } finally { api.complete(); await Promise.all(passes); }
+        expect(api.inspect).not.toHaveBeenCalled();
+        expect(api.getServer('server-2-4')).toMatchObject({ last_players: 0, last_map: 'mp_kino', container_checked_at: new Date(at).toISOString() });
+        expect(api.db.prepare('SELECT COUNT(*) AS n FROM metric_samples').get()).toEqual({ n: 15 });
+        await api.fanoutStatus();
+        expect(api.operations()).toHaveLength(3);
+    });
+
+    it.each(['stale', 'future', 'generation', 'lifecycle', 'missing', 'legacy'])('falls back to inspection for %s heartbeat evidence', async mode => {
+        const api = await setup(1);
+        const host = api.getHost(api.hostIds[0])!;
+        const inventory = structuredClone(host.inventory);
+        if (mode === 'stale') inventory.instanceStates![0].observedAt = new Date(at - 30_000).toISOString();
+        if (mode === 'future') inventory.instanceStates![0].observedAt = new Date(at + 1_000).toISOString();
+        if (mode === 'generation') inventory.instanceStates![0].generation = 2;
+        if (mode === 'missing') inventory.instanceStates = [];
+        if (mode === 'legacy') inventory.capabilities = [];
+        if (mode === 'lifecycle') api.db.prepare(`INSERT INTO server_deployments(id,server,action,state,stage,started_at,submitted_at,finished_at,triggered_by)
+            VALUES('stop','server-0-0','stop','done','done','started',?,?,'test')`).run(new Date(at + 1).toISOString(), new Date(at + 2).toISOString());
+        api.heartbeat(host.id, inventory);
+        await api.inspectForMonitoring('server-0-0');
+        expect(api.inspect).toHaveBeenCalledOnce();
+    });
+
+    it('retains heartbeat sample time, reports stopped containers and keeps forced probes fresh', async () => {
+        const api = await setup(1);
+        vi.setSystemTime(at + 10_000);
+        expect(await api.inspectForMonitoring('server-0-0')).toMatchObject({ observedAt: new Date(at).toISOString(), result: { ok: true } });
+        await api.inspectForMonitoring('server-0-0', true);
+        expect(api.inspect).toHaveBeenCalledOnce();
+        const host = api.getHost(api.hostIds[0])!;
+        api.heartbeat(host.id, { ...host.inventory, instanceStates: [{ ...host.inventory.instanceStates![0], healthy: false, state: 'exited' }] });
+        await api.fanoutStatus();
+        expect(api.getServer('server-0-0')?.container_state).toBe('exited');
+        expect(api.operations()).toEqual([]);
+    });
+
+    it.each(['generation', 'manual', 'lifecycle'])('discards batch replies overtaken by a %s change, including queued servers', async change => {
+        const api = await setup();
+        const pass = api.fanoutStatus();
+        try {
+            await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
+            vi.setSystemTime(at + 1000);
+            if (change === 'generation') api.db.prepare('UPDATE native_instances SET generation=2').run();
+            if (change === 'manual') api.db.prepare("UPDATE servers SET last_checked_at=?,last_error='timeout:manual'").run(new Date().toISOString());
+            if (change === 'lifecycle') api.db.prepare(`INSERT INTO server_deployments(id,server,action,state,stage,started_at,submitted_at,triggered_by)
+                SELECT name,name,'stop','done','done','started',?,'test' FROM servers`).run(new Date().toISOString());
+        } finally { api.complete(); await pass; }
+        for (let i = 0; i < 5; i++) expect(api.getServer(`server-0-${i}`)?.last_seen_at).toBeNull();
+        expect(api.operations()).toHaveLength(1);
+    });
+
+    it('records each failed or omitted result without retrying the batch per server', async () => {
+        const api = await setup();
+        const pass = api.fanoutStatus();
+        try {
+            await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
+            const operation = api.operations()[0];
+            api.reportHostOperation(operation.hostId, operation.id, { state: 'completed', result: { ok: false, results: [
+                { instanceId: 'instance-0-0', generation: 1, reply: { ok: false, reason: 'authentication', text: '' } }
+            ] } });
+        } finally { api.complete(); await pass; }
+        expect(api.getServer('server-0-0')?.last_error).toBe('bad-reply:wrong RCON password');
+        expect(api.getServer('server-0-4')?.last_error).toContain('timeout:');
+        expect(api.operations()).toHaveLength(1);
+    });
+
+    it.each([true, false])('uses batch capability=%s while missing heartbeat state still gets inspected', async capable => {
+        const api = await setup(1);
+        const host = api.getHost(api.hostIds[0])!;
+        api.heartbeat(host.id, { ...host.inventory, instanceStates: [], capabilities: capable ? ['rcon.status-batch'] : [] });
+        const pass = api.fanoutStatus();
+        try {
+            await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
+            expect(api.operations()[0].kind).toBe(capable ? 'rcon.status-batch' : 'rcon.execute');
+        } finally { api.complete(); await pass; }
+        expect(api.inspect).toHaveBeenCalledOnce();
+        expect(api.getServer('server-0-0')?.last_players).toBe(0);
+    });
+});

tests/unit/native-hosts.test.ts

@@ -94,11 +94,30 @@ describe('native host enrollment and scoped authentication', () => {
         const query = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('FROM ec_host_operations o WHERE host_id=?'))!;
         prepare.mockRestore();
         // Assert the actual poll's access path, avoiding hardware-dependent timing thresholds.
-        const plan = handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(api.host.id, 1, 1) as { detail: string }[];
+        const plan = handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(api.host.id, 1, 1, new Date().toISOString()) as { detail: string }[];
         expect(plan.find(row => row.detail.includes('SEARCH cancellation'))?.detail)
             .toContain('(host_id=? AND kind=?)');
     });
 
+    it.each(['accepted', 'running'])('withholds %s health work and replays its original ID after the recovery lease', async state => {
+        const api = await enrolled();
+        vi.useFakeTimers({ toFake: ['Date'] });
+        const at = Date.now();
+        api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] });
+        const busy = api.queueHostOperation(api.host.id, 'instance.inspect', {});
+        api.pendingHostOperations(api.host.id, 'health');
+        api.reportHostOperation(api.host.id, busy.id, { state });
+        const next = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'status' });
+        expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([next.id]);
+        api.reportHostOperation(api.host.id, next.id, { state: 'completed' });
+        expect(api.pendingHostOperations(api.host.id, 'health')).toEqual([]);
+        vi.setSystemTime(at + 30_001);
+        expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([busy.id]);
+        expect(api.pendingHostOperations(api.host.id, 'health')).toEqual([]);
+        api.reportHostOperation(api.host.id, busy.id, { state: 'unknown' });
+        expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([busy.id]);
+    });
+
     it('reserves health delivery while a file backlog is pending and bounds background batches', async () => {
         const api = await enrolled();
         api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] });

tests/unit/provisioning-history-performance.test.ts

@@ -14,6 +14,71 @@ afterEach(async () => {
 });
 
 describe('provisioning reads with retained lifecycle history', () => {
+    it('uses indexed per-server submissions and separate lifecycle indexes for fleet reads', async () => {
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const { latestGameSubmission, latestGameSubmissions } = await import('../../src/lib/server/fleet/lifecycle-evidence.js');
+        const { observationReader } = await import('../../src/lib/server/fleet/status.js');
+        const { runningVersions } = await import('../../src/lib/server/integrations/plutonium.js');
+        const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
+        const handle = db();
+        const now = Date.parse('2026-10-05T12:00:00Z');
+        const iso = (offset: number) => new Date(now + offset).toISOString();
+        handle.transaction(() => {
+            const server = handle.prepare('INSERT INTO servers(name, komodo_name) VALUES (?, ?)');
+            for (let i = 0; i < 16; i++) server.run(`server-${i}`, `server-${i}`);
+            server.run('empty', 'empty');
+            const insert = handle.prepare(`INSERT INTO server_deployments
+                (id, server, action, state, stage, started_at, finished_at, submitted_at, server_deleted_at, triggered_by)
+                VALUES (?, ?, ?, 'done', 'Complete', ?, ?, ?, ?, 'test')`);
+            for (let i = 0; i < 6_000; i++) {
+                const at = iso(-100_000_000 - i);
+                insert.run(`history-${i}`, `server-${i % 16}`, 'restart', at, at, at, null);
+            }
+            for (const [i, action] of ['create', 'start', 'restart', 'rebuild'].entries()) {
+                insert.run(`current-${i}`, `server-${i}`, action, iso(-1000), iso(-1000), iso(-1000), null);
+            }
+            insert.run('stop', 'server-0', 'stop', iso(0), iso(0), iso(0), null);
+            insert.run('destroy', 'server-1', 'destroy', iso(0), iso(0), iso(0), null);
+            insert.run('deleted', 'server-2', 'restart', iso(0), iso(0), iso(0), iso(0));
+            insert.run('unsubmitted', 'server-3', 'rebuild', iso(0), null, null, null);
+            handle.prepare('INSERT INTO restart_rollout(server, requested_at, healthy_at) VALUES (?, ?, ?)')
+                .run('server-0', iso(-500), iso(0));
+            handle.prepare('INSERT INTO restart_rollout(server, requested_at) VALUES (?, ?)').run('server-1', iso(-2000));
+        })();
+        const submissions = latestGameSubmissions();
+        expect(submissions.get('empty')).toBeNull();
+        expect(submissions.get('server-0')).toBe(iso(-500));
+        for (const i of [1, 2, 3]) expect(submissions.get(`server-${i}`)).toBe(iso(-1000));
+        for (const [server, at] of submissions) expect(at).toBe(latestGameSubmission(server));
+
+        // Capture real snapshot queries so plan checks catch regressions at call sites.
+        for (const read of [observationReader, runningVersions]) {
+            const prepare = vi.spyOn(handle, 'prepare');
+            read();
+            const queries = prepare.mock.calls.map(([sql]) => sql).filter(sql => sql.includes('submitted_at'));
+            prepare.mockRestore();
+            expect(queries).toHaveLength(1);
+            const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all() as { detail: string }[];
+            const details = plan.map(row => row.detail).join('\n');
+            expect(details).toContain('SEARCH server_deployments USING COVERING INDEX idx_server_deployments_submission');
+            expect(details).not.toMatch(/SCAN server_deployments\b/);
+        }
+        const prepare = vi.spyOn(handle, 'prepare');
+        fleetOperations(now);
+        const queries = prepare.mock.calls.map(([sql]) => sql).filter(sql => sql.includes('server_deployments'));
+        prepare.mockRestore();
+        expect(queries).toHaveLength(1);
+        const statement = handle.prepare(queries[0]);
+        expect(statement.columns().map(column => column.name)).toEqual(['id', 'server', 'action', 'state', 'stage', 'error', 'started_at']);
+        const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all(iso(-86_400_000)) as { detail: string }[];
+        const details = plan.map(row => row.detail).join('\n');
+        expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_state');
+        expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_finished_at');
+        expect(details).toContain('SEARCH server_deployments USING INTEGER PRIMARY KEY (rowid=?)');
+        expect(details).toContain('SCAN server_deployments USING COVERING INDEX idx_server_deployments_server');
+        expect(details).not.toMatch(/SCAN server_deployments(?:\n|$)/);
+    });
+
     it('indexes the server and queue reads used by collection matching and health polling', async () => {
         const { db } = await import('../../src/lib/server/db/index.js');
         const { latestGameSubmission } = await import('../../src/lib/server/fleet/lifecycle-evidence.js');
@@ -70,7 +135,8 @@ describe('provisioning reads with retained lifecycle history', () => {
             expect(handle.prepare('SELECT message FROM server_deployment_events').all()).toEqual([{ message: 'Creation complete' }]);
             const indexes = handle.pragma('index_list(server_deployments)') as { name: string }[];
             expect(indexes.map(index => index.name)).toEqual(expect.arrayContaining([
-                'idx_server_deployments_server', 'idx_server_deployments_state', 'idx_server_deployments_started_at'
+                'idx_server_deployments_server', 'idx_server_deployments_state', 'idx_server_deployments_started_at',
+                'idx_server_deployments_submission', 'idx_server_deployments_finished_at'
             ]));
             expect(handle.pragma('foreign_key_check')).toEqual([]);
             closeDb();

tests/unit/server-status.test.ts

@@ -75,7 +75,7 @@ describe('database fleet and container readiness', () => {
         expect(getServer('managed')).toMatchObject({ last_error: 'timeout:manual failure', last_seen_at: null, last_checked_at: new Date(at + 10_000).toISOString() });
     });
 
-    it('refreshes a ten-server fleet while one check is slow, without duplicating or exceeding four checks', async () => {
+    it('refreshes a ten-server host while one check is slow, without duplicating or exceeding two checks', async () => {
         const { db, inspect, rcon, fanoutStatus, getServer } = await setup();
         const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
         vi.useFakeTimers({ toFake: ['Date'] });
@@ -109,7 +109,7 @@ describe('database fleet and container readiness', () => {
             await vi.waitFor(() => expect(Date.parse(getServer('fast-8')?.last_seen_at ?? '')).toBeGreaterThanOrEqual(now + 30_000));
             expect(rcon).toHaveBeenCalledTimes(18);
             expect(inspect.mock.calls.filter(([name]) => name === 'managed')).toHaveLength(1);
-            expect(peak).toBe(4);
+            expect(peak).toBe(2);
             release({ ok: true, state: 'running' });
             await Promise.all([first, second]);
             expect(rcon).toHaveBeenCalledTimes(19);
@@ -120,6 +120,64 @@ describe('database fleet and container readiness', () => {
         }
     });
 
+    it.each(['native', 'komodo'])('gives each %s host two slots through inspection and RCON, independently of a stalled host', async backend => {
+        const { db, inspect, rcon, fanoutStatus, getServer } = await setup();
+        db.prepare("DELETE FROM servers WHERE name='managed'").run();
+        const hosts = await import('../../src/lib/server/hosts/index.js');
+        for (let host = 0; host < 3; host++) {
+            const registered = hosts.createHost({ name: `Host ${host}` });
+            hosts.enrollHost(hosts.createEnrollment(registered.id).token, { protocolVersion: 1 });
+            for (let instance = 0; instance < 4; instance++) {
+                const name = `host-${host}-${instance}`;
+                db.prepare('INSERT INTO servers (name, komodo_name, komodo_server) VALUES (?, ?, ?)')
+                    .run(name, name, backend === 'komodo' ? registered.id : null);
+                if (backend === 'native') db.prepare(`INSERT INTO native_instances (id,server,host_id,library_id,image,port,created_at)
+                    VALUES (?, ?, ?, 'library', 'image', ?, ?)`).run(name, name, registered.id, 28960 + instance, new Date().toISOString());
+            }
+        }
+        let releaseSlow!: () => void;
+        let releaseFast!: () => void;
+        const slow = new Promise<void>(resolve => { releaseSlow = resolve; });
+        const fast = new Promise<void>(resolve => { releaseFast = resolve; });
+        const active = new Map<string, number>();
+        const peak = new Map<string, number>();
+        inspect.mockResolvedValue({ ok: true, state: 'running' });
+        rcon.mockImplementation(async name => {
+            const host = name.slice(0, -2);
+            active.set(host, (active.get(host) ?? 0) + 1);
+            peak.set(host, Math.max(peak.get(host) ?? 0, active.get(host)!));
+            try {
+                await (host === 'host-0' ? slow : fast);
+                if (name === 'host-0-0') throw new Error('simulated failed probe');
+                return { ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'status',
+                    status: { players: 2, map: 'zombie_theater', gametype: 'zom', roster: [] } };
+            } finally { active.set(host, active.get(host)! - 1); }
+        });
+        const passes = [fanoutStatus()];
+        try {
+            await vi.waitFor(() => expect(rcon).toHaveBeenCalledTimes(6));
+            // A slot covers the whole check: waiting for RCON cannot admit a third inspection.
+            expect(inspect).toHaveBeenCalledTimes(6);
+            expect([...active.values()]).toEqual([2, 2, 2]);
+            passes.push(fanoutStatus());
+            releaseFast();
+            await vi.waitFor(() => {
+                expect(getServer('host-1-3')?.last_players).toBe(2);
+                expect(getServer('host-2-3')?.last_players).toBe(2);
+            });
+            expect(getServer('host-0-2')?.container_checked_at).toBeNull();
+            expect(rcon).toHaveBeenCalledTimes(10);
+        } finally {
+            releaseSlow(); releaseFast();
+            await Promise.all(passes);
+        }
+        expect([...peak.values()]).toEqual([2, 2, 2]);
+        expect(inspect).toHaveBeenCalledTimes(12);
+        expect(rcon).toHaveBeenCalledTimes(12);
+        expect(getServer('host-0-0')?.last_error).toBe('send-error:simulated failed probe');
+        expect(getServer('host-0-3')?.last_players).toBe(2);
+    });
+
     it('expires early results before recording a fleet collection delayed by a slow check', async () => {
         const { db, inspect, rcon, fanoutStatus } = await setup();
         vi.useFakeTimers({ toFake: ['Date'] });