Keep SSH remote job running when its PTY session hangs up

providers/ssh/src/airflow/providers/ssh/utils/remote_job.py

@@ -223,10 +223,14 @@ def build_posix_wrapper_command(
 exit 0
 '
 
+# Redirect stdin from /dev/null too, not just stdout/stderr: a fresh setsid session
+# leader that keeps the launching terminal on any fd re-acquires it as its controlling
+# terminal, so a hangup (SSH session with a PTY dropping) would SIGHUP the detached job
+# and defeat the whole point of running it in its own session.
 if command -v setsid >/dev/null 2>&1; then
-  setsid bash -c "$job_script" >/dev/null 2>&1 &
+  setsid bash -c "$job_script" </dev/null >/dev/null 2>&1 &
 else
-  nohup bash -c "$job_script" >/dev/null 2>&1 &
+  nohup bash -c "$job_script" </dev/null >/dev/null 2>&1 &
 fi
 echo "{paths.job_id}"
 """

providers/ssh/tests/unit/ssh/utils/test_remote_job.py

@@ -397,11 +397,6 @@ def test_kill_terminates_whole_job_tree(self, tmp_path):
         pgid = self._await_recorded_pid(paths)
         self._assert_kill_tears_down(paths, pgid, marker)
 
-    # Same environment-dependent process-group race that #69384 added reruns for on the
-    # sibling test; that marker was dropped in #69490 when this pty variant was written.
-    # The launch still depends on how the runner schedules the setsid fork, so keep main
-    # green on a fresh draw - the first attempt's assertion text stays in the CI log.
-    @pytest.mark.flaky(reruns=5)
     def test_kill_terminates_whole_job_tree_under_job_control(self, tmp_path):
         """With job control on, setsid(1) forks and the launcher's ``$!`` would name the
         short-lived setsid parent, not the job -- the condition the old wrapper orphaned
@@ -414,8 +409,9 @@ def test_kill_terminates_whole_job_tree_under_job_control(self, tmp_path):
         self._run_bash_mc_under_pty(
             wrapper + "\necho SUBMIT_DONE\n",
             b"SUBMIT_DONE",
-            # The job records its pid only after setsid(2) has put it in its own session, so
-            # a non-empty pid file is proof the pty hangup below can no longer reach it.
+            # Hang up only once the job is up (pid file written), so the pgrep below sees a
+            # started job. The job survives the hangup regardless: the wrapper detaches its
+            # stdin from the terminal, so the setsid session never adopts the pty.
             detached=lambda: pid_path.exists() and bool(pid_path.read_text().strip()),
         )
         pgid = self._await_recorded_pid(paths)