Validate Gemini batch results_folder in the worker that writes results

[shahar1]

AI Summary

In deferrable mode, GenAIGeminiCreateBatchJobOperator / GenAIGeminiCreateEmbeddingsBatchJobOperator write the results file in execute_complete(), which runs in a fresh resume worker where execute() — and its results_folder pre-flight check (moved there by #70362) — never ran. A folder missing in that worker (heterogeneous workers, or a relative path resolved against a different CWD) surfaced as a raw FileNotFoundError after the billable batch job already succeeded.

The existing check is extracted into _validate_results_folder() (raise relocated verbatim) and called both from execute() (unchanged fail-fast before creating the job) and from _prepare_results_for_xcom() before downloading, so the process that writes the file is the one that validates the folder.

related: #70362

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Fable 5)

Generated-by: Claude Code (Fable 5) following the guidelines

[potiuk]

The reasoning here is the valuable part. In deferrable mode the results file is written in execute_complete(), which runs in a fresh resume worker where execute() never ran — so the pre-flight check was in a process that had no bearing on the one doing the write. A folder missing on that worker, whether from a heterogeneous fleet or a relative path resolved against a different CWD, surfaced as a raw FileNotFoundError after the billable batch job had already succeeded. Expensive failure, unhelpful error.

Validating immediately before the write is the right placement.

Two things I checked rather than assumed:

The exception ratchet is unchanged — each operator gains a raise in the helper and loses the inline one, so the file stays at 20 and matches its entry.

The two new tests share a name, which would silently shadow if they lived in the same class. They don't — one is in TestGenAIGeminiCreateBatchJobOperator, the other in TestGenAIGeminiCreateEmbeddingsBatchJobOperator, so both run. Asserting download_file.assert_not_called() alongside the raise is what makes them worth having: it pins that the failure happens before the download rather than merely somewhere.

One nit, not worth holding this up: _validate_results_folder is byte-identical in both classes and can now drift independently. The file already duplicates _wait_until_complete and _prepare_results_for_xcom the same way, so it matches local structure — a shared base is a bigger refactor than this fix should carry.

Rebased onto main before merging.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting