fix(ingest): process socket log payloads synchronously and return verified promotion counts to client

.gitignore

@@ -22,14 +22,15 @@ out/
 *.db
 *.sqlite
 *.sqlite3
-client_state.json
-*.tmp
-
 # Live configuration with generated private keys & tokens (samples are tracked)
 server_config.json
 client_config.json
+client_state.json
+local_collector_state.json
 test_*.json
 test_*.db
+test_pipeline_master_ca/
+test_pipeline_certs/
 master_ca_certs/
 .env
 *.env

src/Server.py

@@ -1879,20 +1879,12 @@ async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.Str
         if client_id and "server" not in log_payload:
             log_payload["server"] = client_id
 
-        INGEST_QUEUE.push(log_payload)
+        res = process_ingested_logs(
-        if len(INGEST_QUEUE._buffer) >= INGEST_QUEUE.batch_size:
+            log_payload,
-            res = INGEST_QUEUE.flush(
+            db_path=SERVER_STATE["config"]["db_path"],
-                db_path=SERVER_STATE["config"]["db_path"],
+            window_hours=SERVER_STATE["config"]["evaluation_window_hours"],
-                window_hours=SERVER_STATE["config"]["evaluation_window_hours"],
+            min_runs=SERVER_STATE["config"]["min_persistence_runs"]
-                min_runs=SERVER_STATE["config"]["min_persistence_runs"]
+        )
-            )
-        else:
-            res = {
-                "status": "success",
-                "buffered": True,
-                "processed": len(log_payload.get("logs", [])),
-                "queue_size": len(INGEST_QUEUE._buffer)
-            }
 
         resp_bytes = json.dumps(res).encode("utf-8")
         writer.write(struct.pack(">I", len(resp_bytes)) + resp_bytes)