Implement streaming SRM detector with late events

Implement a streaming detector for sample ratio mismatch (SRM) across many concurrent experiments. Input is two topic-partitioned streams: assignments (experiment_id, user_id, variant, ts) and pageviews (user_id, ts). Requirements: 1) deduplicate per (experiment_id, user_id) using idempotent state; 2) maintain rolling counts per variant in O(1) memory per experiment (no raw buffering), supporting late/out-of-order events up to 24 hours; 3) every minute, compute a chi-square goodness-of-fit with Yates correction versus the target split, raise an alert if p < 1e-4 and absolute diff ≥ 0.3 percentage points; 4) guard against bot bursts by excluding users with >N assignments/min; 5) complexity and pseudocode for a single-threaded worker and how you’d shard it; 6) explain how you would validate the detector in replay without leaking ground truth.