feat: opt-in time-based closing for tumbling windows on idle streams (#790)

[saigautam9]

Closes #790 (draft — opening early to get design feedback before finishing the app-loop wiring).

Problem

Time-based windows only expire when a new message advances event time, so a stream that goes quiet never emits its final window. The current workaround is publishing dummy "heartbeat" messages on a timer to force a tumble.

Approach

Opt-in close_on_idle flag on tumbling_window(). When enabled, windows close on wall-clock time once window_end + grace_ms passes — no new message required. The close-time math is unchanged; only the clock that drives it changes (event time → real time).

Scoped to final(closing_strategy="partition"), the one strategy where a whole partition can be expired by a single time threshold (expire_by_partition) instead of scanning every key. A new TimeWindow.expire_idle(transaction, now_ms) reuses that existing partition expiry:

def expire_idle(self, transaction, now_ms):
    if not self._close_on_idle:
        return
    max_expired_window_end = now_ms - self._grace_ms
    yield from self.expire_by_partition(transaction, max_expired_window_end, self.collect)

Default is False, so existing pipelines are unaffected (event-time only).

Open questions for maintainers

1. Emitting downstream on idle. compose_all only exposes per-topic root executors — there's no handle to the pipeline stage right after the window. What's the preferred way to route idle-closed windows downstream? This is the main thing gating a non-draft PR.
2. Checkpoint with no offset. An idle close consumes no new Kafka offset. Force a checkpoint flush, or defer to the next real message?
3. Partition ownership after rebalance. Idle closes must only fire for partitions this consumer still owns. Is there an existing assignment hook to read from, or should this check the assignment directly?

Still to do

• App-loop hook calling expire_idle from the empty-poll (rows is None) branch
• Rebalance edge-case test
• Docs

Tests

Added under TestTumblingWindowCloseOnIdle: closes on wall clock past window_end, respects grace, no-op when disabled, and validation rejects the key strategy and current(). Full windows suite passes locally (205 passed).

Happy to change direction — if the dummy-heartbeat approach built into the framework is preferred instead, I can do that.