[Python] Bound and harden the Watch transform

[Eliaaazzz]

Stacked on apache#39023, the Watch transform MVP. The base of this PR is the
python-watch-transform branch, so it shows only the single follow-up commit.
It documents the event-time contract, makes late data observable, and adds an
opt-in bound on the per-input dedup state. The default emission and watermark
behavior are unchanged.

Watermark and event-time

• Documents the event-time/watermark contract on the module docstring and on
  PollResult: an explicit watermark if the poll supplies one, otherwise the
  earliest event time of the round's new outputs, held when a round is empty,
  and released to MAX_TIMESTAMP on complete().
• Adds a throttled warning when an output is emitted behind the current
  watermark, so out-of-order late data is observable instead of being silently
  dropped downstream. Out-of-order sources should supply
  PollResult.with_watermark.

Scalability

The per-input dedup set keeps one hash per distinct output and is unbounded by
default, so it grows with the number of distinct outputs an input produces.

• Adds opt-in Watch.with_dedup_horizon. Outputs more than the horizon behind
  the watermark are dropped and their dedup state collected, so the set tracks a
  sliding event-time window. A re-listed aged output is suppressed rather than
  re-emitted, because the suppression cutoff equals the prior round's eviction
  cutoff and leaves no resurrection gap, so a full-relisting poll stays
  exactly-once with an advancing watermark. The trade-off is that a genuinely
  new output arriving more than the horizon behind the watermark is dropped as
  late and reported through a throttled warning. Off by default. The horizon is
  validated non-negative.
• Reuses the parent dedup map on idle rounds to drop the O(N) per-round copy.

Testing

30 tests and 5 subtests pass on the in-memory DirectRunner, covering the
watermark contract, the out-of-order late warning, dedup horizon eviction,
retention, the at-cutoff boundary, order-independence, suppression, re-emission,
idle-round reuse, and an end-to-end relisting plus suppression case. yapf
0.43.0, isort 7.0.0, and pylint are clean.

[Eliaaazzz]

Validated on Dataflow Runner v2 (us-east1, streaming). The poll re-lists a perpetual old output every round; with with_dedup_horizon it should be emitted once then suppressed.

All 3 seeds: total=11 distinct=11 ancient=1 — the re-listed old output is emitted exactly once, no duplicates across distributed checkpoints, watermark advances to MAX and the job reaches JOB_STATE_DONE.

[Eliaaazzz]

Superseded by #2, which solves the unbounded-state problem with a timestamp high-water-mark cursor (O(1) per-input state) instead of a hash set with time-based GC (O(window)). Closing in favor of that design.