function positiveInteger(value, fallback) { const parsed = Number(value); return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; } export function createShadowObservationDispatcher({ observe, onCompleted = async () => {}, onFailed = async () => {}, onSkipped = async () => {}, maxConcurrent = 2, maxQueued = 100, schedule = setImmediate, now = Date.now, logger = console, } = {}) { const enabled = typeof observe === 'function'; const concurrency = positiveInteger(maxConcurrent, 2); const queueLimit = positiveInteger(maxQueued, 100); const queue = []; let active = 0; let drainScheduled = false; function reportCallbackFailure(label, error) { logger.warn( `[orchestrator-shadow] ${label} callback failed:`, error instanceof Error ? error.message : error, ); } function scheduleDrain() { if (!enabled || drainScheduled || !queue.length || active >= concurrency) return; drainScheduled = true; schedule(drain); } function runObservation(entry) { active += 1; const context = { enqueuedAt: entry.enqueuedAt, startedAt: now(), }; void Promise.resolve() .then(() => observe(entry.input)) .then((result) => onCompleted(entry.input, result, context)) .catch((error) => onFailed(entry.input, error, context)) .catch((error) => reportCallbackFailure('failure', error)) .finally(() => { active -= 1; scheduleDrain(); }); } function drain() { drainScheduled = false; while (active < concurrency && queue.length) { runObservation(queue.shift()); } } return { dispatch(input) { if (!enabled) return false; if (queue.length >= queueLimit) { const context = { enqueuedAt: now(), reason: 'shadow_queue_full', }; void Promise.resolve() .then(() => onSkipped(input, context)) .catch((error) => reportCallbackFailure('skip', error)); return false; } queue.push({ input, enqueuedAt: now() }); scheduleDrain(); return true; }, status() { return { enabled, active, queued: queue.length, maxConcurrent: concurrency, maxQueued: queueLimit, }; }, }; } export const shadowDispatcherInternals = { positiveInteger, };