Fix native health polling and isolate file operation traffic

tests/unit/native-hosts.test.ts

@@ -15,6 +15,24 @@ async function enrolled() {
 }
 
 describe('native host enrollment and scoped authentication', () => {
+    it('reserves health delivery while a file backlog is pending and bounds background batches', async () => {
+        const api = await enrolled();
+        api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] });
+        const files = Array.from({ length: 30 }, (_, i) => api.queueHostOperation(api.host.id, 'files.read', { path: `file-${i}` }));
+        const status = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'status' });
+        const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {});
+        const mutation = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'map_restart' });
+        expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([status.id]);
+        api.reportHostOperation(api.host.id, status.id, { state: 'completed', result: {} });
+        expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([inspect.id]);
+        expect(api.pendingHostOperations(api.host.id, 'read').map(op => op.id)).toEqual(files.slice(0, 4).map(op => op.id));
+        expect(api.pendingHostOperations(api.host.id, 'write').map(op => op.id)).toEqual([mutation.id]);
+        // Rolling upgrades: older agents still receive health through read.
+        api.heartbeat(api.host.id, inventory);
+        files.forEach(file => api.reportHostOperation(api.host.id, file.id, { state: 'completed', result: {} }));
+        expect(api.pendingHostOperations(api.host.id, 'read').map(op => op.id)).toEqual([inspect.id]);
+    });
+
     it('separates observation and mutation delivery and wakes long polls on newly queued work', async () => {
         const api = await enrolled();
         const inspect = api.queueHostOperation(api.host.id, 'host.inspect', {});

tests/unit/native-observations.test.ts

@@ -0,0 +1,76 @@
+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(); });
+
+async function setup() {
+    const hosts = await import('../../src/lib/server/hosts/index.js');
+    const host = hosts.createHost({ name: 'Observation host' });
+    hosts.enrollHost(hosts.createEnrollment(host.id).token, { protocolVersion: 1 });
+    const { db } = await import('../../src/lib/server/db/index.js');
+    db().prepare("INSERT INTO servers(name,komodo_name) VALUES('test','')").run();
+    db().prepare(`INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at)
+        VALUES('instance','test',?,'library','image',28960,?)`).run(host.id, new Date().toISOString());
+    const native = await import('../../src/lib/server/backends/native.js');
+    const instance = native.nativeInstance('test')!;
+    return { ...hosts, ...native, db: db(), host, instance };
+}
+
+describe('overlapping native observations', () => {
+    it.each(['files.read', 'files.list', 'files.stat', 'logs.read', 'instance.inspect'])('shares %s only until completion', async kind => {
+        const api = await setup();
+        const requests = Array.from({ length: 4 }, () => api.nativeRequest(api.instance, kind, { path: 'same' }));
+        const operations = api.listHostOperations(api.host.id);
+        expect(operations).toHaveLength(1);
+        api.reportHostOperation(api.host.id, operations[0].id, { state: 'completed', result: { value: 'first' } });
+        expect(await Promise.all(requests)).toEqual(Array(4).fill({ value: 'first' }));
+        const fresh = api.nativeRequest(api.instance, kind, { path: 'same' });
+        const next = api.listHostOperations(api.host.id)[0];
+        expect(next.id).not.toBe(operations[0].id);
+        api.reportHostOperation(api.host.id, next.id, { state: 'completed', result: { value: 'changed' } });
+        expect(await fresh).toEqual({ value: 'changed' });
+    });
+
+    it('does not merge different paths, mutations, or reads across lifecycle submissions', async () => {
+        const api = await setup();
+        const requests = [api.nativeRequest(api.instance, 'files.read', { path: 'one' }),
+            api.nativeRequest(api.instance, 'files.read', { path: 'two' }),
+            api.nativeRequest(api.instance, 'rcon.execute', { command: 'map_restart' }),
+            api.nativeRequest(api.instance, 'rcon.execute', { command: 'map_restart' })];
+        api.db.prepare("INSERT INTO restart_rollout(server,requested_at) VALUES('test',?)").run(new Date().toISOString());
+        requests.push(api.nativeRequest(api.instance, 'files.read', { path: 'one' }));
+        const operations = api.listHostOperations(api.host.id);
+        expect(operations).toHaveLength(5);
+        for (const operation of operations) api.reportHostOperation(api.host.id, operation.id, { state: 'completed', result: {} });
+        await Promise.all(requests);
+    });
+
+    it('releases failed observations so the next request can recover', async () => {
+        const api = await setup();
+        const first = api.nativeRequest(api.instance, 'files.read', { path: 'one' });
+        const rejected = expect(first).rejects.toThrow('unavailable');
+        api.reportHostOperation(api.host.id, api.listHostOperations(api.host.id)[0].id, { state: 'failed', error: 'unavailable' });
+        await rejected;
+        const next = api.nativeRequest(api.instance, 'files.read', { path: 'one' });
+        expect(api.listHostOperations(api.host.id)).toHaveLength(2);
+        api.reportHostOperation(api.host.id, api.listHostOperations(api.host.id)[0].id, { state: 'completed', result: {} });
+        await next;
+    });
+
+    it('keeps replacement generations separate and rejects the old result', async () => {
+        const api = await setup();
+        const old = api.nativeRequest(api.instance, 'files.read', { path: 'one' });
+        const rejected = expect(old).rejects.toThrow('Instance was replaced');
+        api.db.prepare("UPDATE native_instances SET generation=2 WHERE id='instance'").run();
+        const current = api.nativeRequest(api.nativeInstance('test')!, 'files.read', { path: 'one' });
+        const operations = api.listHostOperations(api.host.id);
+        expect(operations).toHaveLength(2);
+        for (const operation of operations) api.reportHostOperation(api.host.id, operation.id, {
+            state: 'completed', result: { generation: operation.generation }
+        });
+        await rejected;
+        expect(await current).toEqual({ generation: 2 });
+    });
+});

tests/unit/rcon-status.test.ts

@@ -5,6 +5,7 @@ import { komodoConfigured, rconConfigured } from '../../src/lib/server/config.js
 import { getServer, type ServerRow } from '../../src/lib/server/fleet/servers.js';
 import { latestGameSubmission } from '../../src/lib/server/fleet/lifecycle-evidence.js';
 import { rconGetPlayerCount, rconPlutoniumVersion, rconStatusResult } from '../../src/lib/server/integrations/rcon.js';
+import { nativeInstance, nativeRequest } from '../../src/lib/server/backends/native.js';
 
 vi.mock('node:dgram', () => ({ createSocket: vi.fn() }));
 vi.mock('../../src/lib/server/config.js', () => ({
@@ -14,6 +15,7 @@ vi.mock('../../src/lib/server/config.js', () => ({
 vi.mock('../../src/lib/server/fleet/servers.js', () => ({ getServer: vi.fn(), setServerField: vi.fn() }));
 vi.mock('../../src/lib/server/fleet/lifecycle-evidence.js', () => ({ latestGameSubmission: vi.fn() }));
 vi.mock('../../src/lib/server/integrations/komodo.js', () => ({ komodoResolveGamePort: vi.fn() }));
+vi.mock('../../src/lib/server/backends/native.js', () => ({ nativeInstance: vi.fn(), nativeRequest: vi.fn() }));
 
 class FakeSocket extends EventEmitter {
     send = vi.fn();
@@ -47,6 +49,28 @@ afterEach(() => {
 });
 
 describe('RCON status transport', () => {
+    it.each([
+        ['timeout', 'timeout', 'Agent received no reply'],
+        ['unreachable', 'send-error', 'Agent could not reach'],
+        ['unconfigured', 'no-config', 'Agent RCON password'],
+        ['authentication', 'bad-reply', 'wrong RCON password'],
+        ['invalid-reply', 'bad-reply', 'Agent RCON reply is missing']
+    ])('preserves the native agent %s diagnosis', async (agentReason, reason, detail) => {
+        server.native_instance_id = 'instance';
+        vi.mocked(nativeInstance).mockReturnValue({ id: 'instance', generation: 1, host_id: 'host', port: 28960 } as never);
+        vi.mocked(nativeRequest).mockResolvedValue({ ok: false, reason: agentReason, text: '' });
+        expect(await rconStatusResult(server.name)).toMatchObject({ ok: false, reason, detail: expect.stringContaining(detail) });
+        expect(nativeRequest).toHaveBeenCalledOnce();
+        expect(sockets).toHaveLength(0);
+    });
+
+    it('uses the successful native reply after agent-side recovery', async () => {
+        server.native_instance_id = 'instance';
+        vi.mocked(nativeInstance).mockReturnValue({ id: 'instance', generation: 1, host_id: 'host', port: 28960 } as never);
+        vi.mocked(nativeRequest).mockResolvedValue({ ok: true, text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n' });
+        expect(await rconStatusResult(server.name)).toMatchObject({ ok: true, status: { players: 0, map: 'mp_kino' } });
+    });
+
     it('shares overlapping health and occupancy probes and probes again after completion', async () => {
         const health = rconStatusResult(server.name);
         const occupancy = rconGetPlayerCount(server.name);

tests/unit/server-observation.test.ts

@@ -30,13 +30,22 @@ describe('shared server observation', () => {
         expect(fallback).toMatchObject({ status_source: 'rcon', health: 'online', last_players: 4, last_seen_at: at(40_000) });
     });
 
-    it.each(['starting', 'restarting', 'unhealthy', 'exited', 'unknown'])('telemetry cannot override container state %s', state => {
+    it.each(['starting', 'restarting', 'unhealthy', 'exited'])('telemetry cannot override container state %s', state => {
         const selected = selectServerObservation(server({ health: state === 'starting' || state === 'restarting' ? 'starting' : 'offline', container_state: state }), report(), now);
         expect(selected.health).not.toBe('online');
         expect(selected.last_players).toBeNull();
         expect(selected.roster).toBeNull();
     });
 
+    it('uses fresh RCON with unknown container state, but does not substitute telemetry or extend freshness', () => {
+        const raw = server({ container_state: 'unknown', container_checked_at: at(-60_000) });
+        expect(selectServerObservation(raw, report(), now)).toMatchObject({ health: 'online', last_players: 4, status_source: 'rcon' });
+        expect(selectServerObservation(raw, report(46_000), now + 46_000).health).toBe('stale');
+        expect(selectServerObservation(raw, report(121_000), now + 121_000).health).toBe('offline');
+        expect(selectServerObservation({ ...raw, telemetry_after: at() }, report(), now).health).toBe('offline');
+        expect(selectServerObservation({ ...raw, last_seen_at: null }, report(), now).health).toBe('offline');
+    });
+
     it('requires fresh container evidence and reports after the latest lifecycle submission', () => {
         expect(selectServerObservation(server({ container_checked_at: at(-46_000) }), report(), now).health).not.toBe('online');
         expect(selectServerObservation(server({ health: 'provisioning' }), report(), now).health).toBe('provisioning');

tests/unit/server-status.test.ts

@@ -379,7 +379,7 @@ describe('database fleet and container readiness', () => {
         expect((await (await GET()).json()).servers[0]).toMatchObject({ health: 'online', last_error: null });
     });
 
-    it.each(['unhealthy', 'exited', undefined])('does not poll RCON with container state %s', async (state) => {
+    it.each(['unhealthy', 'exited', 'stopped', 'missing'])('does not poll RCON with container state %s', async (state) => {
         const { inspect, rcon, fanoutStatus } = await setup();
         inspect.mockResolvedValue({ ok: false, state, reason: 'not ready' });
         await postTelemetry(gameEvent('heartbeat', 15, 3, 'managed'));
@@ -389,6 +389,30 @@ describe('database fleet and container readiness', () => {
         expect((await (await GET()).json()).servers[0]).toMatchObject({ health: 'offline', last_players: null });
     });
 
+    it.each(['unknown', undefined, 'throw'])('automatically recovers through RCON when inspection returns %s', async state => {
+        const { inspect, rcon, fanoutStatus, getServer } = await setup();
+        const { GET } = await import('../../src/routes/api/servers/+server.js');
+        const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');
+        if (state === 'throw') inspect.mockRejectedValue(new Error('Inspection timed out'));
+        else inspect.mockResolvedValue({ ok: false, state, reason: 'Inspection timed out' });
+        await postTelemetry(gameEvent('heartbeat', 15, 3, 'managed'));
+        rcon.mockResolvedValue({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'status',
+            status: { players: 2, map: 'zombie_theater', gametype: 'zom', roster: [] } });
+        const events: unknown[] = [];
+        const unsubscribe = subscribe(event => events.push(event));
+        try { await fanoutStatus(); } finally { unsubscribe(); }
+        expect(rcon).toHaveBeenCalledOnce();
+        expect(getServer('managed')?.container_state).toBe('unknown');
+        expect((await (await GET()).json()).servers[0]).toMatchObject({ health: 'online', last_players: 2, status_source: 'rcon' });
+        expect(events).toContainEqual(expect.objectContaining({ type: 'server-status', entries: [expect.objectContaining({
+            health: 'online', players: 2, containerState: 'unknown'
+        })] }));
+
+        rcon.mockResolvedValue({ ok: false, endpoint: null, reason: 'timeout', detail: 'no reply' });
+        await fanoutStatus();
+        expect((await (await GET()).json()).servers[0]).toMatchObject({ health: 'stale', last_error: 'timeout:no reply' });
+    });
+
     it('waits for an EC restart request to settle before trusting the old healthy container', async () => {
         const { db, inspect, rcon, fanoutStatus, getServer } = await setup();
         db.prepare("INSERT INTO restart_rollout (server, requested_at) VALUES ('managed', ?)").run(new Date().toISOString());

tests/unit/watchdog.test.ts

@@ -68,6 +68,41 @@ async function sustainedFailure() {
 }
 
 describe('Watchdog detection and ownership', () => {
+    it('scans files every five minutes while retaining minute health checks and immediate manual scans', async () => {
+        const drift = await import('../../src/lib/server/fleet/drift.js');
+        const scan = vi.spyOn(drift, 'prepareDriftCampaign');
+        await watchdog.watchdogTick();
+        expect(scan).toHaveBeenCalledTimes(1);
+        for (let minute = 1; minute < 5; minute++) {
+            advance(watchdog.WATCHDOG_SCAN_MS);
+            await watchdog.watchdogTick();
+        }
+        expect(scan).toHaveBeenCalledTimes(1);
+        expect(komodo.komodoInspectHealthy).toHaveBeenCalledTimes(5);
+        advance(watchdog.WATCHDOG_SCAN_MS);
+        await watchdog.watchdogTick();
+        expect(scan).toHaveBeenCalledTimes(2);
+        await watchdog.watchdogTick(true);
+        expect(scan).toHaveBeenCalledTimes(3);
+        advance(watchdog.WATCHDOG_SCAN_MS);
+        database.db().prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date().toISOString(), server);
+        await watchdog.watchdogTick();
+        expect(scan).toHaveBeenCalledTimes(4);
+        scan.mockRestore();
+    });
+
+    it('retries failed file scans on the next health scan instead of retaining successful freshness', async () => {
+        const drift = await import('../../src/lib/server/fleet/drift.js');
+        const scan = vi.spyOn(drift, 'prepareDriftCampaign').mockRejectedValueOnce(new Error('Agent unavailable'));
+        await watchdog.watchdogTick();
+        expect(watchdog.watchdogOverview().scanError).toContain('Agent unavailable');
+        advance(watchdog.WATCHDOG_SCAN_MS);
+        await watchdog.watchdogTick();
+        expect(scan).toHaveBeenCalledTimes(2);
+        expect(watchdog.watchdogOverview().scanError).toBeNull();
+        scan.mockRestore();
+    });
+
     it('defaults to automatic recovery and debounces manual scans without accelerating the failure window', async () => {
         expect(watchdog.watchdogOverview().enabled).toBe(true);
         down();