feat(events): record a created deployment with the uuid it was given (#7115)

`Create Crew Deployment` fires before the API call that creates the
deployment, so it counts creation ATTEMPTS and cannot carry the uuid - the
call that creates the deployment is the call that returns it. Live effect:
`create_deployment` reads 76,015 events with 0 carrying a uuid (0.000%), so
deployments cannot be joined to anything.

Moving the existing span after the response would fix the uuid and silently
redefine the metric, turning attempts into successes; the deployment churn
figures are built on attempts. So this adds a second span rather than moving
the first.

`Crew Deployment Created` fires after `_validate_response`, which raises
SystemExit on failure - a failed create therefore still counts as an attempt
and reports no creation. Both creation paths, git remote and zip upload,
converge on that line and both return the uuid.

Emits no `deploy:created` feature count: the attempt span already does, and a
second emit would double the deployment count that origin-independent
aggregation depends on.

Tests cover the emitter (uuid carried, distinct span name, no second feature
count, absent-vs-empty uuid) and the call site across both creation paths plus
the failure path.

The warehouse consumer must be widened BEFORE this merges or it delivers
nothing: `mv_span_fanout_forward` filters on a hard-coded 13-name allowlist,
and `mv_fanout_deployment_spans`'s `multiIf` ends in a catch-all `remove_crew`
arm that would mislabel the new span. Runbook prepared separately.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

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

@@ -375,7 +375,16 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
             response = self._create_crew_from_zip(env_vars, repository, confirm)
 
         self._validate_response(response)
-        self._display_creation_success(response.json())
+        json_response = response.json()
+        # After _validate_response, not before: it raises SystemExit on a failed
+        # create, so the span cannot fire for a deployment that was not made. This
+        # is the first point at which the uuid exists -- the pre-flight span at the
+        # top of this method counts the attempt and cannot carry it.
+        created_uuid = json_response.get("uuid")
+        self._telemetry.crew_deployment_created_span(
+            uuid=str(created_uuid) if created_uuid else None, source=source
+        )
+        self._display_creation_success(json_response)
 
     def _prepare_git_repository(self) -> git.Repository | None:
         """Prepare Git for deploy while preserving remote deploy when possible."""

lib/cli/tests/deploy/test_deploy_main.py

@@ -587,6 +587,101 @@ class TestDeployCommand(unittest.TestCase):
             self.assertIn("Deployment created successfully!", fake_out.getvalue())
             self.assertIn("new-uuid", fake_out.getvalue())
 
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    @pytest.mark.timeout(180)
+    def test_create_crew_reports_the_created_uuid_from_the_git_path(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        """The attempt span cannot carry the uuid; the post-success span must."""
+        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 = ""
+
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_response.is_success = True
+        mock_response.json.return_value = {"uuid": "new-uuid", "status": "created"}
+        self.mock_client.create_crew.return_value = mock_response
+
+        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_created_span.assert_called_once_with(
+            uuid="new-uuid", source="cli"
+        )
+        telemetry.create_crew_deployment_span.assert_called_once_with(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")
+    def test_create_crew_reports_the_created_uuid_from_the_zip_path(
+        self, mock_repository, mock_fetch_env, mock_create_project_zip
+    ):
+        """The two creation paths converge, so the uuid must arrive from both."""
+        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.return_value = Path("/tmp/test_project.zip")
+
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_response.is_success = True
+        mock_response.json.return_value = {"uuid": "zip-uuid", "status": "created"}
+        self.mock_client.create_crew_from_zip.return_value = mock_response
+
+        with patch.object(self.deploy_command, "_telemetry") as telemetry:
+            with patch("sys.stdout", new=StringIO()):
+                self.deploy_command.create_crew(skip_validate=True, confirm=True)
+
+        telemetry.crew_deployment_created_span.assert_called_once_with(
+            uuid="zip-uuid", source="cli"
+        )
+
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("builtins.input")
+    def test_a_failed_create_counts_the_attempt_but_reports_no_creation(
+        self, mock_input, mock_repository, mock_fetch_env
+    ):
+        """The whole point of two spans: a failed create is an attempt, not a success.
+
+        Collapsing them into one post-success span would silently convert the
+        creation-attempt metric into a creation-success metric, which the churn
+        figures are built on.
+        """
+        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 = ""
+
+        mock_response = MagicMock()
+        mock_response.status_code = 500
+        mock_response.is_success = False
+        mock_response.json.return_value = {"error": "boom"}
+        self.mock_client.create_crew.return_value = mock_response
+
+        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.create_crew_deployment_span.assert_called_once_with(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")

lib/crewai-core/src/crewai_core/telemetry.py

@@ -350,6 +350,39 @@ class Telemetry:
         self._safe_telemetry_procedure(_operation)
         self.feature_usage_span("deploy:created")
 
+    def crew_deployment_created_span(
+        self, uuid: str | None = None, source: DeploySource = "cli"
+    ) -> None:
+        """Records that a crew deployment was confirmed created, with its uuid.
+
+        Distinct from :meth:`create_crew_deployment_span`, which fires *before*
+        the API call and so counts creation **attempts**. The uuid cannot be on
+        that span: the call that creates the deployment is the call that returns
+        the uuid, so it does not exist yet. Attribution therefore needs a second
+        span, emitted once the response has validated.
+
+        Emits no feature count on purpose. ``create_crew_deployment_span``
+        already emits ``deploy:created``; a second emit would double the
+        deployment count that origin-independent aggregation depends on.
+
+        Args:
+            uuid: The deployment that was created.
+            source: Where the deployment was initiated from.
+        """
+
+        from crewai_core.version import get_crewai_version
+
+        def _operation() -> None:
+            tracer = self.provider.get_tracer(TRACER_NAME)
+            span = tracer.start_span("Crew Deployment Created")
+            self._add_attribute(span, "crewai_version", get_crewai_version())
+            if uuid:
+                self._add_attribute(span, "uuid", uuid)
+            self._add_attribute(span, "source", source)
+            close_span(span)
+
+        self._safe_telemetry_procedure(_operation)
+
     def get_crew_logs_span(
         self, uuid: str | None, log_type: str = "deployment"
     ) -> None:

lib/crewai-core/tests/test_telemetry_deploy.py

@@ -8,7 +8,7 @@ deployments", from the feature-usage aggregation, regardless of origin.
 from __future__ import annotations
 
 from collections.abc import Iterator
-from typing import Any
+from typing import Any, cast
 from unittest.mock import MagicMock, patch
 
 from crewai_core.telemetry import Telemetry
@@ -79,6 +79,61 @@ class TestCreateDeployment:
         ]
 
 
+class TestCrewDeploymentCreated:
+    """The post-success span: the only one that can carry the created uuid."""
+
+    def test_carries_the_uuid_and_defaults_to_cli(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        instance, span = telemetry
+        instance.crew_deployment_created_span("dep-abc")
+        attributes = _attributes(span)
+        assert attributes["uuid"] == "dep-abc"
+        assert attributes["source"] == "cli"
+
+    def test_is_a_separate_span_from_the_attempt(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        """Two names, because they count different things.
+
+        ``Create Crew Deployment`` counts attempts and fires before the API call;
+        this one fires only after one succeeded. Collapsing them would turn the
+        attempt metric into a success metric, which the churn figures rely on.
+        """
+        instance, _ = telemetry
+        # The fixture patches `provider` with a MagicMock; the declared type is a
+        # real TracerProvider, so narrow it rather than reaching through it.
+        tracer = cast(MagicMock, instance.provider).get_tracer.return_value
+        # feature_usage_span opens a span of its own; patch it out so this asserts
+        # on the two deployment spans rather than on emission order in general.
+        with patch.object(instance, "feature_usage_span"):
+            instance.create_crew_deployment_span()
+            instance.crew_deployment_created_span("dep-abc")
+        assert [call.args[0] for call in tracer.start_span.call_args_list] == [
+            "Create Crew Deployment",
+            "Crew Deployment Created",
+        ]
+
+    def test_does_not_emit_a_second_deploy_created_count(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        """A second emit would double every deployment in the feature count."""
+        instance, _ = telemetry
+        with patch.object(instance, "feature_usage_span") as feature:
+            instance.crew_deployment_created_span("dep-abc")
+        feature.assert_not_called()
+
+    def test_omits_the_uuid_key_rather_than_writing_an_empty_one(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        """Absent must stay distinguishable from empty, as on the sibling spans."""
+        instance, span = telemetry
+        instance.crew_deployment_created_span(None, source="tui")
+        attributes = _attributes(span)
+        assert "uuid" not in attributes
+        assert attributes["source"] == "tui"
+
+
 class TestStartDeployment:
     def test_defaults_to_cli_and_keeps_the_uuid(
         self, telemetry: tuple[Telemetry, MagicMock]