[AmoghAtreya]

WatchFilePattern kept its newest-timestamp state under each file path. Beam partitions that state by key, so each path was compared only with the pipeline start time. Any file newer than startup was treated as the latest model, and a poll that saw several of them emitted all of them. RunInference reads this side input with AsSingleton, which rejects more than one value.

Every watched file now shares one key, and the newest modification time seen so far is remembered. Each poll is reduced to the single newest file. A later file is emitted only when its modification time is newer than that. Files already present at pipeline construction stay on the model configured on the handler.

A model update still needs a new filename. The file watcher reports a path once, and RunInference reloads only when the path changes. A rollback has to be uploaded under a new name with a newer modification time.

Fixes #40638.

Test plan

• utils_test.py: several files in one poll, out of order, emit only the newest model, and AsSingleton accepts that value.
• An older file that arrives in a later poll is not emitted.
• A newer file replaces an older one.
• Files at or before pipeline construction are not emitted.
• CHANGES.md notes the fix.

[github-actions]

📋 Review Summary

This PR fixes a bug in WatchFilePattern (#40638) where model files were keyed by their individual paths before stateful timestamp comparison, causing older models to be emitted and multiple files within a single poll to break AsSingleton side input consumption in RunInference. By keying all matched files with a shared global key, reducing each triggered pane via CombinePerKey(_LatestFileCombineFn()), and tracking the latest modification timestamp in _EmitLatestModel, the transform now reliably emits at most one newest model per pane.

🔍 General Feedback

• The pipeline topology change—applying WindowInto with Repeatedly(AfterProcessingTime(1)) and DISCARDING accumulation mode before CombinePerKey and the stateful _EmitLatestModel ParDo—is clean and preserves windowing/triggering strategies downstream for AsSingleton side inputs.
• Deterministic tie-breaking by file path in _is_newer ensures CombineFn commutativity and associativity.
• Unit test coverage in utils_test.py thoroughly exercises combine order independence, tie-breaking, pre-start filtering, single-poll reduction, and multi-poll state progression via TestStream.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Assigning reviewers:

R: @tvalentyn for label python.
Backup reviewer: claudevdm (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).