Retain completed status batches as routine observations

agent/ec_agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.13"
+VERSION = "1.4.14"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1393,7 +1393,8 @@ class Agent:
 
         with ThreadPoolExecutor(max_workers=2) as executor:
             results = list(executor.map(probe, identities))
-        return {"ok": all(row["reply"].get("ok") for row in results), "results": results}
+        # Completion describes the batch; individual failures remain in each reply.
+        return {"ok": True, "results": results}
 
     def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3):
         if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):

agent/tests/test_agent.py

@@ -114,6 +114,12 @@ class JournalRetentionTests(unittest.TestCase):
             self.assertFalse(accepted)
             self.assertEqual(saved["id"], row["id"])
 
+    def test_completed_batch_with_failed_targets_is_pruned_after_acknowledgment(self):
+        result = {"ok": True, "results": [{"instanceId": INSTANCE, "generation": 1,
+                                         "reply": {"ok": False, "reason": "timeout", "text": ""}}]}
+        self.record("rcon.status-batch", {"instances": [{"instanceId": INSTANCE, "generation": 1}]}, result)
+        self.assertEqual(self.prune(), 1)
+
     def test_errors_failures_and_unfinished_operations_are_retained(self):
         retained = [self.record(result=result) for result in ({"ok": False}, {"healthy": False}, {"error": "partial result"}, {"error": False}, {"error": []})]
         retained.append(self.record(error="read did not finish"))
@@ -644,7 +650,7 @@ class AgentTests(unittest.TestCase):
         self.assertFalse(thread.is_alive())
         self.assertEqual(peak, 2)
         self.assertEqual(len(results[0]["results"]), 5)
-        self.assertFalse(results[0]["ok"])
+        self.assertTrue(results[0]["ok"])
         self.assertEqual(results[0]["results"][2]["reply"]["reason"], "unreachable")
         self.assertTrue(results[0]["results"][4]["reply"]["ok"])
 

src/lib/server/hosts/observations.ts

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

src/lib/server/hosts/schema.ts

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

static/agent/ec-agent.py

@@ -32,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.13"
+VERSION = "1.4.14"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1393,7 +1393,8 @@ class Agent:
 
         with ThreadPoolExecutor(max_workers=2) as executor:
             results = list(executor.map(probe, identities))
-        return {"ok": all(row["reply"].get("ok") for row in results), "results": results}
+        # Completion describes the batch; individual failures remain in each reply.
+        return {"ok": True, "results": results}
 
     def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3):
         if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.4.13",
+  "version": "1.4.14",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "2f3449058cd4b3ab9e3ff8f541fa3c65ad2d94aabe20694cdec3a6e59c8a4fd6",
+  "sha256": "3e2e11389f412a0f1b2746c85b9afa626c5e0fcaa4801bd13f1de283abf8830e",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }

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

@@ -63,6 +63,36 @@ describe('routine host observation retention', () => {
         expect(api.pruneHostObservations()).toBe(0);
     });
 
+    it('keeps per-server failures in routine completed batches and prunes them on schedule', async () => {
+        const api = await setup();
+        const results = [{ instanceId: 'one', generation: 1, reply: { ok: false, reason: 'timeout', text: '' } }];
+        const batches = [false, true].map(ok => api.add('rcon.status-batch', { result: { ok, results } }));
+        const failed = api.add('rcon.status-batch', { state: 'failed', error: 'Batch failed' });
+        const malformed = api.add('rcon.status-batch', { result: { ok: false } });
+        expect(api.listHostOperationHistory(1, 'important').operations.map(op => op.id).sort()).toEqual([failed, malformed].sort());
+        for (const id of batches) expect(api.getHostOperation(id)?.result).toMatchObject({ results });
+        expect(api.pruneHostObservations()).toBe(2);
+        for (const id of batches) expect(api.getHostOperationStatus(id)).toBeNull();
+    });
+
+    it('reclassifies existing completed batches once without hiding operation errors', async () => {
+        const api = await setup();
+        const result = { ok: false, results: [{ instanceId: 'one', generation: 1, reply: { ok: false, reason: 'timeout' } }] };
+        const batch = api.add('rcon.status-batch', { result });
+        const retained = [api.add('rcon.status-batch', { result, error: 'Incomplete' }),
+            api.add('rcon.status-batch', { result, state: 'failed' }),
+            api.add('rcon.status-batch', { result: { ...result, error: 'Incomplete' } })];
+        api.handle.exec("UPDATE ec_host_operations SET observation_success=0; DELETE FROM ec_host_schema_migrations WHERE name='completed-status-batches'");
+        api.closeDb();
+        expect(api.listHostOperationHistory(1, 'important').operations.map(op => op.id).sort()).toEqual(retained.sort());
+        expect(api.getHostOperation(batch)?.result).toEqual(result);
+        expect(api.pruneHostObservations()).toBe(1);
+        // The migration is not a recurring scan or reclassification of retained evidence.
+        api.db().prepare("UPDATE ec_host_operations SET state='completed',error=NULL,result=? WHERE id=?").run(JSON.stringify(result), retained[0]);
+        api.closeDb();
+        expect(api.pruneHostObservations()).toBe(0);
+    });
+
     it('classifies full serialized RCON payloads and deletes their events and immutable bytes together', async () => {
         const api = await setup();
         const read = api.add('rcon.execute', { serialized: true, payload: { command: 'status' }, routing: { command: 'map_restart' } });