Ensure logging is configured on remote task workers

src/prefect/logging/configuration.py

@@ -65,6 +65,29 @@ def load_logging_config(path: Path) -> dict[str, Any]:
     return flatdict_to_dict(flat_config)
 
 
+def is_logging_configured() -> bool:
+    """
+    Check whether Prefect logging has already been configured in this process.
+
+    Returns `True` if `setup_logging` has been called at least once, meaning
+    handlers like `APILogHandler` are attached to the Prefect loggers.
+    """
+    return bool(PROCESS_LOGGING_CONFIG)
+
+
+def ensure_logging_setup() -> None:
+    """
+    Ensure Prefect logging is configured in this process, calling
+    `setup_logging` only if it has not already been called.
+
+    Use this in remote execution environments (e.g. Dask/Ray workers) where
+    the normal SDK entry point (`import prefect`) may not have triggered
+    logging configuration.
+    """
+    if not PROCESS_LOGGING_CONFIG:
+        setup_logging()
+
+
 def setup_logging(incremental: bool | None = None) -> dict[str, Any]:
     """
     Sets up logging.

src/prefect/task_engine.py

@@ -73,6 +73,7 @@
     TerminationSignal,
     UpstreamTaskError,
 )
+from prefect.logging.configuration import ensure_logging_setup
 from prefect.logging.loggers import get_logger, patch_print, task_run_logger
 from prefect.results import (
     ResultRecord,
@@ -818,6 +819,9 @@ def initialize_run(
         """
 
         with hydrated_context(self.context):
+            if self.context is not None:
+                ensure_logging_setup()
+
             with SyncClientContext.get_or_create() as client_ctx:
                 self._client = client_ctx.client
                 self._is_started = True
@@ -1445,6 +1449,9 @@ async def initialize_run(
         """
 
         with hydrated_context(self.context):
+            if self.context is not None:
+                ensure_logging_setup()
+
             async with AsyncClientContext.get_or_create():
                 self._client = get_client()
                 self._is_started = True

tests/test_logging.py

@@ -33,6 +33,8 @@
 from prefect.logging import LogEavesdropper
 from prefect.logging.configuration import (
     DEFAULT_LOGGING_SETTINGS_PATH,
+    ensure_logging_setup,
+    is_logging_configured,
     load_logging_config,
     setup_logging,
 )
@@ -285,6 +287,33 @@ def test_setup_logging_applies_root_config_when_no_prior_configuration(
     assert called_config["root"]["handlers"] == ["console"]
 
 
+def test_is_logging_configured_returns_false_when_not_configured(
+    dictConfigMock: MagicMock,
+):
+    assert is_logging_configured() is False
+
+
+def test_is_logging_configured_returns_true_after_setup(dictConfigMock: MagicMock):
+    setup_logging()
+    assert is_logging_configured() is True
+
+
+def test_ensure_logging_setup_calls_setup_logging_when_not_configured(
+    dictConfigMock: MagicMock,
+):
+    ensure_logging_setup()
+    dictConfigMock.assert_called_once()
+
+
+def test_ensure_logging_setup_is_idempotent(dictConfigMock: MagicMock):
+    ensure_logging_setup()
+    ensure_logging_setup()
+    ensure_logging_setup()
+    # setup_logging should only be called once since PROCESS_LOGGING_CONFIG
+    # is populated after the first call
+    dictConfigMock.assert_called_once()
+
+
 def test_setting_aliases_respected_for_logging_config(tmp_path: Path):
     logging_config_content = """
 loggers: