Read chat logs incrementally and drain retention (Refs #62)

agent/ec_agent.py

@@ -35,7 +35,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.8.4"
+VERSION = "1.8.5"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1497,7 +1497,7 @@ class Agent:
                 if line.startswith("PRETTY_NAME="):
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
-        capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
+        capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs", "logs.read-offset"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
         plutonium_cache = self.plutonium_cache_inventory()
         return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
@@ -2231,6 +2231,8 @@ class Agent:
         # swapping either the original link or a target component cannot escape.
         with managed_open(app, relative) as stream:
             stat = os.fstat(stream.fileno())
+            if payload.get("mode") == "incremental":
+                return self.read_log_chunk(stream, stat, payload.get("cursor"))
             start = max(0, stat.st_size - 256 * 1024)
             stream.seek(start)
             lines = stream.read(256 * 1024).decode("utf-8", "replace").splitlines()
@@ -2238,6 +2240,32 @@ class Agent:
             lines = lines[1:]
         return {"lines": lines[-count:], "size": stat.st_size, "mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)), "truncated": start > 0}
 
+    @staticmethod
+    def read_log_chunk(stream, stat, cursor):
+        file_id, modified = f"{stat.st_dev}:{stat.st_ino}", f"{stat.st_mtime_ns}:{stat.st_ctime_ns}"
+        size = stat.st_size
+        mtime = datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat()
+        if cursor is not None and (not isinstance(cursor, dict)
+                or any(type(cursor.get(key)) is not int or cursor[key] < 0 for key in ("offset", "size", "anchorLength"))
+                or cursor["anchorLength"] > min(256, cursor["offset"])
+                or any(not isinstance(cursor.get(key), str) for key in ("fileId", "modified", "anchorHash"))):
+            raise AgentError("Invalid log cursor")
+        if cursor and cursor["fileId"] == file_id and cursor["modified"] == modified and cursor["size"] == size and cursor["offset"] == size:
+            return {"contentBase64": "", "size": size, "mtime": mtime, "truncated": size > 0, "cursor": cursor, "reset": False}
+        reset = not cursor or cursor["fileId"] != file_id or cursor["offset"] > size
+        if not reset:
+            stream.seek(cursor["offset"] - cursor["anchorLength"])
+            anchor = stream.read(cursor["anchorLength"])
+            reset = len(anchor) != cursor["anchorLength"] or digest(anchor) != cursor["anchorHash"]
+        start = max(0, size - 256 * 1024) if reset else cursor["offset"]
+        stream.seek(start)
+        content = stream.read(min(256 * 1024, size - start))
+        offset = start + len(content)
+        stream.seek(max(0, offset - 256))
+        anchor = stream.read(min(256, offset))
+        return {"contentBase64": base64.b64encode(content).decode(), "size": size, "mtime": mtime, "truncated": start > 0, "reset": reset,
+                "cursor": {"fileId": file_id, "modified": modified, "size": size, "offset": offset, "anchorLength": len(anchor), "anchorHash": digest(anchor)}}
+
     def library_dir(self, game, library_id):
         if game not in ("t5", "t6"):
             raise AgentError("Unsupported library game")

agent/tests/test_agent.py

@@ -3,6 +3,7 @@ import importlib.util
 import http.client
 import io
 import json
+import os
 from pathlib import Path
 import tempfile
 import threading
@@ -2699,6 +2700,71 @@ class AgentTests(unittest.TestCase):
         self.agent.rcon = lambda *_: {"ok": True, "players": 0}
         self.assertEqual(self.agent.execute(operation)["state"], "completed")
 
+    def test_incremental_log_reads_append_bytes_without_rereading_unchanged_content(self):
+        self.create()
+        app = self.agent.instance_dir(INSTANCE, 1) / "app"
+        (app / "logs").mkdir()
+        path = app / "logs/games_sp.log"
+        path.write_bytes(b"first\n")
+        payload = {"path": "logs/games_sp.log", "mode": "incremental"}
+        result = self.agent.execute(self.operation("logs.read", payload))
+        self.assertEqual(result["state"], "completed")
+        first = json.loads(result["result"])
+        self.assertEqual(base64.b64decode(first["contentBase64"]), b"first\n")
+        self.assertTrue(first["reset"])
+        # An unchanged stat must not read even the cursor anchor.
+        with path.open("rb") as stream, patch.object(stream, "read", side_effect=AssertionError("unchanged file was read")):
+            unchanged = self.agent.read_log_chunk(stream, os.fstat(stream.fileno()), first["cursor"])
+        self.assertEqual(unchanged["contentBase64"], "")
+        with path.open("ab") as stream:
+            stream.write("café\n".encode())
+        appended = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": first["cursor"]})
+        self.assertEqual(base64.b64decode(appended["contentBase64"]), "café\n".encode())
+        self.assertFalse(appended["reset"])
+        self.assertEqual(appended["cursor"]["offset"], len("first\ncafé\n".encode()))
+        self.assertIn("logs.read-offset", self.agent.inventory()["capabilities"])
+
+    def test_incremental_log_rotation_and_copy_truncate_reset_the_cursor(self):
+        self.create()
+        app = self.agent.instance_dir(INSTANCE, 1) / "app"
+        (app / "logs").mkdir()
+        path = app / "logs/games_sp.log"
+        header = b"same game header\n" * 30
+        path.write_bytes(header + b"original\n")
+        payload = {"path": "logs/games_sp.log", "mode": "incremental"}
+        result = self.agent.read_log(INSTANCE, 1, payload)
+        path.rename(path.with_suffix(".old"))
+        for content in (header + b"rotated!\n", header + b"rewritten and longer\n", b"short\n", b""):
+            with self.subTest(content=content):
+                path.write_bytes(content)
+                result = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": result["cursor"]})
+                self.assertTrue(result["reset"])
+                self.assertEqual(base64.b64decode(result["contentBase64"]), content)
+
+    def test_incremental_log_backlogs_are_bounded_and_preserve_split_utf8_bytes(self):
+        self.create()
+        app = self.agent.instance_dir(INSTANCE, 1) / "app"
+        (app / "logs").mkdir()
+        path = app / "logs/games_sp.log"
+        window = 256 * 1024
+        path.write_bytes(b"x" * (window + 10))
+        payload = {"path": "logs/games_sp.log", "mode": "incremental"}
+        initial = self.agent.read_log(INSTANCE, 1, payload)
+        self.assertTrue(initial["truncated"])
+        self.assertEqual(len(base64.b64decode(initial["contentBase64"])), window)
+        data = b"y" * (window - 1) + "é\n".encode()
+        with path.open("ab") as stream:
+            stream.write(data)
+        first = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": initial["cursor"]})
+        second = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": first["cursor"]})
+        self.assertFalse(first["reset"])
+        self.assertEqual(len(base64.b64decode(first["contentBase64"])), window)
+        self.assertEqual(base64.b64decode(first["contentBase64"]) + base64.b64decode(second["contentBase64"]), data)
+        self.assertEqual(second["cursor"]["offset"], second["size"])
+        for cursor in ({}, {**second["cursor"], "offset": -1}, {**second["cursor"], "anchorLength": 257}, {**second["cursor"], "offset": True}):
+            with self.subTest(cursor=cursor), self.assertRaisesRegex(AgentError, "Invalid log cursor"):
+                self.agent.read_log(INSTANCE, 1, {**payload, "cursor": cursor})
+
     def test_game_log_can_follow_internal_relative_plutainer_symlink(self):
         self.create()
         app = self.agent.instance_dir(INSTANCE, 1) / "app"
@@ -2709,6 +2775,8 @@ class AgentTests(unittest.TestCase):
         result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log", "lines": 2}))
         self.assertEqual(result["state"], "completed")
         self.assertEqual(json.loads(result["result"])["lines"], ["second", "third"])
+        incremental = self.agent.read_log(INSTANCE, 1, {"path": "logs/games_sp.log", "mode": "incremental"})
+        self.assertEqual(base64.b64decode(incremental["contentBase64"]), b"first\nsecond\nthird\n")
         with self.assertRaises(AgentError):
             module.managed_write(app, "logs/games_sp.log", b"must-not-write-through-link")
 
@@ -2722,6 +2790,8 @@ class AgentTests(unittest.TestCase):
         result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log"}))
         self.assertEqual(result["state"], "failed")
         self.assertIn("outside", result["error"])
+        with self.assertRaisesRegex(AgentError, "outside"):
+            self.agent.read_log(INSTANCE, 1, {"path": "logs/games_sp.log", "mode": "incremental"})
 
     def test_console_log_falls_back_to_owned_docker_output_including_stderr(self):
         self.create()

docs/guides/native-hosts.md

@@ -76,7 +76,19 @@ successful monitoring and file-read records. The default is one day; zero keeps
 them indefinitely. Actions, failures, negative checks, and unresolved operations
 are retained. Agent 1.4.9 also cleans up successful routine journal records after
 the controller confirms their final receipt. Older unconfirmed journal entries
-remain until a replay supplies that confirmation. Cleanup runs in bounded batches.
+remain until a replay supplies that confirmation. Controller cleanup runs every
+five minutes and drains eligible records in batches, yielding between batches.
+
+Log-read operation history stores metadata only. The waiting reader receives the
+log content directly; saved match chat remains with its match. Background cleanup
+also removes log bodies left in controller history by earlier versions.
+
+Agent 1.8.5 adds incremental game-log reads for chat collection. After the initial
+tail, the controller requests only appended bytes, handles rotation and truncation,
+and buffers incomplete lines. Unchanged logs need no content transfer or chat
+parsing. Background collection skips offline hosts, stopped servers and known
+empty servers, with one final read after players leave. Opening chat still reads
+on demand. Older agents use tail reads until they update.
 
 ### New game files and cached versions
 

src/lib/server/backends/files.ts

@@ -83,3 +83,11 @@ export async function nativeTail(path: string, lines: number) {
     if (!target) return undefined;
     return nativeRequest<import('../fleet/logs.js').TailResult | null>(target.instance, 'logs.read', { path: target.path, lines });
 }
+
+export async function nativeLogChunk(path: string, cursor?: import('../fleet/logs.js').LogCursor) {
+    const target = remote(path);
+    if (!target) return undefined;
+    const incremental = getHost(target.instance.host_id)?.inventory.capabilities?.includes('logs.read-offset');
+    return nativeRequest<import('../fleet/logs.js').LogChunk | import('../fleet/logs.js').TailResult | null>(target.instance, 'logs.read',
+        { path: target.path, ...(incremental ? { mode: 'incremental', cursor } : { lines: 2000 }) });
+}

src/lib/server/fleet/chat.ts

@@ -4,26 +4,38 @@ import { enrichGameChatTimeline, type ChatMatch, type ChatRoundReport, type Cont
 import { parseGameEvent } from '../../telemetry.js';
 import { telemetryRetention, telemetryCutoff } from './telemetry-requests.js';
 import { getServer, listServers, type ServerRow } from './servers.js';
-import { gameLogKey, resolveLogPath, tailLog, type TailResult } from './logs.js';
+import { gameLogKey, resolveLogPath, readLogChunk, type LogCursor, type TailResult } from './logs.js';
+import { listHosts } from '../hosts/index.js';
+import { observationReader } from './status.js';
 
 /** Historical context is scoped to this registration, including round-one matches. */
 export function serverChatTimeline(server: ServerRow, tail: TailResult): ContextualChatEntry[] {
+    const unscoped = enrichGameChatTimeline(tail.lines, { mtime: tail.mtime, matches: [], reports: [] });
+    const times = unscoped.flatMap(entry => entry.context.realTime ? [entry.context.realTime] : []);
+    if (!times.length) return unscoped;
+    const since = times.reduce((a, b) => a < b ? a : b), until = times.reduce((a, b) => a > b ? a : b);
     const handle = db();
     const matches = handle.prepare(`SELECT id, started_at AS startedAt, ended_at AS endedAt,
         last_received_at AS lastReceivedAt, status, start_observed AS startObserved
         FROM leaderboard_matches WHERE server = ? AND server_deleted_at IS NULL
-        ORDER BY id DESC LIMIT 2000`).all(server.name) as Array<Omit<ChatMatch, 'startObserved'> & { startObserved: number }>;
+            AND started_at <= ? AND (COALESCE(ended_at,last_received_at) >= ? OR status='active')
+        ORDER BY id DESC LIMIT 2000`).all(server.name, until, since) as Array<Omit<ChatMatch, 'startObserved'> & { startObserved: number }>;
 
     // Diagnostic history survives deletion. Its receipt time must belong to the
     // current registration so a reused server name cannot inherit old rounds.
     const cutoff = telemetryCutoff();
+    const roundSince = new Date(Date.parse(since) - 120_000).toISOString();
+    const matchSince = matches.reduce((earliest, match) => match.startedAt < earliest ? match.startedAt : earliest, since);
     const reports = handle.prepare(`SELECT received_at AS receivedAt, event, round,
         player_count AS playerCount FROM telemetry_requests
         WHERE server = ? AND status = 200 AND received_at >= ?
             AND julianday(received_at) >= julianday(?)
+            AND (received_at >= ? OR (event='game_start' AND received_at >= ?))
+            AND received_at <= ?
             AND event IN ('game_start', 'round_change', 'heartbeat', 'game_end')
             AND player_count >= 0 AND (event = 'game_start' OR round >= 0)
-        ORDER BY id DESC LIMIT ?`).all(server.name, cutoff, server.created_at,
+        ORDER BY id DESC LIMIT ?`).all(server.name, cutoff, server.created_at, roundSince, matchSince,
+            new Date(Date.parse(until) + 120_000).toISOString(),
             (telemetryRetention().maxPerServer || 10000)) as ChatRoundReport[];
 
     // Saving gameplay does not depend on diagnostic logging succeeding. Keep the
@@ -77,6 +89,10 @@ function archiveChat(entries: ContextualChatEntry[]): void {
 export function currentMatchChat(server: ServerRow, tail: TailResult | null): ContextualChatEntry[] {
     const entries = tail ? serverChatTimeline(server, tail) : [];
     archiveChat(entries);
+    return currentChatEntries(server, entries);
+}
+
+function currentChatEntries(server: ServerRow, entries: ContextualChatEntry[]): ContextualChatEntry[] {
     const match = db().prepare(`SELECT id, started_at AS startedAt FROM leaderboard_matches
         WHERE server = ? AND server_deleted_at IS NULL ORDER BY id DESC LIMIT 1`)
         .get(server.name) as { id: number; startedAt: string } | undefined;
@@ -101,6 +117,12 @@ export function currentMatchChat(server: ServerRow, tail: TailResult | null): Co
 }
 
 const pending = new Map<string, Promise<{ entries: ContextualChatEntry[]; missing: boolean }>>();
+interface ChatReadState {
+    key: string; cursor?: LogCursor; lines: string[]; partial: Buffer; skipLine: boolean;
+    signature?: string; entries: ContextualChatEntry[]; contextVersion?: string; wasOccupied: boolean;
+}
+const readers = new Map<string, ChatReadState>();
+const failures = new Map<string, number>();
 
 /** Shared by the live API and the worker; collection does not require a viewer. */
 export function collectServerChat(server: ServerRow): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> {
@@ -110,26 +132,94 @@ export function collectServerChat(server: ServerRow): Promise<{ entries: Context
     const work = (async () => {
         const path = resolveLogPath(server.name, gameLogKey(server.name));
         if (!path) throw new Error('Invalid server log path');
-        const tail = await tailLog(path, 2000);
+        let saved = readers.get(server.name);
+        if (saved?.key !== key) saved = undefined;
+        // Publish the new cursor only after its messages have been processed.
+        const state: ChatReadState = saved ? { ...saved } : { key, lines: [], partial: Buffer.alloc(0), skipLine: false, entries: [], wasOccupied: false };
+        const tail = await readLogChunk(path, state.cursor);
         const current = getServer(server.name);
         // A remote read may finish after destruction or replacement.
         if (!current || current.destroyed_at || current.created_at !== server.created_at
             || current.native_instance_id !== server.native_instance_id || current.instance_generation !== server.instance_generation) {
             return { entries: [], missing: true };
         }
-        return { entries: currentMatchChat(current, tail), missing: tail === null };
+        if (!tail) {
+            readers.delete(server.name);
+            return { entries: currentChatEntries(current, []), missing: true };
+        }
+        let changed = false;
+        if ('cursor' in tail) {
+            if (tail.reset) { state.lines = []; state.partial = Buffer.alloc(0); state.skipLine = tail.truncated; state.entries = []; }
+            state.cursor = tail.cursor;
+            const bytes = Buffer.concat([state.partial, Buffer.from(tail.contentBase64, 'base64')]);
+            const end = bytes.lastIndexOf(10) + 1;
+            if (end) {
+                const lines = bytes.subarray(0, end).toString('utf8').split('\n').slice(0, -1).map(line => line.replace(/\r$/, ''));
+                if (state.skipLine) lines.shift();
+                state.skipLine = false;
+                state.lines = [...state.lines, ...lines].slice(-2000);
+                let length = 0, start = state.lines.length;
+                while (start > 0 && (length += Buffer.byteLength(state.lines[start - 1]) + 1) <= 256 * 1024) start--;
+                state.lines = state.lines.slice(start);
+                changed = true;
+            }
+            state.partial = Buffer.from(bytes.subarray(end));
+            if (state.partial.length > 256 * 1024) { state.partial = Buffer.alloc(0); state.skipLine = true; }
+            // File mtime anchors the final line only after the backlog has been read.
+            if (tail.cursor.offset < tail.size || state.partial.length) {
+                state.signature = undefined;
+                readers.set(server.name, state);
+                return { entries: currentChatEntries(current, state.entries), missing: false };
+            }
+        } else {
+            // Older agents remain readable during a rolling update.
+            changed = state.signature !== `${tail.size}:${tail.mtime}`;
+            if (changed) state.lines = tail.lines;
+        }
+        const signature = `${tail.size}:${tail.mtime}`;
+        const needsContext = state.entries.some(entry => entry.type === 'message' && entry.matchId === undefined);
+        const contextVersion = needsContext ? (db().prepare('SELECT received_at FROM server_telemetry WHERE server=?').get(server.name) as { received_at: string } | undefined)?.received_at : undefined;
+        if (changed || state.signature === undefined || (needsContext && contextVersion !== state.contextVersion)) {
+            state.entries = serverChatTimeline(current, { ...tail, lines: state.lines });
+            archiveChat(state.entries);
+            state.contextVersion = contextVersion;
+        }
+        state.signature = signature;
+        readers.set(server.name, state);
+        return { entries: currentChatEntries(current, state.entries), missing: false };
     })().finally(() => pending.delete(key));
     pending.set(key, work);
     return work;
 }
 
 export async function collectFleetChat(): Promise<void> {
-    const servers = listServers().filter(server => !server.destroyed_at);
+    const hosts = new Map(listHosts().map(host => [host.id, host]));
+    const observe = observationReader();
+    const live = listServers().filter(server => !server.destroyed_at);
+    const names = new Set(live.map(server => server.name));
+    for (const name of readers.keys()) if (!names.has(name)) readers.delete(name);
+    for (const name of failures.keys()) if (!names.has(name)) failures.delete(name);
+    const servers = live.filter(server => {
+        if (server.host_id && !hosts.get(server.host_id)?.connected) return false;
+        if (server.container_state && !['running', 'unknown'].includes(server.container_state)) return false;
+        const players = observe(server).last_players ?? server.last_players;
+        // One final read after players leave retains messages written at match end.
+        return players !== 0 || readers.get(server.name)?.wasOccupied;
+    });
     // Bound agent traffic and isolate failures to their own server.
     async function collect() {
         for (let server = servers.shift(); server; server = servers.shift()) {
-            try { await collectServerChat(server); }
-            catch (error) { console.error(`[chat] Could not collect chat for ${server.name}`, error); }
+            try {
+                await collectServerChat(server);
+                const state = readers.get(server.name);
+                if (state) state.wasOccupied = (observe(server).last_players ?? server.last_players ?? 0) > 0;
+                failures.delete(server.name);
+            } catch (error) {
+                if (Date.now() - (failures.get(server.name) ?? -Infinity) >= 5 * 60_000) {
+                    console.error(`[chat] Could not collect chat for ${server.name}`, error);
+                    failures.set(server.name, Date.now());
+                }
+            }
         }
     }
     await Promise.all(Array.from({ length: Math.min(4, servers.length) }, collect));

src/lib/server/fleet/logs.ts

@@ -1,7 +1,8 @@
 import { open, stat } from 'node:fs/promises';
+import { createHash } from 'node:crypto';
 import { resolve, sep } from 'node:path';
 import { rootOfServer, gameOfServer } from '../config.js';
-import { nativeTail } from '../backends/files.js';
+import { nativeTail, nativeLogChunk } from '../backends/files.js';
 
 // Whitelisted log files exposed to the UI, mapped to their path relative
 // to each server's root directory. Adding a new supported log = one entry
@@ -48,6 +49,48 @@ export interface TailResult {
     truncated: boolean; // true when we didn't read from the start of the file
 }
 
+export interface LogCursor {
+    fileId: string; offset: number; size: number; modified: string;
+    anchorLength: number; anchorHash: string;
+}
+export interface LogChunk {
+    contentBase64: string; size: number; mtime: string; truncated: boolean;
+    cursor: LogCursor; reset: boolean;
+}
+
+/** Bounded byte reads; an anchor before the offset detects copy-truncate and regrowth. */
+export async function readLogChunk(path: string, cursor?: LogCursor): Promise<LogChunk | TailResult | null> {
+    const remote = await nativeLogChunk(path, cursor);
+    if (remote !== undefined) return remote;
+    let file;
+    try { file = await open(path, 'r'); }
+    catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null; throw error; }
+    try {
+        const st = await file.stat();
+        if (!st.isFile()) return null;
+        const fileId = `${st.dev}:${st.ino}`, modified = `${st.mtimeMs}:${st.ctimeMs}`, size = st.size;
+        const mtime = st.mtime.toISOString();
+        if (cursor?.fileId === fileId && cursor.modified === modified && cursor.size === size && cursor.offset === size) {
+            return { contentBase64: '', size, mtime, truncated: cursor.offset > 0, cursor, reset: false };
+        }
+        const hash = (value: Buffer) => createHash('sha256').update(value).digest('hex');
+        let reset = !cursor || cursor.fileId !== fileId || cursor.offset > size;
+        if (cursor && !reset) {
+            const anchor = Buffer.alloc(cursor.anchorLength);
+            const read = await file.read(anchor, 0, anchor.length, cursor.offset - anchor.length);
+            reset = read.bytesRead !== anchor.length || hash(anchor) !== cursor.anchorHash;
+        }
+        const start = reset ? Math.max(0, size - READ_WINDOW_BYTES) : cursor!.offset;
+        const content = Buffer.alloc(Math.min(READ_WINDOW_BYTES, size - start));
+        const read = await file.read(content, 0, content.length, start);
+        const offset = start + read.bytesRead;
+        const anchor = Buffer.alloc(Math.min(256, offset));
+        await file.read(anchor, 0, anchor.length, offset - anchor.length);
+        return { contentBase64: content.subarray(0, read.bytesRead).toString('base64'), size, mtime, truncated: start > 0, reset,
+            cursor: { fileId, modified, size, offset, anchorLength: anchor.length, anchorHash: hash(anchor) } };
+    } finally { await file.close(); }
+}
+
 const DEFAULT_TAIL_LINES = 200;
 const MAX_TAIL_LINES = 2000;
 const READ_WINDOW_BYTES = 256 * 1024; // 256 KB — plenty for a few hundred lines

src/lib/server/hosts/index.ts

@@ -8,6 +8,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 { logResultMetadata } from './log-results.js';
 import { parseHostInventory } from './request-context.js';
 import { fleetSettings } from '../settings.js';
 import { recordInstanceResources, validateInstanceResources } from './calibration.js';
@@ -19,6 +20,8 @@ export const AGENT_PROTOCOL_VERSION = 1;
 export const HEARTBEAT_MAX_AGE_MS = 90_000;
 const now = () => new Date().toISOString();
 const workEvents = new EventEmitter();
+// Hold reply bodies only while a reader is awaiting this operation. No TTL cache or disk copy.
+const logReaders = new Map<string, Set<{ received: boolean; result?: unknown }>>();
 workEvents.setMaxListeners(0);
 export function waitForHostWork(hostId: string, signal: AbortSignal, timeoutMs = 20_000): Promise<void> {
     return new Promise(resolve => {
@@ -599,6 +602,7 @@ export function getHostOperationPayload(hostId: string, id: string): Buffer {
     return Buffer.from(JSON.stringify(unpackFileJson(JSON.parse(row.payload.toString()))));
 }
 export function reportHostOperationStatus(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperationStatus {
+    let liveLogResult: { value: unknown } | undefined;
     const operation = database().transaction(() => {
         requireHost(hostId);
         const operation = getHostOperationStatus(id);
@@ -612,7 +616,7 @@ export function reportHostOperationStatus(hostId: string, id: string, input: { s
         let result = input.result === undefined ? null : JSON.stringify(input.result);
         if (result !== null && Buffer.byteLength(result) > 128 * 1024 * 1024) throw new Error('Operation result exceeds the limit.');
         if (result !== null) {
-            const packed = packFileJson(input.result);
+            const packed = packFileJson(operation.kind === 'logs.read' ? logResultMetadata(input.result) : input.result);
             result = JSON.stringify(packed.value);
             referenceBlobs('operation-result', id, packed.hashes);
         }
@@ -620,6 +624,7 @@ export function reportHostOperationStatus(hostId: string, id: string, input: { s
             operation.kind === 'rcon.execute' ? JSON.parse(getHostOperationPayload(hostId, id).toString('utf8')) : {}, input.result);
         database().prepare('UPDATE ec_host_operations SET state=?,result=?,error=?,updated_at=?,observation_success=? WHERE id=?')
             .run(state, result, error, now(), Number(observation), id);
+        if (operation.kind === 'logs.read' && state === 'completed') liveLogResult = { value: input.result ?? null };
         if (operation.kind === 'host.decommission') {
             if (state === 'failed') database().prepare("UPDATE ec_host_decommissions SET stage='failed',error=?,updated_at=? WHERE operation_id=?").run(error ?? 'The agent could not start cleanup. Repair the installation, then retry deletion.', now(), id);
             else database().prepare("UPDATE ec_host_decommissions SET stage='running',updated_at=? WHERE operation_id=? AND stage IN ('queued','running')").run(now(), id);
@@ -632,6 +637,10 @@ export function reportHostOperationStatus(hostId: string, id: string, input: { s
         }
         return getHostOperationStatus(id)!;
     })();
+    if (liveLogResult) for (const reader of logReaders.get(id) ?? []) {
+        reader.received = true;
+        reader.result = liveLogResult.value;
+    }
     queueMicrotask(() => workEvents.emit(`operation:${id}`));
     return operation;
 }
@@ -640,6 +649,21 @@ export function reportHostOperation(hostId: string, id: string, input: { state:
     return getHostOperation(id)!;
 }
 export async function waitHostOperationStatus(id: string, timeoutMs = 30_000): Promise<HostOperationStatus> {
+    if (getHostOperationStatus(id)?.kind !== 'logs.read') return waitOperationStatus(id, timeoutMs);
+    const reader: { received: boolean; result?: unknown } = { received: false };
+    const readers = logReaders.get(id) ?? new Set();
+    readers.add(reader); logReaders.set(id, readers);
+    try {
+        const operation = await waitOperationStatus(id, timeoutMs);
+        if (operation.state !== 'completed') return operation;
+        if (!reader.received) throw new Error('Log reply is no longer available. Request a new read.');
+        return { ...operation, result: reader.result };
+    } finally {
+        readers.delete(reader);
+        if (!readers.size) logReaders.delete(id);
+    }
+}
+async function waitOperationStatus(id: string, timeoutMs: number): Promise<HostOperationStatus> {
     const deadline = Date.now() + timeoutMs;
     while (true) {
         const operation = getHostOperationStatus(id);
@@ -668,8 +692,9 @@ export async function waitHostOperationStatus(id: string, timeoutMs = 30_000): P
     }
 }
 export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise<HostOperation> {
-    await waitHostOperationStatus(id, timeoutMs);
-    return getHostOperation(id)!;
+    const status = await waitHostOperationStatus(id, timeoutMs);
+    const operation = getHostOperation(id)!;
+    return status.kind === 'logs.read' ? { ...operation, result: status.result } : operation;
 }
 
 export function hostPortAvailable(hostId: string, candidate: number, bindAddress?: string, protocol: 'udp' | 'tcp' = 'udp', instanceId?: string): boolean {

src/lib/server/hosts/log-results.ts

@@ -0,0 +1,14 @@
+/** Log contents belong to the waiting reader, never to operation history. */
+export function logResultMetadata(value: unknown): Record<string, unknown> | null {
+    if (value === null || value === undefined) return null;
+    const result = value as Record<string, unknown>;
+    return {
+        retained: false,
+        ...(typeof result.size === 'number' ? { size: result.size } : {}),
+        ...(typeof result.mtime === 'string' ? { mtime: result.mtime.slice(0, 100) } : {}),
+        ...(typeof result.truncated === 'boolean' ? { truncated: result.truncated } : {}),
+        ...(typeof result.reset === 'boolean' ? { reset: result.reset } : {}),
+        ...(Array.isArray(result.lines) ? { lineCount: result.lines.length } : {}),
+        ...(typeof result.contentBase64 === 'string' ? { bytesRead: Buffer.byteLength(result.contentBase64, 'base64') } : {})
+    };
+}

src/lib/server/pipeline/worker.ts

@@ -1,4 +1,4 @@
-import { pruneHistory, pruneSampleHistory } from '../retention.js';
+import { drainRetention } from '../retention.js';
 import { collectFleetChat } from '../fleet/chat.js';
 import { runAutoscaling, pruneAutoscaling } from '../autoscaling/index.js';
 import { runRebalancing, pruneRebalancing } from '../rebalancing/index.js';
@@ -64,7 +64,7 @@ function reconcileRequesters(): void {
         ['rebalancing', 'Reconciling rebalance moves', runRebalancing],
         ['game-files', 'Removing unused game files', cleanupControllerLibraries],
         ['chat', 'Saving match chat', collectFleetChat],
-        ['retention', 'Cleaning expired history', async () => { pruneHistory(); pruneSampleHistory(); pruneAutoscaling(); pruneRebalancing(); }]
+        ['retention', 'Cleaning expired history', async () => { await drainRetention(); pruneAutoscaling(); pruneRebalancing(); }]
     ];
     for (const [key, label, work] of requesters) {
         if (requesterFlights.has(key)) continue;

src/lib/server/retention.ts

@@ -6,6 +6,7 @@ import { runtimeLaunchSql } from './fleet/runtime-config.js';
 import { ensureHostSchema } from './hosts/schema.js';
 import { ensureGameFileSchema } from './game-files/schema.js';
 import { ensureAuthSchema } from './auth/schema.js';
+import { logResultMetadata } from './hosts/log-results.js';
 import type { FleetSettings } from '../fleet-settings.js';
 
 interface Policy {
@@ -298,11 +299,45 @@ export function sampleRetentionPreview(settings = fleetSettings()) {
 }
 export function pruneSampleHistory() {
     initialize();
-    for (const p of samplePolicies().filter(p => p.enabled)) db().prepare(`DELETE FROM ${p.table} WHERE rowid IN (SELECT rowid FROM ${p.table} WHERE ${p.condition} LIMIT 5000)`).run(p.value);
+    let removed = 0;
+    for (const p of samplePolicies().filter(p => p.enabled)) removed += db().prepare(`DELETE FROM ${p.table} WHERE rowid IN (SELECT rowid FROM ${p.table} WHERE ${p.condition} LIMIT 5000)`).run(p.value).changes;
     const s = fleetSettings();
-    if (s.telemetryMaxPerServer) db().prepare(`DELETE FROM telemetry_requests WHERE id IN (SELECT id FROM
-        (SELECT id,ROW_NUMBER() OVER (PARTITION BY server ORDER BY id DESC) AS rank FROM telemetry_requests) WHERE rank>? LIMIT 5000)`).run(s.telemetryMaxPerServer);
-    if (s.telemetryMaxTotal) db().prepare('DELETE FROM telemetry_requests WHERE id IN (SELECT id FROM telemetry_requests ORDER BY id DESC LIMIT 5000 OFFSET ?)').run(s.telemetryMaxTotal);
+    if (s.telemetryMaxPerServer) removed += db().prepare(`DELETE FROM telemetry_requests WHERE id IN (SELECT id FROM
+        (SELECT id,ROW_NUMBER() OVER (PARTITION BY server ORDER BY id DESC) AS rank FROM telemetry_requests) WHERE rank>? LIMIT 5000)`).run(s.telemetryMaxPerServer).changes;
+    if (s.telemetryMaxTotal) removed += db().prepare('DELETE FROM telemetry_requests WHERE id IN (SELECT id FROM telemetry_requests ORDER BY id DESC LIMIT 5000 OFFSET ?)').run(s.telemetryMaxTotal).changes;
+    return removed;
+}
+
+/** Drain every eligible batch, yielding between transactions so reads and receipts can run. */
+export async function drainRetention(): Promise<number> {
+    let total = 0, logCursor = 0;
+    initialize();
+    while (true) {
+        const removed = pruneHistory() + pruneSampleHistory();
+        total += removed;
+        if (!removed) break;
+        await new Promise<void>(resolve => setImmediate(resolve));
+    }
+    // Remove expired rows first; only compact bodies in history that will survive.
+    if (db().prepare("SELECT 1 FROM settings WHERE key='log_history_metadata_v1'").get()) return total;
+    while (true) {
+        const rows = db().prepare(`SELECT rowid AS rid,id,result FROM ec_host_operations
+            WHERE rowid>? AND kind='logs.read' AND json_valid(result)
+                AND (json_type(result,'$.lines')='array' OR json_type(result,'$.contentBase64')='text'
+                    OR json_type(result,'$.contentSha256')='text')
+            ORDER BY rowid LIMIT 50`).all(logCursor) as { rid: number; id: string; result: string }[];
+        db().transaction(() => {
+            for (const row of rows) {
+                db().prepare('UPDATE ec_host_operations SET result=? WHERE id=?')
+                    .run(JSON.stringify(logResultMetadata(JSON.parse(row.result))), row.id);
+                db().prepare("DELETE FROM file_blob_references WHERE owner_type='operation-result' AND owner_id=?").run(row.id);
+            }
+            if (!rows.length) db().prepare("INSERT OR IGNORE INTO settings(key,value) VALUES('log_history_metadata_v1','1')").run();
+        })();
+        if (!rows.length) return total;
+        logCursor = rows.at(-1)!.rid;
+        await new Promise<void>(resolve => setImmediate(resolve));
+    }
 }
 
 function pruneRoutineObservations(): number {

static/agent/ec-agent.py

@@ -35,7 +35,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.8.4"
+VERSION = "1.8.5"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1497,7 +1497,7 @@ class Agent:
                 if line.startswith("PRETTY_NAME="):
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
-        capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
+        capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs", "logs.read-offset"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
         plutonium_cache = self.plutonium_cache_inventory()
         return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
@@ -2231,6 +2231,8 @@ class Agent:
         # swapping either the original link or a target component cannot escape.
         with managed_open(app, relative) as stream:
             stat = os.fstat(stream.fileno())
+            if payload.get("mode") == "incremental":
+                return self.read_log_chunk(stream, stat, payload.get("cursor"))
             start = max(0, stat.st_size - 256 * 1024)
             stream.seek(start)
             lines = stream.read(256 * 1024).decode("utf-8", "replace").splitlines()
@@ -2238,6 +2240,32 @@ class Agent:
             lines = lines[1:]
         return {"lines": lines[-count:], "size": stat.st_size, "mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)), "truncated": start > 0}
 
+    @staticmethod
+    def read_log_chunk(stream, stat, cursor):
+        file_id, modified = f"{stat.st_dev}:{stat.st_ino}", f"{stat.st_mtime_ns}:{stat.st_ctime_ns}"
+        size = stat.st_size
+        mtime = datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat()
+        if cursor is not None and (not isinstance(cursor, dict)
+                or any(type(cursor.get(key)) is not int or cursor[key] < 0 for key in ("offset", "size", "anchorLength"))
+                or cursor["anchorLength"] > min(256, cursor["offset"])
+                or any(not isinstance(cursor.get(key), str) for key in ("fileId", "modified", "anchorHash"))):
+            raise AgentError("Invalid log cursor")
+        if cursor and cursor["fileId"] == file_id and cursor["modified"] == modified and cursor["size"] == size and cursor["offset"] == size:
+            return {"contentBase64": "", "size": size, "mtime": mtime, "truncated": size > 0, "cursor": cursor, "reset": False}
+        reset = not cursor or cursor["fileId"] != file_id or cursor["offset"] > size
+        if not reset:
+            stream.seek(cursor["offset"] - cursor["anchorLength"])
+            anchor = stream.read(cursor["anchorLength"])
+            reset = len(anchor) != cursor["anchorLength"] or digest(anchor) != cursor["anchorHash"]
+        start = max(0, size - 256 * 1024) if reset else cursor["offset"]
+        stream.seek(start)
+        content = stream.read(min(256 * 1024, size - start))
+        offset = start + len(content)
+        stream.seek(max(0, offset - 256))
+        anchor = stream.read(min(256, offset))
+        return {"contentBase64": base64.b64encode(content).decode(), "size": size, "mtime": mtime, "truncated": start > 0, "reset": reset,
+                "cursor": {"fileId": file_id, "modified": modified, "size": size, "offset": offset, "anchorLength": len(anchor), "anchorHash": digest(anchor)}}
+
     def library_dir(self, game, library_id):
         if game not in ("t5", "t6"):
             raise AgentError("Unsupported library game")

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.8.4",
+  "version": "1.8.5",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "7e7c0049161a6f64743210b787a8ec30bd32d726c30587962bc5c705d94dfbac",
+  "sha256": "56187a244b551af884db6bd6419daf7dd7d931fa1e5d5c5414559bb6e25f1094",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }

tests/unit/host-log-results.test.ts

@@ -0,0 +1,56 @@
+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 api = await import('../../src/lib/server/hosts/index.js');
+    const { db } = await import('../../src/lib/server/db/index.js');
+    const host = api.createHost({ name: 'Log host' });
+    const operation = api.queueHostOperation(host.id, 'logs.read', { path: 'logs/games_sp.log' });
+    return { ...api, host, operation, db };
+}
+
+describe('transient log results', () => {
+    it.each([
+        { lines: ['private log body'], size: 17, mtime: '2026-10-08T12:00:00Z', truncated: false },
+        { contentBase64: Buffer.from('private log body').toString('base64'), size: 17, reset: false, cursor: { offset: 17 } }
+    ])('delivers full replies to current readers while storing only metadata', async result => {
+        const api = await setup();
+        const first = api.waitHostOperationStatus(api.operation.id);
+        const second = api.waitHostOperation(api.operation.id);
+        const receipt = api.reportHostOperationStatus(api.host.id, api.operation.id, { state: 'completed', result });
+        expect((await first).result).toEqual(result);
+        expect((await second).result).toEqual(result);
+        expect(receipt.result).toMatchObject({ retained: false, size: 17 });
+        expect(api.getHostOperation(api.operation.id)?.result).toEqual(receipt.result);
+        const stored = api.db().prepare('SELECT result FROM ec_host_operations WHERE id=?').get(api.operation.id);
+        expect(JSON.stringify(stored)).not.toMatch(/private log body|contentBase64|cursor|"lines"/);
+        expect(api.db().prepare("SELECT * FROM file_blob_references WHERE owner_type='operation-result' AND owner_id=?").all(api.operation.id)).toEqual([]);
+        // Completed reads cannot serve a stale body, including duplicate receipts.
+        await expect(api.waitHostOperationStatus(api.operation.id)).rejects.toThrow('no longer available');
+        api.reportHostOperationStatus(api.host.id, api.operation.id, { state: 'completed', result });
+        await expect(api.waitHostOperationStatus(api.operation.id)).rejects.toThrow('no longer available');
+    });
+
+    it('preserves missing-log results and operation failures', async () => {
+        const api = await setup();
+        const missing = api.waitHostOperationStatus(api.operation.id);
+        api.reportHostOperationStatus(api.host.id, api.operation.id, { state: 'completed', result: null });
+        expect((await missing).result).toBeNull();
+        const failed = api.queueHostOperation(api.host.id, 'logs.read', {});
+        const failure = api.waitHostOperationStatus(failed.id);
+        api.reportHostOperationStatus(api.host.id, failed.id, { state: 'failed', error: 'Read failed' });
+        expect(await failure).toMatchObject({ state: 'failed', error: 'Read failed' });
+    });
+
+    it('discards bodies arriving after the reader has timed out', async () => {
+        const api = await setup();
+        await expect(api.waitHostOperationStatus(api.operation.id, 0)).rejects.toThrow();
+        api.reportHostOperationStatus(api.host.id, api.operation.id, { state: 'completed', result: { lines: ['late body'] } });
+        expect(api.getHostOperation(api.operation.id)?.result).toEqual({ retained: false, lineCount: 1 });
+        await expect(api.waitHostOperationStatus(api.operation.id)).rejects.toThrow('no longer available');
+    });
+});

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

@@ -39,6 +39,37 @@ async function setup() {
 }
 
 describe('routine host observation retention', () => {
+    it('drains more than one batch while yielding and preserving recent or failed operations', async () => {
+        const api = await setup();
+        const { drainRetention } = await import('../../src/lib/server/retention.js');
+        const retained = [api.add('logs.read', { age: 0 }), api.add('logs.read', { state: 'failed' }), api.add('instance.restart')];
+        api.handle.transaction(() => {
+            const insert = api.handle.prepare(`INSERT INTO ec_host_operations
+                (id,host_id,kind,payload,state,created_at,updated_at,observation_success)
+                VALUES(?,?,'logs.read','{}','completed',?,?,1)`);
+            for (let i = 0; i < 1501; i++) insert.run(`read-${i}`, api.host.id, at(2 * day), at(2 * day));
+        })();
+        let yielded = false;
+        setImmediate(() => { yielded = true; });
+        expect(await drainRetention()).toBe(1501);
+        expect(yielded).toBe(true);
+        expect((api.handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[]).map(row => row.id).sort()).toEqual(retained.sort());
+        expect(await drainRetention()).toBe(0);
+    });
+
+    it('strips existing log bodies in batches without deleting recent or protected history', async () => {
+        const api = await setup();
+        const { drainRetention, setRetentionControl } = await import('../../src/lib/server/retention.js');
+        const ids = Array.from({ length: 53 }, () => api.add('logs.read', { age: 0 }));
+        setRetentionControl({ category: 'observation', id: ids[0], protected: true });
+        api.handle.prepare("UPDATE ec_host_operations SET result=? WHERE kind='logs.read'")
+            .run(JSON.stringify({ lines: ['legacy log contents'], size: 20, truncated: false }));
+        expect(await drainRetention()).toBe(0);
+        for (const id of ids) expect(api.getHostOperation(id)?.result).toEqual({ retained: false, lineCount: 1, size: 20, truncated: false });
+        expect(api.handle.prepare("SELECT value FROM settings WHERE key='log_history_metadata_v1'").get()).toEqual({ value: '1' });
+        expect(api.handle.pragma('foreign_key_check')).toEqual([]);
+    });
+
     it('removes aged successful reads, manifests, and read-only RCON while retaining failures, actions, and recent evidence', async () => {
         const api = await setup();
         const removed = ['host.inspect', 'instance.inspect', 'files.read', 'files.stat', 'files.list', 'files.manifest', 'logs.read']

tests/unit/logs.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it, beforeEach, afterEach } from 'vitest';
-import { mkdirSync, rmSync, writeFileSync } from 'node:fs';
+import { appendFileSync, mkdirSync, renameSync, rmSync, statSync, utimesSync, writeFileSync } from 'node:fs';
 import { resolve } from 'node:path';
 import { tmpdir } from 'node:os';
 
@@ -12,7 +12,7 @@ process.env.REL_DEST = 'runtime/plutonium/storage/t5/raw/scripts/sp';
 process.env.BETA_ARCHIVE_DIR = resolve(TMP_ROOT, 'beta-archive');
 process.env.DB_PATH = resolve(TMP_ROOT, 'test.db');
 
-const { resolveLogPath, tailLog, isLogKey } = await import(
+const { resolveLogPath, tailLog, readLogChunk, isLogKey } = await import(
     '../../src/lib/server/fleet/logs.js'
 );
 
@@ -94,3 +94,67 @@ describe('tailLog', () => {
         expect(tail?.lines).toEqual(['a', 'b', 'c']);
     });
 });
+
+describe('incremental log reads', () => {
+    const path = resolve(LOG_DIR, 'games_sp.log');
+    async function read(cursor?: import('../../src/lib/server/fleet/logs.js').LogCursor) {
+        const chunk = await readLogChunk(path, cursor);
+        if (!chunk || !('cursor' in chunk)) throw new Error('Expected a byte chunk');
+        return { ...chunk, bytes: Buffer.from(chunk.contentBase64, 'base64') };
+    }
+
+    it('reads only appended bytes and returns no body for unchanged files', async () => {
+        expect(await readLogChunk(path)).toBeNull();
+        writeFileSync(path, 'first\n');
+        const first = await read();
+        expect(first.bytes.toString()).toBe('first\n');
+        expect(first.reset).toBe(true);
+        const unchanged = await read(first.cursor);
+        expect(unchanged.contentBase64).toBe('');
+        expect(unchanged.cursor).toEqual(first.cursor);
+        appendFileSync(path, 'café\n');
+        const appended = await read(first.cursor);
+        expect(appended.bytes.toString()).toBe('café\n');
+        expect(appended.reset).toBe(false);
+        expect(appended.cursor.offset).toBe(Buffer.byteLength('first\ncafé\n'));
+    });
+
+    it('resets after rotation, truncation and copy-truncate regrowth with a restored mtime', async () => {
+        const header = 'same game header\n'.repeat(30);
+        writeFileSync(path, header + 'original\n');
+        let chunk = await read();
+        renameSync(path, path + '.1');
+        writeFileSync(path, header + 'rotated!\n');
+        chunk = await read(chunk.cursor);
+        expect(chunk.reset).toBe(true);
+        expect(chunk.bytes.toString()).toBe(header + 'rotated!\n');
+        const mtime = statSync(path).mtime;
+        writeFileSync(path, header + 'rewritten and longer\n');
+        utimesSync(path, mtime, mtime);
+        chunk = await read(chunk.cursor);
+        expect(chunk.reset).toBe(true);
+        expect(chunk.bytes.toString()).toBe(header + 'rewritten and longer\n');
+        writeFileSync(path, 'short\n');
+        chunk = await read(chunk.cursor);
+        expect(chunk.reset).toBe(true);
+        expect(chunk.bytes.toString()).toBe('short\n');
+    });
+
+    it('bounds initial reads and drains appended backlogs even when the file stops changing', async () => {
+        const window = 256 * 1024;
+        writeFileSync(path, 'x'.repeat(window + 10));
+        const initial = await read();
+        expect(initial.bytes.length).toBe(window);
+        expect(initial.truncated).toBe(true);
+        const appended = Buffer.from('y'.repeat(window + 20));
+        appendFileSync(path, appended);
+        const first = await read(initial.cursor);
+        const second = await read(first.cursor);
+        expect(first.reset).toBe(false);
+        expect(first.bytes.length).toBe(window);
+        expect(first.cursor.offset).toBeLessThan(first.size);
+        expect(second.bytes.length).toBe(20);
+        expect(Buffer.concat([first.bytes, second.bytes])).toEqual(appended);
+        expect((await read(second.cursor)).contentBase64).toBe('');
+    });
+});

tests/unit/native-observations.test.ts

@@ -6,10 +6,10 @@ let fixture: Fixture;
 beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); });
 afterEach(async () => { await fixture.cleanup(); });
 
-async function setup() {
+async function setup(capabilities: string[] = []) {
     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 });
+    hosts.enrollHost(hosts.createEnrollment(host.id).token, { protocolVersion: 1, capabilities });
     const { db } = await import('../../src/lib/server/db/index.js');
     db().prepare("INSERT INTO servers(name) VALUES('test')").run();
     db().prepare(`INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at)
@@ -20,6 +20,19 @@ async function setup() {
 }
 
 describe('overlapping native observations', () => {
+    it.each([false, true])('negotiates incremental log reads with agent support: %s', async incremental => {
+        const api = await setup(incremental ? ['logs.read-offset'] : []);
+        const { nativeLogChunk } = await import('../../src/lib/server/backends/files.js');
+        const cursor = { fileId: '1:2', offset: 3, size: 3, modified: '4', anchorLength: 3, anchorHash: 'hash' };
+        const request = nativeLogChunk(join(fixture.t5, 'test', 'logs/games_sp.log'), cursor);
+        const operation = api.listHostOperations(api.host.id)[0];
+        expect(operation.payload).toEqual({ path: 'logs/games_sp.log', ...(incremental ? { mode: 'incremental', cursor } : { lines: 2000 }) });
+        const result = incremental ? { contentBase64: '', cursor, reset: false } : { lines: ['legacy tail'] };
+        api.reportHostOperationStatus(api.host.id, operation.id, { state: 'completed', result });
+        expect(await request).toEqual(result);
+        expect(api.getHostOperation(operation.id)?.result).toMatchObject({ retained: false });
+    });
+
     it.each(['files.read', 'files.list', 'files.stat', 'files.manifest', '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' }));

tests/unit/server-chat.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
-import { mkdir, rm, utimes, writeFile } from 'node:fs/promises';
+import { appendFile, mkdir, rm, utimes, writeFile } from 'node:fs/promises';
 import { join } from 'node:path';
 import { newFixture, type Fixture } from '../helpers/fixture.js';
 import { gameEvent } from '../helpers/telemetry.js';
@@ -53,6 +53,78 @@ async function get(server = SERVER) {
 }
 
 describe('server chat API', () => {
+    it('buffers partial UTF-8 lines and avoids context queries when the log is unchanged', async () => {
+        await report('game_start', 0);
+        await report('heartbeat', 40, 4);
+        await log(['60:50 say;abc;0;Player;first'], 50);
+        expect((await (await get()).json()).entries).toHaveLength(1);
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const prepare = vi.spyOn(db(), 'prepare');
+        await get();
+        expect(prepare.mock.calls.some(([sql]) => sql.includes('FROM telemetry_requests'))).toBe(false);
+        expect(prepare.mock.calls.some(([sql]) => sql.includes('last_received_at AS lastReceivedAt'))).toBe(false);
+        prepare.mockRestore();
+        const path = join(fixture.t5, SERVER, 'logs/games_sp.log');
+        const line = Buffer.from('61:00 say;abc;0;Player;café\n');
+        await appendFile(path, line.subarray(0, -2)); // Split inside é, before the newline.
+        await utimes(path, new Date(at(60)), new Date(at(60)));
+        expect((await (await get()).json()).entries).toHaveLength(1);
+        await appendFile(path, line.subarray(-2));
+        await utimes(path, new Date(at(60)), new Date(at(60)));
+        const body = await (await get()).json();
+        expect(body.entries.map((entry: any) => entry.message.message)).toEqual(['first', 'café']);
+        expect(body.entries[1].context.realTime).toBe(at(60));
+        expect((await (await get()).json()).entries).toEqual(body.entries);
+    });
+
+    it('retries appended bytes if archiving fails before publishing the cursor', async () => {
+        await report('game_start', 0);
+        await report('heartbeat', 40, 4);
+        await log(['60:50 say;abc;0;Player;first'], 50);
+        await get();
+        const path = join(fixture.t5, SERVER, 'logs/games_sp.log');
+        await appendFile(path, '61:00 say;abc;0;Player;retry me\n');
+        await utimes(path, new Date(at(60)), new Date(at(60)));
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const handle = db(), original = handle.prepare.bind(handle);
+        const prepare = vi.spyOn(handle, 'prepare').mockImplementation((sql: string) => {
+            if (sql.includes('INSERT INTO match_chat')) throw new Error('Temporary database failure');
+            return original(sql);
+        });
+        vi.spyOn(console, 'error').mockImplementation(() => {});
+        expect((await get()).status).toBe(500);
+        prepare.mockRestore();
+        expect((await (await get()).json()).entries.map((entry: any) => entry.message.message)).toEqual(['first', 'retry me']);
+    });
+
+    it('skips empty, stopped and offline servers and collects one last time after players leave', async () => {
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const { createHost } = await import('../../src/lib/server/hosts/index.js');
+        const logs = await import('../../src/lib/server/fleet/logs.js');
+        const read = vi.spyOn(logs, 'readLogChunk');
+        const { collectFleetChat } = await import('../../src/lib/server/fleet/chat.js');
+        db().prepare("UPDATE servers SET last_players=0,container_state='running'").run();
+        await collectFleetChat();
+        expect(read).not.toHaveBeenCalled();
+        db().prepare('UPDATE servers SET last_players=1 WHERE name=?').run(SERVER);
+        await log(['60:50 say;abc;0;Player;playing'], 50);
+        await collectFleetChat();
+        expect(read).toHaveBeenCalledTimes(1);
+        db().prepare('UPDATE servers SET last_players=0 WHERE name=?').run(SERVER);
+        await collectFleetChat();
+        await collectFleetChat();
+        expect(read).toHaveBeenCalledTimes(2);
+        db().prepare("UPDATE servers SET last_players=1,container_state='stopped' WHERE name=?").run(SERVER);
+        await collectFleetChat();
+        expect(read).toHaveBeenCalledTimes(2);
+        const host = createHost({ name: 'Offline' });
+        db().prepare("UPDATE servers SET container_state='running' WHERE name=?").run(SERVER);
+        db().prepare(`INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at)
+            VALUES('offline-instance',?,?,'library','image',28960,?)`).run(SERVER, host.id, at(0));
+        await collectFleetChat();
+        expect(read).toHaveBeenCalledTimes(2);
+    });
+
     it('keeps real timestamps and historical match context stable across refreshes', async () => {
         await report('game_start', 0);
         await report('round_change', 40, 4);
@@ -100,7 +172,7 @@ describe('server chat API', () => {
 
     it('reports read failures without returning server filesystem paths', async () => {
         const logs = await import('../../src/lib/server/fleet/logs.js');
-        vi.spyOn(logs, 'tailLog').mockRejectedValue(new Error('read failed at /private/server/log'));
+        vi.spyOn(logs, 'readLogChunk').mockRejectedValue(new Error('read failed at /private/server/log'));
         vi.spyOn(console, 'error').mockImplementation(() => {});
         const response = await get();
         expect(response.status).toBe(500);

tests/unit/worker-scheduling.test.ts

@@ -6,7 +6,7 @@ vi.mock('../../src/lib/server/pipeline/beta-runner.js', () => ({ betaTick: vi.fn
 vi.mock('../../src/lib/server/maintenance/index.js', () => ({ runMaintenance: vi.fn() }));
 vi.mock('../../src/lib/server/game-files/index.js', () => ({ cleanupControllerLibraries: vi.fn(async () => {}) }));
 vi.mock('../../src/lib/server/provisioning/worker.js', () => ({ processProvisioning: vi.fn(async () => {}) }));
-vi.mock('../../src/lib/server/retention.js', () => ({ pruneHistory: vi.fn(), pruneSampleHistory: vi.fn() }));
+vi.mock('../../src/lib/server/retention.js', () => ({ drainRetention: vi.fn(async () => 0) }));
 vi.mock('../../src/lib/server/fleet/chat.js', () => ({ collectFleetChat: vi.fn(async () => {}) }));
 
 function deferred() {
@@ -58,19 +58,16 @@ describe('independent provisioning scheduling', () => {
 
     it('runs retention at most every five minutes across ticks and explicit kicks', async () => {
         const s = await setup();
-        const { pruneHistory, pruneSampleHistory } = await import('../../src/lib/server/retention.js');
+        const { drainRetention } = await import('../../src/lib/server/retention.js');
         vi.useFakeTimers();
         try {
             s.startWorker();
-            expect(pruneHistory).toHaveBeenCalledTimes(1);
-            expect(pruneSampleHistory).toHaveBeenCalledTimes(1);
+            expect(drainRetention).toHaveBeenCalledTimes(1);
             await vi.advanceTimersByTimeAsync(285_000);
             await s.kick();
-            expect(pruneHistory).toHaveBeenCalledTimes(1);
-            expect(pruneSampleHistory).toHaveBeenCalledTimes(1);
+            expect(drainRetention).toHaveBeenCalledTimes(1);
             await vi.advanceTimersByTimeAsync(15_000);
-            expect(pruneHistory).toHaveBeenCalledTimes(2);
-            expect(pruneSampleHistory).toHaveBeenCalledTimes(2);
+            expect(drainRetention).toHaveBeenCalledTimes(2);
             expect(s.process.mock.calls.length).toBeGreaterThan(2);
         } finally {
             s.stopWorker(); s.background.resolve();
@@ -80,6 +77,27 @@ describe('independent provisioning scheduling', () => {
         }
     });
 
+    it('keeps a retention backlog running without overlapping it or blocking provisioning', async () => {
+        const s = await setup();
+        const { drainRetention } = await import('../../src/lib/server/retention.js');
+        const cleanup = deferred();
+        vi.mocked(drainRetention).mockImplementationOnce(async () => { await cleanup.promise; return 1501; });
+        vi.useFakeTimers();
+        try {
+            s.startWorker();
+            await vi.advanceTimersByTimeAsync(600_000);
+            expect(drainRetention).toHaveBeenCalledTimes(1);
+            expect(s.process.mock.calls.length).toBeGreaterThan(2);
+            cleanup.resolve();
+            await vi.advanceTimersByTimeAsync(0);
+        } finally {
+            s.stopWorker(); cleanup.resolve(); s.background.resolve();
+            await vi.advanceTimersByTimeAsync(0);
+            vi.useRealTimers();
+            await vi.waitFor(() => expect(s.workerActivity()).toEqual([]));
+        }
+    });
+
     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');