Add fleet and per-host concurrency controls

src/lib/server/hosts/index.ts

@@ -533,9 +533,18 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = '
             // sibling's work, but still recovers a lost receipt or agent restart.
             .all(hostId, Number(host.readiness === 'ready' || host.mode === 'draining'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false),
                 ...(lane === 'health' ? [new Date(Date.now() - 30_000).toISOString()] : [])) as OperationDeliveryRow[];
+        const preparationLimit = fleetSettings().preparationConcurrency;
+        const preparing = new Set((database().prepare(`SELECT DISTINCT host_id FROM ec_host_operations
+            WHERE kind IN ('library.download','library.import') AND last_delivered_at IS NOT NULL
+            AND state IN ('pending','accepted','running','unknown')`).all() as { host_id: string }[]).map(row => row.host_id));
+        const admitted = rows.filter(operation => {
+            if (!['library.download','library.import'].includes(operation.kind) || preparing.has(hostId) || !preparationLimit) return true;
+            if (preparing.size >= preparationLimit) return false;
+            preparing.add(hostId); return true;
+        });
         const deliveredAt = now();
-        for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id);
-        return rows;
+        for (const operation of admitted) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id);
+        return admitted;
     })();
 }
 /** Payloads never change after enqueueing; fetch one delivery's UTF-8 bytes without decoding. */