fix(aggregate transform): address event-time edge cases

Addresses several bugs surfaced during review of the event-time
aggregation feature:

* Stop populating `event_time_prev_buckets` in non-`Diff` modes. The map
  is read only by `Diff` and was previously inserted (and not evicted)
  on every flush in `Auto`/`Sum`/`Latest`/etc., growing memory linearly
  with (unique series in interval) x (intervals since startup).

* Drain remaining event-time buckets when the input stream closes.
  `flush_event_time_buckets` now accepts a `force` flag and a new
  `flush_final` entry-point is wired into the input-closed arm so
  in-flight metrics in still-open windows are emitted on shutdown or
  topology reload, matching system-time semantics.

* Reject events for already-emitted windows. Watermark now records the
  exclusive end (`bucket_key + interval_ms`) of the highest flushed
  bucket, and `is_too_late` no longer subtracts `allowed_lateness_ms`.
  `allowed_lateness_ms` keeps its role of delaying bucket close at
  flush time; once a window is emitted it stays closed. This prevents
  late events from re-creating closed buckets and emitting duplicate
  partial aggregates.

* Drop events whose (kind, value) is incompatible with the configured
  mode (for example an `Incremental` event arriving at a `Mean`
  aggregator) without materialising a bucket. The previous code path
  always created an empty `event_time_buckets` entry, which then
  flushed and advanced the watermark, silently rejecting valid
  in-order events for earlier buckets. Compatibility is now decided
  up front by `will_be_stored`, dropped events emit
  `AggregateEventDropped` (rather than the misleading
  `AggregateEventRecorded`), and `event_time_multi_buckets` is only
  touched in `Mean`/`Stdev` mode.

Adds tests for edge-case behaviour, event-time `Mean` and `Stdev`
happy-path flushing, plus an updated changelog entry describing shipped
behaviour.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: kaarolch

changelog.d/24410_aggregate_event_time_aggregation.feature.md

@@ -1,11 +1,16 @@
-The `aggregate` transform now supports **event-time aggregation** via the new `time_source` configuration option. When set to `EventTime`, metrics are bucketed based on their embedded timestamps rather than processing time.
+The `aggregate` transform now supports **event-time aggregation** via the new `time_source` configuration option. When set to `EventTime`, metrics are bucketed by their embedded timestamps rather than by processing time. This addresses cases where events with different source timestamps but the same processing time were incorrectly collapsed together (e.g. losing samples in sinks like Datadog that overwrite earlier values for an identical timestamp).
 
-Additional configuration options include:
+Additional configuration options for event-time mode:
 
-- `allowed_lateness_ms`: Grace period for accepting late-arriving events within event-time windows
-- `use_system_time_for_missing_timestamps`: Option to fallback to system time or drop events without timestamps
-- `max_future_ms`: Maximum allowed future timestamp offset to reject events with clock skew
+- `allowed_lateness_ms`: Grace period that delays closing of an event-time window so late-arriving events still land in the correct bucket. Once a window is emitted, it is closed and cannot be reopened by later events.
+- `use_system_time_for_missing_timestamps`: Fall back to system time for events that do not carry a timestamp; otherwise such events are dropped.
+- `max_future_ms`: Maximum allowed future-timestamp offset before an event is dropped (clock-skew guard).
 
-This addresses issues where events with different source timestamps but same processing time would be incorrectly split across aggregation windows.
+Behaviour:
+
+- Late events for a window that has already been emitted are dropped (and counted via `component_discarded_events_total`) rather than re-opening the window. The watermark advances to the exclusive end of the latest emitted window.
+- Events whose `(kind, value)` is incompatible with the configured aggregation mode (for example an `Incremental` event arriving at a `Mean`-configured aggregator) are dropped explicitly without affecting the watermark or other in-flight buckets.
+- All remaining event-time buckets are flushed when the input stream closes (shutdown or topology reload) so in-flight metrics are never silently dropped.
+- In `Diff` mode a small rolling window of previous buckets is retained to compute deltas; other modes do not retain previous buckets.
 
 authors: kaarolch