Add deployment step lifecycle logs and error resolution hints (#21020)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Alex Streed <alex.s@prefect.io>
Co-authored-by: alex.s <ajstreed1@gmail.com>
Co-authored-by: Alex Streed <desertaxle@users.noreply.github.com>

Author: devin-ai-integration[bot]

src/prefect/deployments/steps/core.py

@@ -157,7 +157,10 @@ async def run_steps(
             continue
         fqn, inputs = _get_step_fully_qualified_name_and_inputs(step)
         step_name = fqn.split(".")[-1]
-        print_function(f" > Running {step_name} step...")
+        if logger:
+            logger.info("Executing deployment step: %s", step_name)
+        else:
+            print_function(f" > Running {step_name} step...")
 
         # SECURITY: Serialize inputs BEFORE running the step (and thus before templating).
         # This ensures that the event payload contains template strings like
@@ -211,6 +214,9 @@ async def run_steps(
                 upstream_outputs[inputs.get("id")] = step_output
             upstream_outputs.update(step_output)
 
+            if logger:
+                logger.info("Deployment step '%s' completed successfully", step_name)
+
             # Emit success event for this step
             await _emit_pull_step_event(
                 serialized_step,
@@ -230,6 +236,8 @@ async def run_steps(
             )
             raise StepExecutionError(f"Encountered error while running {fqn}") from exc
 
+    if logger:
+        logger.info("All deployment steps completed successfully")
     return upstream_outputs
 
 

src/prefect/deployments/steps/pull.py

@@ -281,7 +281,11 @@ async def pull_with_block(
     try:
         block = await Block.aload(full_slug)
     except Exception:
-        deployment_logger.exception("Unable to load block '%s'", full_slug)
+        deployment_logger.exception(
+            "Failed to load storage block with slug %s."
+            " Verify the block exists and you have access to it.",
+            full_slug,
+        )
         raise
 
     try:

src/prefect/flows.py

@@ -3081,9 +3081,7 @@ async def load_flow_from_flow_run(
         await storage_block.get_directory(from_path=from_path, local_path=".")
 
     if deployment.pull_steps:
-        run_logger.debug(
-            f"Running {len(deployment.pull_steps)} deployment pull step(s)"
-        )
+        run_logger.info(f"Running {len(deployment.pull_steps)} deployment pull step(s)")
 
         from prefect.deployments.steps.core import StepExecutionError, run_steps
 

src/prefect/runner/storage.py

@@ -459,10 +459,15 @@ async def _clone_repo(self):
                 if self._credentials or parsed_url.password or parsed_url.username
                 else exc
             )
-            raise RuntimeError(
-                f"Failed to clone repository {strip_auth_from_url(self._url)!r} with exit code"
+            safe_url = strip_auth_from_url(self._url)
+            error_message = (
+                f"Failed to clone repository {safe_url!r} with exit code"
                 f" {exc.returncode}."
-            ) from exc_chain
+            )
+            hint = _get_git_clone_error_hint(exc)
+            if hint:
+                error_message += f" {hint}"
+            raise RuntimeError(error_message) from exc_chain
 
         if self._commit_sha:
             # Fetch the commit
@@ -699,10 +704,14 @@ async def pull_code(self) -> None:
                 )
             )
         except Exception as exc:
-            raise RuntimeError(
+            error_message = (
                 f"Failed to pull contents from remote storage {self._url!r} to"
                 f" {self.destination!r}"
-            ) from exc
+            )
+            hint = _get_remote_storage_error_hint(exc)
+            if hint:
+                error_message += f". {hint}"
+            raise RuntimeError(error_message) from exc
 
     def to_pull_step(self) -> dict[str, Any]:
         """
@@ -898,6 +907,96 @@ def create_storage_from_source(
         return LocalStorage(path=source, pull_interval=pull_interval)
 
 
+# Pattern-based error hints for git clone failures
+_GIT_CLONE_ERROR_HINTS: list[tuple[str, str]] = [
+    (
+        "Authentication failed",
+        "Hint: Check that your credentials or access token are correct and not expired.",
+    ),
+    (
+        "repository not found",
+        "Hint: Verify the repository URL and that you have access to it.",
+    ),
+    (
+        "Could not resolve host",
+        "Hint: Check your network connectivity and DNS settings.",
+    ),
+    (
+        "Connection refused",
+        "Hint: Check your network connectivity and DNS settings.",
+    ),
+    (
+        "Permission denied",
+        "Hint: Check your SSH key or token permissions.",
+    ),
+    (
+        "destination path",
+        "Hint: A stale working directory may exist. Consider removing it and retrying.",
+    ),
+    (
+        "not found",
+        "Hint: Verify the repository URL and that you have access to it.",
+    ),
+]
+
+
+def _get_git_clone_error_hint(exc: subprocess.CalledProcessError) -> str | None:
+    """Extract a resolution hint from a git clone CalledProcessError's stderr."""
+    stderr = ""
+    if exc.stderr:
+        stderr = (
+            exc.stderr.decode("utf-8", errors="replace")
+            if isinstance(exc.stderr, bytes)
+            else str(exc.stderr)
+        )
+    for pattern, hint in _GIT_CLONE_ERROR_HINTS:
+        if pattern.lower() in stderr.lower():
+            return hint
+    return None
+
+
+# Pattern-based error hints for remote storage failures
+_REMOTE_STORAGE_ERROR_HINTS: list[tuple[str, str]] = [
+    (
+        "NoSuchBucket",
+        "Hint: Verify the bucket name and region.",
+    ),
+    (
+        "AccessDenied",
+        "Hint: Check your storage permissions and credentials.",
+    ),
+    (
+        "403",
+        "Hint: Check your storage permissions and credentials.",
+    ),
+    (
+        "NoSuchKey",
+        "Hint: Verify the storage path exists.",
+    ),
+    (
+        "ConnectionError",
+        "Hint: Check network connectivity and the storage endpoint URL.",
+    ),
+    (
+        "EndpointConnectionError",
+        "Hint: Check network connectivity and the storage endpoint URL.",
+    ),
+    (
+        "ConnectionRefusedError",
+        "Hint: Check network connectivity and the storage endpoint URL.",
+    ),
+]
+
+
+def _get_remote_storage_error_hint(exc: Exception) -> str | None:
+    """Extract a resolution hint from a remote storage exception."""
+    error_str = str(exc).lower()
+    for pattern, hint in _REMOTE_STORAGE_ERROR_HINTS:
+        if pattern.lower() in error_str:
+            return hint
+    return None
+
+
 def _format_token_from_credentials(
     netloc: str,
     credentials: dict[str, Any] | GitCredentials,

tests/deployment/test_steps.py

@@ -720,6 +720,90 @@ def func(*args, **kwargs):
         await run_steps(steps, {}, print_function=mock_print)
         mock_print.assert_any_call("this is a warning")
 
+    async def test_run_steps_uses_print_function_without_logger(self):
+        """Test that run_steps uses print_function (not logger) when no logger
+        is provided, and does not emit lifecycle log messages."""
+        mock_print = MagicMock()
+
+        steps = [
+            {
+                "prefect.deployments.steps.run_shell_script": {
+                    "script": "echo 'hello'",
+                }
+            },
+        ]
+
+        await run_steps(steps, {}, print_function=mock_print)
+
+        # print_function should have been called with the step name
+        mock_print.assert_any_call(" > Running run_shell_script step...")
+
+    async def test_run_steps_logs_lifecycle_with_provided_logger(self):
+        """Test that run_steps uses a provided logger for lifecycle messages."""
+        mock_logger = MagicMock()
+
+        steps = [
+            {
+                "prefect.deployments.steps.run_shell_script": {
+                    "script": "echo 'test'",
+                }
+            },
+        ]
+
+        await run_steps(steps, {}, logger=mock_logger)
+
+        # Verify the provided logger was used for lifecycle messages
+        start_logged = any(
+            "Executing deployment step: %s" in str(c)
+            for c in mock_logger.info.call_args_list
+        )
+        complete_logged = any(
+            "Deployment step '%s' completed successfully" in str(c)
+            for c in mock_logger.info.call_args_list
+        )
+        all_complete_logged = any(
+            "All deployment steps completed successfully" in str(c)
+            for c in mock_logger.info.call_args_list
+        )
+
+        assert start_logged, (
+            f"Expected start log, got: {mock_logger.info.call_args_list}"
+        )
+        assert complete_logged, (
+            f"Expected complete log, got: {mock_logger.info.call_args_list}"
+        )
+        assert all_complete_logged, (
+            f"Expected all-complete log, got: {mock_logger.info.call_args_list}"
+        )
+
+    async def test_run_steps_no_complete_log_on_failure(self, monkeypatch):
+        """Test that the all-complete log is NOT emitted when a step fails."""
+        mock_logger = MagicMock()
+
+        def failing_step(**kwargs):
+            raise RuntimeError("boom")
+
+        monkeypatch.setattr(
+            "prefect.deployments.steps.run_shell_script",
+            failing_step,
+        )
+
+        steps = [
+            {
+                "prefect.deployments.steps.run_shell_script": {
+                    "script": "echo 'test'",
+                }
+            },
+        ]
+
+        with pytest.raises(StepExecutionError):
+            await run_steps(steps, {}, logger=mock_logger)
+
+        info_calls = [str(c) for c in mock_logger.info.call_args_list]
+        assert not any(
+            "All deployment steps completed successfully" in c for c in info_calls
+        )
+
 
 class MockCredentials:
     def __init__(
@@ -1535,7 +1619,8 @@ async def test_block_not_found(self, caplog):
                 }
             )
 
-        assert "Unable to load block 'in-the/wind'" in caplog.text
+        assert "Failed to load storage block with slug in-the/wind" in caplog.text
+        assert "Verify the block exists and you have access to it" in caplog.text
 
     async def test_incorrect_type_of_block(self, caplog):
         """