Summary

• keep session-window periodic emits monotonic by never moving the output watermark behind the current chunk watermark
• apply the same watermark guard to both regular and substream session aggregation transforms
• add focused unit coverage plus a stateless regression test that checks we no longer emit outdated watermark errors for the reproduced long-running pattern

Before / After

When an active session is still open, the periodic-emit path used to overwrite the current watermark with session.win_start. That works only if the current watermark has not already moved past the session start.

Example:

• if current_watermark = 150 and win_start = 200
  • before PR: emitted watermark = 200
  • after PR: emitted watermark = 200
• if current_watermark = 250 and win_start = 200
  • before PR: emitted watermark = 200 which regresses the watermark
  • after PR: emitted watermark = 250 via max(current_watermark, win_start)

So this PR keeps the intended periodic emit behavior, but stops the session path from moving the watermark backwards.