Reuse serialized server events across SSE connections

agent/ec_agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.10"
+VERSION = "1.4.11"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -918,6 +918,8 @@ class Agent:
         self.observation_keep_days = 0  # Older controllers do not authorize cleanup.
         self.library_progress = {}
         self.library_transfers = {}
+        self.inventory_manifest_cache = {}
+        self.inventory_manifest_lock = threading.Lock()
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
         self.executing_id = None
 
@@ -1191,6 +1193,25 @@ class Agent:
                 continue
         return samples
 
+    def inventory_manifest(self, marker):
+        # Inventory only: launch/download verification must still read disk.
+        # Include file identity so atomic replacements invalidate even when
+        # their size and modification time match the previous manifest.
+        with self.inventory_manifest_lock:
+            try:
+                metadata = marker.stat()
+                signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
+                cached = self.inventory_manifest_cache.get(marker)
+                if cached is not None and cached[0] == signature:
+                    return cached[1]
+                self.inventory_manifest_cache.pop(marker, None)
+                value = json.loads(marker.read_text())
+                self.inventory_manifest_cache[marker] = (signature, value)
+                return value
+            except (OSError, ValueError):
+                self.inventory_manifest_cache.pop(marker, None)
+                raise
+
     def inventory(self):
         cpu_usage = self.cpu_usage_percent()
         usage = shutil.disk_usage(self.root)
@@ -1210,13 +1231,18 @@ class Agent:
         libraries, caches = [], []
         progress = {key: value.copy() for key, value in list(self.library_progress.items())}
         progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
-        for marker in (self.root / "libraries").glob("*/*/manifest.json"):
+        manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
+        partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
+        with self.inventory_manifest_lock:
+            for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
+                del self.inventory_manifest_cache[marker]
+        for marker in manifests:
             # A crash between writing a manifest and its atomic promotion must
             # never expose the staging directory as a ready library.
             if marker.parent.name.startswith("."):
                 continue
             try:
-                value = json.loads(marker.read_text())
+                value = self.inventory_manifest(marker)
                 # Rechecking an existing cache must revoke readiness until its
                 # entire contents have passed verification again.
                 if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
@@ -1233,9 +1259,9 @@ class Agent:
             except (OSError, ValueError, KeyError):
                 continue
         ready = {item["id"] for item in libraries}
-        for marker in (self.root / "libraries").glob("*/.*.partial/source.json"):
+        for marker in partial_manifests:
             try:
-                value = json.loads(marker.read_text())
+                value = self.inventory_manifest(marker)
                 library_id = identifier(value["libraryId"])
                 if library_id in ready or library_id in progress:
                     continue

agent/tests/test_agent.py

@@ -386,6 +386,68 @@ class AgentTests(unittest.TestCase):
         self.agent.journal.close()
         self.temp.cleanup()
 
+    def test_inventory_reuses_manifests_until_mtime_changes(self):
+        marker = self.library / "manifest.json"
+        saved = json.loads(marker.read_text())
+        original_read = Path.read_text
+        reads = []
+
+        def read(path, *args, **kwargs):
+            if path == marker:
+                reads.append(path)
+            return original_read(path, *args, **kwargs)
+
+        with patch.object(Path, "read_text", read):
+            self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
+            self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9)
+            self.assertEqual(len(reads), 1)
+            metadata = marker.stat()
+            marker.write_text(json.dumps({**saved, "size": 8}))
+            module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns + 1_000_000_000))
+            self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
+            self.assertEqual(len(reads), 2)
+
+    def test_inventory_reloads_atomic_replacement_with_same_size_and_mtime(self):
+        marker = self.library / "manifest.json"
+        saved = json.loads(marker.read_text())
+        self.agent.inventory()
+        metadata = marker.stat()
+        module.atomic_json(marker, {**saved, "size": 8})
+        module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
+        self.assertEqual(marker.stat().st_size, metadata.st_size)
+        self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
+
+    def test_inventory_discards_invalid_and_removed_manifests(self):
+        marker = self.library / "manifest.json"
+        saved = json.loads(marker.read_text())
+        self.agent.inventory()
+        marker.write_text("invalid JSON")
+        self.assertEqual(self.agent.inventory()["libraryCaches"], [])
+        self.assertNotIn(marker, self.agent.inventory_manifest_cache)
+        module.atomic_json(marker, saved)
+        self.assertEqual(len(self.agent.inventory()["libraryCaches"]), 1)
+        marker.unlink()
+        self.assertEqual(self.agent.inventory()["libraryCaches"], [])
+        self.assertNotIn(marker, self.agent.inventory_manifest_cache)
+        module.atomic_json(marker, {**saved, "size": 8})
+        self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8)
+
+    def test_cached_inventory_preserves_transfer_progress_and_integrity_checks(self):
+        self.agent.inventory()
+        progress = {"libraryVersionId": self.library_id, "status": "verifying", "bytesVerified": 3}
+        module.atomic_json(self.library / "transfer.json", progress)
+        self.assertEqual(self.agent.inventory()["libraryCaches"][0]["status"], "missing")
+        module.atomic_json(self.library / "transfer.json", {**progress, "status": "failed", "bytesVerified": 5})
+        cache = self.agent.inventory()["libraryCaches"][0]
+        self.assertEqual(cache["status"], "failed")
+        self.assertEqual(cache["bytesVerified"], 5)
+        (self.library / "content/game.ff").write_bytes(b"bad-data!")
+        with self.assertRaises(AgentError):
+            self.agent.verify_library(self.library)
+        (self.library / "manifest.json").write_text("invalid JSON")
+        with self.assertRaises(ValueError):
+            self.agent.verify_library(self.library)
+
     def test_library_remove_preserves_other_versions_and_is_idempotent(self):
         other = self.agent.library_dir("t5", "other-library")
         other.mkdir()
@@ -1179,6 +1241,17 @@ class AgentTests(unittest.TestCase):
         self.assertEqual(caches[0]["status"], "missing")
         self.assertEqual(caches[0]["bytesDownloaded"], len(content))
         self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"])
+        (staging / "content/zone/test.ff").write_bytes(content[:5])
+        original_read = Path.read_text
+
+        def read(path, *args, **kwargs):
+            if path == staging / "source.json":
+                self.fail("Unchanged partial manifest should be cached")
+            return original_read(path, *args, **kwargs)
+
+        with patch.object(Path, "read_text", read):
+            partial = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
+        self.assertEqual(partial["bytesDownloaded"], 5)
         self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3}
         caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"]
         self.assertEqual(len(caches), 1)

docs/native-hosts.md

@@ -595,6 +595,10 @@ 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.11 caches parsed library manifests during inventory collection until
+their modification time, size, or file identity changes. Deleted manifests leave
+the cache, while transfer progress and partial-file sizes are sampled each time.
+Launch and download integrity verification continue to read from disk.
 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/hooks.server.ts

@@ -3,9 +3,10 @@ import { migrate } from '$lib/server/db/migrate.js';
 import { startWorker } from '$lib/server/pipeline/worker.js';
 import { interruptDeployments } from '$lib/server/provisioning/history.js';
 import { authenticationHandle } from '$lib/server/auth/guard.js';
-import { emit } from '$lib/server/pipeline/bus.js';
+import { invalidateFleet } from '$lib/server/pipeline/bus.js';
 import { startPerformanceMonitor } from '$lib/server/performance.js';
 import { withOperationSource } from '$lib/server/hosts/operation-context.js';
+import { withHostRequestContext } from '$lib/server/hosts/request-context.js';
 
 let booted = false;
 function boot(): void {
@@ -23,13 +24,13 @@ function boot(): void {
 export const handle: Handle = async (input) => {
     boot();
     const started = performance.now();
-    const response = await withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`,
-        () => authenticationHandle(input));
+    const response = await withHostRequestContext(() => withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`,
+        () => authenticationHandle(input)));
     const elapsed = performance.now() - started;
     response.headers.append('Server-Timing', `app;dur=${elapsed.toFixed(1)}`);
     if (process.env.EC_PERF_LOG === 'true' && elapsed > 250 && !input.event.route.id?.startsWith('/api/agent/')) {
         console.info('[request]', { route: input.event.route.id, method: input.event.request.method, status: response.status, elapsedMs: Math.round(elapsed) });
     }
-    if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) emit({ type: 'fleet-invalidated' });
+    if (input.event.locals.user && !['GET','HEAD','OPTIONS'].includes(input.event.request.method) && response.status < 400) invalidateFleet();
     return response;
 };

src/lib/server/db/index.ts

@@ -391,6 +391,21 @@ export function db(): Database.Database {
         throw err;
     }
 
+    // Cache only runtime queries, after migrations have finished. Statements
+    // belong to this connection, so reopening the DB starts with an empty cache.
+    // Callers share statement options (e.g. raw()); use consistent options for
+    // identical SQL and pass values to run/get/all rather than binding permanently.
+    const statements = new Map<string, Database.Statement>();
+    const prepare = handle.prepare.bind(handle);
+    handle.prepare = ((sql: string) => {
+        let statement = statements.get(sql);
+        if (!statement) {
+            statement = prepare(sql);
+            statements.set(sql, statement);
+        }
+        return statement;
+    }) as typeof handle.prepare;
+
     _db = handle;
     return _db;
 }

src/lib/server/fleet/overview.ts

@@ -2,19 +2,31 @@ import { actionLabels, type ProvisioningAction } from '../../provisioning.js';
 import { db } from '../db/index.js';
 import type { FleetOperation } from '../../fleet-overview.js';
 
-/** Read persisted work, including recent completions for one-shot lobby watches. */
+/** Fleet clients need only active work and unresolved failures. */
 export function fleetOperations(now = Date.now()): FleetOperation[] {
+    return readOperations(now);
+}
+
+/** Recent completions are loaded only by the server's own page. */
+export function serverOperations(server: string, now = Date.now()): FleetOperation[] {
+    return readOperations(now, server);
+}
+
+function readOperations(now: number, server?: string): FleetOperation[] {
     const since = new Date(now - 86_400_000).toISOString();
+    const bindings = { server: server ?? null, since };
+    const scope = (column: string) => server === undefined ? '' : `AND ${column} = @server`;
     const handle = db();
     const operations: FleetOperation[] = [];
-    const slots = handle.prepare('SELECT server, requested_at, error FROM restart_rollout WHERE healthy_at IS NULL')
-        .all() as { server: string; requested_at: string; error: string | null }[];
+    const slots = handle.prepare(`SELECT server, requested_at, error FROM restart_rollout WHERE healthy_at IS NULL ${scope('server')}`)
+        .all(bindings) as { server: string; requested_at: string; error: string | null }[];
     const recovering = new Map(slots.map(slot => [slot.server, slot]));
     const representedRecovery = new Set<string>();
     const campaigns = handle.prepare(`SELECT d.id, d.action, d.started_at, ds.server,
         ds.restart_state, ds.error, ds.restarted_at, ds.players
         FROM deployments d JOIN deployment_servers ds ON ds.deployment_id = d.id
-        WHERE ds.server_deleted_at IS NULL AND (ds.restart_state != 'done' OR ds.checked_at >= ? OR d.finished_at >= ? OR d.started_at >= ?)`).all(since, since, since) as {
+        WHERE ds.server_deleted_at IS NULL ${scope('ds.server')} AND (ds.restart_state != 'done'
+            ${server === undefined ? '' : 'OR ds.checked_at >= @since OR d.finished_at >= @since OR d.started_at >= @since'})`).all(bindings) as {
         id: string; action: string; started_at: string; server: string; restart_state: string;
         error: string | null; restarted_at: string | null; players: number | null;
     }[];
@@ -34,19 +46,19 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
             href: `/campaigns/${encodeURIComponent(row.id)}` });
     }
     // Only the newest lifecycle operation can still require action on a server.
-    // Separate branches let SQLite use state and finish indexes; UNION deduplicates
-    // recent warnings that are also the newest operation. Unary + keeps terminal
-    // branches on finished_at and rowid instead of searching historical states.
+    // Separate branches use state/rowid indexes for fleet reads and the server
+    // index for detail history. Completed work is never loaded for the full fleet.
     const lifecycle = handle.prepare(`SELECT id, server, action, state, stage, error, started_at FROM server_deployments
-        WHERE server_deleted_at IS NULL AND state IN ('queued', 'running', 'waiting', 'awaiting-health')
+        WHERE server_deleted_at IS NULL ${scope('server')} AND state IN ('queued', 'running', 'waiting', 'awaiting-health')
+        ${server === undefined ? '' : `UNION
+        SELECT id, server, action, state, stage, error, started_at FROM server_deployments
+        WHERE server_deleted_at IS NULL AND server = @server
+            AND +state IN ('done', 'warning', 'cancelled', 'failed', 'interrupted') AND finished_at >= @since`}
         UNION
         SELECT id, server, action, state, stage, error, started_at FROM server_deployments
-        WHERE server_deleted_at IS NULL AND +state IN ('done', 'warning', 'cancelled') AND finished_at >= ?
-        UNION
-        SELECT id, server, action, state, stage, error, started_at FROM server_deployments
-        WHERE server_deleted_at IS NULL AND +state IN ('failed', 'interrupted', 'warning')
-            AND rowid IN (SELECT MAX(rowid) FROM server_deployments GROUP BY server)`)
-        .all(since) as { id: string; server: string; action: string; state: string; stage: string;
+        WHERE server_deleted_at IS NULL ${scope('server')} AND +state IN ('failed', 'interrupted')
+            AND rowid IN (SELECT MAX(rowid) FROM server_deployments WHERE 1=1 ${scope('server')} GROUP BY server)`)
+        .all(bindings) as { id: string; server: string; action: string; state: string; stage: string;
             error: string | null; started_at: string }[];
     for (const row of lifecycle) {
         if (row.state === 'awaiting-health') representedRecovery.add(row.server);
@@ -59,8 +71,8 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
             detail: row.error ?? row.stage, href: `/provisioning/${encodeURIComponent(row.id)}` });
     }
     const betas = handle.prepare(`SELECT b.id, b.opened_at, bs.server, bs.stage, bs.error
-        FROM betas b JOIN beta_servers bs ON bs.beta_id = b.id WHERE bs.server_deleted_at IS NULL AND b.state IN ('running', 'ready-promote')`)
-        .all() as { id: number; opened_at: string; server: string; stage: string; error: string | null }[];
+        FROM betas b JOIN beta_servers bs ON bs.beta_id = b.id WHERE bs.server_deleted_at IS NULL ${scope('bs.server')} AND b.state IN ('running', 'ready-promote')`)
+        .all(bindings) as { id: number; opened_at: string; server: string; stage: string; error: string | null }[];
     const stages: Record<string, string> = { queued: 'Waiting for an earlier test wave', deploying: 'Deploying beta',
         'awaiting-empty': 'Waiting for an empty lobby', restarting: 'Restarting', 'awaiting-health': 'Recovering',
         'awaiting-match': 'Observing a match', ready: 'Ready for review', failed: 'Beta needs attention' };
@@ -71,9 +83,9 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
             state: row.stage === 'failed' ? 'attention' : 'active', label: stages[row.stage] ?? row.stage,
             detail: row.error ?? recovering.get(row.server)?.error ?? `Beta ${row.id}`, href: `/betas/${row.id}` });
     }
-    const maintenance = handle.prepare(`SELECT m.* FROM maintenance_jobs m WHERE m.server_deleted_at IS NULL AND m.state = 'failed'
+    const maintenance = handle.prepare(`SELECT m.* FROM maintenance_jobs m WHERE m.server_deleted_at IS NULL ${scope('m.server')} AND m.state = 'failed'
         AND NOT EXISTS (SELECT 1 FROM maintenance_jobs n WHERE n.server = m.server AND n.kind = m.kind AND n.id > m.id)`)
-        .all() as { id: number; server: string; kind: string; detail: string | null; requested_at: string }[];
+        .all(bindings) as { id: number; server: string; kind: string; detail: string | null; requested_at: string }[];
     for (const job of maintenance) {
         operations.push({ key: `maintenance:${job.id}`, group: `maintenance:${job.id}`, kind: 'maintenance',
             server: job.server, state: 'attention', startedAt: job.requested_at,

src/lib/server/fleet/snapshot.ts

@@ -1,26 +1,32 @@
 import { db } from '../db/index.js';
-import { listServers } from './servers.js';
-import { observationReader } from './status.js';
+import { getServer, listServers } from './servers.js';
+import { observationReader, serverObservation } from './status.js';
 import { config, komodoConfigured, rconConfigured } from '../config.js';
 import { runningVersions } from '../integrations/plutonium.js';
 import { listHosts } from '../hosts/index.js';
 import type { NativeInstance } from '../backends/native.js';
+import { serverOperations } from './overview.js';
 
-export function fleetSnapshot() {
+export function fleetSnapshot(serverName?: string) {
+    const bindings = serverName === undefined ? [] : [serverName];
     const rconHost = config.rcon.host;
     const hosts = listHosts();
     const hostsById = new Map(hosts.map(({ inventory: _inventory, ...host }) => [host.id, host]));
-    const versions = new Map(runningVersions().map(version => [version.server, version]));
-    const instances = new Map((db().prepare('SELECT * FROM native_instances WHERE retired_at IS NULL').all() as NativeInstance[]).map(instance => [instance.server, instance]));
+    const versions = new Map(runningVersions(serverName).map(version => [version.server, version]));
+    const instances = new Map((db().prepare(`SELECT * FROM native_instances WHERE retired_at IS NULL ${serverName === undefined ? '' : 'AND server = ?'}`)
+        .all(...bindings) as NativeInstance[]).map(instance => [instance.server, instance]));
     const holds = new Map<string, { target: string; betaId: number }[]>();
     for (const row of db().prepare(`SELECT bsv.server, s.target_rel_path AS target, b.id AS betaId
         FROM beta_servers bsv JOIN betas b ON b.id=bsv.beta_id JOIN beta_scripts bs ON bs.beta_id=b.id
-        JOIN scripts s ON s.id=bs.script_id WHERE bsv.server_deleted_at IS NULL AND b.state IN ('running','ready-promote')`).all() as
+        JOIN scripts s ON s.id=bs.script_id WHERE bsv.server_deleted_at IS NULL AND b.state IN ('running','ready-promote')
+        ${serverName === undefined ? '' : 'AND bsv.server = ?'}`).all(...bindings) as
         { server: string; target: string; betaId: number }[]) {
         const entries = holds.get(row.server) ?? []; entries.push({ target: row.target, betaId: row.betaId }); holds.set(row.server, entries);
     }
-    const observe = observationReader();
-    const servers = listServers().map((s) => {
+    const observe = serverName === undefined ? observationReader() : serverObservation;
+    const selected = serverName === undefined ? undefined : getServer(serverName);
+    const rows = serverName === undefined ? listServers() : selected && !selected.destroyed_at ? [selected] : [];
+    const servers = rows.map((s) => {
         const instance = s.native_instance_id ? instances.get(s.name) : null;
         return {
             ...observe(s),
@@ -36,7 +42,7 @@ export function fleetSnapshot() {
     return {
         servers,
         serverNow: Date.now(),
-        reservedNames: (db().prepare("SELECT j.server FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id WHERE d.state IN ('queued', 'running', 'awaiting-health')").all() as { server: string }[]).map(row => row.server),
+        reservedNames: serverName === undefined ? (db().prepare("SELECT j.server FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id WHERE d.state IN ('queued', 'running', 'awaiting-health')").all() as { server: string }[]).map(row => row.server) : [],
         integrations: {
             native: hosts.length > 0 || !komodoConfigured(),
             hostsReady: hosts.filter(host => host.readiness === 'ready').length,
@@ -49,3 +55,6 @@ export function fleetSnapshot() {
     };
 }
 
+export function serverSnapshot(name: string) {
+    return { ...fleetSnapshot(name), operations: serverOperations(name) };
+}

src/lib/server/game-files/index.ts

@@ -332,9 +332,17 @@ function transferProgress(value: unknown, status: LibraryCacheState): LibraryTra
         bytesPerSecond: progress.bytesPerSecond as number | null ?? null, etaSeconds: progress.etaSeconds as number | null ?? null };
 }
 
+type LibraryCacheMetadata = Pick<GameLibraryVersion, 'state' | 'sizeBytes' | 'manifestSha256'>;
+const publishedCacheMetadata = new WeakMap<ReturnType<typeof database>, Map<string, LibraryCacheMetadata>>();
+
 export function updateHostLibraryCaches(hostId: string, input: unknown, options: { snapshot?: boolean } = {}): void {
     if (!Array.isArray(input) || input.length > 2000) throw new Error('Invalid host library report.');
     const handle = database();
+    // Published metadata is immutable. Keep caches local to the connection and
+    // avoid retaining a publication from an outer transaction that may roll back.
+    const versions = handle.inTransaction ? new Map<string, LibraryCacheMetadata>()
+        : publishedCacheMetadata.get(handle) ?? new Map<string, LibraryCacheMetadata>();
+    if (!handle.inTransaction) publishedCacheMetadata.set(handle, versions);
     const now = new Date().toISOString();
     handle.transaction(() => {
         const reported = new Set<string>();
@@ -344,7 +352,10 @@ export function updateHostLibraryCaches(hostId: string, input: unknown, options:
             if (typeof id !== 'string') throw new Error('Invalid library version ID.');
             if (reported.has(id)) throw new Error('Duplicate version in host cache report.');
             reported.add(id);
-            const version = getLibraryVersion(id); if (!version || version.state !== 'ready') continue;
+            const version = versions.get(id) ?? handle.prepare(`SELECT state, size_bytes AS sizeBytes, manifest_sha256 AS manifestSha256
+                FROM game_file_versions WHERE id=?`).get(id) as LibraryCacheMetadata | undefined;
+            if (!version || version.state !== 'ready') continue;
+            versions.set(id, version);
             let status = raw.status ?? raw.state;
             if (!['missing', 'downloading', 'verifying', 'ready', 'failed', 'outdated'].includes(status)) throw new Error('Invalid library cache state.');
             const checksum = typeof raw.manifestSha256 === 'string' ? raw.manifestSha256 : null;

src/lib/server/hosts/index.ts

@@ -6,6 +6,7 @@ import { db } from '../db/index.js';
 import { ensureHostSchema } from './schema.js';
 import { successfulObservation, routineObservationSql, importantOperationSql, operationTargetSql, safeOperationTarget } from './observations.js';
 import { currentOperationSource } from './operation-context.js';
+import { parseHostInventory } from './request-context.js';
 import { fleetSettings } from '../settings.js';
 import { recordInstanceResources, validateInstanceResources } from './calibration.js';
 import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js';
@@ -38,7 +39,7 @@ function presentDecommission(row: DecommissionRow | undefined): HostDecommission
 }
 function hostFromRow(row: HostRow): Host {
     const enabling = database().prepare('SELECT requested_at FROM ec_host_enable_requests WHERE host_id=?').get(row.id) as { requested_at: string } | undefined;
-    const inventory: HostInventory = JSON.parse(row.inventory);
+    const inventory = parseHostInventory(row.id, row.inventory);
     const connected = !!row.credential_active && !!row.last_heartbeat && Date.now() - Date.parse(row.last_heartbeat) <= HEARTBEAT_MAX_AGE_MS;
     const reasons: string[] = [];
     if (!row.enrolled_at) reasons.push('Agent has not enrolled.');
@@ -304,9 +305,15 @@ export function enrollHost(token: unknown, rawInventory: unknown): { hostId: str
 }
 export function authenticateCredential(token: string): Host | null {
     if (!token || token.length > 200) return null;
-    const row = database().prepare('SELECT id,host_id FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL').get(digest(token)) as { id: string; host_id: string } | undefined;
+    const row = database().prepare('SELECT id,host_id,last_used_at FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL')
+        .get(digest(token)) as { id: string; host_id: string; last_used_at: string | null } | undefined;
     if (!row) return null;
-    database().prepare('UPDATE ec_host_credentials SET last_used_at=? WHERE id=?').run(now(), row.id);
+    const timestamp = Date.now();
+    if (row.last_used_at === null || timestamp - Date.parse(row.last_used_at) >= 60_000) {
+        database().prepare(`UPDATE ec_host_credentials SET last_used_at=? WHERE id=? AND revoked_at IS NULL
+            AND (last_used_at IS NULL OR last_used_at<=?)`)
+            .run(new Date(timestamp).toISOString(), row.id, new Date(timestamp - 60_000).toISOString());
+    }
     return getHost(row.host_id);
 }
 export function revokeCredentials(hostId: string): void {

src/lib/server/hosts/request-context.ts

@@ -0,0 +1,28 @@
+import { AsyncLocalStorage } from 'node:async_hooks';
+import type { HostInventory } from '../../hosts.js';
+
+type InventoryCache = Map<string, { json: string; inventory: HostInventory }>;
+const requests = new AsyncLocalStorage<{ inventories: InventoryCache | null }>();
+
+export function withHostRequestContext<T>(work: () => T | Promise<T>): Promise<T> {
+    const context = { inventories: new Map() as InventoryCache | null };
+    return requests.run(context, async () => {
+        try { return await work(); }
+        finally {
+            // Detached work can inherit this context. Release request data and
+            // stop caching when the response is ready, even if that work continues.
+            context.inventories = null;
+        }
+    });
+}
+
+export function parseHostInventory(hostId: string, json: string): HostInventory {
+    const cache = requests.getStore()?.inventories;
+    const saved = cache?.get(hostId);
+    if (saved?.json === json) return saved.inventory;
+    const inventory: HostInventory = JSON.parse(json);
+    // Compare the stored JSON so heartbeats and transaction rollbacks within
+    // a request remain visible. Other host state is always read afresh.
+    cache?.set(hostId, { json, inventory });
+    return inventory;
+}

src/lib/server/integrations/plutonium.ts

@@ -20,13 +20,18 @@ export function freshRelease(): number | null {
     const age = latest.checkedAt ? Date.now() - Date.parse(latest.checkedAt) : NaN;
     return !latest.error && age >= 0 && age < HOUR ? latest.revision : null;
 }
-export function runningVersions(): PlutoniumVersion[] {
-    const versions = db().prepare('SELECT * FROM plutonium_versions').all() as PlutoniumVersion[];
+export function runningVersions(serverName?: string): PlutoniumVersion[] {
+    const bindings = serverName === undefined ? [] : [serverName];
+    const versions = db().prepare(`SELECT * FROM plutonium_versions ${serverName === undefined ? '' : 'WHERE server = ?'}`)
+        .all(...bindings) as PlutoniumVersion[];
     const target = freshRelease();
     const byServer = new Map(versions.map(version => [version.server, version]));
-    const epochs = latestGameSubmissions();
-    const recovering = new Set((db().prepare('SELECT server FROM restart_rollout WHERE healthy_at IS NULL').all() as { server: string }[]).map(row => row.server));
-    return listServers().map(server => {
+    const epochs = serverName === undefined ? latestGameSubmissions() : new Map([[serverName, latestGameSubmission(serverName)]]);
+    const recovering = new Set((db().prepare(`SELECT server FROM restart_rollout WHERE healthy_at IS NULL ${serverName === undefined ? '' : 'AND server = ?'}`)
+        .all(...bindings) as { server: string }[]).map(row => row.server));
+    const selected = serverName === undefined ? undefined : getServer(serverName);
+    const servers = serverName === undefined ? listServers() : selected && !selected.destroyed_at ? [selected] : [];
+    return servers.map(server => {
         const value = byServer.get(server.name)
             ?? { server: server.name, revision: null, checked_at: null, error: null };
         const age = value.checked_at ? Date.now() - Date.parse(value.checked_at) : NaN;

src/lib/server/pipeline/sweep.ts

@@ -7,6 +7,7 @@ import { pruneHostObservations } from '../hosts/index.js';
 interface PendingRow {
     deployment_id: string; server: string; restart_state: string;
     restarted_at: string | null; files_changed: number;
+    players: number | null; error: string | null;
 }
 type SweepResult = { checked: number; restarted: number; healthy: number; held: number; failed: number };
 let sweepInFlight: Promise<SweepResult> | null = null;
@@ -39,8 +40,10 @@ async function runSweep(): Promise<SweepResult> {
             attempts = attempts + CASE WHEN ? IS NOT NULL AND checked_at IS NOT ? THEN 1 ELSE 0 END,
             restarted_at = COALESCE(restarted_at, ?) WHERE deployment_id = ? AND server = ?`)
             .run(state, operation.checked_at, operation.players, error, operation.checked_at, operation.checked_at, restartedAt, row.deployment_id, row.server);
+        if (state !== row.restart_state || operation.players !== row.players || error !== row.error) {
             emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error });
         }
+    }
     db().prepare(`UPDATE deployments SET finished_at = ? WHERE finished_at IS NULL AND NOT EXISTS (
         SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND server_deleted_at IS NULL AND restart_state IN ('pending','failed','held'))`).run(nowIso());
     db().prepare(`DELETE FROM deployments WHERE started_at < ?

src/lib/ui/stores/fleet.ts

@@ -6,12 +6,14 @@ import { browser } from '$app/environment';
 import { statusByServer, statusReceivedAt, syncServerTime, serverTimeNow, matchEventsByServer, telemetryByServer, uiNow } from './servers.js';
 import { type FleetServer, type FleetOperation } from '$lib/fleet-overview.js';
 
-export const fleet = writable<{ servers: FleetServer[]; operations: FleetOperation[]; receivedAt: number }>({
+// Each server detail view owns a scoped snapshot; it must not replace the shared fleet.
+export function createFleetStore(serverName?: string) {
+    const fleet = writable<{ servers: FleetServer[]; operations: FleetOperation[]; receivedAt: number }>({
         servers: [], operations: [], receivedAt: 0
     });
-export const fleetError = writable('');
-export const fleetRefreshing = writable(false);
-export const fleetIntegrations = writable<{ rcon: boolean; komodo: boolean; native?: boolean; hostsReady?: number; hostsTotal?: number; rconTimeoutMs: number; defaultRconHost: string } | null>(null);
+    const fleetError = writable('');
+    const fleetRefreshing = writable(false);
+    const fleetIntegrations = writable<{ rcon: boolean; komodo: boolean; native?: boolean; hostsReady?: number; hostsTotal?: number; rconTimeoutMs: number; defaultRconHost: string } | null>(null);
 
     const observed = new Map<string, { server: FleetServer; status: unknown; match: unknown; report: unknown; phase: string; value: FleetServer }>();
     const freshnessPhase = derived([fleet, statusByServer, telemetryByServer, uiNow], ([snapshot, live, reports, now]) => {
@@ -23,7 +25,7 @@ const freshnessPhase = derived([fleet, statusByServer, telemetryByServer, uiNow]
         });
         return times.map(time => `${freshStatusTimestamp(time ?? null, now)}:${now - Date.parse(time ?? '') >= 0 && now - Date.parse(time ?? '') <= 120_000}`).join(',');
     });
-export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, freshnessPhase], ([snapshot, live, received, matches, reports]) => {
+    const fleetServers = derived([fleet, statusByServer, statusReceivedAt, matchEventsByServer, telemetryByServer, freshnessPhase], ([snapshot, live, received, matches, reports]) => {
         const now = get(uiNow);
         const next = snapshot.servers.map(server => {
             const candidate = live.get(server.name);
@@ -63,7 +65,7 @@ export const fleetServers = derived([fleet, statusByServer, statusReceivedAt, ma
 
 
     let snapshotServerNow = 0;
-export function hydrateFleet(snapshot: { servers: FleetServer[]; operations: FleetOperation[]; integrations: NonNullable<ReturnType<typeof getIntegrations>>; serverNow: number }) {
+    function hydrateFleet(snapshot: { servers: FleetServer[]; operations: FleetOperation[]; integrations: NonNullable<ReturnType<typeof getIntegrations>>; serverNow: number }) {
         if (!browser || snapshot.serverNow < snapshotServerNow) return;
         snapshotServerNow = snapshot.serverNow;
         syncServerTime(snapshot.serverNow);
@@ -76,7 +78,7 @@ function getIntegrations() { return get(fleetIntegrations); }
 
     let inFlight: Promise<void> | null = null;
     let refreshRequested = false;
-export function refreshFleet(afterCurrent = false): Promise<void> {
+    function refreshFleet(afterCurrent = false): Promise<void> {
         if (!browser) return Promise.resolve();
         if (inFlight) { refreshRequested ||= afterCurrent; return inFlight; }
         inFlight = (async () => {
@@ -87,15 +89,17 @@ export function refreshFleet(afterCurrent = false): Promise<void> {
             try {
                 const results = await Promise.allSettled([
                     (async () => {
-                    const response = await fetch('/api/servers', { signal: controller.signal });
+                        const response = await fetch(serverName === undefined ? '/api/servers' : `/api/servers/${encodeURIComponent(serverName)}`, { signal: controller.signal });
                         if (!response.ok) throw new Error('Could not refresh servers. Retrying automatically.');
                         const servers = await response.json();
                         if (servers.serverNow < snapshotServerNow) return;
                         snapshotServerNow = servers.serverNow;
                         syncServerTime(servers.serverNow); fleetIntegrations.set(servers.integrations);
-                    fleet.update(value => ({ ...value, servers: servers.servers, receivedAt: requestedAt }));
+                        fleet.update(value => ({ ...value, servers: servers.servers, receivedAt: requestedAt,
+                            operations: serverName === undefined ? value.operations : servers.operations }));
                     })(),
                     (async () => {
+                        if (serverName !== undefined) return;
                         const response = await fetch('/api/fleet', { signal: controller.signal });
                         if (!response.ok) throw new Error('Could not refresh operations. Retrying automatically.');
                         const operations = await response.json();
@@ -120,17 +124,46 @@ export function refreshFleet(afterCurrent = false): Promise<void> {
     }
 
     let stop: (() => void) | null = null;
-export function startFleetMonitor(): () => void {
+    function startFleetMonitor(): () => void {
         if (!browser || stop) return () => {};
         const stopPolling = pollVisible(() => refreshFleet(), 60_000, !get(fleet).receivedAt || Date.now() - get(fleet).receivedAt > 30_000);
         let refreshTimer: ReturnType<typeof setTimeout> | undefined;
+        let stopped = false;
+        let pending = false;
+        let refreshing = false;
+        let lastEventRefresh = -Infinity;
+        const schedule = () => {
+            if (stopped || refreshing || refreshTimer !== undefined || !pending || document.visibilityState === 'hidden') return;
+            refreshTimer = setTimeout(async () => {
+                refreshTimer = undefined;
+                refreshing = true;
+                try {
+                    // Wait for any manual/poll refresh, then fetch evidence newer than
+                    // the event. Do not use refreshFleet(true)'s immediate follow-up.
+                    if (inFlight) await inFlight;
+                    if (stopped || document.visibilityState === 'hidden') return;
+                    pending = false;
+                    lastEventRefresh = Date.now();
+                    await refreshFleet();
+                } finally {
+                    refreshing = false;
+                    schedule();
+                }
+            }, Math.max(250, lastEventRefresh + 2000 - Date.now()));
+        };
         const changed = () => {
-        if (refreshTimer || document.visibilityState === 'hidden') return;
-        refreshTimer = setTimeout(() => { refreshTimer = undefined; void refreshFleet(true); }, 250);
+            if (document.visibilityState === 'hidden') return;
+            pending = true;
+            schedule();
         };
         window.addEventListener('ec:fleet-changed', changed);
         stop = () => {
+            stopped = true;
             stopPolling(); clearTimeout(refreshTimer); window.removeEventListener('ec:fleet-changed', changed); stop = null;
         };
         return stop;
     }
+    return { fleet, fleetServers, fleetError, fleetRefreshing, fleetIntegrations, hydrateFleet, refreshFleet, startFleetMonitor };
+}
+
+export const { fleet, fleetServers, fleetError, fleetRefreshing, fleetIntegrations, hydrateFleet, refreshFleet, startFleetMonitor } = createFleetStore();

src/lib/ui/stores/servers.ts

@@ -1,4 +1,4 @@
-import { writable, readonly } from 'svelte/store';
+import { writable, readonly, get } from 'svelte/store';
 import { browser } from '$app/environment';
 import type { GameTelemetry } from '$lib/telemetry.js';
 
@@ -56,13 +56,17 @@ export function serverTimeNow(): number {
     return serverClock ? serverClock.time + performance.now() - serverClock.receivedAt : Date.now();
 }
 
-export function syncServerTime(serverNow?: number): void {
+function updateServerClock(serverNow?: number): void {
     if (typeof serverNow === 'number' && Number.isFinite(serverNow)) {
         // A delayed HTTP snapshot must not rewind time behind a newer stream
         // update and make that update look like it came from the future.
         const time = serverClock ? Math.max(serverNow, serverTimeNow()) : serverNow;
         serverClock = { time, receivedAt: performance.now() };
     }
+}
+
+export function syncServerTime(serverNow?: number): void {
+    updateServerClock(serverNow);
     // Refresh immediately: a suspended tab's interval may be well behind.
     _uiNow.set(serverTimeNow());
 }
@@ -74,35 +78,67 @@ let source: EventSource | null = null;
 export function startServerStream(): (() => void) | undefined {
     if (!browser || source) return;
     syncServerTime();
-    source = new EventSource('/api/events/servers');
-    source.addEventListener('server-telemetry', (evt) => {
+    const stream = source = new EventSource('/api/events/servers');
+    let stopped = false;
+    let frame: number | null = null;
+    let pendingStatus: Map<string, ServerStatus> | null = null;
+    let pendingTelemetry: Map<string, GameTelemetry> | null = null;
+    let pendingMatches: Map<string, GameTelemetry> | null = null;
+    let receivedAt = 0;
+    const flush = () => {
+        frame = null;
+        if (stopped) return;
+        const statuses = pendingStatus, telemetry = pendingTelemetry, matches = pendingMatches;
+        const statusAt = receivedAt;
+        pendingStatus = pendingTelemetry = pendingMatches = null;
+        // Publish the latest clock before exposing reports, including after a
+        // suspended frame. Every changed store is published at most once per frame.
+        syncServerTime();
+        if (statuses) { statusByServer.set(statuses); statusReceivedAt.set(statusAt); }
+        if (matches) matchEventsByServer.set(matches);
+        if (telemetry) telemetryByServer.set(telemetry);
+    };
+    const schedule = () => { if (frame === null) frame = requestAnimationFrame(flush); };
+    stream.addEventListener('server-telemetry', (evt) => {
+        if (stopped) return;
         const data = JSON.parse((evt as MessageEvent).data) as { entries: GameTelemetry[]; snapshot?: boolean; serverNow?: number };
-        syncServerTime(data.serverNow);
-        matchEventsByServer.update((previous) => {
-            const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous);
-            for (const entry of data.entries) {
-                if (entry.event === 'game_start' || entry.event === 'game_end') next.set(entry.server, entry);
+        updateServerClock(data.serverNow);
+        // Merge into one private map per frame. Snapshots replace all earlier
+        // evidence; later deltas still apply in arrival order.
+        if (data.snapshot) {
+            pendingTelemetry = new Map();
+            pendingMatches = new Map();
         }
-            return next;
+        pendingTelemetry ??= new Map(get(telemetryByServer));
+        for (const entry of data.entries) {
+            pendingTelemetry.set(entry.server, entry);
+            if (entry.event === 'game_start' || entry.event === 'game_end') {
+                pendingMatches ??= new Map(get(matchEventsByServer));
+                pendingMatches.set(entry.server, entry);
+            }
+        }
+        schedule();
     });
-        telemetryByServer.update((previous) => {
-            const next = data.snapshot ? new Map<string, GameTelemetry>() : new Map(previous);
-            for (const entry of data.entries) next.set(entry.server, entry);
-            return next;
-        });
-    });
-    source.addEventListener('server-status', (evt) => {
+    stream.addEventListener('server-status', (evt) => {
+        if (stopped) return;
         const data = JSON.parse((evt as MessageEvent).data) as { entries: ServerStatus[]; serverNow?: number; delta?: boolean };
-        syncServerTime(data.serverNow);
-        statusByServer.update(previous => {
-            const m = data.delta ? new Map(previous) : new Map<string, ServerStatus>();
-            for (const e of data.entries) m.set(e.server, { ...e, receivedAt: Date.now() });
-            return m;
-        });
-        statusReceivedAt.set(Date.now());
+        updateServerClock(data.serverNow);
+        receivedAt = Date.now();
+        if (!data.delta) pendingStatus = new Map();
+        pendingStatus ??= new Map(get(statusByServer));
+        for (const entry of data.entries) pendingStatus.set(entry.server, { ...entry, receivedAt });
+        schedule();
     });
     if (tickInterval === null) {
         tickInterval = setInterval(() => { if (typeof document === 'undefined' || document.visibilityState !== 'hidden') _uiNow.set(serverTimeNow()); }, 1000);
     }
-    return () => { source?.close(); source = null; if (tickInterval) clearInterval(tickInterval); tickInterval = null; };
+    return () => {
+        if (stopped) return;
+        stopped = true;
+        if (frame !== null) cancelAnimationFrame(frame);
+        pendingStatus = pendingTelemetry = pendingMatches = null;
+        stream.close(); source = null;
+        if (tickInterval) clearInterval(tickInterval);
+        tickInterval = null;
+    };
 }

src/routes/+layout.svelte

@@ -75,7 +75,7 @@
     $effect(() => {
         if (!sessionReady) return;
         const path = $page.url.pathname;
-        if (!(path === '/' || /^\/(servers|scripts|configs|plugins)\//.test(path) || path === '/betas/new')) return;
+        if (!(path === '/' || /^\/(scripts|configs|plugins)\//.test(path) || path === '/betas/new')) return;
         if ('fleetSnapshot' in $page.data && $page.data.fleetSnapshot) hydrateFleet($page.data.fleetSnapshot);
         return startFleetMonitor();
     });

src/routes/+page.svelte

@@ -1,3 +1,7 @@
+<script module lang="ts">
+    const serverTitleCollator = new Intl.Collator(undefined, { numeric: true });
+</script>
+
 <script lang="ts">
     let { data }: { data: import('./$types').PageData } = $props();
     const snapshot = $derived($fleet.receivedAt ? $fleet : data.fleetSnapshot);
@@ -92,15 +96,21 @@
     let problems = $derived(servers.filter(server => server.health === 'offline' || server.health === 'stale'));
     let starting = $derived(servers.filter(server => server.health === 'starting').length);
     let provisioning = $derived(servers.filter(server => server.health === 'provisioning').length);
-    let filtered = $derived(servers.filter(server => matchesFleet(server, query, filter, game, pinned, activeOperations, observationNow()))
-        .sort((a, b) => {
-            const pinOrder = Number(pinned.includes(b.name)) - Number(pinned.includes(a.name));
-            if (pinOrder) return pinOrder;
-            const rank = (server: FleetServer) => sort === 'players' ? (freshServer(server, observationNow()) ? server.last_players ?? -1 : -1)
-                : sort === 'round' ? currentRound(server, $telemetryByServer.get(server.name), observationNow()) ?? -1
-                : sort === 'duration' ? matchDuration(server) : 0;
-            return rank(b) - rank(a) || serverTitle(a.name).localeCompare(serverTitle(b.name), undefined, { numeric: true });
-        }));
+    let filtered = $derived.by(() => {
+        const now = observationNow();
+        const pinnedNames = new Set(pinned);
+        return servers.filter(server => matchesFleet(server, query, filter, game, pinned, activeOperations, now))
+            .map(server => ({
+                server,
+                title: serverTitle(server.name),
+                pinned: Number(pinnedNames.has(server.name)),
+                rank: sort === 'players' ? (freshServer(server, now) ? server.last_players ?? -1 : -1)
+                    : sort === 'round' ? currentRound(server, $telemetryByServer.get(server.name), now) ?? -1
+                    : sort === 'duration' ? matchDuration(server, now) : 0
+            }))
+            .sort((a, b) => b.pinned - a.pinned || b.rank - a.rank || serverTitleCollator.compare(a.title, b.title))
+            .map(({ server }) => server);
+    });
     let groups = $derived.by(() => {
         const result = new Map<string, { title: string; game: string; servers: FleetServer[]; launch: ReturnType<typeof fleetIdentity>['launch'] }>();
         for (const server of filtered) {
@@ -116,9 +126,9 @@
     let spotlight = $derived(servers.filter(server => currentRound(server, $telemetryByServer.get(server.name), observationNow()) !== null)
         .sort((a, b) => matchDuration(b) - matchDuration(a))[0]);
 
-    function matchDuration(server: FleetServer): number {
-        return freshServer(server, observationNow()) && server.active_since && server.game_started_at && !server.game_ended_at
-            && (server.last_players ?? 0) > 0 ? Math.max(0, observationNow() - Date.parse(server.game_started_at)) : -1;
+    function matchDuration(server: FleetServer, now = observationNow()): number {
+        return freshServer(server, now) && server.active_since && server.game_started_at && !server.game_ended_at
+            && (server.last_players ?? 0) > 0 ? Math.max(0, now - Date.parse(server.game_started_at)) : -1;
     }
     function togglePin(name: string) { pinned = pinned.includes(name) ? pinned.filter(item => item !== name) : [...pinned, name]; }
     function launch(selection: { game: ZombieGame; slug: string } | null = null) {

src/routes/api/agent/heartbeat/+server.ts

@@ -2,7 +2,7 @@ import { json } from '@sveltejs/kit';
 import { heartbeat, pendingCreationCancellations } from '$lib/server/hosts/index.js';
 import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js';
 import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js';
-import { reconcileHostLibraryCleanup, cleanupControllerLibraries } from '$lib/server/game-files/index.js';
+import { reconcileHostLibraryCleanup } from '$lib/server/game-files/index.js';
 import { fleetSettings } from '$lib/server/settings.js';
 export async function POST({ request }: { request: Request }) {
     const authenticated = authenticateAgent(request);
@@ -16,7 +16,6 @@ export async function POST({ request }: { request: Request }) {
         }
         reconcileHostPreparation(host.id);
         reconcileHostLibraryCleanup(host.id);
-        await cleanupControllerLibraries();
         return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15,
             observationKeepDays: fleetSettings().observationKeepDays, serverTime: new Date().toISOString() });
     } catch (cause) { return hostApiError(cause); }

src/routes/api/events/servers/+server.ts

@@ -1,11 +1,14 @@
 import { observationReader } from '$lib/server/fleet/status.js';
 import { listServers } from '$lib/server/fleet/servers.js';
 import { statusEntry } from '$lib/server/pipeline/worker.js';
-import { subscribe } from '$lib/server/pipeline/bus.js';
+import { subscribe, type PipelineEvent } from '$lib/server/pipeline/bus.js';
 import { listTelemetry } from '$lib/server/fleet/telemetry.js';
 
 const encoder = new TextEncoder();
 const HEARTBEAT_MS = 20_000;
+// Bus subscribers receive the same event object. Reuse its complete frame
+// across connections without retaining events after delivery.
+const frames = new WeakMap<PipelineEvent, Uint8Array>();
 
 export async function GET() {
     let heartbeat: NodeJS.Timeout | null = null;
@@ -15,7 +18,12 @@ export async function GET() {
         start(controller) {
             unsubscribe = subscribe((event) => {
                 if (event.type === 'server-status' || event.type === 'server-telemetry') {
-                    controller.enqueue(encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`));
+                    let frame = frames.get(event);
+                    if (!frame) {
+                        frame = encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`);
+                        frames.set(event, frame);
+                    }
+                    controller.enqueue(frame);
                 }
             });
             const observation = observationReader();

src/routes/api/servers/[name]/+server.ts

@@ -1,6 +1,11 @@
 import { kick } from '$lib/server/pipeline/worker.js';
 import { json } from '@sveltejs/kit';
 import { destroyServer } from '$lib/server/provisioning/index.js';
+import { serverSnapshot } from '$lib/server/fleet/snapshot.js';
+
+export function GET({ params }: { params: { name: string } }) {
+    return json(serverSnapshot(params.name));
+}
 
 export async function DELETE({ params }: { params: { name: string } }) {
     const name = params.name;

src/routes/servers/[name]/+page.server.ts

@@ -1,4 +1,5 @@
-import { fleetSnapshot } from '$lib/server/fleet/snapshot.js';
-import { fleetOperations } from '$lib/server/fleet/overview.js';
+import { serverSnapshot } from '$lib/server/fleet/snapshot.js';
 
-export function load() { return { fleetSnapshot: { ...fleetSnapshot(), operations: fleetOperations() } }; }
+export function load({ params }: { params: { name: string } }) {
+    return { serverSnapshot: serverSnapshot(params.name) };
+}

src/routes/servers/[name]/+page.svelte

@@ -1,14 +1,12 @@
 <script lang="ts">
     let { data }: { data: import('./$types').PageData } = $props();
-    const snapshot = $derived($fleet.receivedAt ? $fleet : data.fleetSnapshot);
-    const visibleFleet = $derived($fleet.receivedAt ? $fleetServers : data.fleetSnapshot.servers);
     import RuntimeVersion from '$ui/components/RuntimeVersion.svelte';
     import MatchHistory from '$ui/components/MatchHistory.svelte';
     import RoundTelemetry from '$ui/components/RoundTelemetry.svelte';
     import TelemetryDiagnostics from '$ui/components/TelemetryDiagnostics.svelte';
     import { pollVisible } from '$ui/poll.js';
     import { fetchCampaigns } from '$ui/campaigns.js';
-    import { fleet, fleetServers, fleetError, fleetIntegrations, refreshFleet } from '$ui/stores/fleet.js';
+    import { createFleetStore } from '$ui/stores/fleet.js';
     import { freshServer, fleetIdentity } from '$lib/fleet-overview.js';
     import { serverTitle } from '$ui/server-labels.js';
     import ServerIdentity from '$ui/components/ServerIdentity.svelte';
@@ -109,9 +107,17 @@
     }
 
     let name = $derived($page.params.name ?? '');
+    const detail = $derived(createFleetStore(name));
+    const { fleet, fleetServers, fleetError, fleetIntegrations, refreshFleet } = $derived(detail);
+    $effect(() => {
+        detail.hydrateFleet(data.serverSnapshot);
+        return detail.startFleetMonitor();
+    });
+    const snapshot = $derived($fleet.receivedAt ? $fleet : data.serverSnapshot);
+    const visibleFleet = $derived($fleet.receivedAt ? $fleetServers : data.serverSnapshot.servers);
     let server = $derived((visibleFleet.find(s => s.name === name) as Server | undefined) ?? null);
     const native = $derived(!!server?.native_instance_id);
-    let integrations = $derived($fleetIntegrations ?? data.fleetSnapshot.integrations);
+    let integrations = $derived($fleetIntegrations ?? data.serverSnapshot.integrations);
     let history = $state<DeploymentRow[]>([]);
     let inventory = $state<InventoryEntry[]>([]);
     let komodoProbe = $state<KomodoProbeResult | null>(null);
@@ -610,7 +616,7 @@
     </div>
 {:else if $fleetError}
     <div class="ui-error" role="alert">{$fleetError}<Button onclick={() => refreshFleet(true)}>Retry</Button></div>
-{:else if $fleet.receivedAt || data.fleetSnapshot}
+{:else if $fleet.receivedAt || data.serverSnapshot}
     <EmptyState title="Server not found" description="This server may have been removed from the fleet." />
 {:else}
     <div class="loading">Loading…</div>

static/agent/ec-agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.10"
+VERSION = "1.4.11"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -918,6 +918,8 @@ class Agent:
         self.observation_keep_days = 0  # Older controllers do not authorize cleanup.
         self.library_progress = {}
         self.library_transfers = {}
+        self.inventory_manifest_cache = {}
+        self.inventory_manifest_lock = threading.Lock()
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
         self.executing_id = None
 
@@ -1191,6 +1193,25 @@ class Agent:
                 continue
         return samples
 
+    def inventory_manifest(self, marker):
+        # Inventory only: launch/download verification must still read disk.
+        # Include file identity so atomic replacements invalidate even when
+        # their size and modification time match the previous manifest.
+        with self.inventory_manifest_lock:
+            try:
+                metadata = marker.stat()
+                signature = (metadata.st_mtime_ns, metadata.st_size, metadata.st_dev, metadata.st_ino)
+                cached = self.inventory_manifest_cache.get(marker)
+                if cached is not None and cached[0] == signature:
+                    return cached[1]
+                self.inventory_manifest_cache.pop(marker, None)
+                value = json.loads(marker.read_text())
+                self.inventory_manifest_cache[marker] = (signature, value)
+                return value
+            except (OSError, ValueError):
+                self.inventory_manifest_cache.pop(marker, None)
+                raise
+
     def inventory(self):
         cpu_usage = self.cpu_usage_percent()
         usage = shutil.disk_usage(self.root)
@@ -1210,13 +1231,18 @@ class Agent:
         libraries, caches = [], []
         progress = {key: value.copy() for key, value in list(self.library_progress.items())}
         progress.update({key: transfer.snapshot() for key, transfer in list(self.library_transfers.items())})
-        for marker in (self.root / "libraries").glob("*/*/manifest.json"):
+        manifests = list((self.root / "libraries").glob("*/*/manifest.json"))
+        partial_manifests = list((self.root / "libraries").glob("*/.*.partial/source.json"))
+        with self.inventory_manifest_lock:
+            for marker in self.inventory_manifest_cache.keys() - set(manifests + partial_manifests):
+                del self.inventory_manifest_cache[marker]
+        for marker in manifests:
             # A crash between writing a manifest and its atomic promotion must
             # never expose the staging directory as a ready library.
             if marker.parent.name.startswith("."):
                 continue
             try:
-                value = json.loads(marker.read_text())
+                value = self.inventory_manifest(marker)
                 # Rechecking an existing cache must revoke readiness until its
                 # entire contents have passed verification again.
                 if value["libraryId"] not in progress and (marker.parent / "transfer.json").exists():
@@ -1233,9 +1259,9 @@ class Agent:
             except (OSError, ValueError, KeyError):
                 continue
         ready = {item["id"] for item in libraries}
-        for marker in (self.root / "libraries").glob("*/.*.partial/source.json"):
+        for marker in partial_manifests:
             try:
-                value = json.loads(marker.read_text())
+                value = self.inventory_manifest(marker)
                 library_id = identifier(value["libraryId"])
                 if library_id in ready or library_id in progress:
                     continue

static/agent/manifest.json

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

tests/unit/creation-queue.test.ts

@@ -48,7 +48,11 @@ beforeEach(async () => {
     vi.mocked(komodo.komodoUpdateTagsOnResource).mockResolvedValue({ ok: true });
     const rcon = await import('../../src/lib/server/integrations/rcon.js');
     vi.mocked(rcon.rconGetPlayerCount).mockResolvedValue(0);
-    vi.mocked(rcon.rconPlutoniumVersion).mockResolvedValue({ revision: 5354, error: null });
+    vi.mocked(rcon.rconPlutoniumVersion).mockImplementation(async () => {
+        // Version evidence must be observed strictly after the launch timestamp.
+        await new Promise(resolve => setTimeout(resolve, 2));
+        return { revision: 5354, error: null };
+    });
     vi.mocked(rcon.rconStatusResult).mockImplementation(async server => status(server));
     const { kick } = await import('../../src/lib/server/pipeline/worker.js');
     vi.mocked(kick).mockResolvedValue(undefined);

tests/unit/db-statements.test.ts

@@ -0,0 +1,45 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { newFixture, type Fixture } from '../helpers/fixture.js';
+
+let fixture: Fixture;
+beforeEach(async () => {
+    vi.resetModules();
+    fixture = await newFixture([]);
+});
+afterEach(async () => { await fixture.cleanup(); });
+
+describe('prepared statement caching', () => {
+    it('reuses statements with fresh parameters and remains usable after rollback', async () => {
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const insertSql = 'INSERT INTO settings (key, value) VALUES (?, ?)';
+        const selectSql = 'SELECT value FROM settings WHERE key = ?';
+        const insert = db().prepare(insertSql);
+        const select = db().prepare(selectSql);
+        insert.run('cache-first', 'one');
+        expect(() => db().transaction(() => {
+            db().prepare(insertSql).run('cache-rollback', 'discarded');
+            throw new Error('rollback');
+        })()).toThrow('rollback');
+        db().prepare(insertSql).run('cache-second', 'two');
+
+        expect(db().prepare(insertSql)).toBe(insert);
+        expect(db().prepare(selectSql)).toBe(select);
+        expect(select.get('cache-first')).toEqual({ value: 'one' });
+        expect(select.get('cache-second')).toEqual({ value: 'two' });
+        expect(select.get('cache-rollback')).toBeUndefined();
+    });
+
+    it('prepares fresh statements when the connection is reopened', async () => {
+        const { db, closeDb } = await import('../../src/lib/server/db/index.js');
+        const sql = 'SELECT value FROM settings WHERE key = ?';
+        const oldStatement = db().prepare(sql);
+        db().prepare('INSERT INTO settings (key, value) VALUES (?, ?)').run('cache-persisted', 'saved');
+        closeDb();
+
+        const statement = db().prepare(sql);
+        expect(statement).not.toBe(oldStatement);
+        expect(statement.database).not.toBe(oldStatement.database);
+        expect(statement.get('cache-persisted')).toEqual({ value: 'saved' });
+        expect(db().prepare(sql)).toBe(statement);
+    });
+});

tests/unit/deployment-collections.test.ts

@@ -15,7 +15,11 @@ vi.mock('../../src/lib/server/integrations/komodo.js', () => ({
 }));
 vi.mock('../../src/lib/server/integrations/rcon.js', () => ({
     rconGetPlayerCount: vi.fn(async () => 0),
-    rconPlutoniumVersion: vi.fn(async () => ({ revision: 5354, error: null })),
+    rconPlutoniumVersion: vi.fn(async () => {
+        // Version evidence must be observed strictly after the launch timestamp.
+        await new Promise(resolve => setTimeout(resolve, 2));
+        return { revision: 5354, error: null };
+    }),
     rconStatusResult: vi.fn(async (server: string) => ({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: '',
         status: { players: 0, map: server.startsWith('t6_') ? 'zm_transit' : server.includes('_moon_') ? 'zombie_moon' : 'zombie_theater', gametype: null, roster: [] } }))
 }));

tests/unit/fleet-monitor.test.ts

@@ -0,0 +1,158 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { get } from 'svelte/store';
+import type { FleetOperation } from '../../src/lib/fleet-overview.js';
+
+vi.mock('$app/environment', () => ({ browser: true }));
+
+let documentStub: EventTarget & { visibilityState: string };
+let stop: (() => void) | undefined;
+const changed = () => window.dispatchEvent(new Event('ec:fleet-changed'));
+const visibility = (state: string) => {
+    documentStub.visibilityState = state;
+    documentStub.dispatchEvent(new Event('visibilitychange'));
+};
+const response = (url: string) => Response.json(url === '/api/servers'
+    ? { serverNow: Date.now(), servers: [], integrations: {} } : { operations: [] });
+
+beforeEach(() => {
+    vi.resetModules();
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date('2026-10-05T12:00:00Z'));
+    documentStub = Object.assign(new EventTarget(), { visibilityState: 'visible' });
+    vi.stubGlobal('document', documentStub);
+    vi.stubGlobal('window', new EventTarget());
+    vi.stubGlobal('fetch', vi.fn(async (url: string) => response(url)));
+});
+afterEach(() => {
+    stop?.(); stop = undefined;
+    vi.useRealTimers(); vi.unstubAllGlobals(); vi.restoreAllMocks();
+});
+async function setup() {
+    const store = await import('../../src/lib/ui/stores/fleet.js');
+    store.fleet.set({ servers: [], operations: [], receivedAt: Date.now() });
+    stop = store.startFleetMonitor();
+    return store;
+}
+
+describe('fleet event refreshes', () => {
+    it('refreshes server history through one scoped request without replacing the fleet snapshot', async () => {
+        const store = await import('../../src/lib/ui/stores/fleet.js');
+        const integrations = { rcon: true, komodo: true, rconTimeoutMs: 2000, defaultRconHost: '' };
+        const snapshot = { servers: [], operations: [], integrations, serverNow: Date.now() };
+        store.hydrateFleet(snapshot);
+        const originalFleet = get(store.fleet);
+        const detail = store.createFleetStore('one / two');
+        detail.hydrateFleet(snapshot);
+        stop = detail.startFleetMonitor();
+        const history: FleetOperation[] = [{ key: 'deployment:done', group: 'deployment:done', kind: 'deployment',
+            server: 'one / two', state: 'done', label: 'Complete', detail: 'Complete', href: '/provisioning/done', startedAt: new Date().toISOString() }];
+        vi.mocked(fetch).mockResolvedValue(Response.json({ ...snapshot, operations: history }));
+        changed();
+        await vi.advanceTimersByTimeAsync(250);
+        expect(fetch).toHaveBeenCalledTimes(1);
+        expect(fetch).toHaveBeenCalledWith('/api/servers/one%20%2F%20two', expect.anything());
+        expect(get(detail.fleet).operations).toEqual(history);
+        expect(get(store.fleet)).toBe(originalFleet);
+        expect(get(store.fleet).operations).toEqual([]);
+    });
+
+    it('isolates late responses when navigating between server detail stores', async () => {
+        const { createFleetStore, fleet } = await import('../../src/lib/ui/stores/fleet.js');
+        const first = createFleetStore('first');
+        const second = createFleetStore('second');
+        let finish!: (response: Response) => void;
+        vi.mocked(fetch).mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
+        const oldRequest = first.refreshFleet();
+        const newSnapshot = { serverNow: Date.now(), servers: [], integrations: {}, operations: [] };
+        vi.mocked(fetch).mockResolvedValue(Response.json(newSnapshot));
+        await second.refreshFleet();
+        const current = get(second.fleet);
+        finish(Response.json(newSnapshot)); await oldRequest;
+        expect(get(second.fleet)).toBe(current);
+        expect(get(fleet).receivedAt).toBe(0);
+        expect(fetch).toHaveBeenCalledTimes(2);
+    });
+
+    it('coalesces bursts and refreshes at most once every two seconds with a trailing refresh', async () => {
+        await setup();
+        changed(); changed(); changed();
+        await vi.advanceTimersByTimeAsync(249);
+        expect(fetch).not.toHaveBeenCalled();
+        await vi.advanceTimersByTimeAsync(1);
+        expect(fetch).toHaveBeenCalledTimes(2);
+        for (let i = 0; i < 7; i++) {
+            changed();
+            await vi.advanceTimersByTimeAsync(250);
+            expect(fetch).toHaveBeenCalledTimes(2);
+        }
+        await vi.advanceTimersByTimeAsync(250);
+        expect(fetch).toHaveBeenCalledTimes(4);
+        changed();
+        await vi.advanceTimersByTimeAsync(2000);
+        expect(fetch).toHaveBeenCalledTimes(6);
+        await vi.advanceTimersByTimeAsync(5000);
+        expect(fetch).toHaveBeenCalledTimes(6);
+    });
+
+    it('keeps events during a request without an immediate follow-up when it finishes', async () => {
+        await setup();
+        let finish!: () => void;
+        const gate = new Promise<void>(resolve => { finish = resolve; });
+        vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
+        changed();
+        await vi.advanceTimersByTimeAsync(250);
+        expect(fetch).toHaveBeenCalledTimes(2);
+        changed(); changed();
+        await vi.advanceTimersByTimeAsync(500);
+        finish();
+        await vi.advanceTimersByTimeAsync(1499);
+        expect(fetch).toHaveBeenCalledTimes(2);
+        await vi.advanceTimersByTimeAsync(1);
+        expect(fetch).toHaveBeenCalledTimes(4);
+    });
+
+    it('waits for a slow manual refresh and coalesces events arriving during it', async () => {
+        const { refreshFleet } = await setup();
+        let finish!: () => void;
+        const gate = new Promise<void>(resolve => { finish = resolve; });
+        vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
+        const manual = refreshFleet();
+        changed();
+        await vi.advanceTimersByTimeAsync(3000);
+        changed(); changed();
+        expect(fetch).toHaveBeenCalledTimes(2);
+        finish(); await manual;
+        await vi.advanceTimersByTimeAsync(0);
+        expect(fetch).toHaveBeenCalledTimes(4);
+        await vi.advanceTimersByTimeAsync(3000);
+        expect(fetch).toHaveBeenCalledTimes(4);
+    });
+
+    it('skips queued refreshes while hidden, refreshes on return, and cancels on stop', async () => {
+        await setup();
+        changed(); visibility('hidden');
+        await vi.advanceTimersByTimeAsync(250);
+        changed();
+        await vi.advanceTimersByTimeAsync(3000);
+        expect(fetch).not.toHaveBeenCalled();
+        visibility('visible');
+        await vi.advanceTimersByTimeAsync(0);
+        expect(fetch).toHaveBeenCalledTimes(2);
+        changed(); stop!();
+        await vi.advanceTimersByTimeAsync(3000);
+        expect(fetch).toHaveBeenCalledTimes(2);
+    });
+
+    it('cancels a follow-up waiting for an active request when stopped', async () => {
+        const { refreshFleet } = await setup();
+        let finish!: () => void;
+        const gate = new Promise<void>(resolve => { finish = resolve; });
+        vi.mocked(fetch).mockImplementation(async url => { await gate; return response(String(url)); });
+        const manual = refreshFleet();
+        changed();
+        await vi.advanceTimersByTimeAsync(250);
+        stop!(); finish(); await manual;
+        await vi.advanceTimersByTimeAsync(3000);
+        expect(fetch).toHaveBeenCalledTimes(2);
+    });
+});

tests/unit/fleet-overview.test.ts

@@ -64,8 +64,10 @@ describe('persisted Fleet operations', () => {
         const { db } = await import('../../src/lib/server/db/index.js');
         const handle = db();
         handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at,detail) VALUES ('server','update','failed',5354,?,'Revision not verified')").run(iso());
-        const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
+        const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
         expect(fleetOperations(now)).toEqual([expect.objectContaining({ kind: 'maintenance', state: 'attention', detail: 'Revision not verified' })]);
+        expect(serverOperations('server', now)).toEqual(fleetOperations(now));
+        expect(serverOperations('unrelated', now)).toEqual([]);
         handle.prepare("INSERT INTO maintenance_jobs(server,kind,state,target_revision,requested_at) VALUES ('server','update','queued',5354,?)").run(iso(1000));
         expect(fleetOperations(now)).toEqual([]);
     });
@@ -76,7 +78,7 @@ describe('persisted Fleet operations', () => {
         handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run();
         handle.prepare("INSERT INTO deployments (id, action, started_at) VALUES ('one', 'promote', ?)").run(iso(-2 * 86_400_000));
         handle.prepare("INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state, players) VALUES ('one', 'server', 1, 'pending', 4)").run();
-        const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
+        const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
         expect(fleetOperations(now)).toEqual([expect.objectContaining({ label: 'Waiting for an empty lobby', state: 'active', href: '/campaigns/one' })]);
         handle.prepare("UPDATE deployment_servers SET restarted_at = ?, players = NULL").run(iso());
         handle.prepare("INSERT INTO restart_rollout (server, requested_at) VALUES ('server', ?)").run(iso());
@@ -86,18 +88,20 @@ describe('persisted Fleet operations', () => {
         expect(fleetOperations(now)[0].label).toBe('Recovering');
         db().prepare("UPDATE deployment_servers SET restart_state = 'done', checked_at = ?").run(iso());
         db().prepare('UPDATE restart_rollout SET healthy_at = ?').run(iso());
-        expect(fleetOperations(now)[0].state).toBe('done');
+        expect(fleetOperations(now)).toEqual([]);
+        expect(serverOperations('server', now)[0].state).toBe('done');
     });
 
-    it('keeps lifecycle failures actionable until superseded and retains successful completions', async () => {
+    it('keeps lifecycle failures actionable until superseded and retains completions only for that server', async () => {
         const { db } = await import('../../src/lib/server/db/index.js');
         const handle = db();
         const insert = handle.prepare("INSERT INTO server_deployments (id, server, action, state, stage, started_at, finished_at, triggered_by) VALUES (?, ?, ?, ?, ?, ?, ?, 'test')");
         insert.run('failed', 'server', 'create', 'failed', 'Creating container', iso(-86_400_001), iso(-86_400_001));
-        const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
+        const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
         expect(fleetOperations(now)[0].state).toBe('attention');
         insert.run('retry', 'server', 'create', 'done', 'Ready', iso(), iso());
-        expect(fleetOperations(now)).toEqual([expect.objectContaining({ key: 'deployment:retry', state: 'done' })]);
+        expect(fleetOperations(now)).toEqual([]);
+        expect(serverOperations('server', now)).toEqual([expect.objectContaining({ key: 'deployment:retry', state: 'done' })]);
         insert.run('queued', 'new-server', 'create', 'queued', 'Waiting to launch', iso(), null);
         expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ server: 'new-server', label: 'Waiting to launch', state: 'active' }));
     });
@@ -107,7 +111,7 @@ describe('persisted Fleet operations', () => {
         const handle = db();
         handle.prepare("INSERT INTO servers (name, komodo_name) VALUES ('server', 'server')").run();
         handle.prepare("INSERT INTO restart_rollout (server, requested_at, error) VALUES ('server', ?, 'RCON not answering')").run(iso());
-        const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
+        const { fleetOperations, serverOperations } = await import('../../src/lib/server/fleet/overview.js');
         expect(fleetOperations(now)[0]).toMatchObject({ kind: 'recovery', state: 'attention', detail: 'RCON not answering' });
         handle.prepare("INSERT INTO scripts (id, name, target_rel_path) VALUES (1, 'test', 'test.gsc')").run();
         handle.prepare("INSERT INTO script_versions (id, script_id, version, sha256, size, content) VALUES (1, 1, 1, 'hash', 0, X'')").run();
@@ -115,9 +119,11 @@ describe('persisted Fleet operations', () => {
         handle.prepare("INSERT INTO beta_servers (beta_id, server, had_original, stage) VALUES (1, 'server', 0, 'awaiting-match')").run();
         expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'beta', label: 'Observing a match', href: '/betas/1' }));
         expect(fleetOperations(now)).toContainEqual(expect.objectContaining({ kind: 'recovery', state: 'attention' }));
+        expect(serverOperations('server', now)).toEqual(fleetOperations(now));
+        expect(serverOperations('unrelated', now)).toEqual([]);
     });
 
-    it('preserves lifecycle selection across cutoff, deletion, supersession, and overlapping warning branches', async () => {
+    it('keeps only active work and unresolved failures across deletion and supersession', async () => {
         const { db } = await import('../../src/lib/server/db/index.js');
         const handle = db();
         const since = iso(-86_400_000);
@@ -138,13 +144,12 @@ describe('persisted Fleet operations', () => {
                 }
             }
         })();
-        // The former OR query is the compatibility reference for all three branches.
+        // Completed work, including warnings and cancellations, belongs to server history.
         const expected = handle.prepare(`SELECT d.id FROM server_deployments d
             WHERE d.server_deleted_at IS NULL AND (d.state IN ('queued', 'running', 'waiting', 'awaiting-health')
-            OR (d.state IN ('done', 'warning', 'cancelled') AND d.finished_at >= ?)
-            OR (d.state IN ('failed', 'interrupted', 'warning') AND NOT EXISTS
+            OR (d.state IN ('failed', 'interrupted') AND NOT EXISTS
                 (SELECT 1 FROM server_deployments n WHERE n.server = d.server AND n.rowid > d.rowid)))`)
-            .all(since) as { id: string }[];
+            .all() as { id: string }[];
         const { fleetOperations } = await import('../../src/lib/server/fleet/overview.js');
         expect(fleetOperations(now).map(row => row.key).sort()).toEqual(expected.map(row => `deployment:${row.id}`).sort());
     });

tests/unit/fleet-snapshot-scope.test.ts

@@ -0,0 +1,85 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { newFixture, type Fixture } from '../helpers/fixture.js';
+
+let fixture: Fixture;
+const now = Date.parse('2026-10-05T12:00:00Z');
+const iso = (offset = 0) => new Date(now + offset).toISOString();
+beforeEach(async () => {
+    vi.resetModules();
+    vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(now);
+    fixture = await newFixture([]);
+    const { db } = await import('../../src/lib/server/db/index.js');
+    db().transaction(() => {
+        const server = db().prepare('INSERT INTO servers(name, komodo_name) VALUES (?, ?)');
+        const operation = db().prepare(`INSERT INTO server_deployments(id,server,action,state,stage,started_at,finished_at,triggered_by)
+            VALUES(?,?,'restart',?,'Complete',?,?,'test')`);
+        for (const name of ['one', 'two']) {
+            server.run(name, name);
+            for (let i = 0; i < 100; i++) operation.run(`${name}-${i}`, name, 'done', iso(-i), iso(-i));
+            operation.run(`${name}-old`, name, 'done', iso(-86_400_001), iso(-86_400_001));
+            operation.run(`${name}-failed`, name, 'failed', iso(), iso());
+            operation.run(`${name}-active`, name, 'waiting', iso(), null);
+        }
+        db().prepare("INSERT INTO deployments(id,action,started_at,finished_at) VALUES('campaign','push',?,?)").run(iso(), iso());
+        for (const name of ['one', 'two']) {
+            db().prepare(`INSERT INTO deployment_servers(deployment_id,server,files_changed,restart_state,checked_at)
+                VALUES('campaign',?,1,'done',?)`).run(name, iso());
+        }
+    })();
+});
+afterEach(async () => { await fixture.cleanup(); vi.useRealTimers(); vi.restoreAllMocks(); });
+
+describe('fleet and server response scope', () => {
+    it('omits completed history from fleet REST and SSR responses', async () => {
+        const { GET } = await import('../../src/routes/api/fleet/+server.js');
+        const { load } = await import('../../src/routes/+page.server.js');
+        const body = await GET().json();
+        expect(body.operations.map((operation: { key: string }) => operation.key).sort())
+            .toEqual(['deployment:one-active', 'deployment:two-active']);
+        expect(load().fleetSnapshot.operations).toEqual(body.operations);
+    });
+
+    it('loads only the selected server and its recent history in both REST and SSR', async () => {
+        const { GET } = await import('../../src/routes/api/servers/[name]/+server.js');
+        const { load } = await import('../../src/routes/servers/[name]/+page.server.js');
+        const registry = await import('../../src/lib/server/fleet/servers.js');
+        const { fleetSnapshot } = await import('../../src/lib/server/fleet/snapshot.js');
+        const expected = fleetSnapshot().servers.find(server => server.name === 'one');
+        const list = vi.spyOn(registry, 'listServers');
+        const snapshot = load({ params: { name: 'one' } }).serverSnapshot;
+        expect(list).not.toHaveBeenCalled();
+        expect(snapshot.servers).toEqual([expected]);
+        expect(snapshot.operations).toHaveLength(103);
+        expect(snapshot.operations.every(operation => operation.server === 'one')).toBe(true);
+        expect(snapshot.operations.map(operation => operation.key)).not.toContain('deployment:one-old');
+        expect(snapshot.operations).toContainEqual(expect.objectContaining({ key: 'campaign:campaign:one', state: 'done' }));
+        expect(snapshot.operations).toContainEqual(expect.objectContaining({ key: 'deployment:one-failed', state: 'attention' }));
+        const body = await GET({ params: { name: 'one' } }).json();
+        expect(body).toEqual(JSON.parse(JSON.stringify(snapshot)));
+        expect(list).not.toHaveBeenCalled();
+    });
+
+    it('returns no unrelated servers or history for missing and removed servers', async () => {
+        const { serverSnapshot } = await import('../../src/lib/server/fleet/snapshot.js');
+        expect(serverSnapshot('missing')).toMatchObject({ servers: [], operations: [] });
+        const { db } = await import('../../src/lib/server/db/index.js');
+        db().prepare("UPDATE servers SET destroyed_at = ? WHERE name = 'one'").run(iso());
+        expect(serverSnapshot('one').servers).toEqual([]);
+        db().prepare("UPDATE server_deployments SET server_deleted_at = ? WHERE server = 'one'").run(iso());
+        db().prepare("UPDATE deployment_servers SET server_deleted_at = ? WHERE server = 'one'").run(iso());
+        expect(serverSnapshot('one').operations).toEqual([]);
+    });
+
+    it('uses a server index for recent lifecycle history', async () => {
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const { serverOperations } = await import('../../src/lib/server/fleet/overview.js');
+        const prepare = vi.spyOn(db(), 'prepare');
+        serverOperations('one', now);
+        const sql = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('FROM server_deployments'))!;
+        prepare.mockRestore();
+        const plan = db().prepare(`EXPLAIN QUERY PLAN ${sql}`).all({ server: 'one', since: iso(-86_400_000) }) as { detail: string }[];
+        const details = plan.map(row => row.detail).join('\n');
+        expect(details).toMatch(/SEARCH server_deployments USING (?:COVERING )?INDEX idx_server_deployments_server/);
+        expect(details).not.toMatch(/SCAN server_deployments\b/);
+    });
+});

tests/unit/game-file-library.test.ts

@@ -145,6 +145,79 @@ describe('game-file import lifecycle', () => {
     });
 });
 describe('verified host distribution', () => {
+    it('reuses published metadata across host reports without reading or parsing a large manifest', () => {
+        const version = begin();
+        const manifest = JSON.stringify(Array.from({ length: 20_000 }, (_, i) => ({ ...file, path: `zone/file-${i}.ff` })));
+        const size = file.size * 20_000;
+        state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=?,manifest=?,manifest_sha256='large-checksum' WHERE id=?")
+            .run(size, manifest, version.id);
+        const otherHost = createHost({ name: 'Other reporting host' });
+        const parse = JSON.parse;
+        const parsing = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => {
+            if (value === manifest) throw new Error('Cache reports must not deserialize manifests.');
+            return parse(value, reviver);
+        });
+        const preparing = vi.spyOn(state.handle!, 'prepare');
+        try {
+            for (const id of [hostId, otherHost.id, hostId]) {
+                updateHostLibraryCaches(id, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'large-checksum' }]);
+                expect(getHostLibraryCache(id, version.id)).toMatchObject({ status: 'ready', bytesDownloaded: size, bytesTotal: size });
+            }
+            const reads = preparing.mock.calls.map(([sql]) => sql).filter(sql => sql.includes('FROM game_file_versions'));
+            expect(reads).toHaveLength(1);
+            expect(state.handle!.prepare(reads[0]).columns().map(column => column.name)).toEqual(['state', 'sizeBytes', 'manifestSha256']);
+            updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'wrong' }]);
+            expect(getHostLibraryCache(hostId, version.id)?.status).toBe('failed');
+            expect(() => updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'downloading', bytesDownloaded: size + 1 }]))
+                .toThrow('Invalid library download progress');
+        } finally { parsing.mockRestore(); preparing.mockRestore(); }
+    });
+
+    it('does not cache missing or unpublished versions', () => {
+        updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
+        const version = begin();
+        state.handle!.prepare("UPDATE game_file_versions SET id='future' WHERE id=?").run(version.id);
+        updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
+        expect(getHostLibraryCache(hostId, 'future')).toBeNull();
+        state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=123,manifest_sha256='checksum' WHERE id='future'").run();
+        updateHostLibraryCaches(hostId, [{ libraryVersionId: 'future', status: 'ready', manifestSha256: 'checksum' }]);
+        expect(getHostLibraryCache(hostId, 'future')).toMatchObject({ status: 'ready', bytesTotal: 123 });
+    });
+
+    it('does not retain metadata from a rolled-back publication', () => {
+        const version = begin();
+        expect(() => state.handle!.transaction(() => {
+            state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=123,manifest_sha256='temporary' WHERE id=?").run(version.id);
+            updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'temporary' }]);
+            throw new Error('rollback');
+        })()).toThrow('rollback');
+        updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'temporary' }]);
+        expect(getHostLibraryCache(hostId, version.id)).toBeNull();
+        state.handle!.prepare("UPDATE game_file_versions SET state='ready',size_bytes=456,manifest_sha256='published' WHERE id=?").run(version.id);
+        updateHostLibraryCaches(hostId, [{ libraryVersionId: version.id, status: 'ready', manifestSha256: 'published' }]);
+        expect(getHostLibraryCache(hostId, version.id)).toMatchObject({ status: 'ready', bytesTotal: 456 });
+    });
+
+    it('keeps metadata caches separate for each database connection', () => {
+        const seed = () => {
+            listLibraries();
+            legacyLibrary('cache-library', 't6');
+            legacyVersion('cache-version', 'cache-library', '2025-01-01', '2025-01-01');
+        };
+        const report = () => updateHostLibraryCaches(hostId, [{ libraryVersionId: 'cache-version', status: 'ready', manifestSha256: 'checksum' }]);
+        seed(); report();
+        const original = state.handle!;
+        state.handle = new Database(':memory:');
+        try {
+            seed();
+            const preparing = vi.spyOn(state.handle, 'prepare');
+            try {
+                report();
+                expect(preparing.mock.calls.filter(([sql]) => sql.includes('FROM game_file_versions'))).toHaveLength(1);
+            } finally { preparing.mockRestore(); }
+        } finally { state.handle.close(); state.handle = original; }
+    });
+
     it('deduplicates active downloads and blocks readiness without exact manifest verification', async () => {
         const version = await published();
         const operation = queueLibraryDownload(hostId, version.id);

tests/unit/host-preparation.test.ts

@@ -68,6 +68,23 @@ function verified(id: string, version: GameLibraryVersion) {
 }
 
 describe('automatic host preparation', () => {
+    it('accepts heartbeats without running controller library cleanup', async () => {
+        const host = hosts.createHost({ name: 'Heartbeat host' });
+        const { credential } = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
+        const { POST } = await import('../../src/routes/api/agent/heartbeat/+server.js');
+        const cleanup = vi.spyOn(libraries, 'cleanupControllerLibraries').mockRejectedValue(new Error('Cleanup is independent of heartbeats'));
+        try {
+            for (let i = 0; i < 3; i++) {
+                const response = await POST({ request: new Request('https://controller.test/api/agent/heartbeat', {
+                    method: 'POST', headers: { Authorization: `Bearer ${credential}`, 'Content-Type': 'application/json' },
+                    body: JSON.stringify({ inventory: { ...inventory, libraryCaches: [] } })
+                }) });
+                expect(response.status).toBe(200);
+            }
+            expect(cleanup).not.toHaveBeenCalled();
+        } finally { cleanup.mockRestore(); }
+    });
+
     it('keeps re-enabled hosts disabled until a fresh runtime pull and both game libraries are verified', async () => {
         const t5 = await publish('t5');
         const t6 = await publish('t6');

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

@@ -0,0 +1,146 @@
+import { setImmediate } from 'node:timers/promises';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import type { HostInventory } from '../../src/lib/hosts.js';
+import { newFixture, type Fixture } from '../helpers/fixture.js';
+
+let fixture: Fixture;
+const epoch = Date.parse('2026-10-05T12:00:00Z');
+const inventory = { hostname: 'request-cache-host', protocolVersion: 1, dockerRunning: true, directoriesWritable: true,
+    storageAvailable: 100 * 1024 ** 3, runtimeImages: ['plutainer:test'], libraries: [{ id: 'test-library', state: 'ready' }] };
+beforeEach(async () => {
+    vi.resetModules(); vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(epoch);
+    fixture = await newFixture([]);
+});
+afterEach(async () => { vi.restoreAllMocks(); vi.useRealTimers(); await fixture.cleanup(); });
+
+async function enrolled() {
+    const hosts = await import('../../src/lib/server/hosts/index.js');
+    const context = await import('../../src/lib/server/hosts/request-context.js');
+    const host = hosts.createHost({ name: 'Request cache host' });
+    const { credential } = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
+    // Finish enrollment's independent availability refresh before counting parses.
+    await setImmediate();
+    return { ...hosts, ...context, hostId: host.id, credential };
+}
+
+describe('agent credential usage writes', () => {
+    it('writes on first use and at most once per minute, including after reopening the database', async () => {
+        const api = await enrolled();
+        const { db, closeDb } = await import('../../src/lib/server/db/index.js');
+        db().exec(`CREATE TABLE credential_usage_writes (credential_id TEXT, at TEXT);
+            CREATE TRIGGER count_credential_usage AFTER UPDATE OF last_used_at ON ec_host_credentials
+            BEGIN INSERT INTO credential_usage_writes VALUES (NEW.id, NEW.last_used_at); END;`);
+        const writes = () => db().prepare('SELECT at FROM credential_usage_writes').all();
+        const authenticate = () => expect(api.authenticateCredential(api.credential)?.id).toBe(api.hostId);
+        authenticate(); authenticate(); authenticate();
+        expect(writes()).toEqual([{ at: new Date(epoch).toISOString() }]);
+        vi.setSystemTime(epoch + 59_999);
+        authenticate();
+        closeDb(); authenticate();
+        expect(writes()).toHaveLength(1);
+        vi.setSystemTime(epoch + 60_000);
+        authenticate(); authenticate();
+        expect(writes()).toEqual([{ at: new Date(epoch).toISOString() }, { at: new Date(epoch + 60_000).toISOString() }]);
+        expect(api.authenticateCredential('invalid')).toBeNull();
+        expect(writes()).toHaveLength(2);
+    });
+
+    it('checks revocation immediately and records first use of a replacement credential', async () => {
+        const api = await enrolled();
+        const { db } = await import('../../src/lib/server/db/index.js');
+        await api.withHostRequestContext(() => {
+            expect(api.authenticateCredential(api.credential)?.id).toBe(api.hostId);
+            vi.setSystemTime(epoch + 1000);
+            const replacement = api.rotateCredential(api.hostId);
+            expect(api.authenticateCredential(api.credential)).toBeNull();
+            expect(api.authenticateCredential(replacement)?.id).toBe(api.hostId);
+            expect(db().prepare('SELECT last_used_at FROM ec_host_credentials WHERE revoked_at IS NULL').get())
+                .toEqual({ last_used_at: new Date(epoch + 1000).toISOString() });
+            api.revokeCredentials(api.hostId);
+            expect(api.authenticateCredential(replacement)).toBeNull();
+            expect(api.getHost(api.hostId)?.credentialActive).toBe(false);
+        });
+    });
+});
+
+describe('request-scoped host inventory parsing', () => {
+    it('parses once across authentication, host reads, and operation handling within a request', async () => {
+        const api = await enrolled();
+        const serialized = JSON.stringify(inventory);
+        const parse = vi.spyOn(JSON, 'parse');
+        await api.withHostRequestContext(async () => {
+            const authenticated = api.authenticateCredential(api.credential)!;
+            const operation = api.queueHostOperation(api.hostId, 'instance.inspect', { name: 'server' });
+            api.pendingHostOperationRows(api.hostId, 'read', false);
+            api.reportHostOperationStatus(api.hostId, operation.id, { state: 'accepted' });
+            await Promise.resolve();
+            for (let i = 0; i < 10; i++) expect(api.getHost(api.hostId)?.inventory).toBe(authenticated.inventory);
+            expect(api.requireReadyHost(api.hostId).inventory).toBe(authenticated.inventory);
+            expect(api.listHosts()[0].inventory).toBe(authenticated.inventory);
+        });
+        expect(parse.mock.calls.filter(([value]) => value === serialized)).toHaveLength(1);
+        await api.withHostRequestContext(() => api.getHost(api.hostId));
+        expect(parse.mock.calls.filter(([value]) => value === serialized)).toHaveLength(2);
+    });
+
+    it('keeps host metadata fresh and reparses changed inventory, including after rollback', async () => {
+        const api = await enrolled();
+        const { db } = await import('../../src/lib/server/db/index.js');
+        await api.withHostRequestContext(() => {
+            const original = api.getHost(api.hostId)!;
+            const updated = api.updateHost(api.hostId, { name: 'Renamed host', mode: 'draining' });
+            expect(updated).toMatchObject({ name: 'Renamed host', mode: 'draining', readiness: 'draining' });
+            expect(updated.inventory).toBe(original.inventory);
+            const heartbeat = api.heartbeat(api.hostId, { ...inventory, storageAvailable: 123 });
+            expect(heartbeat.inventory).not.toBe(original.inventory);
+            expect(api.getHost(api.hostId)?.inventory).toBe(heartbeat.inventory);
+            expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(123);
+            expect(() => db().transaction(() => {
+                db().prepare('UPDATE ec_hosts SET inventory=? WHERE id=?').run(JSON.stringify({ ...inventory, storageAvailable: 456 }), api.hostId);
+                expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(456);
+                throw new Error('rollback');
+            })()).toThrow('rollback');
+            expect(api.getHost(api.hostId)?.inventory.storageAvailable).toBe(123);
+        });
+    });
+
+    it('isolates overlapping requests and does not cache reads outside a request', async () => {
+        const api = await enrolled();
+        let release!: () => void;
+        const gate = new Promise<void>(resolve => { release = resolve; });
+        let first!: HostInventory;
+        const request = api.withHostRequestContext(async () => {
+            first = api.getHost(api.hostId)!.inventory;
+            await gate;
+            expect(api.getHost(api.hostId)!.inventory).toBe(first);
+        });
+        try {
+            await api.withHostRequestContext(async () => {
+                const second = api.getHost(api.hostId)!.inventory;
+                expect(second).not.toBe(first);
+                await Promise.resolve();
+                expect(api.getHost(api.hostId)!.inventory).toBe(second);
+            });
+            expect(api.getHost(api.hostId)!.inventory).not.toBe(api.getHost(api.hostId)!.inventory);
+        } finally { release(); await request; }
+    });
+
+    it.each([false, true])('releases cached inventory from inherited work after request completion (failed: %s)', async failed => {
+        const api = await enrolled();
+        let release!: () => void;
+        const gate = new Promise<void>(resolve => { release = resolve; });
+        let detached!: Promise<void>;
+        let original!: HostInventory;
+        const request = api.withHostRequestContext(() => {
+            original = api.getHost(api.hostId)!.inventory;
+            detached = gate.then(() => {
+                const later = api.getHost(api.hostId)!.inventory;
+                expect(later).not.toBe(original);
+                expect(api.getHost(api.hostId)!.inventory).not.toBe(later);
+            });
+            if (failed) throw new Error('request failed');
+        });
+        if (failed) await expect(request).rejects.toThrow('request failed'); else await request;
+        release(); await detached;
+    });
+});

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

@@ -70,10 +70,10 @@ describe('provisioning reads with retained lifecycle history', () => {
         expect(queries).toHaveLength(1);
         const statement = handle.prepare(queries[0]);
         expect(statement.columns().map(column => column.name)).toEqual(['id', 'server', 'action', 'state', 'stage', 'error', 'started_at']);
-        const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all(iso(-86_400_000)) as { detail: string }[];
+        const plan = handle.prepare(`EXPLAIN QUERY PLAN ${queries[0]}`).all() as { detail: string }[];
         const details = plan.map(row => row.detail).join('\n');
         expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_state');
-        expect(details).toContain('SEARCH server_deployments USING INDEX idx_server_deployments_finished_at');
+        expect(queries[0]).not.toContain('finished_at >=');
         expect(details).toContain('SEARCH server_deployments USING INTEGER PRIMARY KEY (rowid=?)');
         expect(details).toContain('SCAN server_deployments USING COVERING INDEX idx_server_deployments_server');
         expect(details).not.toMatch(/SCAN server_deployments(?:\n|$)/);

tests/unit/sweep.test.ts

@@ -16,6 +16,53 @@ afterEach(async () => {
 });
 
 describe('sweep', () => {
+    it('emits server state only when state, players, or error changes', async () => {
+        const { sweep } = await import('../../src/lib/server/pipeline/sweep.js');
+        const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js');
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
+        await seedServerRegistry();
+        db().prepare("INSERT INTO deployments (id, action, started_at, triggered_by) VALUES ('changes', 'test', ?, 'ui')")
+            .run(new Date().toISOString());
+        db().prepare(`INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state)
+            VALUES ('changes', 't5_zombie_theater_1', 1, 'pending')`).run();
+        const events = vi.fn();
+        const unsubscribe = subscribe(event => { if (event.type === 'server-state') events(event); });
+        try {
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(1);
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(1);
+
+            // Fresh check timestamps still update the campaign, without an SSE event.
+            const checkedAt = new Date().toISOString();
+            db().prepare('UPDATE server_deployments SET checked_at = ?').run(checkedAt);
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(1);
+            expect(db().prepare("SELECT checked_at, attempts FROM deployment_servers WHERE deployment_id = 'changes'").get())
+                .toEqual({ checked_at: checkedAt, attempts: 1 });
+
+            db().prepare('UPDATE server_deployments SET players = 2').run();
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(2);
+            expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'pending', players: 2 }));
+            db().prepare("UPDATE server_deployments SET error = 'Waiting for players'").run();
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(3);
+            expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ error: 'Waiting for players' }));
+            db().prepare("UPDATE server_deployments SET state = 'failed'").run();
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(4);
+            expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'failed' }));
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(4);
+            db().prepare("UPDATE server_deployments SET state = 'done'").run();
+            await sweep();
+            expect(events).toHaveBeenCalledTimes(5);
+            expect(events).toHaveBeenLastCalledWith(expect.objectContaining({ state: 'done', error: null }));
+        } finally { unsubscribe(); }
+    });
+
     it('queues pending rows in Provisioning even when integrations are unset', async () => {
         const { sweep } = await import('../../src/lib/server/pipeline/sweep.js');
         const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js');

tests/unit/telemetry.test.ts

@@ -90,6 +90,45 @@ it('timestamps initial telemetry and subsequent status and telemetry events with
     }
 });
 
+it('serializes each live server event once across connections and releases cancelled subscribers', async () => {
+    const { GET } = await import('../../src/routes/api/events/servers/+server.js');
+    const { emit } = await import('../../src/lib/server/pipeline/bus.js');
+    const readers = await Promise.all([GET(), GET()].map(async response => (await response).body!.getReader()));
+    const stringify = vi.spyOn(JSON, 'stringify');
+    try {
+        for (const reader of readers) {
+            await reader.read(); // Initial comment and per-connection snapshots.
+            await reader.read();
+            await reader.read();
+        }
+        for (const event of [
+            { type: 'server-status' as const, delta: true, entries: [] },
+            { type: 'server-telemetry' as const, entries: [{ ...payload, receivedAt: new Date().toISOString() }] }
+        ]) {
+            stringify.mockClear();
+            emit(event);
+            expect(stringify).toHaveBeenCalledTimes(1);
+            const [first, second] = await Promise.all(readers.map(reader => reader.read()));
+            expect(first.value).toBe(second.value);
+            const frame = new TextDecoder().decode(first.value);
+            expect(frame.startsWith(`event: ${event.type}\ndata: `)).toBe(true);
+            expect(JSON.parse(frame.split('\n')[1].slice(6))).toEqual({ ...event, serverNow: expect.any(Number) });
+        }
+        await readers[0].cancel();
+        stringify.mockClear();
+        emit({ type: 'server-status', entries: [] });
+        expect(stringify).toHaveBeenCalledTimes(1);
+        expect((await readers[1].read()).done).toBe(false);
+        await readers[1].cancel();
+        stringify.mockClear();
+        emit({ type: 'server-telemetry', entries: [] });
+        expect(stringify).not.toHaveBeenCalled();
+    } finally {
+        stringify.mockRestore();
+        await Promise.all(readers.map(reader => reader.cancel()));
+    }
+});
+
 it.each(['round_change', 'heartbeat'] as const)('persists %s stats, broadcasts updates, and leaves RCON safety data untouched', async (event) => {
     const payload = gameEvent(event, 17);
     const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');

tests/unit/ui-server-clock.test.ts

@@ -8,9 +8,16 @@ import { gameEvent } from '../helpers/telemetry.js';
 vi.mock('$app/environment', () => ({ browser: true }));
 
 const epoch = Date.parse('2026-09-19T12:00:00Z');
-let source: EventTarget;
+let source: EventTarget & { close: () => void };
 let stop: (() => void) | undefined;
 let snapshot: { serverNow: number; servers: FleetServer[]; integrations: object };
+let frames: Map<number, FrameRequestCallback>;
+
+function flushFrame() {
+    const callbacks = [...frames.values()];
+    frames.clear();
+    for (const callback of callbacks) callback(performance.now());
+}
 
 function server(timestamp = epoch, changes: Partial<FleetServer> = {}): FleetServer {
     return {
@@ -35,12 +42,20 @@ function status(timestamp: number, changes: Partial<ServerStatus> = {}): ServerS
 
 function emit(serverNow: number, entries: ServerStatus[]) {
     source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ serverNow, entries }) }));
+    flushFrame();
 }
 
 beforeEach(() => {
     vi.resetModules();
     vi.useFakeTimers({ toFake: ['Date', 'performance', 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval'] });
     vi.setSystemTime(epoch);
+    frames = new Map();
+    let frameId = 0;
+    vi.stubGlobal('requestAnimationFrame', vi.fn((callback: FrameRequestCallback) => {
+        frames.set(++frameId, callback);
+        return frameId;
+    }));
+    vi.stubGlobal('cancelAnimationFrame', vi.fn((id: number) => { frames.delete(id); }));
     snapshot = { serverNow: epoch, servers: [server()], integrations: {} };
     vi.stubGlobal('fetch', vi.fn(async (url: string) => new Response(JSON.stringify(
         url === '/api/servers' ? snapshot : { operations: [] }
@@ -66,6 +81,130 @@ async function setup() {
     return { ...clock, ...fleet };
 }
 
+function queueEvent(type: string, data: object) {
+    source.dispatchEvent(new MessageEvent(type, { data: JSON.stringify(data) }));
+}
+
+describe('animation-frame stream batching', () => {
+    it('publishes each changed store once for a burst and preserves lifecycle evidence through heartbeats', async () => {
+        const stores = await setup();
+        emit(epoch, [status(epoch), status(epoch, { server: 'two' })]);
+        const before = get(stores.statusByServer);
+        await vi.advanceTimersByTimeAsync(1000);
+        const watched = [stores.statusByServer, stores.statusReceivedAt, stores.telemetryByServer, stores.matchEventsByServer, stores.uiNow];
+        const changes = watched.map(() => vi.fn());
+        const unsubscribe = watched.map((store, index) => store.subscribe(changes[index]));
+        changes.forEach(change => change.mockClear());
+        vi.mocked(requestAnimationFrame).mockClear();
+        try {
+            for (let i = 0; i < 100; i++) {
+                queueEvent('server-status', { delta: true, serverNow: epoch + 1000 + i,
+                    entries: [status(epoch + 1000, { players: i })] });
+                queueEvent('server-telemetry', { serverNow: epoch + 1000 + i,
+                    entries: [{ ...gameEvent(i === 0 ? 'game_start' : 'heartbeat', i, 1), receivedAt: new Date(epoch + 1000 + i).toISOString() }] });
+            }
+            expect(requestAnimationFrame).toHaveBeenCalledTimes(1);
+            changes.forEach(change => expect(change).not.toHaveBeenCalled());
+            expect(get(stores.statusByServer)).toBe(before);
+            expect(before.get('t5_zombie_theater_1')?.players).toBe(0);
+            flushFrame();
+            changes.forEach(change => expect(change).toHaveBeenCalledTimes(1));
+            expect(get(stores.statusByServer).size).toBe(2);
+            expect(get(stores.statusByServer).get('t5_zombie_theater_1')?.players).toBe(99);
+            expect(get(stores.telemetryByServer).get('t5_zombie_theater_1')).toMatchObject({ event: 'heartbeat', round: 99 });
+            expect(get(stores.matchEventsByServer).get('t5_zombie_theater_1')?.event).toBe('game_start');
+
+            queueEvent('server-telemetry', { entries: [{ ...gameEvent('heartbeat', 100, 1), receivedAt: new Date(epoch + 1100).toISOString() }] });
+            flushFrame();
+            expect(requestAnimationFrame).toHaveBeenCalledTimes(2);
+            expect(changes[2]).toHaveBeenCalledTimes(2);
+            expect(changes[3]).toHaveBeenCalledTimes(1);
+        } finally { unsubscribe.forEach(unsubscribe => unsubscribe()); }
+    });
+
+    it('applies snapshots and subsequent deltas in order, including empty snapshots', async () => {
+        const { statusByServer, telemetryByServer, matchEventsByServer } = await setup();
+        const telemetry = (event: GameEvent['event'], name: string) => ({ ...gameEvent(event, 1, 1, name), receivedAt: new Date(epoch).toISOString() });
+        emit(epoch, [status(epoch, { server: 'old' })]);
+        queueEvent('server-telemetry', { entries: [telemetry('game_start', 'old')] });
+        flushFrame();
+
+        queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'discarded' })] });
+        queueEvent('server-status', { entries: [status(epoch, { server: 'two' })] });
+        queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'three' })] });
+        queueEvent('server-telemetry', { entries: [telemetry('game_end', 'discarded')] });
+        queueEvent('server-telemetry', { snapshot: true, entries: [telemetry('heartbeat', 'two')] });
+        queueEvent('server-telemetry', { entries: [telemetry('game_end', 'three')] });
+        queueEvent('server-telemetry', { entries: [telemetry('heartbeat', 'three')] });
+        flushFrame();
+        expect([...get(statusByServer).keys()]).toEqual(['two', 'three']);
+        expect([...get(telemetryByServer).keys()]).toEqual(['two', 'three']);
+        expect([...get(matchEventsByServer).keys()]).toEqual(['three']);
+        expect(get(matchEventsByServer).get('three')?.event).toBe('game_end');
+        expect(get(telemetryByServer).get('three')?.event).toBe('heartbeat');
+
+        queueEvent('server-status', { entries: [] });
+        queueEvent('server-telemetry', { snapshot: true, entries: [] });
+        flushFrame();
+        expect(get(statusByServer).size).toBe(0);
+        expect(get(telemetryByServer).size).toBe(0);
+        expect(get(matchEventsByServer).size).toBe(0);
+    });
+
+    it('retains event arrival times so buffered status cannot overwrite a newer HTTP snapshot', async () => {
+        const { fleetServers, refreshFleet, statusByServer, statusReceivedAt } = await setup();
+        await vi.advanceTimersByTimeAsync(1000);
+        queueEvent('server-status', { delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 1 })] });
+        await vi.advanceTimersByTimeAsync(1000);
+        snapshot = { ...snapshot, serverNow: epoch + 2000, servers: [server(epoch + 2000, { last_players: 4 })] };
+        await refreshFleet();
+        queueEvent('server-status', { delta: true, serverNow: epoch + 2000, entries: [status(epoch + 2000, { server: 'two' })] });
+        flushFrame();
+        expect(get(statusByServer).get('t5_zombie_theater_1')?.receivedAt).toBe(epoch + 1000);
+        expect(get(statusReceivedAt)).toBe(epoch + 2000);
+        expect(get(fleetServers)[0].last_players).toBe(4);
+    });
+
+    it('accounts for frame delay without rewinding the server clock', async () => {
+        const { uiNow, statusByServer } = await setup();
+        vi.stubGlobal('document', { visibilityState: 'hidden' });
+        queueEvent('server-status', { serverNow: epoch + 5000, entries: [status(epoch + 5000)] });
+        await vi.advanceTimersByTimeAsync(2000);
+        queueEvent('server-telemetry', { serverNow: epoch + 4000, entries: [] });
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(get(uiNow)).toBe(epoch);
+        expect(get(statusByServer).size).toBe(0);
+        expect(frames.size).toBe(1);
+        flushFrame();
+        expect(get(uiNow)).toBe(epoch + 8000);
+        expect(get(statusByServer).size).toBe(1);
+    });
+
+    it('cancels queued frames and discards buffered events when stopping and restarting', async () => {
+        const { statusByServer, telemetryByServer, startServerStream } = await setup();
+        const oldSource = source;
+        const close = vi.spyOn(oldSource, 'close');
+        queueEvent('server-status', { entries: [status(epoch)] });
+        queueEvent('server-telemetry', { entries: [{ ...gameEvent('game_start'), receivedAt: new Date(epoch).toISOString() }] });
+        const callback = [...frames.values()][0];
+        const oldStop = stop!;
+        oldStop();
+        expect(close).toHaveBeenCalledTimes(1);
+        expect(cancelAnimationFrame).toHaveBeenCalledTimes(1);
+        expect(frames.size).toBe(0);
+        callback(0);
+        expect(get(statusByServer).size).toBe(0);
+        expect(get(telemetryByServer).size).toBe(0);
+        stop = startServerStream();
+        oldStop();
+        oldSource.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({ entries: [status(epoch)] }) }));
+        queueEvent('server-status', { delta: true, entries: [status(epoch, { server: 'new' })] });
+        flushFrame();
+        expect([...get(statusByServer).keys()]).toEqual(['new']);
+        expect(get(telemetryByServer).size).toBe(0);
+    });
+});
+
 describe('Fleet report clock', () => {
     it('accepts newer navigation snapshots while keeping stream evidence newer than the snapshot', async () => {
         const { fleetServers, hydrateFleet } = await setup();
@@ -86,6 +225,7 @@ describe('Fleet report clock', () => {
         source.dispatchEvent(new MessageEvent('server-status', { data: JSON.stringify({
             delta: true, serverNow: epoch + 1000, entries: [status(epoch + 1000, { players: 2 })]
         }) }));
+        flushFrame();
         expect(get(statusByServer).size).toBe(2);
         expect(get(fleetServers).find(server => server.name === 't5_zombie_theater_1')?.last_players).toBe(2);
         const changes = vi.fn();
@@ -103,6 +243,7 @@ describe('Fleet report clock', () => {
             source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + seconds * 1000,
                 entries: [{ ...gameEvent(event, 2, players), receivedAt: at(seconds),
                     timing: { matchId: 1, players: {}, server: timing } }] }) }));
+            flushFrame();
             return get(fleetServers)[0];
         };
         const active = { activeSince: at(1), inactiveSince: null, gameStartedAt: at(1), gameEndedAt: null };
@@ -132,15 +273,17 @@ describe('Fleet report clock', () => {
             entries: [{ ...gameEvent('heartbeat', 2, 1), receivedAt: new Date(epoch).toISOString(),
                 timing: { matchId: 1, players: {}, server: { activeSince: startedAt, inactiveSince: null,
                     gameStartedAt: startedAt, gameEndedAt: null } } }] }) }));
+        flushFrame();
         expect(get(fleetServers)[0]).toMatchObject({ active_since: startedAt, game_started_at: startedAt });
     });
 
-    it('updates count and roster immediately from telemetry and expires them together', async () => {
+    it('updates count and roster on the next frame and expires them together', async () => {
         snapshot.servers = [server(epoch, { container_state: 'running', container_checked_at: new Date(epoch).toISOString(), last_players: 4 })];
         const { fleetServers, uiNow } = await setup();
         await vi.advanceTimersByTimeAsync(1000);
         source.dispatchEvent(new MessageEvent('server-telemetry', { data: JSON.stringify({ serverNow: epoch + 1000,
             entries: [{ ...gameEvent('heartbeat', 15, 3), receivedAt: new Date(epoch + 1000).toISOString() }] }) }));
+        flushFrame();
         expect(get(fleetServers)[0]).toMatchObject({ health: 'online', last_players: 3, status_source: 'telemetry' });
         expect(get(fleetServers)[0].roster).toHaveLength(3);
         expect(matchesFleet(get(fleetServers)[0], 'Player 3', 'occupied', '', [], [], get(uiNow))).toBe(true);
@@ -149,12 +292,15 @@ describe('Fleet report clock', () => {
         expect(matchesFleet(get(fleetServers)[0], '', 'occupied', '', [], [], get(uiNow))).toBe(false);
     });
 
-    it('applies game lifecycle events immediately and clears them after an empty report', async () => {
+    it('applies game lifecycle events on the next frame and clears them after an empty report', async () => {
         const { fleetServers } = await setup();
-        const report = (event: string, time: number) => source.dispatchEvent(new MessageEvent('server-telemetry', {
+        const report = (event: string, time: number) => {
+            source.dispatchEvent(new MessageEvent('server-telemetry', {
                 data: JSON.stringify({ serverNow: time, entries: [{ server: 't5_zombie_theater_1', event,
                     receivedAt: new Date(time).toISOString(), players: [] }] })
             }));
+            flushFrame();
+        };
         emit(epoch, [status(epoch, { players: 1 })]);
         report('game_end', epoch + 1000);
         expect(get(fleetServers)[0].game_ended_at).toBe(new Date(epoch + 1000).toISOString());
@@ -249,6 +395,7 @@ describe('Fleet report clock', () => {
         source.dispatchEvent(new MessageEvent('server-telemetry', {
             data: JSON.stringify({ serverNow: epoch, entries: [], snapshot: true })
         }));
+        flushFrame();
         expect(get(uiNow)).toBe(epoch);
         await vi.advanceTimersByTimeAsync(1000);
         expect(get(uiNow)).toBe(epoch + 1000);

tests/unit/worker-scheduling.test.ts

@@ -31,6 +31,30 @@ async function setup() {
 }
 
 describe('independent provisioning scheduling', () => {
+    it('runs controller library cleanup on worker ticks without overlapping a slow cleanup', async () => {
+        const s = await setup();
+        const { cleanupControllerLibraries } = await import('../../src/lib/server/game-files/index.js');
+        const cleanup = deferred();
+        vi.mocked(cleanupControllerLibraries).mockImplementationOnce(() => cleanup.promise);
+        vi.useFakeTimers();
+        try {
+            s.startWorker();
+            expect(cleanupControllerLibraries).toHaveBeenCalledTimes(1);
+            await vi.advanceTimersByTimeAsync(30_000);
+            expect(cleanupControllerLibraries).toHaveBeenCalledTimes(1);
+            expect(s.process.mock.calls.length).toBeGreaterThan(1);
+            cleanup.resolve();
+            await vi.advanceTimersByTimeAsync(0);
+            await vi.advanceTimersByTimeAsync(15_000);
+            expect(cleanupControllerLibraries).toHaveBeenCalledTimes(2);
+        } finally {
+            s.stopWorker(); cleanup.resolve(); s.background.resolve();
+            await vi.advanceTimersByTimeAsync(0);
+            vi.useRealTimers();
+            await vi.waitFor(() => expect(s.workerActivity()).toEqual([]));
+        }
+    });
+
     it('admits and checks new work while campaign, beta and maintenance observers are still waiting', async () => {
         const s = await setup();
         try {