feat(cli): record why a deployment create failed

`crewai deploy create` counts every attempt (`Create Crew Deployment`) and every
success (`Crew Deployment Created`), but the gap between them carried no cause:
among clients able to emit the success span, the CLI succeeds 96.7% of the time
and the run TUI 36.4%, and nothing said why. A third span, `Crew Deployment
Failed`, now fires for every failure after the attempt is counted, with a closed
vocabulary `reason` (api_4xx, api_5xx, invalid_response, network_error,
zip_error, user_declined, unexpected), the HTTP `status_code` when the API
answered, and the existing `source`. Never the error message.

The request path is factored into `_request_crew_creation`; every exception is
classified, reported and re-raised unchanged, so CLI and TUI behaviour is the
same as before. HTTP failures are classified before `_validate_response`, which
still prints and exits as it did.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

lib/cli/src/crewai_cli/deploy/main.py

@@ -1,11 +1,14 @@
+import json
 from pathlib import Path
 import subprocess
 from typing import Any
 from urllib.parse import quote
 import webbrowser
+import zipfile
 
 from crewai_core.plus_api import CreateCrewPayload
-from crewai_core.telemetry import DeploySource
+from crewai_core.telemetry import DeployFailureReason, DeploySource
+import httpx
 from rich.console import Console
 
 from crewai_cli import git
@@ -126,6 +129,37 @@ def _deployment_page_url(base_url: str, json_response: dict[str, Any]) -> str |
     )
 
 
+def _creation_failure_reason(exc: BaseException) -> DeployFailureReason:
+    """Classify an exception raised while requesting a deployment, for telemetry.
+
+    Only the archive step raises ``ValueError`` / ``OSError`` inside that request
+    (the project name is validated at construction), so those read as zip errors.
+    """
+    if isinstance(exc, (KeyboardInterrupt, EOFError)):
+        return "user_declined"
+    if isinstance(exc, httpx.HTTPError):
+        return "network_error"
+    if isinstance(exc, (ValueError, OSError, zipfile.BadZipFile)):
+        return "zip_error"
+    return "unexpected"
+
+
+def _response_failure_reason(response: httpx.Response) -> DeployFailureReason | None:
+    """Classify a create response that ``_validate_response`` will reject.
+
+    Mirrors its checks in the same order; ``None`` means the response will pass.
+    """
+    try:
+        response.json()
+    except (json.JSONDecodeError, ValueError):
+        return "invalid_response"
+    if response.status_code >= 500:
+        return "api_5xx"
+    if not response.is_success:
+        return "api_4xx"
+    return None
+
+
 def _needs_lockfile_for_deploy(project_root: Path | None = None) -> bool:
     """Return True when deploy should create the project's first lockfile."""
     root = project_root or Path.cwd()
@@ -442,18 +476,22 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
             return
         self._telemetry.create_crew_deployment_span(source=source)
         console.print("Creating deployment...", style="bold blue")
-        env_vars = fetch_and_json_env_file()
-        repository = self._prepare_git_repository()
-        remote_repo_url = repository.origin_url() if repository else None
-
-        if remote_repo_url:
-            self._confirm_input(env_vars, remote_repo_url, confirm)
-            payload = self._create_payload(env_vars, remote_repo_url)
-            response = self.plus_api_client.create_crew(payload)
-        else:
-            _display_git_remote_help()
-            response = self._create_crew_from_zip(env_vars, repository, confirm)
+        try:
+            response = self._request_crew_creation(confirm)
+        except BaseException as exc:
+            # Report and re-raise unchanged: the CLI and the run TUI already
+            # decide how each failure is shown, this only explains the gap
+            # between attempts and successes.
+            self._telemetry.crew_deployment_failed_span(
+                _creation_failure_reason(exc), source=source
+            )
+            raise
 
+        failure_reason = _response_failure_reason(response)
+        if failure_reason is not None:
+            self._telemetry.crew_deployment_failed_span(
+                failure_reason, source=source, status_code=response.status_code
+            )
         self._validate_response(response)
         json_response = response.json()
         # After _validate_response, not before: it raises SystemExit on a failed
@@ -466,6 +504,20 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
         )
         self._display_creation_success(json_response)
 
+    def _request_crew_creation(self, confirm: bool) -> httpx.Response:
+        """Ask the Enterprise API to create the deployment, from git or from a ZIP."""
+        env_vars = fetch_and_json_env_file()
+        repository = self._prepare_git_repository()
+        remote_repo_url = repository.origin_url() if repository else None
+
+        if remote_repo_url:
+            self._confirm_input(env_vars, remote_repo_url, confirm)
+            payload = self._create_payload(env_vars, remote_repo_url)
+            return self.plus_api_client.create_crew(payload)
+
+        _display_git_remote_help()
+        return self._create_crew_from_zip(env_vars, repository, confirm)
+
     def _prepare_git_repository(self) -> git.Repository | None:
         """Prepare Git for deploy while preserving remote deploy when possible."""
         try:
@@ -544,7 +596,7 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
         env_vars: dict[str, str],
         repository: git.Repository | None,
         confirm: bool,
-    ) -> Any:
+    ) -> httpx.Response:
         """Create a deployment by uploading a project ZIP archive."""
         if not self.project_name:
             raise ValueError("project_name is required to create a ZIP deployment")

lib/cli/tests/deploy/test_deploy_main.py

@@ -825,6 +825,199 @@ class TestDeployCommand(unittest.TestCase):
         telemetry.create_crew_deployment_span.assert_called_once_with(source="cli")
         telemetry.crew_deployment_created_span.assert_not_called()
 
+    # --- why a create failed (the Crew Deployment Failed span) ---------------
+
+    def _git_path(self, mock_input, mock_repository, mock_fetch_env):
+        mock_fetch_env.return_value = {"ENV_VAR": "value"}
+        mock_repository.return_value.origin_url.return_value = (
+            "https://github.com/test/repo.git"
+        )
+        mock_repository.return_value.create_initial_commit_if_needed.return_value = (
+            False
+        )
+        mock_input.return_value = ""
+
+    @staticmethod
+    def _api_response(status_code: int, body: object) -> MagicMock:
+        response = MagicMock()
+        response.status_code = status_code
+        response.is_success = 200 <= status_code < 300
+        if isinstance(body, Exception):
+            response.json.side_effect = body
+        else:
+            response.json.return_value = body
+        return response
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_a_rejected_create_reports_the_api_class_and_status(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        """A 4xx names the class and carries the exact code; the message never leaves."""
+        self._git_path(mock_input, mock_repository, mock_fetch_env)
+        self.mock_client.create_crew.return_value = self._api_response(
+            422, {"name": ["has already been taken"]}
+        )
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                with self.assertRaises(SystemExit):
+                    self.deploy_command.create_crew(skip_validate=True)
+
+        telemetry.crew_deployment_failed_span.assert_called_once_with(
+            "api_4xx", source="cli", status_code=422
+        )
+        telemetry.crew_deployment_created_span.assert_not_called()
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_a_server_error_reports_api_5xx(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        self._git_path(mock_input, mock_repository, mock_fetch_env)
+        self.mock_client.create_crew.return_value = self._api_response(
+            503, {"error": "upstream unavailable"}
+        )
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                with self.assertRaises(SystemExit):
+                    self.deploy_command.create_crew(skip_validate=True)
+
+        telemetry.crew_deployment_failed_span.assert_called_once_with(
+            "api_5xx", source="cli", status_code=503
+        )
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_a_non_json_success_body_reports_invalid_response(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        """_validate_response rejects it, so it is a failure with a 2xx attached."""
+        self._git_path(mock_input, mock_repository, mock_fetch_env)
+        self.mock_client.create_crew.return_value = self._api_response(
+            200, ValueError("not json")
+        )
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                with self.assertRaises(SystemExit):
+                    self.deploy_command.create_crew(skip_validate=True)
+
+        telemetry.crew_deployment_failed_span.assert_called_once_with(
+            "invalid_response", source="cli", status_code=200
+        )
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_a_transport_failure_reports_network_error_and_propagates(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        """No response, so no status; the exception reaches the caller unchanged."""
+        self._git_path(mock_input, mock_repository, mock_fetch_env)
+        self.mock_client.create_crew.side_effect = httpx.ConnectError("refused")
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                with pytest.raises(httpx.ConnectError, match="refused"):
+                    self.deploy_command.create_crew(skip_validate=True)
+
+        telemetry.crew_deployment_failed_span.assert_called_once_with(
+            "network_error", source="cli"
+        )
+        telemetry.crew_deployment_created_span.assert_not_called()
+
+    @patch("crewai_cli.deploy.main.create_project_zip")
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    def test_a_failed_archive_reports_zip_error(
+        self, mock_repository, mock_fetch_env, mock_create_project_zip
+    ):
+        mock_fetch_env.return_value = {"ENV_VAR": "value"}
+        mock_repository.side_effect = ValueError("not a Git repository")
+        initialized_repository = MagicMock()
+        initialized_repository.origin_url.return_value = None
+        mock_repository.initialize.return_value = initialized_repository
+        mock_create_project_zip.side_effect = ValueError(
+            "No deployable project files were found."
+        )
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                with pytest.raises(ValueError, match="No deployable project files"):
+                    self.deploy_command.create_crew(skip_validate=True, confirm=True)
+
+        telemetry.crew_deployment_failed_span.assert_called_once_with(
+            "zip_error", source="cli"
+        )
+        self.mock_client.create_crew_from_zip.assert_not_called()
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_an_abort_at_the_prompt_reports_user_declined(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        self._git_path(mock_input, mock_repository, mock_fetch_env)
+        mock_input.side_effect = KeyboardInterrupt
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                with self.assertRaises(KeyboardInterrupt):
+                    self.deploy_command.create_crew(skip_validate=True)
+
+        telemetry.crew_deployment_failed_span.assert_called_once_with(
+            "user_declined", source="cli"
+        )
+        self.mock_client.create_crew.assert_not_called()
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_a_failure_from_the_run_tui_keeps_its_source(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        """The TUI succeeds far less often than the CLI; the split must survive."""
+        self._git_path(mock_input, mock_repository, mock_fetch_env)
+        self.mock_client.create_crew.return_value = self._api_response(
+            403, {"error": "forbidden"}
+        )
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                with self.assertRaises(SystemExit):
+                    self.deploy_command.create_crew(
+                        skip_validate=True, confirm=True, source="tui"
+                    )
+
+        telemetry.crew_deployment_failed_span.assert_called_once_with(
+            "api_4xx", source="tui", status_code=403
+        )
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_a_successful_create_reports_no_failure(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        self._git_path(mock_input, mock_repository, mock_fetch_env)
+        self.mock_client.create_crew.return_value = self._api_response(
+            201, {"uuid": "new-uuid", "status": "created"}
+        )
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                self.deploy_command.create_crew(skip_validate=True)
+
+        telemetry.crew_deployment_failed_span.assert_not_called()
+        telemetry.crew_deployment_created_span.assert_called_once_with(
+            uuid="new-uuid", source="cli"
+        )
+
     @patch("crewai_cli.deploy.main.create_project_zip")
     @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
     @patch("crewai_cli.deploy.main.git.Repository")