Reuse fixed SQL statements for observation pruning

agent/ec_agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.11"
+VERSION = "1.4.14"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1054,7 +1054,7 @@ class Agent:
             raise AgentError("Container ownership or generation does not match")
         state = info.get("State", {})
         status = state.get("Status", "unknown")
-        healthy = state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
+        healthy = state.get("Status") == "running" and state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
         if status == "running" and not healthy:
             status = state.get("Health", {}).get("Status", "unknown")
         return {"exists": True, "running": state.get("Running") is True, "healthy": healthy, "state": status, "containerId": info.get("Id"), "instanceId": instance_id, "generation": generation, "startedAt": state.get("StartedAt"), "creationOperation": labels.get(LABEL + ".operation"), "specSha256": labels.get(LABEL + ".spec-sha256")}
@@ -1152,7 +1152,7 @@ class Agent:
                 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"
+                healthy = state.get("Status") == "running" and 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")
@@ -1368,7 +1368,7 @@ class Agent:
         def probe(identity):
             instance_id, generation = identity
             target = {"instanceId": instance_id, "generation": generation}
-            reply = {"ok": False, "reason": "timeout", "text": ""}
+            reply = {"ok": False, "reason": "deadline", "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)
@@ -1382,7 +1382,7 @@ class Agent:
                             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)
+                                reply = self.rcon(instance_id, generation, "status", deadline=deadline, status_attempts=1)
                         except (AgentError, OSError, ValueError, KeyError):
                             reply = {"ok": False, "reason": "unreachable", "text": ""}
                         finally:
@@ -1393,9 +1393,10 @@ class Agent:
 
         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}
+        # Completion describes the batch; individual failures remain in each reply.
+        return {"ok": True, "results": results}
 
-    def rcon(self, instance_id, generation, command, deadline=None):
+    def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3):
         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)
@@ -1407,11 +1408,12 @@ class Agent:
         family = socket.AF_INET6 if ":" in host else socket.AF_INET
         # 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):
+        attempts = status_attempts if command == "status" else 1
+        for attempt in range(attempts):
             if deadline is not None and time.monotonic() >= deadline:
-                return {"ok": False, "reason": "timeout", "players": None, "text": ""}
+                return {"ok": False, "reason": "deadline" if attempt == 0 else "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":
+            if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == attempts - 1:
                 return result
             time.sleep(min(0.25 * 2 ** attempt, max(0, deadline - time.monotonic())) if deadline is not None else 0.25 * 2 ** attempt)
 

agent/tests/test_agent.py

@@ -114,6 +114,12 @@ class JournalRetentionTests(unittest.TestCase):
             self.assertFalse(accepted)
             self.assertEqual(saved["id"], row["id"])
 
+    def test_completed_batch_with_failed_targets_is_pruned_after_acknowledgment(self):
+        result = {"ok": True, "results": [{"instanceId": INSTANCE, "generation": 1,
+                                         "reply": {"ok": False, "reason": "timeout", "text": ""}}]}
+        self.record("rcon.status-batch", {"instances": [{"instanceId": INSTANCE, "generation": 1}]}, result)
+        self.assertEqual(self.prune(), 1)
+
     def test_errors_failures_and_unfinished_operations_are_retained(self):
         retained = [self.record(result=result) for result in ({"ok": False}, {"healthy": False}, {"error": "partial result"}, {"error": False}, {"error": []})]
         retained.append(self.record(error="read did not finish"))
@@ -583,7 +589,7 @@ class AgentTests(unittest.TestCase):
             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")):
+        for status, running, health in (("exited", False, "healthy"), ("running", True, "unhealthy"), ("running", True, "healthy"), ("restarting", True, "healthy"), ("paused", True, "healthy")):
             info = self.resource_container()
             info["State"].update({"Status": status, "Running": running, "Health": {"Status": health}})
             observations = []
@@ -593,7 +599,9 @@ class AgentTests(unittest.TestCase):
             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.assertEqual(observations[0]["healthy"], status == "running" and running and health == "healthy")
+            self.docker.containers[self.agent.container_name(INSTANCE, 1)] = info
+            self.assertEqual(self.agent.inspect(INSTANCE, 1)["healthy"], observations[0]["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"
@@ -611,7 +619,7 @@ class AgentTests(unittest.TestCase):
         lock = threading.Lock()
         active, peak = 0, 0
 
-        def rcon(instance_id, generation, command, deadline):
+        def rcon(instance_id, generation, command, deadline, status_attempts):
             nonlocal active, peak
             with lock:
                 active += 1
@@ -620,6 +628,7 @@ class AgentTests(unittest.TestCase):
                     ready.set()
             try:
                 self.assertEqual(command, "status")
+                self.assertEqual(status_attempts, 1)
                 self.assertTrue(release.wait(2))
                 if instance_id == "00000000-0000-4000-8000-000000000002":
                     raise AgentError("local failure")
@@ -641,10 +650,47 @@ class AgentTests(unittest.TestCase):
         self.assertFalse(thread.is_alive())
         self.assertEqual(peak, 2)
         self.assertEqual(len(results[0]["results"]), 5)
-        self.assertFalse(results[0]["ok"])
+        self.assertTrue(results[0]["ok"])
         self.assertEqual(results[0]["results"][2]["reply"]["reason"], "unreachable")
         self.assertTrue(results[0]["results"][4]["reply"]["ok"])
 
+    def test_batch_uses_one_attempt_but_individual_status_still_retries(self):
+        saved = {"rconPassword": "secret", "port": 28960}
+        timeout = {"ok": False, "reason": "timeout", "text": ""}
+        operation = {"kind": "rcon.status-batch", "payload": {"instances": [{"instanceId": INSTANCE, "generation": 1}]}}
+        with patch.object(self.agent, "spec", return_value=saved), patch.object(self.agent, "rcon_attempt", return_value=timeout) as attempt, patch.object(module.time, "sleep") as sleep:
+            self.assertEqual(self.agent.dispatch(operation)["results"][0]["reply"], timeout)
+            attempt.assert_called_once()
+            sleep.assert_not_called()
+            attempt.reset_mock()
+            self.assertEqual(self.agent.rcon(INSTANCE, 1, "status"), timeout)
+            self.assertEqual(attempt.call_count, 3)
+            self.assertEqual(sleep.call_count, 2)
+            attempt.reset_mock()
+            self.assertEqual(self.agent.rcon(INSTANCE, 1, "status", deadline=0, status_attempts=1)["reason"], "deadline")
+            attempt.assert_not_called()
+
+    def test_batch_reaches_healthy_servers_after_four_unresponsive_servers(self):
+        clock = [0]
+        checked = []
+        targets = [{"instanceId": f"00000000-0000-4000-8000-{i:012d}", "generation": 1} for i in range(16)]
+
+        def attempt(family, host, port, password, command, deadline):
+            checked.append(port)
+            if port < 28964:
+                clock[0] += 3
+                return {"ok": False, "reason": "timeout", "text": ""}
+            return {"ok": True, "text": "status"}
+
+        # Serial execution is a conservative budget test; concurrency is tested separately.
+        with patch.object(module, "ThreadPoolExecutor") as pool, patch.object(module.time, "monotonic", side_effect=lambda: clock[0]), patch.object(self.agent, "spec", side_effect=lambda instance, generation: {"port": 28960 + int(instance[-12:]), "rconPassword": "secret"}), patch.object(self.agent, "rcon_attempt", side_effect=attempt), patch.object(module.time, "sleep") as sleep:
+            pool.return_value.__enter__.return_value.map.side_effect = map
+            result = self.agent.rcon_status_batch({"payload": {"instances": targets}})
+        self.assertEqual(len(checked), 16)
+        self.assertEqual(sum(row["reply"]["ok"] for row in result["results"]), 12)
+        self.assertEqual(clock[0], 12)
+        sleep.assert_not_called()
+
     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})
@@ -656,7 +702,7 @@ class AgentTests(unittest.TestCase):
             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")
+            self.assertEqual(result["results"][0]["reply"]["reason"], "deadline")
             probe.assert_not_called()
         for targets in ([], [None], [{"instanceId": INSTANCE, "generation": 1}] * 2):
             with self.assertRaises(AgentError):

docs/maintenance.md

@@ -102,9 +102,18 @@ reachable. Finished scheduled and legacy maintenance activity is retained for
 
 ## Health recovery and configuration drift
 
-The health observer (the watchdog) scans once per minute. Three observations over
-at least two minutes are required before an automatic health restart; starting or
-restarting containers get five minutes. Unknown Komodo responses are observations,
+The status monitor collects container and game health every 30 seconds. Once per
+minute, the watchdog evaluates that completed evidence without issuing its own
+routine health probes. Samples expire after 45 seconds; missing or stale evidence
+leaves health unknown and cannot justify recovery. Samples retain their original
+timestamps and instance identity, and each failure sample counts at most once.
+A monitoring gap or changed instance resets the sustained failure window. After
+an application restart, routine watchdog checks wait for a new status-monitor pass.
+**Check now** and recovery verification still request fresh evidence, and
+Provisioning retains its checks before executing recovery actions.
+
+Three observations over at least two minutes are required before an automatic
+health restart; starting or restarting containers get five minutes. Unknown Komodo responses are observations,
 not proof of a stopped container. Repeated manual checks do not accelerate this
 window. A successful routine or manual game-status probe between scans resets
 the failure streak. Missing RCON configuration, unresolved ports, and invalid

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

@@ -1,3 +1,4 @@
+import { createHash } from 'node:crypto';
 import { config } from '../config.js';
 import { db } from '../db/index.js';
 import { getHost, queueHostOperation, waitHostOperationStatus } from '../hosts/index.js';
@@ -7,6 +8,7 @@ import { rconStatusResult, agentRconStatusResult, type AgentRconReply } from '..
 import { withOperationSource } from '../hosts/operation-context.js';
 import { getServer } from './servers.js';
 import { latestGameSubmission } from './lifecycle-evidence.js';
+import { ONLINE_MAX_MS } from '../../server-freshness.js';
 
 const MAX_AGE_MS = 30_000;
 const MAX_SERVERS = 1024;
@@ -21,6 +23,7 @@ interface Entry<T> {
 }
 
 export class MonitoringChangedError extends Error {}
+export class MonitoringSkippedError extends Error {}
 
 function identity(server: string, probe: Probe): Identity {
     const row = getServer(server);
@@ -100,6 +103,62 @@ 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>>>>();
 
+interface MonitoringStatus {
+    identity: string;
+    container: Observation<Awaited<ReturnType<typeof komodoInspectHealthy>>>;
+    game?: Observation<Awaited<ReturnType<typeof rconStatusResult>>>;
+}
+// Keep failed samples too, independently of the successful-probe cache. Reading
+// this snapshot must never dispatch a request or renew an observation's age.
+const statuses = new Map<string, { snapshot?: MonitoringStatus }>();
+
+export function monitoringIdentity(server: string): string {
+    const container = identity(server, 'inspect');
+    const game = identity(server, 'rcon');
+    // Identity includes connection configuration; persist only its fingerprint.
+    return createHash('sha256').update(JSON.stringify([container.key, game.key, game.port])).digest('hex');
+}
+
+/** Capture ownership before a status pass; publish only evidence for that instance. */
+export function monitoringStatusWriter(server: string) {
+    let before: string;
+    try { before = identity(server, 'inspect').key; }
+    catch { return () => {}; }
+    const slot = { snapshot: statuses.get(server)?.snapshot };
+    statuses.delete(server);
+    statuses.set(server, slot);
+    while (statuses.size > MAX_SERVERS) statuses.delete(statuses.keys().next().value!);
+    return (container: MonitoringStatus['container'], game?: MonitoringStatus['game']) => {
+        // A forced check or newer pass supersedes this writer, even while its
+        // network requests are still in flight.
+        if (statuses.get(server) !== slot) return;
+        try {
+            if (identity(server, 'inspect').key !== before) return;
+            const current = slot.snapshot;
+            if (current && current.container.observedAt > container.observedAt) return;
+            slot.snapshot = { identity: monitoringIdentity(server), container, game };
+        } catch { statuses.delete(server); }
+    };
+}
+
+/** Latest completed status-monitor pass, with stale or superseded evidence excluded. */
+export function monitoringStatus(server: string): MonitoringStatus | null {
+    const saved = statuses.get(server)?.snapshot;
+    if (!saved) return null;
+    const fresh = (at: string) => {
+        const age = Date.now() - Date.parse(at);
+        return age >= 0 && age <= ONLINE_MAX_MS;
+    };
+    try {
+        if (saved.identity !== monitoringIdentity(server)
+            || !fresh(saved.container.observedAt)
+            || newerEvidence(identity(server, 'inspect'), Date.parse(saved.container.observedAt))) return null;
+        const game = saved.game && fresh(saved.game.observedAt)
+            && !newerEvidence(identity(server, 'rcon'), Date.parse(saved.game.observedAt)) ? saved.game : undefined;
+        return { ...saved, game };
+    } catch { return null; }
+}
+
 /** 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);
@@ -141,6 +200,7 @@ export function prepareMonitoringRcon(servers: string[]): Map<string, () => Prom
                     const operation = withOperationSource('Status monitor', () => queueHostOperation(instance.host_id, 'rcon.status-batch', {
                         instances: instances.map(value => ({ instanceId: value.id, generation: value.generation }))
                     }));
+                    // Allow delivery and receipt overhead beyond the agent's 20-second budget.
                     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 }> };
@@ -153,8 +213,12 @@ export function prepareMonitoringRcon(servers: string[]): Map<string, () => Prom
                 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.');
+                if (replies[0].reply.reason === 'deadline') throw new MonitoringSkippedError('Host status batch budget expired before this server was checked.');
                 return agentRconStatusResult(replies[0].reply, endpoint);
-            } catch (error) { return { ok: false as const, endpoint, reason: 'timeout' as const, detail: (error as Error).message }; }
+            } catch (error) {
+                if (error instanceof MonitoringSkippedError) throw 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
@@ -175,10 +239,12 @@ export function prepareMonitoringRcon(servers: string[]): Map<string, () => Prom
 
 /** Routine monitoring only. Manual diagnostics and disruption gates use fresh probes directly. */
 export function inspectForMonitoring(server: string, fresh = false) {
+    if (fresh) statuses.delete(server);
     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) {
+    if (fresh) statuses.delete(server);
     return observe(server, fresh, 'rcon', rcon, rconStatusResult);
 }

src/lib/server/hosts/index.ts

@@ -255,14 +255,17 @@ 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.');
+    if (inventory.instanceStates !== undefined) {
+        if (!Array.isArray(inventory.instanceStates) || inventory.instanceStates.length > 1000) {
+            throw new Error('Invalid instance state inventory.');
+        }
+        // A malformed server observation must not discard the host heartbeat.
+        inventory.instanceStates = inventory.instanceStates.filter(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)));
     }
     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.');
@@ -425,11 +428,11 @@ export function pruneHostObservations(): number {
         const rows = handle.prepare(`SELECT id FROM ec_host_operations WHERE observation_success=1 AND state='completed'
             AND updated_at<? ORDER BY updated_at LIMIT 500`).all(cutoff) as { id: string }[];
         if (!rows.length) return 0;
-        const ids = rows.map(row => row.id);
-        const placeholders = ids.map(() => '?').join(',');
-        handle.prepare(`DELETE FROM ec_host_operation_events WHERE operation_id IN (${placeholders})`).run(...ids);
-        handle.prepare(`DELETE FROM ec_host_operation_payloads WHERE operation_id IN (${placeholders})`).run(...ids);
-        return handle.prepare(`DELETE FROM ec_host_operations WHERE id IN (${placeholders})`).run(...ids).changes;
+        // Fixed SQL text prevents each batch size from adding three cached statements.
+        const ids = JSON.stringify(rows.map(row => row.id));
+        handle.prepare('DELETE FROM ec_host_operation_events WHERE operation_id IN (SELECT value FROM json_each(?))').run(ids);
+        handle.prepare('DELETE FROM ec_host_operation_payloads WHERE operation_id IN (SELECT value FROM json_each(?))').run(ids);
+        return handle.prepare('DELETE FROM ec_host_operations WHERE id IN (SELECT value FROM json_each(?))').run(ids).changes;
     })();
 }
 /** Heartbeats can tell an executing create to stop at its next safe command boundary. */

src/lib/server/hosts/observations.ts

@@ -11,7 +11,9 @@ export function safeOperationTarget(kind: string, value: string | null): string
 export function successfulObservation(kind: string, payload: Record<string, unknown>, result: unknown): boolean {
     const routine = observationKinds.includes(kind) || kind === 'rcon.execute' && commands.includes(String(payload.command));
     const value = result && typeof result === 'object' ? result as Record<string, unknown> : {};
-    return routine && value.ok !== false && value.healthy !== false && (value.error == null || value.error === '');
+    // Older agents set ok=false when any target fails, despite completing the batch.
+    const completedBatch = kind === 'rcon.status-batch' && Array.isArray(value.results);
+    return routine && (completedBatch || value.ok !== false) && value.healthy !== false && (value.error == null || value.error === '');
 }
 
 /** Uses the operation alias o; serialized payloads may live outside the routing row. */
@@ -22,7 +24,9 @@ export const routineObservationSql = `(CASE WHEN o.kind IN (${observationKinds.m
 
 export const successfulObservationSql = `(o.state='completed' AND COALESCE(o.error,'')='' AND ${routineObservationSql}
     AND CASE WHEN o.result IS NULL THEN 1 WHEN json_valid(o.result)
-        THEN json_extract(o.result,'$.ok') IS NOT 0 AND json_extract(o.result,'$.healthy') IS NOT 0
+        THEN (json_extract(o.result,'$.ok') IS NOT 0
+            OR (o.kind='rcon.status-batch' AND json_type(o.result,'$.results')='array'))
+            AND json_extract(o.result,'$.healthy') IS NOT 0
             AND COALESCE(json_extract(o.result,'$.error'),'')='' ELSE 0 END)`;
 
 export const importantOperationSql = `(o.routine=0 OR o.state IN ('failed','unknown')

src/lib/server/hosts/schema.ts

@@ -112,6 +112,15 @@ export function ensureHostSchema(handle: Database.Database): void {
     handle.exec(`CREATE INDEX IF NOT EXISTS ec_host_operations_routine_history ON ec_host_operations(routine,created_at);
         CREATE INDEX IF NOT EXISTS ec_host_operations_important_history ON ec_host_operations(created_at)
             WHERE ${importantOperationSql.replaceAll('o.', '')}`);
+    handle.exec('CREATE TABLE IF NOT EXISTS ec_host_schema_migrations (name TEXT PRIMARY KEY)');
+    if (!handle.prepare('SELECT 1 FROM ec_host_schema_migrations WHERE name=?').get('completed-status-batches')) {
+        handle.transaction(() => {
+            // Reclassify existing completed batches without discarding their per-server replies.
+            handle.exec(`UPDATE ec_host_operations AS o SET observation_success=1
+                WHERE o.kind='rcon.status-batch' AND o.observation_success=0 AND ${successfulObservationSql}`);
+            handle.prepare('INSERT INTO ec_host_schema_migrations(name) VALUES(?)').run('completed-status-batches');
+        })();
+    }
     const hostColumns = handle.prepare('PRAGMA table_info(ec_hosts)').all() as { name: string }[];
     if (!hostColumns.some(column => column.name === 'deleted_at')) handle.exec('ALTER TABLE ec_hosts ADD COLUMN deleted_at TEXT');
     const allowanceColumns = handle.prepare('PRAGMA table_info(ec_resource_allowances)').all() as { name: string }[];

src/lib/server/maintenance/recovery.ts

@@ -4,7 +4,8 @@ import { komodoConfigured } from '../config.js';
 import { nativeInstance } from '../backends/native.js';
 import { getServer, listServers } from '../fleet/servers.js';
 import { hasActiveProvisioning } from '../fleet/provisioning.js';
-import { inspectForMonitoring, rconForMonitoring, MonitoringChangedError } from '../fleet/health-observations.js';
+import { inspectForMonitoring, rconForMonitoring, monitoringStatus, monitoringIdentity, MonitoringChangedError } from '../fleet/health-observations.js';
+import { ONLINE_MAX_MS } from '../../server-freshness.js';
 import { latestGameSubmission, telemetryEvidence, telemetryReport } from '../fleet/status-evidence.js';
 import { prepareDriftCampaign, createDriftRebuildCampaign, previewDrift } from '../fleet/drift.js';
 import { cancelProvisioning, requestWatchdogLifecycle, escalateWatchdogRestart } from '../provisioning/requests.js';
@@ -41,28 +42,34 @@ function transition(id: string, state: WatchdogState, summary: string) {
     event(id, summary);
 }
 function detect(server: string, kind: 'health' | 'drift', summary: string, details: WatchdogEvidence): WatchdogIncident {
+    const sampledAt = kind === 'health' ? details.observedAt ?? iso() : iso();
+    const sampledTime = Date.parse(sampledAt);
     const previous = active(server, kind);
     if (previous) {
         // Repeated UI scans must not manufacture a sustained failure.
-        const newSample = Date.now() - Date.parse(previous.last_seen_at) >= WATCHDOG_SCAN_MS;
+        const sampleGap = sampledTime - Date.parse(previous.last_seen_at);
         const actionable = (details as WatchdogEvidence & { actionable?: boolean }).actionable;
-        const priorActionable = (JSON.parse(previous.details) as { actionable?: boolean }).actionable;
+        const prior = JSON.parse(previous.details) as WatchdogEvidence & { actionable?: boolean };
+        const newSample = sampleGap > 0
+            && Date.now() - Date.parse(prior.countedAt ?? previous.last_seen_at) >= WATCHDOG_SCAN_MS;
+        const priorActionable = prior.actionable;
         // A successful routine/manual probe between watchdog samples breaks
         // the failure streak, even if our next probe happens to fail again.
         const after = latestGameSubmission(server);
         const successful = [getServer(server)?.last_seen_at, telemetryReport(server)?.receivedAt]
             .some(at => at && Date.parse(at) > Date.parse(previous.last_seen_at)
-                && Date.parse(at) <= Date.now() && (!after || Date.parse(at) > Date.parse(after)));
-        const reset = kind === 'health' && (successful || (actionable === true && priorActionable !== true));
+                && Date.parse(at) <= sampledTime && (!after || Date.parse(at) > Date.parse(after)));
+        const reset = kind === 'health' && (successful || (actionable === true && priorActionable !== true)
+            || prior.identity !== details.identity || sampleGap > WATCHDOG_SCAN_MS + ONLINE_MAX_MS);
         db().prepare('UPDATE watchdog_incidents SET details=?, last_seen_at=?, failures=?, first_seen_at=? WHERE id=?')
-            .run(JSON.stringify(details), newSample || reset ? iso() : previous.last_seen_at,
-                reset ? 1 : previous.failures + (newSample ? 1 : 0), reset ? iso() : previous.first_seen_at, previous.id);
+            .run(JSON.stringify({ ...details, countedAt: newSample || reset ? iso() : prior.countedAt ?? previous.last_seen_at }), newSample || reset ? sampledAt : previous.last_seen_at,
+                reset ? 1 : previous.failures + (newSample ? 1 : 0), reset ? sampledAt : previous.first_seen_at, previous.id);
         if (previous.state === 'open') transition(previous.id, 'open', summary);
         return incident(previous.id)!;
     }
     const id = randomUUID();
     db().prepare(`INSERT INTO watchdog_incidents(id,server,kind,state,summary,details,first_seen_at,last_seen_at)
-        VALUES (?,?,?,'open',?,?,?,?)`).run(id, server, kind, summary, JSON.stringify(details), iso(), iso());
+        VALUES (?,?,?,'open',?,?,?,?)`).run(id, server, kind, summary, JSON.stringify({ ...details, countedAt: iso() }), sampledAt, sampledAt);
     event(id, summary);
     return incident(id)!;
 }
@@ -222,23 +229,29 @@ async function reconcileActions(): Promise<void> {
     }
 }
 
-interface Observation { healthy: boolean; actionable: boolean; reason: string; containerState: string | null; players?: number; source?: 'rcon' | 'telemetry' }
+interface Observation { healthy: boolean; actionable: boolean; reason: string; containerState: string | null; observedAt: string; identity: string; players?: number; source?: 'rcon' | 'telemetry' }
 async function observe(server: string, operationId?: string, fresh = false): Promise<Observation | null> {
     const epoch = latestGameSubmission(server);
     const suppressed = () => (operationId ? hasActiveProvisioning(server, operationId) : scanSuppressed(server))
         || protectedWork(server) || getServer(server)?.desired_state !== 'running' || epoch !== latestGameSubmission(server);
-    const inspected = await inspectForMonitoring(server, fresh || !!operationId).catch(error => {
+    // Routine watchdog passes only consume the status monitor. Manual checks and
+    // recovery verification still obtain fresh, independently verified evidence.
+    const snapshot = !fresh && !operationId ? monitoringStatus(server) : null;
+    if (!fresh && !operationId && !snapshot) return null;
+    const inspected = snapshot?.container ?? await inspectForMonitoring(server, true).catch(error => {
         if (error instanceof MonitoringChangedError) return null;
         throw error;
     });
     if (!inspected) return null;
     const health = inspected.result;
     if (suppressed()) return null;
+    const identity = snapshot?.identity ?? monitoringIdentity(server);
+    const containerEvidence = { identity, observedAt: inspected.observedAt };
     const state = health.state?.toLowerCase() ?? null;
-    if (!health.ok) return { healthy: false, actionable: !!state && ['stopped','exited','created','dead','unhealthy','restarting','starting'].includes(state), reason: health.reason, containerState: state };
+    if (!health.ok) return { ...containerEvidence, healthy: false, actionable: !!state && ['stopped','exited','created','dead','unhealthy','restarting','starting'].includes(state), reason: health.reason, containerState: state };
     const telemetry = telemetryEvidence(server);
-    if (telemetry) return { healthy: true, actionable: false, reason: 'Komodo healthy with fresh telemetry', containerState: state, players: telemetry.players, source: 'telemetry' };
-    const probed = await rconForMonitoring(server, fresh || !!operationId).catch(error => {
+    if (telemetry) return { identity, observedAt: telemetry.receivedAt, healthy: true, actionable: false, reason: 'Container healthy with fresh telemetry', containerState: state, players: telemetry.players, source: 'telemetry' };
+    const probed = snapshot ? snapshot.game : await rconForMonitoring(server, true).catch(error => {
         if (error instanceof MonitoringChangedError) return null;
         throw error;
     });
@@ -246,9 +259,10 @@ async function observe(server: string, operationId?: string, fresh = false): Pro
     const rcon = probed.result;
     if (suppressed()) return null;
     const latest = telemetryEvidence(server);
-    if (rcon.ok || latest) return { healthy: true, actionable: false, reason: 'Komodo healthy with fresh game status', containerState: state,
+    const gameEvidence = { identity: snapshot?.identity ?? monitoringIdentity(server), observedAt: latest?.receivedAt ?? probed.observedAt };
+    if (rcon.ok || latest) return { ...gameEvidence, healthy: true, actionable: false, reason: 'Container healthy with fresh game status', containerState: state,
         players: latest?.players ?? (rcon.ok ? rcon.status.players : 0), source: latest ? 'telemetry' : 'rcon' };
-    return { healthy: false, actionable: rcon.reason === 'timeout' || rcon.reason === 'send-error',
+    return { ...gameEvidence, healthy: false, actionable: rcon.reason === 'timeout' || rcon.reason === 'send-error',
         reason: `Game status unavailable: ${rcon.reason}; waiting for player-count verification before recovery`, containerState: state };
 }
 async function scanServer(server: string, forceScan: boolean): Promise<void> {
@@ -258,13 +272,14 @@ async function scanServer(server: string, forceScan: boolean): Promise<void> {
     const observation = await observe(server, undefined, forceScan);
     // Probes can overlap an API request; recheck ownership before queuing anything.
     if (!observation || scanSuppressed(server) || protectedWork(server) || getServer(server)?.desired_state === 'stopped') return;
+    if (observation.identity !== monitoringIdentity(server)) return;
     if (!scanSuppressed(server, 'health') && observation.healthy) {
         const current = active(server, 'health');
         if (current) transition(current.id, 'resolved', 'Fresh game status and Komodo health verified');
     } else if (!scanSuppressed(server, 'health')) {
         const problem = detect(server, 'health', observation.reason, observation);
         const startup = ['starting','restarting'].includes(observation.containerState ?? '');
-        const oldEnough = Date.now() - Date.parse(problem.first_seen_at) >= (startup ? WATCHDOG_RECOVERY_MS : WATCHDOG_FAILURE_MS);
+        const oldEnough = Date.parse(observation.observedAt) - Date.parse(problem.first_seen_at) >= (startup ? WATCHDOG_RECOVERY_MS : WATCHDOG_FAILURE_MS);
         if (problem.state === 'open' && observation.actionable && oldEnough && problem.failures >= 3 && maintenanceAllowed('health')) {
             const result = db().transaction(() => {
                 const next = requestWatchdogLifecycle(server, 'restart', problem.id);

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, prepareMonitoringRcon, MonitoringChangedError } from '../fleet/health-observations.js';
+import { inspectForMonitoring, rconForMonitoring, prepareMonitoringRcon, monitoringStatus, monitoringStatusWriter, MonitoringChangedError, MonitoringSkippedError } 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';
@@ -149,6 +149,10 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
     let updates: Partial<Record<keyof ServerRow, string | number | null>> = {};
     const flush = () => { setServerFields(srv.name, updates); updates = {}; };
     if (!isProvisioning(srv.name)) {
+        const previousGame = monitoringStatus(srv.name)?.game;
+        const publishObservation = monitoringStatusWriter(srv.name);
+        let gameObservation: Awaited<ReturnType<typeof rconForMonitoring>> | undefined;
+        const settling = restartSettling(srv.name);
         const epoch = latestGameSubmission(srv.name);
         const unchanged = () => {
             const current = getServer(srv.name);
@@ -157,7 +161,7 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
                 && current.native_instance_id === srv.native_instance_id && current.instance_generation === srv.instance_generation;
         };
         const inspectionStartedAt = new Date().toISOString();
-        const inspected = restartSettling(srv.name)
+        const inspected = settling
             ? { result: { ok: false as const, state: 'restarting', reason: 'Restart requested' }, observedAt: inspectionStartedAt }
             : await inspectForMonitoring(srv.name).catch(err => err instanceof MonitoringChangedError ? null
                 : { result: { ok: false as const, state: 'unknown', reason: (err as Error).message }, observedAt: inspectionStartedAt });
@@ -177,11 +181,19 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
             } else if (unavailable || !telemetryEvidence(srv.name)) {
                 flush();
                 const probeStartedAt = new Date().toISOString();
-                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 });
+                const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name)).catch(err => {
+                    if (err instanceof MonitoringSkippedError) {
+                        // The container was checked, but game evidence retains its original age.
+                        if (unchanged()) publishObservation(inspected, previousGame);
+                        return null;
+                    }
+                    if (err instanceof MonitoringChangedError) return null;
+                    return { result: { ok: false as const, endpoint: null, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }, observedAt: probeStartedAt };
+                });
                 if (!probe) return null;
                 const result = probe.result;
                 if (unchanged() && (getServer(srv.name)?.last_checked_at ?? '') <= probe.observedAt) {
+                    gameObservation = probe;
                     const now = probe.observedAt;
                     updates.last_checked_at = now;
                     updates.last_reply_snippet = result.rawReply ?? null;
@@ -196,6 +208,7 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
                     } else updates.last_error = encodeError(result.reason, result.detail);
                 }
             }
+            if (!settling && unchanged()) publishObservation(inspected, gameObservation);
         }
     } else {
         updates.last_error = null;

src/lib/watchdog.ts

@@ -17,6 +17,9 @@ export interface WatchdogIncident {
     operation_stage?: string | null;
 }
 export interface WatchdogEvidence {
+    observedAt?: string;
+    countedAt?: string;
+    identity?: string;
     reason?: string;
     containerState?: string | null;
     files?: DriftPlanFile[];

static/agent/ec-agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.11"
+VERSION = "1.4.14"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1054,7 +1054,7 @@ class Agent:
             raise AgentError("Container ownership or generation does not match")
         state = info.get("State", {})
         status = state.get("Status", "unknown")
-        healthy = state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
+        healthy = state.get("Status") == "running" and state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
         if status == "running" and not healthy:
             status = state.get("Health", {}).get("Status", "unknown")
         return {"exists": True, "running": state.get("Running") is True, "healthy": healthy, "state": status, "containerId": info.get("Id"), "instanceId": instance_id, "generation": generation, "startedAt": state.get("StartedAt"), "creationOperation": labels.get(LABEL + ".operation"), "specSha256": labels.get(LABEL + ".spec-sha256")}
@@ -1152,7 +1152,7 @@ class Agent:
                 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"
+                healthy = state.get("Status") == "running" and 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")
@@ -1368,7 +1368,7 @@ class Agent:
         def probe(identity):
             instance_id, generation = identity
             target = {"instanceId": instance_id, "generation": generation}
-            reply = {"ok": False, "reason": "timeout", "text": ""}
+            reply = {"ok": False, "reason": "deadline", "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)
@@ -1382,7 +1382,7 @@ class Agent:
                             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)
+                                reply = self.rcon(instance_id, generation, "status", deadline=deadline, status_attempts=1)
                         except (AgentError, OSError, ValueError, KeyError):
                             reply = {"ok": False, "reason": "unreachable", "text": ""}
                         finally:
@@ -1393,9 +1393,10 @@ class Agent:
 
         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}
+        # Completion describes the batch; individual failures remain in each reply.
+        return {"ok": True, "results": results}
 
-    def rcon(self, instance_id, generation, command, deadline=None):
+    def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3):
         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)
@@ -1407,11 +1408,12 @@ class Agent:
         family = socket.AF_INET6 if ":" in host else socket.AF_INET
         # 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):
+        attempts = status_attempts if command == "status" else 1
+        for attempt in range(attempts):
             if deadline is not None and time.monotonic() >= deadline:
-                return {"ok": False, "reason": "timeout", "players": None, "text": ""}
+                return {"ok": False, "reason": "deadline" if attempt == 0 else "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":
+            if result["ok"] or result.get("reason") not in ("timeout", "unreachable") or attempt == attempts - 1:
                 return result
             time.sleep(min(0.25 * 2 ** attempt, max(0, deadline - time.monotonic())) if deadline is not None else 0.25 * 2 ** attempt)
 

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.4.11",
+  "version": "1.4.14",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "4c6cc180f3c45a436878a0278d62b6840526150e039fdf51bafadef63759b131",
+  "sha256": "3e2e11389f412a0f1b2746c85b9afa626c5e0fcaa4801bd13f1de283abf8830e",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }

tests/unit/health-observations.test.ts

@@ -48,6 +48,74 @@ function deferred<T>() {
 }
 
 describe('shared monitoring observations', () => {
+    it('retains failures for watchdog consumption without caching them as reusable probes', async () => {
+        const api = await setup();
+        api.rcon.mockResolvedValue({ ok: false, endpoint: null, reason: 'timeout', detail: 'No reply' });
+        const publish = api.monitoringStatusWriter(server);
+        publish(await api.inspectForMonitoring(server), await api.rconForMonitoring(server));
+        expect(api.monitoringStatus(server)).toMatchObject({
+            container: { observedAt: new Date(now).toISOString(), result: healthy },
+            game: { observedAt: new Date(now).toISOString(), result: { ok: false, reason: 'timeout' } }
+        });
+        vi.setSystemTime(now + 40_000);
+        expect(api.monitoringStatus(server)?.game?.observedAt).toBe(new Date(now).toISOString());
+        expect(api.inspect).toHaveBeenCalledTimes(1);
+        expect(api.rcon).toHaveBeenCalledTimes(1);
+        vi.setSystemTime(now + 45_001);
+        expect(api.monitoringStatus(server)).toBeNull();
+        vi.setSystemTime(now - 1);
+        expect(api.monitoringStatus(server)).toBeNull();
+    });
+
+    it.each(['registration', 'generation', 'lifecycle', 'endpoint', 'manual', 'offline'] as const)('invalidates a completed status snapshot after %s changes', async change => {
+        const api = await setup();
+        const host = await native(api);
+        const publish = api.monitoringStatusWriter(server);
+        publish(await api.inspectForMonitoring(server), await api.rconForMonitoring(server));
+        expect(api.monitoringStatus(server)).not.toBeNull();
+        if (change === 'registration') api.database.prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date(now + 1).toISOString(), server);
+        if (change === 'generation') api.database.prepare('UPDATE native_instances SET generation=2').run();
+        if (change === 'lifecycle') api.database.prepare('INSERT INTO restart_rollout(server,requested_at) VALUES(?,?)').run(server, new Date(now + 1).toISOString());
+        if (change === 'endpoint') api.database.prepare('UPDATE native_instances SET port=28961').run();
+        if (change === 'manual') api.database.prepare('UPDATE servers SET container_checked_at=? WHERE name=?').run(new Date(now + 1).toISOString(), server);
+        if (change === 'offline') api.database.prepare('UPDATE ec_hosts SET last_heartbeat=? WHERE id=?').run(new Date(now - 300_000).toISOString(), host.host.id);
+        expect(api.monitoringStatus(server)).toBeNull();
+    });
+
+    it('does not publish a status pass across an instance replacement', async () => {
+        const api = await setup();
+        await native(api);
+        const publish = api.monitoringStatusWriter(server);
+        const container = await api.inspectForMonitoring(server);
+        const game = await api.rconForMonitoring(server);
+        api.database.prepare('UPDATE native_instances SET generation=2').run();
+        publish(container, game);
+        expect(api.monitoringStatus(server)).toBeNull();
+    });
+
+    it.each(['inspect', 'rcon'] as const)('invalidates the routine snapshot when a forced %s check starts', async kind => {
+        const api = await setup();
+        const publish = api.monitoringStatusWriter(server);
+        const container = await api.inspectForMonitoring(server);
+        const game = await api.rconForMonitoring(server);
+        publish(container, game);
+        expect(api.monitoringStatus(server)).not.toBeNull();
+        await (kind === 'inspect' ? api.inspectForMonitoring : api.rconForMonitoring)(server, true);
+        publish(container, game);
+        expect(api.monitoringStatus(server)).toBeNull();
+    });
+
+    it('excludes old game evidence even when the container sample is fresh', async () => {
+        const api = await setup();
+        const game = await api.rconForMonitoring(server);
+        vi.setSystemTime(now + 46_000);
+        const publish = api.monitoringStatusWriter(server);
+        publish(await api.inspectForMonitoring(server), game);
+        expect(api.monitoringStatus(server)?.container.result).toEqual(healthy);
+        expect(api.monitoringStatus(server)?.game).toBeUndefined();
+        expect(api.rcon).toHaveBeenCalledTimes(1);
+    });
+
     it.each(['inspect', 'rcon'] as const)('reuses %s results until 30 seconds after request start without refreshing their timestamp', async kind => {
         const api = await setup();
         const call = kind === 'inspect' ? api.inspectForMonitoring : api.rconForMonitoring;

tests/unit/host-observation-retention.test.ts

@@ -63,6 +63,36 @@ describe('routine host observation retention', () => {
         expect(api.pruneHostObservations()).toBe(0);
     });
 
+    it('keeps per-server failures in routine completed batches and prunes them on schedule', async () => {
+        const api = await setup();
+        const results = [{ instanceId: 'one', generation: 1, reply: { ok: false, reason: 'timeout', text: '' } }];
+        const batches = [false, true].map(ok => api.add('rcon.status-batch', { result: { ok, results } }));
+        const failed = api.add('rcon.status-batch', { state: 'failed', error: 'Batch failed' });
+        const malformed = api.add('rcon.status-batch', { result: { ok: false } });
+        expect(api.listHostOperationHistory(1, 'important').operations.map(op => op.id).sort()).toEqual([failed, malformed].sort());
+        for (const id of batches) expect(api.getHostOperation(id)?.result).toMatchObject({ results });
+        expect(api.pruneHostObservations()).toBe(2);
+        for (const id of batches) expect(api.getHostOperationStatus(id)).toBeNull();
+    });
+
+    it('reclassifies existing completed batches once without hiding operation errors', async () => {
+        const api = await setup();
+        const result = { ok: false, results: [{ instanceId: 'one', generation: 1, reply: { ok: false, reason: 'timeout' } }] };
+        const batch = api.add('rcon.status-batch', { result });
+        const retained = [api.add('rcon.status-batch', { result, error: 'Incomplete' }),
+            api.add('rcon.status-batch', { result, state: 'failed' }),
+            api.add('rcon.status-batch', { result: { ...result, error: 'Incomplete' } })];
+        api.handle.exec("UPDATE ec_host_operations SET observation_success=0; DELETE FROM ec_host_schema_migrations WHERE name='completed-status-batches'");
+        api.closeDb();
+        expect(api.listHostOperationHistory(1, 'important').operations.map(op => op.id).sort()).toEqual(retained.sort());
+        expect(api.getHostOperation(batch)?.result).toEqual(result);
+        expect(api.pruneHostObservations()).toBe(1);
+        // The migration is not a recurring scan or reclassification of retained evidence.
+        api.db().prepare("UPDATE ec_host_operations SET state='completed',error=NULL,result=? WHERE id=?").run(JSON.stringify(result), retained[0]);
+        api.closeDb();
+        expect(api.pruneHostObservations()).toBe(0);
+    });
+
     it('classifies full serialized RCON payloads and deletes their events and immutable bytes together', async () => {
         const api = await setup();
         const read = api.add('rcon.execute', { serialized: true, payload: { command: 'status' }, routing: { command: 'map_restart' } });
@@ -111,6 +141,31 @@ describe('routine host observation retention', () => {
         expect(api.handle.prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.restart'").get()).toEqual({ count: 2000 });
     });
 
+    it('reuses the same three delete statements across different batch sizes', async () => {
+        const api = await setup();
+        const statements = new Map<string, ReturnType<typeof api.handle.prepare>>();
+        for (const count of [1, 2, 7, 50, 500]) {
+            api.handle.transaction(() => {
+                const insert = api.handle.prepare(`INSERT INTO ec_host_operations
+                    (id,host_id,kind,payload,state,created_at,updated_at,observation_success)
+                    VALUES(?,?,'files.stat','{}','completed',?,?,1)`);
+                for (let i = 0; i < count; i++) insert.run(`read-${count}-${i}`, api.host.id, at(2 * day), at(2 * day));
+            })();
+            const prepare = vi.spyOn(api.handle, 'prepare');
+            expect(api.pruneHostObservations()).toBe(count);
+            const deletes = prepare.mock.calls.map(([sql]) => sql).filter(sql => sql.startsWith('DELETE FROM ec_host_'));
+            prepare.mockRestore();
+            expect(deletes).toHaveLength(3);
+            for (const sql of deletes) {
+                const statement = api.handle.prepare(sql);
+                if (statements.has(sql)) expect(statement).toBe(statements.get(sql));
+                statements.set(sql, statement);
+            }
+        }
+        expect(statements.size).toBe(3);
+        expect(api.handle.pragma('foreign_key_check')).toEqual([]);
+    });
+
     it('backfills legacy observation classification once while preserving actions and negative or invalid results', async () => {
         const api = await setup();
         const read = api.add('files.manifest');

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

@@ -47,11 +47,14 @@ async function setup(count = 5, hostCount = 1) {
 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 hosts = await import('../../src/lib/server/hosts/index.js');
+        const wait = vi.spyOn(hosts, 'waitHostOperationStatus');
         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(wait).toHaveBeenCalledWith(operation.id, 25_000);
                 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([]);
@@ -59,6 +62,10 @@ describe('host status observations', () => {
         } 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.monitoringStatus('server-2-4')).toMatchObject({
+            container: { observedAt: new Date(at).toISOString(), result: { ok: true } },
+            game: { observedAt: api.getServer('server-2-4')?.last_seen_at, result: { ok: true } }
+        });
         expect(api.db.prepare('SELECT COUNT(*) AS n FROM metric_samples').get()).toEqual({ n: 15 });
         await api.fanoutStatus();
         expect(api.operations()).toHaveLength(3);
@@ -90,6 +97,9 @@ describe('host status observations', () => {
         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.monitoringStatus('server-0-0')).toMatchObject({ container: {
+            observedAt: new Date(at).toISOString(), result: { ok: false, state: 'exited' }
+        } });
         expect(api.operations()).toEqual([]);
     });
 
@@ -108,6 +118,32 @@ describe('host status observations', () => {
         expect(api.operations()).toHaveLength(1);
     });
 
+    it('preserves previous game evidence for deadline skips while recording actual timeouts', async () => {
+        const api = await setup(3);
+        const first = api.fanoutStatus();
+        await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
+        api.complete(); await first;
+        const previous = api.getServer('server-0-0')!;
+        const previousGame = api.monitoringStatus('server-0-0')?.game;
+        vi.setSystemTime(at + 31_000);
+        const pass = api.fanoutStatus();
+        try {
+            await vi.waitFor(() => expect(api.operations()).toHaveLength(2));
+            const operation = api.operations().find(op => op.state !== 'completed')!;
+            api.reportHostOperation(operation.hostId, operation.id, { state: 'completed', result: { ok: false, results: [
+                { instanceId: 'instance-0-0', generation: 1, reply: { ok: false, reason: 'deadline', text: '' } },
+                { instanceId: 'instance-0-1', generation: 1, reply: { ok: false, reason: 'timeout', text: '' } },
+                { instanceId: 'instance-0-2', generation: 1, reply: { ok: true, text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n' } }
+            ] } });
+        } finally { api.complete(); await pass; }
+        expect(api.getServer('server-0-0')).toMatchObject({ last_checked_at: previous.last_checked_at,
+            last_seen_at: previous.last_seen_at, last_error: previous.last_error, last_players: previous.last_players,
+            last_reply_snippet: previous.last_reply_snippet });
+        expect(api.monitoringStatus('server-0-0')?.game).toEqual(previousGame);
+        expect(api.getServer('server-0-1')?.last_error).toContain('timeout:');
+        expect(Date.parse(api.getServer('server-0-2')!.last_seen_at!)).toBeGreaterThanOrEqual(at + 31_000);
+    });
+
     it('records each failed or omitted result without retrying the batch per server', async () => {
         const api = await setup();
         const pass = api.fanoutStatus();
@@ -120,6 +156,7 @@ describe('host status observations', () => {
         } 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.monitoringStatus('server-0-4')?.game?.result).toMatchObject({ ok: false, reason: 'timeout' });
         expect(api.operations()).toHaveLength(1);
     });
 

tests/unit/native-hosts.test.ts

@@ -15,6 +15,19 @@ async function enrolled() {
 }
 
 describe('native host enrollment and scoped authentication', () => {
+    it('keeps the host online and delivers work when individual state observations are invalid', async () => {
+        const api = await enrolled();
+        const good = { instanceId: 'healthy-server', generation: 1, state: 'running', healthy: true, observedAt: new Date().toISOString() };
+        const operation = api.queueHostOperation(api.host.id, 'host.inspect', {});
+        const host = api.heartbeat(api.host.id, { ...inventory, instanceStates: [good,
+            { ...good, instanceId: 'crashing', state: 'restarting' },
+            { ...good, instanceId: 'paused', state: 'paused' }, null,
+            { ...good, generation: 0 }, { ...good, observedAt: 'invalid' }] });
+        expect(host.inventory.instanceStates).toEqual([good]);
+        expect(host.readiness).toBe('ready');
+        expect(api.pendingHostOperations(api.host.id).map(row => row.id)).toContain(operation.id);
+    });
+
     it('queues serialized launch files and processes receipts without hydrating plugin payloads', async () => {
         const api = await enrolled();
         const { db } = await import('../../src/lib/server/db/index.js');

tests/unit/watchdog.test.ts

@@ -58,6 +58,10 @@ const onlyIncident = () => { expect(incidents()).toHaveLength(1); return inciden
 const operations = () => database.db().prepare('SELECT id, action, state FROM server_deployments ORDER BY rowid').all() as { id: string; action: string; state: string }[];
 function advance(ms: number) { vi.setSystemTime(Date.now() + ms); }
 function down() { vi.mocked(komodo.komodoInspectHealthy).mockResolvedValue({ ok: false, state: 'exited', reason: 'Container exited' }); }
+async function pollStatus() {
+    const worker = await vi.importActual<typeof import('../../src/lib/server/pipeline/worker.js')>('../../src/lib/server/pipeline/worker.js');
+    await worker.fanoutStatus();
+}
 async function sustainedFailure() {
     down();
     await watchdog.watchdogTick(true);
@@ -68,24 +72,148 @@ async function sustainedFailure() {
 }
 
 describe('Watchdog detection and ownership', () => {
+    it('waits for status-monitor evidence without issuing its own routine probes', async () => {
+        down();
+        for (let minute = 0; minute < 4; minute++) {
+            await watchdog.watchdogTick();
+            advance(watchdog.WATCHDOG_SCAN_MS);
+        }
+        expect(komodo.komodoInspectHealthy).not.toHaveBeenCalled();
+        expect(rcon.rconStatusResult).not.toHaveBeenCalled();
+        expect(incidents()).toEqual([]);
+        expect(operations()).toEqual([]);
+    });
+
+    it.each(['container', 'game'])('requests recovery from sustained %s failures collected by status polling', async kind => {
+        if (kind === 'container') down();
+        else vi.mocked(rcon.rconStatusResult).mockResolvedValue(missingRcon);
+        for (let minute = 0; minute < 3; minute++) {
+            await pollStatus();
+            await watchdog.watchdogTick();
+            if (minute < 2) {
+                expect(onlyIncident().failures).toBe(minute + 1);
+                expect(operations()).toEqual([]);
+                advance(watchdog.WATCHDOG_SCAN_MS);
+            }
+        }
+        expect(onlyIncident()).toMatchObject({ failures: 3, state: 'recovering' });
+        expect(operations()).toEqual([expect.objectContaining({ action: 'restart', state: 'waiting' })]);
+        expect(komodo.komodoInspectHealthy).toHaveBeenCalledTimes(3);
+        expect(rcon.rconStatusResult).toHaveBeenCalledTimes(kind === 'game' ? 3 : 0);
+    });
+
+    it('does not count the same sample again or use stale data, and resets after a monitoring gap', async () => {
+        down();
+        await pollStatus();
+        await watchdog.watchdogTick();
+        const first = onlyIncident();
+        advance(30_000);
+        database.db().prepare("DELETE FROM settings WHERE key='watchdog_last_scan'").run();
+        await watchdog.watchdogTick();
+        advance(60_000);
+        await watchdog.watchdogTick();
+        expect(onlyIncident()).toMatchObject({ failures: 1, last_seen_at: first.last_seen_at });
+        expect(komodo.komodoInspectHealthy).toHaveBeenCalledTimes(1);
+        advance(60_000);
+        await pollStatus();
+        await watchdog.watchdogTick();
+        expect(onlyIncident()).toMatchObject({ failures: 1, first_seen_at: new Date().toISOString() });
+        expect(operations()).toEqual([]);
+    });
+
+    it('uses sample times across polling offsets without counting early or losing sustained failures', async () => {
+        down();
+        await pollStatus();
+        advance(20_000);
+        await watchdog.watchdogTick();
+        advance(10_000);
+        await pollStatus();
+        advance(50_000);
+        // The first failure is stale until the monitor supplies another sample.
+        await pollStatus();
+        await watchdog.watchdogTick();
+        expect(onlyIncident().failures).toBe(2);
+        advance(40_000);
+        await pollStatus();
+        advance(20_000);
+        await watchdog.watchdogTick();
+        expect(onlyIncident()).toMatchObject({ failures: 3, state: 'recovering' });
+        expect(operations()).toHaveLength(1);
+    });
+
+    it('breaks a routine failure streak on a successful status poll between watchdog passes', async () => {
+        vi.mocked(rcon.rconStatusResult).mockResolvedValue(missingRcon);
+        await pollStatus();
+        await watchdog.watchdogTick();
+        advance(30_000);
+        vi.mocked(rcon.rconStatusResult).mockResolvedValueOnce(healthyRcon);
+        await pollStatus();
+        advance(30_000);
+        await pollStatus();
+        await watchdog.watchdogTick();
+        expect(onlyIncident()).toMatchObject({ failures: 1, first_seen_at: new Date().toISOString() });
+        expect(operations()).toEqual([]);
+        expect(rcon.rconStatusResult).toHaveBeenCalledTimes(3);
+    });
+
+    it('uses fresh telemetry alongside the polled container and waits for polled RCON when telemetry expires', async () => {
+        const { saveTelemetry } = await import('../../src/lib/server/fleet/telemetry.js');
+        saveTelemetry({ ...gameEvent('heartbeat', 17, 4, server), receivedAt: new Date().toISOString() });
+        await pollStatus();
+        await watchdog.watchdogTick();
+        expect(incidents()).toEqual([]);
+        expect(rcon.rconStatusResult).not.toHaveBeenCalled();
+        advance(watchdog.WATCHDOG_SCAN_MS);
+        await watchdog.watchdogTick();
+        expect(incidents()).toEqual([]);
+        expect(rcon.rconStatusResult).not.toHaveBeenCalled();
+        vi.mocked(rcon.rconStatusResult).mockResolvedValue(missingRcon);
+        await pollStatus();
+        advance(30_000);
+        database.db().prepare("DELETE FROM settings WHERE key='watchdog_last_scan'").run();
+        await watchdog.watchdogTick();
+        expect(onlyIncident()).toMatchObject({ failures: 1, state: 'open' });
+        expect(rcon.rconStatusResult).toHaveBeenCalledTimes(1);
+    });
+
+    it('starts a new failure window for a replacement registration', async () => {
+        down();
+        await pollStatus();
+        await watchdog.watchdogTick();
+        advance(watchdog.WATCHDOG_SCAN_MS);
+        await pollStatus();
+        await watchdog.watchdogTick();
+        expect(onlyIncident().failures).toBe(2);
+        advance(watchdog.WATCHDOG_SCAN_MS);
+        database.db().prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date().toISOString(), server);
+        await pollStatus();
+        await watchdog.watchdogTick();
+        expect(onlyIncident()).toMatchObject({ failures: 1, first_seen_at: new Date().toISOString() });
+        expect(operations()).toEqual([]);
+    });
+
     it('scans files every five minutes while retaining minute health checks and immediate manual scans', async () => {
         const drift = await import('../../src/lib/server/fleet/drift.js');
         const scan = vi.spyOn(drift, 'prepareDriftCampaign');
+        await pollStatus();
         await watchdog.watchdogTick();
         expect(scan).toHaveBeenCalledTimes(1);
         for (let minute = 1; minute < 5; minute++) {
             advance(watchdog.WATCHDOG_SCAN_MS);
+            await pollStatus();
             await watchdog.watchdogTick();
         }
         expect(scan).toHaveBeenCalledTimes(1);
         expect(komodo.komodoInspectHealthy).toHaveBeenCalledTimes(5);
         advance(watchdog.WATCHDOG_SCAN_MS);
+        await pollStatus();
         await watchdog.watchdogTick();
         expect(scan).toHaveBeenCalledTimes(2);
         await watchdog.watchdogTick(true);
         expect(scan).toHaveBeenCalledTimes(3);
         advance(watchdog.WATCHDOG_SCAN_MS);
         database.db().prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date().toISOString(), server);
+        await pollStatus();
         await watchdog.watchdogTick();
         expect(scan).toHaveBeenCalledTimes(4);
         scan.mockRestore();
@@ -94,9 +222,11 @@ describe('Watchdog detection and ownership', () => {
     it('retries failed file scans on the next health scan instead of retaining successful freshness', async () => {
         const drift = await import('../../src/lib/server/fleet/drift.js');
         const scan = vi.spyOn(drift, 'prepareDriftCampaign').mockRejectedValueOnce(new Error('Agent unavailable'));
+        await pollStatus();
         await watchdog.watchdogTick();
         expect(watchdog.watchdogOverview().scanError).toContain('Agent unavailable');
         advance(watchdog.WATCHDOG_SCAN_MS);
+        await pollStatus();
         await watchdog.watchdogTick();
         expect(scan).toHaveBeenCalledTimes(2);
         expect(watchdog.watchdogOverview().scanError).toBeNull();
@@ -112,6 +242,7 @@ describe('Watchdog detection and ownership', () => {
         expect(onlyIncident().failures).toBe(1);
         expect(operations()).toEqual([]);
         advance(watchdog.WATCHDOG_SCAN_MS);
+        await pollStatus();
         await watchdog.watchdogTick();
         expect(onlyIncident().failures).toBe(2);
         advance(watchdog.WATCHDOG_SCAN_MS - 1);