Aggregate with an explicit lateness policy

Process arrivals in listed order. Use nonnegative integer event time and signed integer value. Before each unique event, compute watermark as maximum accepted event time minus lateness; reject an event as late when its time is strictly below that prior watermark. Equality is accepted. Accepted events join half-open windows via time // width * width. Identical IDs are ignored, conflicting ID payloads fail, and late IDs are returned for audit. This is a simple local max-seen estimator, not a distributed watermark implementation; timestamp plausibility checking is an extension.