diff --git a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
index 39d52deb1dd13..f772eb14ab656 100644
--- a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
+++ b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
@@ -351,6 +351,12 @@ def _execute_in_fork(command_to_exec: CommandType, celery_task_id: str | None =
         if celery_task_id:
             args.external_executor_id = celery_task_id
 
+        # Reset SIGCHLD to default to prevent billiard's inherited handler from stealing 
+        # the exit status of child processes spawned by the supervisor.
+        import signal
+        if hasattr(signal, "SIGCHLD"):
+            signal.signal(signal.SIGCHLD, signal.SIG_DFL)
+
         setproctitle(f"airflow task supervisor: {command_to_exec}")
         log.debug("calling func '%s' with args %s", args.func.__name__, args)
         args.func(args)
diff --git a/providers/celery/tests/unit/celery/executors/test_celery_executor.py b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
index 28b0840590235..4df915418dd82 100644
--- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
+++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
@@ -1513,3 +1513,24 @@ def test_team_app_name_includes_team_name(self):
         team_conf = ExecutorConf(team_name="team_beta")
         celery_app = celery_executor_utils.create_celery_app(team_conf)
         assert "team_beta" in celery_app.main
+
+
+class TestCeleryExecutorUtils:
+    @mock.patch("airflow.providers.celery.executors.celery_executor_utils.setproctitle")
+    @mock.patch("signal.signal")
+    def test_execute_in_fork_resets_sigchld(self, mock_signal, mock_setproctitle):
+        """
+        Test that _execute_in_fork resets SIGCHLD to SIG_DFL to prevent
+        billiard from intercepting child process exit statuses.
+        """
+        from airflow.providers.celery.executors import celery_executor_utils
+        import signal
+        
+        args = mock.MagicMock()
+        command_to_exec = ["airflow", "tasks", "run", "dummy"]
+        
+        celery_executor_utils._execute_in_fork(command_to_exec, celery_task_id="test_id", args=args)
+        
+        if hasattr(signal, "SIGCHLD"):
+            mock_signal.assert_called_once_with(signal.SIGCHLD, signal.SIG_DFL)
+        args.func.assert_called_once_with(args)