/** Batch lifecycle boundary. A successful load promise does not imply GPU readiness. */ export function createPipeModelBatches({ load, remove, onChange = () => {}, now = Date.now, retryMs = 30000 }) { const records = new Map() const state = () => { const covered = new Set(), incomplete = new Set() let pending = 0, failed = 0, batches = 0, instances = 0 const byKind = {} for (const record of records.values()) { if (record.status === 'pending') pending += 1 if (record.status === 'failed') failed += 1 if (record.status !== 'ready') { for (const instance of record.group.instances) incomplete.add(instance.id) continue } batches += 1 instances += record.group.instances.length byKind[record.group.kind] = (byKind[record.group.kind] || 0) + record.group.instances.length for (const instance of record.group.instances) covered.add(instance.id) } return { covered: [...covered].filter((id) => !incomplete.has(id)), pending, failed, batches, instances, byKind } } const changed = () => onChange(state()) function dispose(record) { record.active = false if (record.model) { remove(record.model); record.model = null } } async function sync(groups) { const wanted = new Set(groups.map((group) => group.key)) for (const [key, record] of records) { if (!wanted.has(key)) { dispose(record); records.delete(key) } } const jobs = [] for (const group of groups) { const fingerprint = JSON.stringify(group) const previous = records.get(group.key) if (previous && previous.fingerprint === fingerprint && (previous.status !== 'failed' || now() - previous.failedAt < retryMs)) continue if (previous) dispose(previous) const record = { group, fingerprint, active: true, status: 'pending', model: null } records.set(group.key, record) const error = () => { if (!record.active) return record.status = 'failed' record.failedAt = now() if (record.model) { remove(record.model); record.model = null } changed() } jobs.push((async () => { try { const model = await load(group, { alive: () => record.active, ready: () => { if (record.active && record.status === 'pending') { record.status = 'ready'; changed() } }, error, }) if (!record.active || record.status === 'failed') remove(model) else record.model = model } catch { error() } })()) } changed() await Promise.all(jobs) return state() } function retain(predicate) { for (const [key, record] of records) { if (!predicate(record.group)) { dispose(record); records.delete(key) } } changed() } return { sync, state, retain, clear: () => sync([]) } }