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/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/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}` : ''}`;

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/rcon.ts

@@ -367,6 +367,30 @@ export function formatRawReply(buf: Buffer): string {
     return lines.join('\n');
 }
 
+export interface AgentRconReply { ok?: boolean; reason?: string; text: string }
+
+export function agentRconStatusResult(reply: AgentRconReply, endpoint: { host: string; port: number }): RconStatusResult {
+    if (reply.ok === false) {
+        const failures: Record<string, { reason: RconFailure; detail: string }> = {
+            unconfigured: { reason: 'no-config', detail: 'Agent RCON password is not configured.' },
+            unreachable: { reason: 'send-error', detail: 'Agent could not reach the local RCON endpoint.' },
+            timeout: { reason: 'timeout', detail: 'Agent received no reply from the local RCON endpoint.' },
+            authentication: { reason: 'bad-reply', detail: 'wrong RCON password' },
+            'invalid-reply': { reason: 'bad-reply', detail: 'Agent RCON reply is missing the 0xff prefix.' }
+        };
+        return { ok: false, endpoint, ...(failures[reply.reason ?? ''] ?? {
+            reason: 'bad-reply' as const, detail: 'Agent RCON query failed.'
+        }) };
+    }
+    const players = parsePlayerCount(reply.text);
+    if (players === null) return { ok: false, endpoint, reason: 'bad-reply', detail: 'Agent returned an unrecognized RCON status.', rawReply: reply.text.slice(0, 4000) };
+    const info = parseInfoLine(reply.text); const header = parseStatusHeader(reply.text);
+    return { ok: true, endpoint, status: { players,
+        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
@@ -383,25 +407,7 @@ export async function rconStatusResult(server: string): Promise<RconStatusResult
         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) {
+                return agentRconStatusResult(reply, endpoint);
-                    const failures: Record<string, { reason: RconFailure; detail: string }> = {
-                        unconfigured: { reason: 'no-config', detail: 'Agent RCON password is not configured.' },
-                        unreachable: { reason: 'send-error', detail: 'Agent could not reach the local RCON endpoint.' },
-                        timeout: { reason: 'timeout', detail: 'Agent received no reply from the local RCON endpoint.' },
-                        authentication: { reason: 'bad-reply', detail: 'wrong RCON password' },
-                        'invalid-reply': { reason: 'bad-reply', detail: 'Agent RCON reply is missing the 0xff prefix.' }
-                    };
-                    return { ok: false, endpoint, ...(failures[reply.reason ?? ''] ?? {
-                        reason: 'bad-reply' as const, detail: 'Agent RCON query failed.'
-                    }) };
-                }
-                const players = parsePlayerCount(reply.text);
-                if (players === null) return { ok: false, endpoint, reason: 'bad-reply', detail: 'Agent returned an unrecognized RCON status.', rawReply: reply.text.slice(0, 4000) };
-                const info = parseInfoLine(reply.text); const header = parseStatusHeader(reply.text);
-                return { ok: true, endpoint, status: { players,
-                    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) };
             } 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';
@@ -96,7 +96,7 @@ function drainStatusQueue(host: string, queue: StatusQueue): void {
     if (!queue.active && !queue.queued.length) statusQueues.delete(host);
 }
 
-function queueStatus(srv: ServerRow): 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;
@@ -110,7 +110,7 @@ function queueStatus(srv: ServerRow): Promise<number | null> {
     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));
@@ -124,7 +124,9 @@ function queueStatus(srv: ServerRow): Promise<number | null> {
 export async function fanoutStatus(): Promise<void> {
     const servers = listServers();
     if (!servers.length) return;
-    const results = await Promise.all(servers.map(queueStatus));
+    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
@@ -140,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;
@@ -175,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/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);
+    });
+});