import { fleetSettings } from '../settings.js'; import { db } from '../db/index.js'; import { ensureHostSchema } from './schema.js'; import type { CapacityCalibration, CapacitySample, CapacitySamplePage, HostInventory, ResourceAllowance } from '../../hosts.js'; const BOOTSTRAP_CPU = 0.5; const BOOTSTRAP_MEMORY = 1024 ** 3; const WINDOW_MS = 24 * 60 * 60_000; const MAX_SAMPLES = 10_000; type Phase = 'startup' | 'running'; interface AllowanceRow { scope: string; phase: Phase; cpu_cores: number; memory_bytes: number; samples: number; window_start: string; observed_at: string; computed_at: string; } function database() { const handle = db(); ensureHostSchema(handle); return handle; } /** Read retained minute peaks without contacting agents or recalculating allowances. */ export function capacitySamples(hostId: string, options: { scope: 'host' | 'fleet'; phase: 'all' | Phase; page: number }): CapacitySamplePage { const handle = database(); const clauses = ['s.bucket >= ?']; const values: (string | number)[] = [new Date(Date.now() - WINDOW_MS).toISOString()]; if (options.scope === 'host') { clauses.push('s.host_id = ?'); values.push(hostId); } if (options.phase !== 'all') { clauses.push('s.phase = ?'); values.push(options.phase); } const where = clauses.join(' AND '); const total = (handle.prepare(`SELECT COUNT(*) AS count FROM ec_instance_resource_samples s WHERE ${where}`).get(...values) as { count: number }).count; const pageSize = 50; const page = Math.min(options.page, Math.max(1, Math.ceil(total / pageSize))); const samples = handle.prepare(`SELECT s.host_id AS hostId, h.name AS hostName, s.instance_id AS instanceId, i.server, s.generation, s.phase, s.bucket AS timestamp, s.cpu_cores AS cpuCores, s.memory_bytes AS memoryBytes FROM ec_instance_resource_samples s LEFT JOIN ec_hosts h ON h.id=s.host_id LEFT JOIN native_instances i ON i.id=s.instance_id AND i.host_id=s.host_id WHERE ${where} ORDER BY s.bucket DESC, s.host_id, s.instance_id, s.generation, s.phase LIMIT ? OFFSET ?`) .all(...values, pageSize, (page - 1) * pageSize) as CapacitySample[]; return { samples, total, page, pageSize }; } export function validateInstanceResources(inventory: HostInventory): void { if (inventory.instanceResourcesError !== undefined && (typeof inventory.instanceResourcesError !== 'string' || inventory.instanceResourcesError.length > 500)) throw new Error('Invalid instance resource sampling error.'); const samples = inventory.instanceResources; if (samples === undefined) return; if (!Array.isArray(samples) || samples.length > 1000) throw new Error('Invalid instance resource samples.'); const seen = new Set(); for (const sample of samples) { if (!sample || typeof sample !== 'object' || typeof sample.instanceId !== 'string' || !sample.instanceId || sample.instanceId.length > 100 || !Number.isSafeInteger(sample.generation) || sample.generation < 1 || !Number.isSafeInteger(sample.uptimeSeconds) || sample.uptimeSeconds < 0 || typeof sample.cpuCores !== 'number' || !Number.isFinite(sample.cpuCores) || sample.cpuCores < 0 || !Number.isSafeInteger(sample.memoryBytes) || sample.memoryBytes <= 0 || (inventory.cpuCount !== undefined && sample.cpuCores > inventory.cpuCount + 0.01) || (inventory.memoryTotal !== undefined && sample.memoryBytes > inventory.memoryTotal) || seen.has(sample.instanceId)) throw new Error('Invalid instance resource sample.'); seen.add(sample.instanceId); } } /** Heartbeat ingestion owns all learning work. Launch requests only read cached allowances. */ export function recordInstanceResources(hostId: string, inventory: HostInventory, timestamp: string): void { if (!inventory.instanceResources?.length || !inventory.dockerRunning || !inventory.cpuCount || !inventory.memoryTotal) return; const handle = database(); const bucket = timestamp.slice(0, 16) + ':00.000Z'; const observed = new Map(); const owned = handle.prepare('SELECT 1 FROM native_instances WHERE id=? AND host_id=? AND generation=? AND retired_at IS NULL'); const insert = handle.prepare(`INSERT INTO ec_instance_resource_samples (host_id,instance_id,generation,phase,bucket,cpu_cores,memory_bytes) VALUES(?,?,?,?,?,?,?) ON CONFLICT(host_id,instance_id,generation,phase,bucket) DO UPDATE SET cpu_cores=MAX(cpu_cores,excluded.cpu_cores), memory_bytes=MAX(memory_bytes,excluded.memory_bytes)`); for (const sample of inventory.instanceResources) { // An agent can only train from its currently owned generation; foreign, // retired and delayed pre-rebuild samples cannot contaminate the model. if (!owned.get(sample.instanceId, hostId, sample.generation)) continue; const phase: Phase = sample.uptimeSeconds < 300 ? 'startup' : 'running'; insert.run(hostId, sample.instanceId, sample.generation, phase, bucket, sample.cpuCores, sample.memoryBytes); const prior = observed.get(phase); observed.set(phase, { cpu: Math.max(prior?.cpu ?? 0, sample.cpuCores), memory: Math.max(prior?.memory ?? 0, sample.memoryBytes) }); } for (const [phase, peak] of observed) for (const scope of [hostId, '*']) { const previous = handle.prepare('SELECT * FROM ec_resource_allowances WHERE scope=? AND phase=?').get(scope, phase) as AllowanceRow | undefined; if (!previous || previous.computed_at < bucket) { const rows = handle.prepare(`SELECT cpu_cores, memory_bytes, bucket FROM ec_instance_resource_samples WHERE ${scope === '*' ? '' : 'host_id=? AND '}phase=? AND bucket>=? ORDER BY bucket DESC LIMIT ?`) .all(...(scope === '*' ? [] : [scope]), phase, new Date(Date.parse(timestamp) - WINDOW_MS).toISOString(), MAX_SAMPLES) as { cpu_cores: number; memory_bytes: number; bucket: string }[]; const percentile = (values: number[]) => values.sort((a, b) => a - b)[Math.ceil(values.length * 0.95) - 1]; const cpu = Math.max(percentile(rows.map(row => row.cpu_cores)), peak.cpu); const memory = Math.ceil(Math.max(percentile(rows.map(row => row.memory_bytes)), peak.memory)); handle.prepare(`INSERT INTO ec_resource_allowances (scope,phase,cpu_cores,memory_bytes,samples,window_start,observed_at,computed_at) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(scope,phase) DO UPDATE SET cpu_cores=excluded.cpu_cores,memory_bytes=excluded.memory_bytes,samples=excluded.samples, window_start=excluded.window_start,observed_at=excluded.observed_at,computed_at=excluded.computed_at`) .run(scope, phase, cpu, memory, rows.length, rows.at(-1)!.bucket, timestamp, timestamp); if (scope === '*' && fleetSettings().instanceResourceKeepDays) handle.prepare('DELETE FROM ec_instance_resource_samples WHERE bucket=?') .all(hostId, '*', new Date(Date.now() - WINDOW_MS).toISOString()) as AllowanceRow[]; const allowance = (phase: Phase): ResourceAllowance => { const local = rows.find(row => row.scope === hostId && row.phase === phase); const fleet = rows.find(row => row.scope === '*' && row.phase === phase); const learned = (row?: AllowanceRow) => !!row && row.samples >= (phase === 'startup' ? 3 : 20) && Date.parse(row.observed_at) - Date.parse(row.window_start) >= (phase === 'startup' ? 120_000 : 300_000); const selected = learned(local) ? local : learned(fleet) ? fleet : undefined; return { // A few higher observations can increase a fleet/bootstrap allowance // immediately. Decreasing it requires the evidence thresholds above. cpuCores: Math.max(0.01, selected?.cpu_cores ?? BOOTSTRAP_CPU, local?.cpu_cores ?? 0, selected ? 0 : fleet?.cpu_cores ?? 0), memoryBytes: Math.max(selected?.memory_bytes ?? BOOTSTRAP_MEMORY, local?.memory_bytes ?? 0, selected ? 0 : fleet?.memory_bytes ?? 0), source: selected ? selected.scope === hostId ? 'host' : 'fleet' : 'bootstrap', samples: selected?.samples ?? local?.samples ?? fleet?.samples ?? 0, observedAt: selected?.observed_at ?? local?.observed_at ?? fleet?.observed_at ?? null }; }; const running = allowance('running'); const startup = allowance('startup'); if (running.source === 'bootstrap' && (running.cpuCores > startup.cpuCores || running.memoryBytes > startup.memoryBytes)) startup.source = 'bootstrap'; startup.cpuCores = Math.max(startup.cpuCores, running.cpuCores); startup.memoryBytes = Math.max(startup.memoryBytes, running.memoryBytes); return { running, startup }; }