Fix SSHRemoteJobOperator custom remote_base_dir cleanup validation

[axelray-dev]

Summary

Fixes inconsistent validation of custom remote_base_dir on SSHRemoteJobOperator cleanup (#69813).

The operator accepts a custom base directory when constructing job paths, but cleanup validation only allowed the hardcoded defaults (/tmp/airflow-ssh-jobs and the Windows temp default). Jobs with a custom base then failed at the end with ValueError: Invalid job directory ... Expected path under '/tmp/airflow-ssh-jobs'.

Changes

• Thread an optional expected base into _validate_job_dir and the cleanup command builders, defaulting to the existing platform defaults.
• Pass the operator-configured base (RemoteJobPaths.base_dir / remote_base_dir) into cleanup so custom bases validate the same way as defaults.
• Keep prefix-plus-separator safety checks so cleanup still rejects paths outside the configured base.
• Add unit tests for custom-base cleanup success and rejection on both POSIX and Windows.

Verification

• python -m pytest providers/ssh/tests/unit/ssh/utils/test_remote_job.py -q
• 31 passed

Backwards compatibility

• Default behavior is unchanged when remote_base_dir is unset.
• Custom bases that already pass init validation now also pass cleanup validation.

Fixes #69813

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[achimgaedke]

Suggestion: Remove the _validate_job_dir from the build_posix_cleanup_command - which achieves the same as the PR, but is less code.

[kaxil]

The new tests pin the builders, but this line is the wiring #69813 actually hit: if the operator change were reverted and only the utils change kept, every test in the PR would still pass. test_execute_complete_with_cleanup in operators/test_ssh_remote_job.py is easy to clone into a variant with remote_base_dir="/tmp-data/airflow-ssh-jobs" and a matching event job_dir, which would cover the reported scenario end to end.

[kaxil]

If validation fails here, the ValueError escapes _cleanup_remote_job before the retry loop and fails a task whose remote job already finished, which is the failure mode from #69813 and contradicts the log-and-leave contract in the docstring above. That's still reachable after this fix: remote_base_dir is templated, and templates are re-rendered when the operator resumes after deferral, so a value like {{ var.value.ssh_jobs_base }} re-renders against resume-time state while event["job_dir"] keeps the submission-time base the trigger was constructed with. If the Variable changes while the job runs, cleanup rejects a valid path and the task fails.

The cheap guard is to catch the ValueError and route it through the same warn-and-leave path used when retries are exhausted. A fuller fix would thread the submission-time base (self._paths.base_dir) through the trigger event alongside job_dir and validate against that, which also makes the self._paths fallback on the line above unnecessary, since cleanup always runs on a fresh instance where _paths is None. Happy for either to land as a follow-up if you'd rather keep this PR minimal.

[kaxil]

On the suggestion to drop this validation from the cleanup builders instead: I'd keep it. By the time cleanup runs, job_dir is whatever came back in the trigger event (execute_complete passes event["job_dir"] straight through), and it ends up in rm -rf. This prefix check is the only thing pinning an event-supplied path under the configured base.

[AchimGaedkeLynker]

Hmm, interesting. I thought I could trust the return values of the trigger as it comes from the validated operator argument. I probably haven't conquered the full complexity of the deferral mechanism.

The eternal question of "how paranoid should I be" was (rightfully) preventing me from diving into a fix/PR straight away.

[kaxil]

expected_base or POSIX_DEFAULT_BASE_DIR treats a falsy value as unset, but RemoteJobPaths.__post_init__ only substitutes the default when base_dir is None. Since remote_base_dir is templated and only validated pre-render, a value that renders empty builds job_dir directly under / at submission, then cleanup validates it against the hardcoded default here and raises. Checking expected_base is None keeps the two consistent.

[axelray-dev]

Thanks for the detailed notes.

I agree we should keep the cleanup validation. Once job_dir comes back from the trigger event, that prefix check is basically the only guard before rm -rf.

I'll update this PR to:

• use expected_base is None instead of truthiness
• catch cleanup validation errors and warn/leave instead of failing the task
• add an operator-level custom remote_base_dir cleanup test so the execute_complete path is covered