Active-window batcher residency is unbounded and escalation multiplies it

[SimonHeybrock]

#1271 bounds the rate-aware batcher's retained gated overflow (data time + bytes). That was the unbounded term behind the TBL OOM (#378), but it is the smaller half of batcher memory at high rates: nothing bounds the active window — the per-stream buckets holding one batch length of in-flight traffic.

At DREAM's specified ceiling of 1e8 events/s (~763 MB/s across banks), peak residency in healthy, keeping-up operation:

batch length	active window	overflow (bounded by #1271)
1.0 s	~545 MB	≤ 512 MB, typically ~0
8.0 s	~5450 MB	≤ 512 MB, typically ~0

Residency scales linearly with batch length, and AdaptiveMessageBatcher escalates the batch length because processing is slow — so the footprint grows 8x (base → ceiling of the escalation grid) exactly when the service is already struggling. The overflow bound cannot help: this data is inside the window the batcher is legitimately assembling, not backlog.

The same mismatch exists one stage upstream: BackgroundMessageSource.max_queue_size bounds a count of consume batches, while payloads span four orders of magnitude (~8 kB monitor counts to 33.5 MB ad00 frames), so the queue's footprint varies by the same factor (noted in its docstring since #1271).

Candidate directions, each with real trade-offs — this needs a design decision, not a quick patch:

• Cap AdaptiveMessageBatcher.max_level against a memory budget instead of the fixed 3, so escalation cannot multiply residency past what the host affords.
• Bound total batcher residency (window + overflow) rather than only the overflow, shedding within the window under pressure.
• Shed upstream at the consumer, in bytes rather than batch count, so oversized traffic never reaches the batcher.

Interacts with the freshness policy established in #1271 (core/rate_aware_batcher.py module docstring): any bound here should shed stale-first and stay structurally selective (gated stream kinds only), and with the escalation policy, whose whole purpose is to trade latency for throughput — a memory cap on escalation re-introduces the overload it was escaping, so the failure mode it picks must be chosen deliberately.

[SimonHeybrock]

Symptom of the missing bound, found in review of #1271: a producer whose timestamp stops advancing never converges a rate estimate (zero diffs are filtered) and never moves the HWM past the timeout threshold, so as the only traffic in a service nothing ever closes the window. The stall backstop saw the anchor inside the window, treated it as quietness and re-armed, and the bucket (or _future, if the frozen stamp sits just past the window) grew at the full ingest rate with nothing delivered. #1284 bounds that symptom at a couple of stall intervals' worth of traffic by firing the timeout close when the buffer has grown since the previous stall check; it does not touch the residency scaling above.