Add configurable retention for successful routine operations

agent/ec_agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.8"
+VERSION = "1.4.9"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -516,6 +516,19 @@ class ReadWriteGate:
                 self.condition.notify_all()
 
 
+def read_operation(operation):
+    if not isinstance(operation, dict):
+        return False
+    if operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
+        return True
+    payload = operation.get("payload")
+    return operation.get("kind") == "rcon.execute" and isinstance(payload, dict) and payload.get("command") in ("status", "getstatus", "version")
+
+
+def valid_observation_keep_days(value):
+    return type(value) is int and 0 <= value <= 36500
+
+
 class Journal:
     def __init__(self, path, recover=True):
         self.db = sqlite3.connect(path)
@@ -525,8 +538,14 @@ class Journal:
         self.db.execute("PRAGMA journal_mode=WAL")
         self.db.execute("PRAGMA synchronous=FULL")
         self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
-        if "checkpoint" not in {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}:
+        columns = {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}
+        if "checkpoint" not in columns:
             self.db.execute("ALTER TABLE operations ADD COLUMN checkpoint TEXT")
+        if "reported_at" not in columns:
+            self.db.execute("ALTER TABLE operations ADD COLUMN reported_at REAL")
+        if "retention_eligible" not in columns:
+            self.db.execute("ALTER TABLE operations ADD COLUMN retention_eligible INTEGER NOT NULL DEFAULT 0")
+        self.db.execute("CREATE INDEX IF NOT EXISTS operations_observation_retention ON operations(updated,id) WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1")
         self.db.execute("UPDATE operations SET state='unknown', error='Agent restarted during execution; inspect before reconciliation', updated=? WHERE state='running'", (time.time(),))
         self.db.commit()
 
@@ -551,10 +570,37 @@ class Journal:
         return dict(row)
 
     def update(self, op_id, state, result=None, error=None):
-        self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=? WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id))
+        self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=?, reported_at=NULL, retention_eligible=0 WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id))
         self.db.commit()
         return self.get(op_id)
 
+    def confirm_completed(self, row):
+        """Only an acknowledged, unchanged completion can become disposable."""
+        if row["state"] != "completed":
+            return
+        eligible = False
+        try:
+            result = json.loads(row["result"]) if row.get("result") is not None else None
+            eligible = read_operation(json.loads(row["request"])) and not row.get("error") and not (
+                isinstance(result, dict) and (result.get("ok") is False or result.get("healthy") is False or result.get("error") not in (None, ""))
+            )
+        except (ValueError, TypeError, KeyError):
+            pass  # Malformed historical records are retained for inspection.
+        self.db.execute("""UPDATE operations SET reported_at=COALESCE(reported_at,?), retention_eligible=?
+            WHERE id=? AND state='completed' AND updated=? AND result IS ? AND error IS ?""",
+            (time.time(), int(bool(eligible)), row["id"], row["updated"], row.get("result"), row.get("error")))
+        self.db.commit()
+
+    def prune_observations(self, keep_days):
+        if not valid_observation_keep_days(keep_days) or keep_days == 0:
+            return 0
+        cutoff = time.time() - keep_days * 86400
+        removed = self.db.execute("""DELETE FROM operations WHERE id IN (
+            SELECT id FROM operations WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1
+                AND updated<? ORDER BY updated,id LIMIT 500)""", (cutoff,)).rowcount
+        self.db.commit()
+        return removed
+
     def close(self):
         self.db.close()
 
@@ -868,6 +914,7 @@ class Agent:
         self.read_claim_lock = threading.Lock()
         self.read_claims = set()
         self.cancelled_creations = frozenset()
+        self.observation_keep_days = 0  # Older controllers do not authorize cleanup.
         self.library_progress = {}
         self.library_transfers = {}
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
@@ -2118,7 +2165,7 @@ class Agent:
                     return self.journal.update(operation["id"], "failed", error="Interrupted library download could not be resumed")
             if kind == "files.stage":
                 return self.journal.update(operation["id"], "completed", self.stage(instance_id, generation, payload, operation["id"]))
-            if kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read") or kind == "rcon.execute" and payload.get("command") in ("status", "getstatus", "version"):
+            if self.read_operation(operation):
                 return self.journal.update(operation["id"], "completed", self.dispatch(operation))
             if kind in ("files.activate", "files.rollback") and checkpoint:
                 stage = self.instance_dir(instance_id, generation) / "stages" / identifier(payload["stageId"]) / "manifest.json"
@@ -2172,7 +2219,7 @@ class Agent:
             pass
         return row
 
-    def report(self, row):
+    def report(self, row, journal=None):
         payload = {"state": row["state"]}
         if row.get("result"):
             payload["result"] = json.loads(row["result"])
@@ -2182,13 +2229,19 @@ class Agent:
                 payload["state"] = "running"
         if row.get("error"):
             payload["error"] = row["error"]
-        return self.request("POST", "/api/agent/operations/" + urllib.parse.quote(row["id"], safe=""), payload)
+        response = self.request("POST", "/api/agent/operations/" + urllib.parse.quote(row["id"], safe=""), payload)
+        confirmed = response.get("operation") if isinstance(response, dict) else None
+        if row["state"] == "completed" and isinstance(confirmed, dict) and confirmed.get("id") == row["id"] and confirmed.get("state") == "completed":
+            (journal if journal is not None else self.journal).confirm_completed(row)
+        return response
 
     def heartbeat(self):
         response = self.request("POST", "/api/agent/heartbeat", {"inventory": self.inventory(), "operationId": self.current_operation})
         # Only the heartbeat thread replaces this immutable snapshot; it never
         # mutates the operation journal or races Docker cleanup against create.
         self.cancelled_creations = frozenset(value for value in response.get("cancelledCreations", []) if isinstance(value, str))
+        keep_days = response.get("observationKeepDays")
+        self.observation_keep_days = keep_days if valid_observation_keep_days(keep_days) else 0
         host = response.get("host", {})
         if isinstance(host.get("portMin"), int) and isinstance(host.get("portMax"), int):
             self.config["portRange"] = {"min": host["portMin"], "max": host["portMax"]}
@@ -2205,7 +2258,7 @@ class Agent:
 
     @staticmethod
     def read_operation(operation):
-        return operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
+        return read_operation(operation)
 
     def instance_lock(self, operation):
         key = operation.get("instanceId")
@@ -2219,12 +2272,19 @@ class Agent:
         # Each lane owns its SQLite connection and never writes executing_id,
         # checkpoints, or current_operation used by the mutation state machine.
         journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
+        next_prune = time.monotonic()
         try:
             while not self.stopping.is_set():
                 try:
                     if self.update_pending():
                         self.stopping.wait(3)
                         continue
+                    if lane == "read" and time.monotonic() >= next_prune:
+                        next_prune = time.monotonic() + 60
+                        with self.activity(read_only=True) as acquired:
+                            if acquired and not self.update_pending():
+                                if journal.prune_observations(self.observation_keep_days) == 500:
+                                    next_prune = time.monotonic() + 15
                     response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20")
                     handled = False
                     for operation in response.get("operations", []):
@@ -2242,7 +2302,7 @@ class Agent:
                                     continue
                                 row, _ = journal.accept(operation)
                                 if row["state"] in ("accepted", "unknown", "running"):
-                                    self.report(journal.update(operation["id"], "running"))
+                                    self.report(journal.update(operation["id"], "running"), journal=journal)
                                     with self.read_gate.read(), self.instance_lock(operation):
                                         try:
                                             result = self.dispatch(operation)
@@ -2251,7 +2311,7 @@ class Agent:
                                             row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable")
                                 # Receipt delivery can take seconds; it must not
                                 # hold the instance lock needed by health checks.
-                                self.report(row)
+                                self.report(row, journal=journal)
                                 handled = True
                         finally:
                             with self.read_claim_lock:

agent/tests/test_agent.py

@@ -57,6 +57,115 @@ class Response(io.BytesIO):
         self.headers = headers or {}
 
 
+class JournalRetentionTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.path = Path(self.temp.name) / "operations.sqlite3"
+        self.journal = Journal(self.path)
+        self.now = 2000000
+        self.sequence = 0
+
+    def tearDown(self):
+        self.journal.close()
+        self.temp.cleanup()
+
+    def record(self, kind="files.stat", payload=None, result=None, state="completed", error=None, at=None, confirmed=True):
+        self.sequence += 1
+        operation = {"id": f"operation-{self.sequence}", "kind": kind, "payload": payload or {}}
+        self.journal.accept(operation)
+        with patch.object(module.time, "time", return_value=self.now - 172800 if at is None else at):
+            row = self.journal.update(operation["id"], state, result, error)
+        if confirmed:
+            self.journal.confirm_completed(row)
+        return row
+
+    def prune(self, days=1):
+        with patch.object(module.time, "time", return_value=self.now):
+            return self.journal.prune_observations(days)
+
+    def test_cutoff_requires_an_older_confirmed_success(self):
+        old = self.record(at=self.now - 86401)
+        boundary = self.record(at=self.now - 86400)
+        recent = self.record(at=self.now - 10)
+        unconfirmed = self.record(confirmed=False)
+        self.assertEqual(self.prune(), 1)
+        with self.assertRaisesRegex(AgentError, "Unknown operation"):
+            self.journal.get(old["id"])
+        for row in (boundary, recent, unconfirmed):
+            self.assertEqual(self.journal.get(row["id"])["state"], "completed")
+
+    def test_zero_and_invalid_retention_preserve_every_record(self):
+        row = self.record()
+        for days in (0, -1, 36501, True, False, None, "1", 1.0, 1.5):
+            with self.subTest(days=days):
+                self.assertEqual(self.prune(days), 0)
+                self.assertEqual(self.journal.get(row["id"])["state"], "completed")
+
+    def test_all_read_kinds_are_eligible_but_mutations_keep_their_replay_journal(self):
+        for kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
+            self.record(kind, result={"healthy": True})
+        for command in ("status", "getstatus", "version"):
+            self.record("rcon.execute", {"command": command}, {"ok": True})
+        mutations = [self.record(kind) for kind in ("instance.restart", "instance.stop", "instance.create", "files.stage", "files.activate", "credentials.rotate")]
+        mutations.append(self.record("rcon.execute", {"command": "map_restart"}, {"ok": True}))
+        self.assertEqual(self.prune(), 10)
+        for row in mutations:
+            saved, accepted = self.journal.accept(json.loads(row["request"]))
+            self.assertFalse(accepted)
+            self.assertEqual(saved["id"], row["id"])
+
+    def test_errors_failures_and_unfinished_operations_are_retained(self):
+        retained = [self.record(result=result) for result in ({"ok": False}, {"healthy": False}, {"error": "partial result"}, {"error": False}, {"error": []})]
+        retained.append(self.record(error="read did not finish"))
+        retained.extend(self.record(state=state) for state in ("accepted", "running", "unknown", "failed"))
+        self.assertEqual(self.prune(), 0)
+        for row in retained:
+            self.assertEqual(self.journal.get(row["id"])["state"], row["state"])
+
+    def test_old_receipt_cannot_confirm_a_changed_or_reexecuted_result(self):
+        old = self.record(confirmed=False)
+        with patch.object(module.time, "time", return_value=self.now - 90000):
+            current = self.journal.update(old["id"], "completed", {"new": "result"})
+        self.journal.confirm_completed(old)
+        self.assertIsNone(self.journal.get(old["id"])["reported_at"])
+        self.assertEqual(self.prune(), 0)
+        self.journal.confirm_completed(current)
+        self.assertEqual(self.prune(), 1)
+
+    def test_pruning_uses_bounded_indexed_batches(self):
+        template = self.record()
+        self.journal.db.executemany("""INSERT INTO operations(id,fingerprint,request,state,result,updated,reported_at,retention_eligible)
+            VALUES(?,?,?,'completed',NULL,?,?,1)""", [
+                (f"batch-{index}", template["fingerprint"], template["request"], self.now - 172800, self.now - 172800)
+                for index in range(504)
+            ])
+        self.journal.db.commit()
+        self.assertEqual(self.prune(), 500)
+        self.assertEqual(self.journal.db.execute("SELECT COUNT(*) FROM operations").fetchone()[0], 5)
+        self.assertEqual(self.prune(), 5)
+        plan = self.journal.db.execute("""EXPLAIN QUERY PLAN SELECT id FROM operations
+            WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1 AND updated<? ORDER BY updated,id LIMIT 500""", (self.now,)).fetchall()
+        self.assertIn("operations_observation_retention", " ".join(str(tuple(row)) for row in plan))
+
+    def test_migration_keeps_legacy_completions_until_their_receipt_is_reacknowledged(self):
+        self.journal.close()
+        self.path.unlink()
+        legacy = module.sqlite3.connect(self.path)
+        legacy.execute("CREATE TABLE operations(id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
+        request = {"id": "legacy-read", "kind": "files.list", "instanceId": None, "generation": None, "payload": {}}
+        canonical = json.dumps(request, sort_keys=True, separators=(",", ":"))
+        legacy.execute("INSERT INTO operations VALUES(?,?,?,'completed',?,NULL,?)", (request["id"], module.digest(canonical.encode()), canonical, '{"entries":[]}', self.now - 172800))
+        legacy.commit()
+        legacy.close()
+        self.journal = Journal(self.path)
+        row, accepted = self.journal.accept(request)
+        self.assertFalse(accepted)
+        self.assertIsNone(row["reported_at"])
+        self.assertEqual(self.prune(), 0)
+        self.journal.confirm_completed(row)
+        self.assertEqual(self.prune(), 1)
+
+
 class ManifestTests(unittest.TestCase):
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory()
@@ -1418,6 +1527,85 @@ class AgentTests(unittest.TestCase):
         self.agent.heartbeat()
         self.assertEqual(self.agent.config["portRange"], {"min": 30000, "max": 30100})
 
+    def test_heartbeat_requires_valid_retention_and_never_changes_the_journal(self):
+        self.assertEqual(self.agent.observation_keep_days, 0)
+        with patch.object(self.agent.journal, "prune_observations", side_effect=AssertionError("heartbeat must not use the journal")), patch.object(self.agent.journal, "confirm_completed", side_effect=AssertionError("heartbeat must not use the journal")):
+            for value, expected in ((1, 1), (36500, 36500), (0, 0), (True, 0), (False, 0), (1.0, 0), ("1", 0), (-1, 0), (36501, 0), (None, 0)):
+                with self.subTest(value=value):
+                    self.agent.observation_keep_days = 3
+                    self.agent.request = lambda *_, value=value: {"observationKeepDays": value}
+                    self.agent.heartbeat()
+                    self.assertEqual(self.agent.observation_keep_days, expected)
+            self.agent.observation_keep_days = 3
+            self.agent.request = lambda *_: {}
+            self.agent.heartbeat()
+            self.assertEqual(self.agent.observation_keep_days, 0)
+
+    def test_lost_completed_receipt_is_retained_and_replay_reports_the_saved_result(self):
+        operation = self.operation("files.stat", {"path": "configs/server.cfg"})
+        self.agent.journal.accept(operation)
+        with patch.object(module.time, "time", return_value=module.time.time() - 172800):
+            row = self.agent.journal.update(operation["id"], "completed", {"size": 8})
+        self.agent.request = MagicMock(side_effect=AgentError("receipt lost"))
+        with self.assertRaisesRegex(AgentError, "receipt lost"):
+            self.agent.report(row)
+        self.assertIsNone(self.agent.journal.get(operation["id"])["reported_at"])
+        self.assertEqual(self.agent.journal.prune_observations(1), 0)
+        with patch.object(self.agent, "dispatch", side_effect=AssertionError("completed read must not be executed again")):
+            replay = self.agent.execute(operation)
+        self.assertEqual(replay["result"], row["result"])
+        for response in ({}, None, {"operation": {"id": "different", "state": "completed"}}, {"operation": {"id": operation["id"], "state": "running"}}):
+            self.agent.request = lambda *_, response=response: response
+            self.agent.report(replay)
+            self.assertIsNone(self.agent.journal.get(operation["id"])["reported_at"])
+        self.agent.request = lambda *_: {"operation": {"id": operation["id"], "state": "completed"}}
+        self.agent.report(replay)
+        self.assertIsNotNone(self.agent.journal.get(operation["id"])["reported_at"])
+        self.assertEqual(self.agent.journal.prune_observations(1), 1)
+
+    def test_threaded_read_lane_confirms_and_prunes_through_its_own_connection(self):
+        self.create()
+        old_operation = self.operation("files.list", op_id="old-observation")
+        self.agent.journal.accept(old_operation)
+        with patch.object(module.time, "time", return_value=module.time.time() - 172800):
+            old = self.agent.journal.update(old_operation["id"], "completed", {"entries": []})
+        self.agent.journal.confirm_completed(old)
+        self.agent.observation_keep_days = 1
+        operation = self.operation("files.stat", {"path": "configs/server.cfg"}, "thread-observation")
+        polls = 0
+        failures = []
+
+        def request(method, path, payload=None):
+            nonlocal polls
+            if method == "POST":
+                return {"operation": {"id": operation["id"], "state": payload["state"]}}
+            polls += 1
+            if polls == 1:
+                return {"operations": [operation]}
+            self.agent.stopping.set()
+            return {"operations": []}
+
+        def run():
+            try:
+                self.agent.read_loop()
+            except BaseException as error:
+                failures.append(error)
+
+        self.agent.request = request
+        with patch.object(self.agent.journal, "confirm_completed", side_effect=AssertionError("main journal used by reader")), patch.object(self.agent.journal, "prune_observations", side_effect=AssertionError("main journal used by reader")):
+            thread = threading.Thread(target=run)
+            thread.start()
+            thread.join(timeout=5)
+            self.agent.stopping.set()
+            thread.join(timeout=5)
+        self.assertFalse(thread.is_alive())
+        self.assertEqual(failures, [])
+        self.assertEqual(polls, 2)
+        self.assertIsNotNone(self.agent.journal.get(operation["id"])["reported_at"])
+        with self.assertRaisesRegex(AgentError, "Unknown operation"):
+            self.agent.journal.get(old_operation["id"])
+        self.assertEqual(self.agent.journal.get("create")["state"], "completed")
+
     def test_download_content_permissions_allow_container_read_access_with_private_umask(self):
         content = b"complete-game-file"
         manifest = self.manifest(content)

src/lib/fleet-settings.ts

@@ -1,5 +1,6 @@
 export interface FleetSettings {
     deployKeepDays: number;
+    observationKeepDays: number;
     restartConcurrency: number;
     createConcurrency: number;
     destroyConcurrency: number;
@@ -7,6 +8,7 @@ export interface FleetSettings {
 
 export const defaultFleetSettings: FleetSettings = {
     deployKeepDays: 7,
+    observationKeepDays: 1,
     restartConcurrency: 1,
     createConcurrency: 1,
     destroyConcurrency: 1
@@ -14,6 +16,7 @@ export const defaultFleetSettings: FleetSettings = {
 
 export const fleetSettingLabels: Record<keyof FleetSettings, string> = {
     deployKeepDays: 'Campaign history retention',
+    observationKeepDays: 'Routine operation history retention',
     restartConcurrency: 'Concurrent restarts',
     createConcurrency: 'Concurrent server startups',
     destroyConcurrency: 'Concurrent server destructions'
@@ -27,11 +30,14 @@ export function validateFleetSettings(input: unknown): FleetSettings {
     }
     for (const key of Object.keys(defaultFleetSettings) as (keyof FleetSettings)[]) {
         const value = values[key];
-        if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 1) {
-            throw new Error(`${fleetSettingLabels[key]} must be a positive whole number.`);
+        const minimum = key === 'observationKeepDays' ? 0 : 1;
+        if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < minimum) {
+            throw new Error(`${fleetSettingLabels[key]} must be ${minimum === 0 ? 'zero or a positive whole number' : 'a positive whole number'}.`);
+        }
+        if ((key === 'deployKeepDays' || key === 'observationKeepDays') && value > 36500) {
+            throw new Error(`${fleetSettingLabels[key]} must be at most 36500 days.`);
         }
-        if (key === 'deployKeepDays' && value > 36500) throw new Error('Campaign history retention must be at most 36500 days.');
     }
-    return { deployKeepDays: values.deployKeepDays as number, restartConcurrency: values.restartConcurrency as number,
+    return { deployKeepDays: values.deployKeepDays as number, observationKeepDays: values.observationKeepDays as number, restartConcurrency: values.restartConcurrency as number,
         createConcurrency: values.createConcurrency as number, destroyConcurrency: values.destroyConcurrency as number };
 }

src/lib/server/db/fleet-settings.ts

@@ -8,8 +8,9 @@ export function initializeFleetSettings(handle: Database.Database): void {
     const saved = handle.prepare('SELECT value FROM settings WHERE key = ?').get(FLEET_SETTINGS_KEY) as { value: string } | undefined;
     if (saved) {
         const settings = JSON.parse(saved.value);
-        if (!Object.hasOwn(settings, 'destroyConcurrency')) {
-            const upgraded = validateFleetSettings({ ...settings, destroyConcurrency: defaultFleetSettings.destroyConcurrency });
+        if (!Object.hasOwn(settings, 'destroyConcurrency') || !Object.hasOwn(settings, 'observationKeepDays')) {
+            const upgraded = validateFleetSettings({ destroyConcurrency: defaultFleetSettings.destroyConcurrency,
+                observationKeepDays: defaultFleetSettings.observationKeepDays, ...settings });
             handle.prepare('UPDATE settings SET value = ? WHERE key = ?').run(JSON.stringify(upgraded), FLEET_SETTINGS_KEY);
         }
         return;

src/lib/server/hosts/index.ts

@@ -4,6 +4,8 @@ import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypt
 import { isIP } from 'node:net';
 import { db } from '../db/index.js';
 import { ensureHostSchema } from './schema.js';
+import { successfulObservation } from './observations.js';
+import { fleetSettings } from '../settings.js';
 import { recordInstanceResources, validateInstanceResources } from './calibration.js';
 import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js';
 import type { Host, HostDecommission, HostDeletion, HostInventory, HostMode, HostOperation, HostOperationHistoryPage, HostOperationState } from '../../hosts.js';
@@ -372,6 +374,24 @@ export function listHostOperationHistory(page = 1): HostOperationHistoryPage {
         return { operations, page: currentPage, pageSize, total, totalPages };
     })();
 }
+
+/** Bound each cleanup pass so retained history cannot stall monitoring or receipts. */
+export function pruneHostObservations(): number {
+    const days = fleetSettings().observationKeepDays;
+    if (!Number.isSafeInteger(days) || days <= 0 || days > 36500) return 0;
+    const handle = database();
+    const cutoff = new Date(Date.now() - days * 86_400_000).toISOString();
+    return handle.transaction(() => {
+        const rows = handle.prepare(`SELECT id FROM ec_host_operations WHERE observation_success=1 AND state='completed'
+            AND updated_at<? ORDER BY updated_at LIMIT 500`).all(cutoff) as { id: string }[];
+        if (!rows.length) return 0;
+        const ids = rows.map(row => row.id);
+        const placeholders = ids.map(() => '?').join(',');
+        handle.prepare(`DELETE FROM ec_host_operation_events WHERE operation_id IN (${placeholders})`).run(...ids);
+        handle.prepare(`DELETE FROM ec_host_operation_payloads WHERE operation_id IN (${placeholders})`).run(...ids);
+        return handle.prepare(`DELETE FROM ec_host_operations WHERE id IN (${placeholders})`).run(...ids).changes;
+    })();
+}
 /** Heartbeats can tell an executing create to stop at its next safe command boundary. */
 export function pendingCreationCancellations(hostId: string): string[] {
     return (database().prepare(`SELECT json_extract(payload,'$.createOperationId') AS id FROM ec_host_operations
@@ -479,7 +499,10 @@ export function reportHostOperationStatus(hostId: string, id: string, input: { s
         const error = input.error === undefined || input.error === null ? null : String(input.error).slice(0, 4000);
         const 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.');
-        database().prepare('UPDATE ec_host_operations SET state=?,result=?,error=?,updated_at=? WHERE id=?').run(state, result, error, now(), id);
+        const observation = state === 'completed' && !error && successfulObservation(operation.kind,
+            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 === '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);

src/lib/server/hosts/observations.ts

@@ -0,0 +1,18 @@
+export const observationKinds = ['host.inspect', 'instance.inspect', 'files.read', 'files.list', 'files.stat', 'files.manifest', 'logs.read'];
+const commands = ['status', 'getstatus', 'version'];
+
+export function successfulObservation(kind: string, payload: Record<string, unknown>, result: unknown): boolean {
+    const routine = observationKinds.includes(kind) || kind === 'rcon.execute' && commands.includes(String(payload.command));
+    const value = result && typeof result === 'object' ? result as Record<string, unknown> : {};
+    return routine && value.ok !== false && value.healthy !== false && (value.error == null || value.error === '');
+}
+
+/** Uses the operation alias o; serialized payloads may live outside the routing row. */
+export const routineObservationSql = `(o.kind IN (${observationKinds.map(kind => `'${kind}'`).join(',')})
+    OR (o.kind='rcon.execute' AND json_extract(COALESCE(
+        (SELECT p.payload FROM ec_host_operation_payloads p WHERE p.operation_id=o.id),o.payload),'$.command') IN ('status','getstatus','version')))`;
+
+export const successfulObservationSql = `(o.state='completed' AND COALESCE(o.error,'')='' AND ${routineObservationSql}
+    AND CASE WHEN o.result IS NULL THEN 1 WHEN json_valid(o.result)
+        THEN json_extract(o.result,'$.ok') IS NOT 0 AND json_extract(o.result,'$.healthy') IS NOT 0
+            AND COALESCE(json_extract(o.result,'$.error'),'')='' ELSE 0 END)`;

src/lib/server/hosts/schema.ts

@@ -1,4 +1,5 @@
 import type Database from 'better-sqlite3';
+import { successfulObservationSql } from './observations.js';
 
 const initialized = new WeakSet<Database.Database>();
 export function ensureHostSchema(handle: Database.Database): void {
@@ -28,6 +29,7 @@ export function ensureHostSchema(handle: Database.Database): void {
             payload TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'pending'
                 CHECK (state IN ('pending','accepted','running','completed','failed','unknown')),
             result TEXT, error TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, last_delivered_at TEXT,
+            observation_success INTEGER NOT NULL DEFAULT 0,
             UNIQUE(host_id, idempotency_key)
         );
         CREATE INDEX IF NOT EXISTS ec_host_operations_queue ON ec_host_operations(host_id, state, created_at);
@@ -45,6 +47,7 @@ export function ensureHostSchema(handle: Database.Database): void {
             id INTEGER PRIMARY KEY AUTOINCREMENT, operation_id TEXT NOT NULL REFERENCES ec_host_operations(id),
             state TEXT NOT NULL, created_at TEXT NOT NULL, error TEXT
         );
+        CREATE INDEX IF NOT EXISTS ec_host_operation_events_operation ON ec_host_operation_events(operation_id);
         CREATE TABLE IF NOT EXISTS ec_host_ports (
             host_id TEXT NOT NULL REFERENCES ec_hosts(id), instance_id TEXT NOT NULL,
             bind_address TEXT NOT NULL, protocol TEXT NOT NULL CHECK (protocol IN ('udp','tcp')),
@@ -82,6 +85,14 @@ export function ensureHostSchema(handle: Database.Database): void {
     `);
     const columns = handle.prepare('PRAGMA table_info(ec_host_operations)').all() as { name: string }[];
     if (!columns.some(column => column.name === 'last_delivered_at')) handle.exec('ALTER TABLE ec_host_operations ADD COLUMN last_delivered_at TEXT');
+    if (!columns.some(column => column.name === 'observation_success')) {
+        handle.transaction(() => {
+            handle.exec('ALTER TABLE ec_host_operations ADD COLUMN observation_success INTEGER NOT NULL DEFAULT 0');
+            handle.exec(`UPDATE ec_host_operations AS o SET observation_success=1 WHERE ${successfulObservationSql}`);
+        })();
+    }
+    handle.exec(`CREATE INDEX IF NOT EXISTS ec_host_operations_observation_retention ON ec_host_operations(updated_at)
+        WHERE observation_success=1 AND state='completed'`);
     const hostColumns = handle.prepare('PRAGMA table_info(ec_hosts)').all() as { name: string }[];
     if (!hostColumns.some(column => column.name === 'deleted_at')) handle.exec('ALTER TABLE ec_hosts ADD COLUMN deleted_at TEXT');
     const allowanceColumns = handle.prepare('PRAGMA table_info(ec_resource_allowances)').all() as { name: string }[];

src/lib/server/pipeline/sweep.ts

@@ -2,6 +2,7 @@ import { fleetSettings } from '../settings.js';
 import { db } from '../db/index.js';
 import { emit } from './bus.js';
 import { requestRestart } from '../provisioning/index.js';
+import { pruneHostObservations } from '../hosts/index.js';
 
 interface PendingRow {
     deployment_id: string; server: string; restart_state: string;
@@ -18,6 +19,7 @@ export function sweep(): Promise<SweepResult> {
     return sweepInFlight;
 }
 async function runSweep(): Promise<SweepResult> {
+    pruneHostObservations();
     const pending = db().prepare(`SELECT ds.* FROM deployment_servers ds JOIN deployments d ON d.id = ds.deployment_id
         WHERE ds.server_deleted_at IS NULL AND ds.restart_state IN ('pending','failed') ORDER BY d.started_at, d.id, ds.server`).all() as PendingRow[];
     const result: SweepResult = { checked: 0, restarted: 0, healthy: 0, held: 0, failed: 0 };

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

@@ -3,6 +3,7 @@ import { heartbeat, pendingCreationCancellations } from '$lib/server/hosts/index
 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 { fleetSettings } from '$lib/server/settings.js';
 export async function POST({ request }: { request: Request }) {
     const authenticated = authenticateAgent(request);
     try {
@@ -16,6 +17,7 @@ export async function POST({ request }: { request: Request }) {
         reconcileHostPreparation(host.id);
         reconcileHostLibraryCleanup(host.id);
         await cleanupControllerLibraries();
-        return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15, serverTime: new Date().toISOString() });
+        return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15,
+            observationKeepDays: fleetSettings().observationKeepDays, serverTime: new Date().toISOString() });
     } catch (cause) { return hostApiError(cause); }
 }

src/routes/settings/+page.svelte

@@ -54,6 +54,9 @@
                     <label for="retention">Campaign history retention (days)</label>
                     <input id="retention" type="number" min="1" max="36500" step="1" required bind:value={settings.deployKeepDays} aria-describedby="retention-help" />
                     <p id="retention-help">Default: 7 days. Removes eligible old campaign records and file diffs on the next cleanup check. Pending, failed, held, and health-recovery-linked campaigns are kept. Provisioning history is retained.</p>
+                    <label for="observation-retention">Routine operation history retention (days)</label>
+                    <input id="observation-retention" type="number" min="0" max="36500" step="1" required bind:value={settings.observationKeepDays} aria-describedby="observation-retention-help" />
+                    <p id="observation-retention-help">Default: 1 day. Removes successful monitoring and file-read operation records on the next cleanup check. Failures and actions are retained. Set to 0 to keep these records indefinitely.</p>
                     <label for="restarts">Concurrent restarts</label>
                     <input id="restarts" type="number" min="1" step="1" required bind:value={settings.restartConcurrency} aria-describedby="restarts-help" />
                     <p id="restarts-help">Default: 1 across the fleet. Campaigns, betas, maintenance, and lifecycle actions share recovery slots until container and game health are verified.</p>

static/agent/ec-agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.8"
+VERSION = "1.4.9"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -516,6 +516,19 @@ class ReadWriteGate:
                 self.condition.notify_all()
 
 
+def read_operation(operation):
+    if not isinstance(operation, dict):
+        return False
+    if operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
+        return True
+    payload = operation.get("payload")
+    return operation.get("kind") == "rcon.execute" and isinstance(payload, dict) and payload.get("command") in ("status", "getstatus", "version")
+
+
+def valid_observation_keep_days(value):
+    return type(value) is int and 0 <= value <= 36500
+
+
 class Journal:
     def __init__(self, path, recover=True):
         self.db = sqlite3.connect(path)
@@ -525,8 +538,14 @@ class Journal:
         self.db.execute("PRAGMA journal_mode=WAL")
         self.db.execute("PRAGMA synchronous=FULL")
         self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
-        if "checkpoint" not in {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}:
+        columns = {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}
+        if "checkpoint" not in columns:
             self.db.execute("ALTER TABLE operations ADD COLUMN checkpoint TEXT")
+        if "reported_at" not in columns:
+            self.db.execute("ALTER TABLE operations ADD COLUMN reported_at REAL")
+        if "retention_eligible" not in columns:
+            self.db.execute("ALTER TABLE operations ADD COLUMN retention_eligible INTEGER NOT NULL DEFAULT 0")
+        self.db.execute("CREATE INDEX IF NOT EXISTS operations_observation_retention ON operations(updated,id) WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1")
         self.db.execute("UPDATE operations SET state='unknown', error='Agent restarted during execution; inspect before reconciliation', updated=? WHERE state='running'", (time.time(),))
         self.db.commit()
 
@@ -551,10 +570,37 @@ class Journal:
         return dict(row)
 
     def update(self, op_id, state, result=None, error=None):
-        self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=? WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id))
+        self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=?, reported_at=NULL, retention_eligible=0 WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id))
         self.db.commit()
         return self.get(op_id)
 
+    def confirm_completed(self, row):
+        """Only an acknowledged, unchanged completion can become disposable."""
+        if row["state"] != "completed":
+            return
+        eligible = False
+        try:
+            result = json.loads(row["result"]) if row.get("result") is not None else None
+            eligible = read_operation(json.loads(row["request"])) and not row.get("error") and not (
+                isinstance(result, dict) and (result.get("ok") is False or result.get("healthy") is False or result.get("error") not in (None, ""))
+            )
+        except (ValueError, TypeError, KeyError):
+            pass  # Malformed historical records are retained for inspection.
+        self.db.execute("""UPDATE operations SET reported_at=COALESCE(reported_at,?), retention_eligible=?
+            WHERE id=? AND state='completed' AND updated=? AND result IS ? AND error IS ?""",
+            (time.time(), int(bool(eligible)), row["id"], row["updated"], row.get("result"), row.get("error")))
+        self.db.commit()
+
+    def prune_observations(self, keep_days):
+        if not valid_observation_keep_days(keep_days) or keep_days == 0:
+            return 0
+        cutoff = time.time() - keep_days * 86400
+        removed = self.db.execute("""DELETE FROM operations WHERE id IN (
+            SELECT id FROM operations WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1
+                AND updated<? ORDER BY updated,id LIMIT 500)""", (cutoff,)).rowcount
+        self.db.commit()
+        return removed
+
     def close(self):
         self.db.close()
 
@@ -868,6 +914,7 @@ class Agent:
         self.read_claim_lock = threading.Lock()
         self.read_claims = set()
         self.cancelled_creations = frozenset()
+        self.observation_keep_days = 0  # Older controllers do not authorize cleanup.
         self.library_progress = {}
         self.library_transfers = {}
         self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
@@ -2118,7 +2165,7 @@ class Agent:
                     return self.journal.update(operation["id"], "failed", error="Interrupted library download could not be resumed")
             if kind == "files.stage":
                 return self.journal.update(operation["id"], "completed", self.stage(instance_id, generation, payload, operation["id"]))
-            if kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read") or kind == "rcon.execute" and payload.get("command") in ("status", "getstatus", "version"):
+            if self.read_operation(operation):
                 return self.journal.update(operation["id"], "completed", self.dispatch(operation))
             if kind in ("files.activate", "files.rollback") and checkpoint:
                 stage = self.instance_dir(instance_id, generation) / "stages" / identifier(payload["stageId"]) / "manifest.json"
@@ -2172,7 +2219,7 @@ class Agent:
             pass
         return row
 
-    def report(self, row):
+    def report(self, row, journal=None):
         payload = {"state": row["state"]}
         if row.get("result"):
             payload["result"] = json.loads(row["result"])
@@ -2182,13 +2229,19 @@ class Agent:
                 payload["state"] = "running"
         if row.get("error"):
             payload["error"] = row["error"]
-        return self.request("POST", "/api/agent/operations/" + urllib.parse.quote(row["id"], safe=""), payload)
+        response = self.request("POST", "/api/agent/operations/" + urllib.parse.quote(row["id"], safe=""), payload)
+        confirmed = response.get("operation") if isinstance(response, dict) else None
+        if row["state"] == "completed" and isinstance(confirmed, dict) and confirmed.get("id") == row["id"] and confirmed.get("state") == "completed":
+            (journal if journal is not None else self.journal).confirm_completed(row)
+        return response
 
     def heartbeat(self):
         response = self.request("POST", "/api/agent/heartbeat", {"inventory": self.inventory(), "operationId": self.current_operation})
         # Only the heartbeat thread replaces this immutable snapshot; it never
         # mutates the operation journal or races Docker cleanup against create.
         self.cancelled_creations = frozenset(value for value in response.get("cancelledCreations", []) if isinstance(value, str))
+        keep_days = response.get("observationKeepDays")
+        self.observation_keep_days = keep_days if valid_observation_keep_days(keep_days) else 0
         host = response.get("host", {})
         if isinstance(host.get("portMin"), int) and isinstance(host.get("portMax"), int):
             self.config["portRange"] = {"min": host["portMin"], "max": host["portMax"]}
@@ -2205,7 +2258,7 @@ class Agent:
 
     @staticmethod
     def read_operation(operation):
-        return operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
+        return read_operation(operation)
 
     def instance_lock(self, operation):
         key = operation.get("instanceId")
@@ -2219,12 +2272,19 @@ class Agent:
         # Each lane owns its SQLite connection and never writes executing_id,
         # checkpoints, or current_operation used by the mutation state machine.
         journal = Journal(self.root / "agent" / "operations.sqlite3", recover=False)
+        next_prune = time.monotonic()
         try:
             while not self.stopping.is_set():
                 try:
                     if self.update_pending():
                         self.stopping.wait(3)
                         continue
+                    if lane == "read" and time.monotonic() >= next_prune:
+                        next_prune = time.monotonic() + 60
+                        with self.activity(read_only=True) as acquired:
+                            if acquired and not self.update_pending():
+                                if journal.prune_observations(self.observation_keep_days) == 500:
+                                    next_prune = time.monotonic() + 15
                     response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20")
                     handled = False
                     for operation in response.get("operations", []):
@@ -2242,7 +2302,7 @@ class Agent:
                                     continue
                                 row, _ = journal.accept(operation)
                                 if row["state"] in ("accepted", "unknown", "running"):
-                                    self.report(journal.update(operation["id"], "running"))
+                                    self.report(journal.update(operation["id"], "running"), journal=journal)
                                     with self.read_gate.read(), self.instance_lock(operation):
                                         try:
                                             result = self.dispatch(operation)
@@ -2251,7 +2311,7 @@ class Agent:
                                             row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable")
                                 # Receipt delivery can take seconds; it must not
                                 # hold the instance lock needed by health checks.
-                                self.report(row)
+                                self.report(row, journal=journal)
                                 handled = True
                         finally:
                             with self.read_claim_lock:

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.4.8",
+  "version": "1.4.9",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "45eb0ef619fb0ceef2ac843657fd7fdaf8cfa610951f0ddfabdbf5942abc6e8c",
+  "sha256": "c0a02509a8307b2abb75e5450bf3f39596252f26e041ed938d662bc7127bec62",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }

tests/unit/fleet-settings.test.ts

@@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { newFixture, type Fixture } from '../helpers/fixture.js';
 
 let fixture: Fixture;
-const defaults = { deployKeepDays: 7, restartConcurrency: 1, createConcurrency: 1, destroyConcurrency: 1 };
+const defaults = { deployKeepDays: 7, observationKeepDays: 1, restartConcurrency: 1, createConcurrency: 1, destroyConcurrency: 1 };
 beforeEach(async () => {
     vi.resetModules();
     fixture = await newFixture([]);
@@ -14,7 +14,7 @@ describe('fleet settings API and persistence', () => {
     it('returns defaults and saves all settings across a module reload and database reopen', async () => {
         const { GET, PUT } = await import('../../src/routes/api/settings/+server.js');
         expect(await GET().json()).toEqual(defaults);
-        const saved = { deployKeepDays: 30, restartConcurrency: 3, createConcurrency: 2, destroyConcurrency: 4 };
+        const saved = { deployKeepDays: 30, observationKeepDays: 5, restartConcurrency: 3, createConcurrency: 2, destroyConcurrency: 4 };
         const response = await PUT({ request: new Request('http://localhost/api/settings', { method: 'PUT', body: JSON.stringify(saved) }) });
         expect(response.status).toBe(200);
         expect(await response.json()).toEqual(saved);
@@ -37,7 +37,7 @@ describe('fleet settings API and persistence', () => {
         vi.stubEnv('DEPLOYMENT_RESTART_CONCURRENCY', '4');
         vi.stubEnv('DEPLOYMENT_CREATE_CONCURRENCY', '2');
         const { fleetSettings } = await import('../../src/lib/server/settings.js');
-        expect(fleetSettings()).toEqual({ deployKeepDays: 14, restartConcurrency: 4, createConcurrency: 2, destroyConcurrency: 1 });
+        expect(fleetSettings()).toEqual({ deployKeepDays: 14, observationKeepDays: 1, restartConcurrency: 4, createConcurrency: 2, destroyConcurrency: 1 });
         closeDb();
         vi.stubEnv('DEPLOY_KEEP_DAYS', '7');
         expect(fleetSettings().deployKeepDays).toBe(14);
@@ -53,18 +53,47 @@ describe('fleet settings API and persistence', () => {
         vi.stubEnv('DEPLOYMENT_RESTART_CONCURRENCY', '9');
         vi.stubEnv('DEPLOYMENT_CREATE_CONCURRENCY', '8');
         const { GET } = await import('../../src/routes/api/settings/+server.js');
-        const upgraded = { ...previous, destroyConcurrency: 1 };
+        const upgraded = { ...previous, destroyConcurrency: 1, observationKeepDays: 1 };
         expect(await GET().json()).toEqual(upgraded);
         expect(JSON.parse((db().prepare("SELECT value FROM settings WHERE key = 'fleet_settings'").get() as { value: string }).value)).toEqual(upgraded);
         closeDb();
         expect(await GET().json()).toEqual(upgraded);
     });
 
+    it('adds routine observation retention to modern saved settings without changing any existing values', async () => {
+        const { db, closeDb } = await import('../../src/lib/server/db/index.js');
+        const previous = { deployKeepDays: 45, restartConcurrency: 3, createConcurrency: 2, destroyConcurrency: 4 };
+        db().prepare("UPDATE settings SET value = ? WHERE key = 'fleet_settings'").run(JSON.stringify(previous));
+        db().prepare("INSERT INTO settings(key,value) VALUES('unrelated','keep')").run();
+        closeDb();
+        vi.stubEnv('DEPLOY_KEEP_DAYS', 'bad');
+        const { GET } = await import('../../src/routes/api/settings/+server.js');
+        const upgraded = { ...previous, observationKeepDays: 1 };
+        expect(await GET().json()).toEqual(upgraded);
+        expect(JSON.parse((db().prepare("SELECT value FROM settings WHERE key = 'fleet_settings'").get() as { value: string }).value)).toEqual(upgraded);
+        expect(db().prepare("SELECT value FROM settings WHERE key='unrelated'").get()).toEqual({ value: 'keep' });
+        closeDb();
+        expect(await GET().json()).toEqual(upgraded);
+    });
+
+    it.each([0, 36500])('persists routine observation retention of %s days across database reopen', async observationKeepDays => {
+        const { GET, PUT } = await import('../../src/routes/api/settings/+server.js');
+        const saved = { ...defaults, observationKeepDays };
+        const response = await PUT({ request: new Request('http://localhost/api/settings', { method: 'PUT', body: JSON.stringify(saved) }) });
+        expect(response.status).toBe(200);
+        expect(await response.json()).toEqual(saved);
+        const { closeDb } = await import('../../src/lib/server/db/index.js');
+        closeDb();
+        expect(await GET().json()).toEqual(saved);
+    });
+
     it.each([
         'null', '[]', '{}', '{',
         ...['deployKeepDays', 'restartConcurrency', 'createConcurrency', 'destroyConcurrency'].flatMap(key =>
             [0, -1, 1.5, '2', true, null, 9007199254740992].map(value => JSON.stringify({ ...defaults, [key]: value }))),
+        ...[-1, 1.5, '2', true, null, 36501, 9007199254740992].map(value => JSON.stringify({ ...defaults, observationKeepDays: value })),
         JSON.stringify({ deployKeepDays: 7, restartConcurrency: 1, createConcurrency: 1 }),
+        JSON.stringify({ deployKeepDays: 7, restartConcurrency: 1, createConcurrency: 1, destroyConcurrency: 1 }),
         JSON.stringify({ ...defaults, deployKeepDays: 36501 }),
         JSON.stringify({ ...defaults, unknown: 1 })
     ])('rejects invalid input without partially saving: %s', async body => {

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

@@ -0,0 +1,155 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { newFixture, type Fixture } from '../helpers/fixture.js';
+
+const now = Date.parse('2026-10-05T16:00:00Z');
+const day = 86_400_000;
+const at = (age: number) => new Date(now - age).toISOString();
+const inventory = { protocolVersion: 1, dockerRunning: true, directoriesWritable: true,
+    storageAvailable: 100 * 1024 ** 3, runtimeImages: ['plutainer:test'], libraries: [{ id: 'library', state: 'ready' }] };
+let fixture: Fixture;
+
+beforeEach(async () => {
+    vi.resetModules(); fixture = await newFixture([]);
+    vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(now);
+});
+afterEach(async () => { vi.useRealTimers(); vi.restoreAllMocks(); await fixture.cleanup(); });
+
+async function setup() {
+    const hosts = await import('../../src/lib/server/hosts/index.js');
+    const { db, closeDb } = await import('../../src/lib/server/db/index.js');
+    const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js');
+    const host = hosts.createHost({ name: 'Retention host' });
+    const credentials = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
+    const handle = db();
+    function add(kind: string, options: { state?: string; payload?: Record<string, unknown>; result?: unknown;
+        error?: string; age?: number; serialized?: boolean; routing?: Record<string, unknown> } = {}) {
+        const payload = options.payload ?? {};
+        const operation = options.serialized
+            ? hosts.queueHostOperationSerialized(host.id, kind, Buffer.from(JSON.stringify(payload)), {}, options.routing ?? {})
+            : hosts.queueHostOperation(host.id, kind, payload);
+        const state = options.state ?? 'completed';
+        if (state !== 'pending') hosts.reportHostOperationStatus(host.id, operation.id, {
+            state, result: options.result === undefined ? { ok: true } : options.result, error: options.error
+        });
+        handle.prepare('UPDATE ec_host_operations SET created_at=?,updated_at=? WHERE id=?')
+            .run(at(10 * day), at(options.age ?? 2 * day), operation.id);
+        return operation.id;
+    }
+    return { ...hosts, handle, db, closeDb, host, credentials, fleetSettings, saveFleetSettings, add };
+}
+
+describe('routine host observation retention', () => {
+    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']
+            .map(kind => api.add(kind));
+        removed.push(...['status', 'getstatus', 'version'].map(command => api.add('rcon.execute', { payload: { command } })));
+        const retained = [
+            ...['pending', 'accepted', 'running', 'failed', 'unknown'].map(state => api.add('files.read', { state })),
+            ...['instance.restart', 'instance.stop', 'instance.remove', 'files.stage', 'files.activate', 'files.rollback', 'runtime.pull', 'unknown.read']
+                .map(kind => api.add(kind)),
+            api.add('rcon.execute', { payload: { command: 'map_restart' } }),
+            api.add('rcon.execute', { payload: { command: 'status; map_restart' } }),
+            api.add('rcon.execute', { payload: { command: 'status' }, result: { ok: false, reason: 'timeout' } }),
+            api.add('instance.inspect', { result: { healthy: false, state: 'stopped' } }),
+            api.add('files.read', { error: 'Read failed' }),
+            api.add('files.read', { result: { error: 'Read failed' } }),
+            api.add('files.stat', { age: day }),
+            api.add('files.read', { age: day - 1 })
+        ];
+        expect(api.pruneHostObservations()).toBe(removed.length);
+        const ids = (api.handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[]).map(row => row.id);
+        expect(ids.sort()).toEqual(retained.sort());
+        expect(api.pruneHostObservations()).toBe(0);
+    });
+
+    it('classifies full serialized RCON payloads and deletes their events and immutable bytes together', async () => {
+        const api = await setup();
+        const read = api.add('rcon.execute', { serialized: true, payload: { command: 'status' }, routing: { command: 'map_restart' } });
+        const action = api.add('rcon.execute', { serialized: true, payload: { command: 'map_restart' }, routing: { command: 'status' } });
+        expect(api.handle.prepare('SELECT COUNT(*) AS count FROM ec_host_operation_payloads').get()).toEqual({ count: 2 });
+        expect(api.pruneHostObservations()).toBe(1);
+        expect(api.getHostOperationStatus(read)).toBeNull();
+        expect(api.getHostOperation(action)?.payload).toEqual({ command: 'map_restart' });
+        expect(api.handle.prepare('SELECT DISTINCT operation_id FROM ec_host_operation_events').all()).toEqual([{ operation_id: action }]);
+        expect(api.handle.prepare('SELECT operation_id FROM ec_host_operation_payloads').all()).toEqual([{ operation_id: action }]);
+        expect(api.handle.pragma('foreign_key_check')).toEqual([]);
+    });
+
+    it('honors disabled retention and changes to the saved period using the last receipt time', async () => {
+        const api = await setup();
+        const old = api.add('files.read', { age: 4 * day });
+        const recentReceipt = api.add('files.read', { age: 2 * day });
+        api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 0 });
+        expect(api.pruneHostObservations()).toBe(0);
+        api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 3 });
+        expect(api.pruneHostObservations()).toBe(1);
+        expect(api.getHostOperationStatus(old)).toBeNull();
+        expect(api.getHostOperationStatus(recentReceipt)).not.toBeNull();
+        api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 1 });
+        expect(api.pruneHostObservations()).toBe(1);
+    });
+
+    it('bounds each cleanup batch and seeks through the retention index despite extensive retained action history', async () => {
+        const api = await setup();
+        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(?,?,?,'{}','completed',?,?,?)`);
+            for (let index = 0; index < 2000; index++) insert.run(`action-${index}`, api.host.id, 'instance.restart', at(10 * day), at(10 * day), 0);
+            for (let index = 0; index < 503; index++) insert.run(`read-${index}`, api.host.id, 'files.read', at(2 * day), at(2 * day), 1);
+        })();
+        const prepare = vi.spyOn(api.handle, 'prepare');
+        expect(api.pruneHostObservations()).toBe(500);
+        const query = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('SELECT id FROM ec_host_operations WHERE observation_success'))!;
+        prepare.mockRestore();
+        const plan = api.handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(at(day)) as { detail: string }[];
+        expect(plan.some(row => row.detail.includes('USING INDEX ec_host_operations_observation_retention'))).toBe(true);
+        expect(plan.some(row => row.detail.includes('TEMP B-TREE'))).toBe(false);
+        expect(api.pruneHostObservations()).toBe(3);
+        expect(api.pruneHostObservations()).toBe(0);
+        expect(api.handle.prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.restart'").get()).toEqual({ count: 2000 });
+    });
+
+    it('backfills legacy observation classification once while preserving actions and negative or invalid results', async () => {
+        const api = await setup();
+        const read = api.add('files.manifest');
+        const serializedRead = api.add('rcon.execute', { serialized: true, payload: { command: 'status' } });
+        const retained = [api.add('instance.restart'), api.add('rcon.execute', { payload: { command: 'map_restart' } }),
+            api.add('instance.inspect', { result: { healthy: false } }), api.add('files.read', { result: { ok: false } }),
+            api.add('files.read', { error: 'Unavailable' }), api.add('files.read', { result: { error: 'Unavailable' } }),
+            api.add('files.stat', { state: 'unknown' })];
+        const malformed = api.add('files.read');
+        api.handle.prepare('UPDATE ec_host_operations SET result=? WHERE id=?').run('{not-json', malformed);
+        retained.push(malformed);
+        api.handle.exec('DROP INDEX ec_host_operations_observation_retention');
+        api.handle.exec('ALTER TABLE ec_host_operations DROP COLUMN observation_success');
+        api.closeDb();
+
+        expect(api.pruneHostObservations()).toBe(2);
+        expect(api.getHostOperationStatus(read)).toBeNull();
+        expect(api.getHostOperationStatus(serializedRead)).toBeNull();
+        const handle = api.db();
+        expect((handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[]).map(row => row.id).sort()).toEqual(retained.sort());
+        // Subsequent boots honor the recorded classification instead of repeatedly
+        // deriving it from historical rows that were deliberately preserved.
+        handle.prepare("UPDATE ec_host_operations SET result='{}' WHERE id=?").run(malformed);
+        api.closeDb();
+        expect(api.pruneHostObservations()).toBe(0);
+        expect(api.getHostOperationStatus(malformed)).not.toBeNull();
+    });
+
+    it('returns the current shared retention setting to an agent on every heartbeat', async () => {
+        const api = await setup();
+        const { POST } = await import('../../src/routes/api/agent/heartbeat/+server.js');
+        for (const observationKeepDays of [7, 0]) {
+            api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays });
+            const response = await POST({ request: new Request('https://ec.test/api/agent/heartbeat', {
+                method: 'POST', headers: { authorization: `Bearer ${api.credentials.credential}`, 'Content-Type': 'application/json' },
+                body: JSON.stringify({ inventory })
+            }) });
+            expect(response.status).toBe(200);
+            expect(await response.json()).toMatchObject({ observationKeepDays, heartbeatSeconds: 15 });
+        }
+    });
+});