feat(telemetry): count deployments from any origin and record where they started (#6974)

Deployments were only countable through two span types that no aggregation
reads, and cli_usage:deploy counted the TUI button rather than deployments.
Emit deploy:created and deploy:pushed alongside the existing spans so a
deployment is countable from the feature-usage aggregation no matter how it
was started, and tag Create Crew Deployment / Start Deployment with
source=cli|tui so the two origins stay distinguishable.

Separately, the TUI's `t` and `d` key bindings dispatch straight to
action_view_traces / action_deploy_crew, which never recorded anything -
only on_button_pressed did. Every keyboard-driven trace view and deploy was
therefore invisible. Move the recording into the actions, which both input
paths funnel through, and past the completed guard so a mid-run keypress
that does nothing is not counted as usage.


Claude-Session: https://claude.ai/code/session_01ASfWmW3RGy4qAQm6s8U9jH

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

docs/edge/ar/telemetry.mdx

@@ -59,7 +59,7 @@ os.environ['OTEL_SDK_DISABLED'] = 'true'
 | نعم       | بيانات تنفيذ الاختبار | تشمل: مفتاح ومعرّف الطاقم المُولّد عشوائياً، عدد التكرارات، اسم النموذج المستخدم، درجة الجودة (عدد عشري)، وقت التنفيذ (بالثواني). كلها غير شخصية. |
 | نعم       | بيانات دورة حياة المهمة | تشمل: أوقات الإنشاء وبدء/انتهاء التنفيذ، معرّفات الطاقم والمهمة. مخزنة كنطاقات مع طوابع زمنية. لا بيانات شخصية. |
 | نعم       | سمات LLM | تشمل: الاسم، model_name، model، top_k، temperature، واسم فئة LLM. كلها بيانات تقنية غير شخصية. |
-| نعم       | محاولة نشر الطاقم باستخدام CLI الخاص بـ CrewAI | تشمل: حقيقة إجراء النشر ومعرّف الطاقم، وما إذا كان يحاول سحب السجلات، لا بيانات أخرى. |
+| نعم       | محاولة نشر الطاقم باستخدام CLI الخاص بـ CrewAI | تشمل: حقيقة إجراء النشر ومعرّف الطاقم، وما إذا كان يحاول سحب السجلات، وما إذا بدأ النشر من أمر CLI أو من واجهة التشغيل TUI. لا تُسجَّل محتويات المشروع أو الطاقم. لا توجد بيانات شخصية. |
 | نعم       | بيئة التنفيذ | تشمل: مساعد البرمجة بالذكاء الاصطناعي الذي يشغّل العملية إن وُجد (واحد من قائمة ثابتة مثل `claude_code` أو `codex` أو `cursor` أو `unknown`)، ومكان تشغيل العملية (واحد من قائمة ثابتة مثل `ci` أو `container` أو `serverless` أو `interactive`)، و`project_id` من ملف `pyproject.toml` عند ضبطه. يتحقق الاكتشاف فقط مما إذا كانت متغيرات البيئة المعروفة مضبوطة، ولا يقرأ قيمها أبدًا. لا بيانات شخصية. |
 | نعم       | إشارات دورة حياة التدفق | تشمل: بدء التدفق، وما إذا اكتمل أو فشل، وما إذا فشلت إحدى دواله، وما إذا توقّف مؤقتًا لطلب إدخال أو ملاحظات بشرية، وما إذا كان البدء استئنافًا لتشغيل سابق، وما إذا فشلت دورة محادثة، ومدة تشغيل التدفق، وما إذا كان التدفق من التدفقات التي يشغّلها CrewAI داخليًا أم من كتابتك. يُسجَّل اسم التدفق (يجب ألا يتضمن معلومات شخصية)، كما هو الحال بالفعل عند إنشاء التدفق وتنفيذه. لا تُسجَّل أبدًا أسماء الدوال أو رسائل الأخطاء أو حالة التدفق. لا بيانات شخصية. |
 | لا        | بيانات الوكيل الموسّعة | تشمل: وصف الهدف، نص الخلفية، معرّف ملف موجهات i18n. يجب على المستخدمين التأكد من عدم تضمين معلومات شخصية في حقول النص. |

docs/edge/en/telemetry.mdx

@@ -59,7 +59,7 @@ own tracer provider, which is independent of the one described here.
 | Yes       | Test Execution Data | Includes: crew's randomly generated key and ID, number of iterations, model name used, quality score (float), execution time (in seconds). All non-personal. |
 | Yes       | Task Lifecycle Data | Includes: creation and execution start/end times, crew and task identifiers. Stored as spans with timestamps. No personal data. |
 | Yes       | LLM Attributes | Includes: name, model_name, model, top_k, temperature, and class name of the LLM. All technical, non-personal data. |
-| Yes       | Crew Deployment attempt using crewAI CLI | Includes: The fact a deploy is being made and crew id, and if it's trying to pull logs, no other data. |
+| Yes       | Crew Deployment attempt using crewAI CLI | Includes: The fact a deploy is being made and crew id, whether it's trying to pull logs, and whether the deploy was started from a CLI command or from the run TUI. No project or crew contents. No personal data. |
 | Yes       | Execution Environment | Includes: which AI coding assistant is running the process, if any (one of a fixed list such as `claude_code`, `codex`, `cursor`, or `unknown`), where the process runs (one of a fixed list such as `ci`, `container`, `serverless`, `interactive`), and the `project_id` from your `pyproject.toml` when one is configured. Detection reads only whether known environment variables are set, never their values. No personal data. |
 | Yes       | Flow Lifecycle Signals | Includes: that a flow started, whether it completed or failed, whether one of its methods failed, whether it paused for human input or feedback, whether the start was a resumed run, whether a conversation turn failed, how long the flow ran, and whether the flow is one CrewAI runs internally or one you wrote. The flow name is recorded (should not include personal info), as it already is for flow creation and execution. Method names, error messages and flow state are never recorded. No personal data. |
 | No        | Agent's Expanded Data | Includes: goal description, backstory text, i18n prompt file identifier. Users should ensure no personal info is included in text fields. |

docs/edge/ko/telemetry.mdx

@@ -57,7 +57,7 @@ provider로 등록하지 않습니다. 이를 통해 양방향이 분리됩니
 | 예      | 테스트 실행 데이터                         | crew의 랜덤 생성 키와 ID, 반복 횟수, 사용된 모델명, 품질 점수(실수), 실행 시간(초 단위)이 포함됩니다. 모두 비개인 정보입니다. |
 | 예      | 작업 라이프사이클 데이터                   | 생성 및 실행 시작/종료 시각, crew 및 작업 식별자가 포함됩니다. 타임스탬프를 포함한 span으로 저장됩니다. 개인 정보 없음. |
 | 예      | LLM 속성                                  | LLM의 이름, model_name, 모델, top_k, temperature 및 클래스명이 포함됩니다. 모두 기술적이고 비개인 정보입니다. |
-| 예      | crewAI CLI를 통한 Crew 배포 시도           | 배포가 시도되고 있고 crew id가 포함되며, 로그를 가져오려고 하는 경우에만 해당. 다른 데이터 없음. |
+| 예      | crewAI CLI를 통한 Crew 배포 시도 | 포함 항목: 배포가 시도되고 있다는 사실과 crew id, 로그를 가져오려고 하는지 여부, 그리고 배포가 CLI 명령에서 시작되었는지 실행 TUI에서 시작되었는지 여부. 프로젝트나 crew의 내용은 기록되지 않습니다. 개인 정보 없음. |
 | 예      | 실행 환경                                  | 포함: 프로세스를 실행 중인 AI 코딩 어시스턴트(있는 경우, `claude_code`, `codex`, `cursor`, `unknown` 등 고정 목록 중 하나), 프로세스가 실행되는 위치(`ci`, `container`, `serverless`, `interactive` 등 고정 목록 중 하나), 그리고 `pyproject.toml`에 설정된 경우 `project_id`. 감지는 알려진 환경 변수의 설정 여부만 확인하며 값은 읽지 않음. 개인 데이터 없음. |
 | 예      | Flow 라이프사이클 신호 | 포함 항목: flow의 시작, 완료 또는 실패 여부, 해당 메서드의 실패 여부, 사람의 입력이나 피드백을 위해 일시 중지되었는지 여부, 해당 시작이 재개된 실행인지 여부, 대화 턴의 실패 여부, flow 실행 시간, 그리고 해당 flow가 CrewAI가 내부적으로 실행하는 것인지 사용자가 작성한 것인지 여부. flow 이름은 기록되며(개인 정보 포함 불가), 이는 flow 생성 및 실행에서 이미 그러합니다. 메서드 이름, 오류 메시지, flow 상태는 절대 기록하지 않습니다. 개인 정보 없음. |
 | 아니오  | 에이전트 확장 데이터                      | 목표 설명, 배경 이야기 텍스트, i18n 프롬프트 파일 식별자가 포함됩니다. 사용자들은 텍스트 필드에 개인 정보가 포함되지 않도록 해야 합니다. |

docs/edge/pt-BR/telemetry.mdx

@@ -59,7 +59,7 @@ por meio do próprio tracer provider, que é independente do descrito aqui.
 | Sim    | Dados de Execução de Testes                | Inclui: chave e ID aleatórias do crew, número de iterações, nome do modelo usado, score de qualidade (float), tempo de execução (em segundos). Tudo não pessoal.         |
 | Sim    | Dados do Ciclo de Vida da Tarefa           | Inclui: horários de criação, início/fim de execução, identificadores de crew e tarefa. Armazenado como spans com timestamps. Sem dados pessoais.            |
 | Sim    | Atributos do LLM                           | Inclui: nome, model_name, model, top_k, temperatura e nome da classe do LLM. Todos técnicos, sem dados pessoais.                  |
-| Sim    | Tentativa de Deploy do Crew pelo CLI do crewAI | Inclui: O fato de um deploy estar sendo realizado e o crew id, e se está tentando buscar logs, sem mais dados.                          |
+| Sim    | Tentativa de Deploy do Crew pelo CLI do crewAI | Inclui: O fato de um deploy estar sendo realizado e o crew id, se está tentando buscar logs, e se o deploy foi iniciado por um comando do CLI ou pela TUI de execução. Nenhum conteúdo do projeto ou do crew. Nenhum dado pessoal. |
 | Sim    | Ambiente de Execução                       | Inclui: qual assistente de código com IA está executando o processo, se houver (um de uma lista fixa como `claude_code`, `codex`, `cursor` ou `unknown`), onde o processo é executado (um de uma lista fixa como `ci`, `container`, `serverless`, `interactive`) e o `project_id` do seu `pyproject.toml` quando houver um configurado. A detecção lê apenas se variáveis de ambiente conhecidas estão definidas, nunca seus valores. Sem dados pessoais. |
 | Sim    | Sinais de Ciclo de Vida do Flow | Inclui: que um flow iniciou, se foi concluído ou falhou, se um de seus métodos falhou, se pausou para entrada ou feedback humano, se o início foi uma execução retomada, se um turno de conversa falhou, por quanto tempo o flow executou e se o flow é um que o CrewAI executa internamente ou um que você escreveu. O nome do flow é registrado (não deve incluir info pessoal), como já ocorre na criação e execução do flow. Nomes de métodos, mensagens de erro e estado do flow nunca são registrados. Sem dados pessoais. |
 | Não    | Dados Expandidos do Agente                 | Inclui: descrição do objetivo, texto da história, identificador de arquivo i18n prompt. Usuários devem garantir que não haja info pessoal nesses campos de texto.            |

lib/cli/src/crewai_cli/crew_run_tui.py

@@ -1027,6 +1027,9 @@ FooterKey .footer-key--key {
     def action_view_traces(self) -> None:
         if self._status != "completed":
             return
+        # Recorded here rather than in on_button_pressed so the `t` key binding
+        # is counted too, and only once the action can actually do something.
+        self._record_tui_button_click("view_traces")
         if self._trace_url:
             import webbrowser
 
@@ -1115,6 +1118,9 @@ FooterKey .footer-key--key {
     def action_deploy_crew(self) -> None:
         if self._status != "completed":
             return
+        # Recorded here rather than in on_button_pressed so the `d` key binding
+        # is counted too, and only once the action can actually do something.
+        self._record_tui_button_click("deploy")
         self._want_deploy = True
         self._unsubscribe()
         self.exit(self._crew_result)
@@ -1130,10 +1136,8 @@ FooterKey .footer-key--key {
 
     def on_button_pressed(self, event: Button.Pressed) -> None:
         if event.button.id in ("btn-traces", "btn-traces-done"):
-            self._record_tui_button_click("view_traces")
             self.action_view_traces()
         elif event.button.id == "btn-deploy":
-            self._record_tui_button_click("deploy")
             self.action_deploy_crew()
 
     def _scroll_to_result(self) -> None:

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

@@ -5,6 +5,7 @@ from urllib.parse import quote
 import webbrowser
 
 from crewai_core.plus_api import CreateCrewPayload
+from crewai_core.telemetry import DeploySource
 from rich.console import Console
 
 from crewai_cli import git
@@ -285,17 +286,23 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
 
         return _deployment_identifier(status_response)
 
-    def deploy(self, uuid: str | None = None, skip_validate: bool = False) -> None:
+    def deploy(
+        self,
+        uuid: str | None = None,
+        skip_validate: bool = False,
+        source: DeploySource = "cli",
+    ) -> None:
         """
         Deploy a crew using either UUID or project name.
 
         Args:
             uuid (Optional[str]): The UUID of the crew to deploy.
             skip_validate (bool): Skip pre-deploy validation checks.
+            source (DeploySource): Where the deployment was initiated from.
         """
         if not _prepare_project_for_deploy(skip_validate):
             return
-        self._telemetry.start_deployment_span(uuid)
+        self._telemetry.start_deployment_span(uuid, source=source)
         console.print("Starting deployment...", style="bold blue")
         repository = self._prepare_git_repository()
         remote_repo_url = repository.origin_url() if repository else None
@@ -337,17 +344,23 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
             raise ValueError("Deployment status response did not include a uuid")
         return str(uuid)
 
-    def create_crew(self, confirm: bool = False, skip_validate: bool = False) -> None:
+    def create_crew(
+        self,
+        confirm: bool = False,
+        skip_validate: bool = False,
+        source: DeploySource = "cli",
+    ) -> None:
         """
         Create a new crew deployment.
 
         Args:
             confirm (bool): Whether to skip the interactive confirmation prompt.
             skip_validate (bool): Skip pre-deploy validation checks.
+            source (DeploySource): Where the deployment was initiated from.
         """
         if not _prepare_project_for_deploy(skip_validate):
             return
-        self._telemetry.create_crew_deployment_span()
+        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()

lib/cli/src/crewai_cli/run_crew.py

@@ -512,14 +512,14 @@ def _chain_deploy() -> None:
         from crewai_cli.deploy.main import DeployCommand
 
         console.print("\nStarting deployment…\n", style="bold #FF5A50")
-        DeployCommand().create_crew(confirm=True, skip_validate=True)
+        DeployCommand().create_crew(confirm=True, skip_validate=True, source="tui")
     except AuthenticationRequiredError:
         from crewai_cli.authentication.main import AuthenticationCommand
 
         console.print()
         AuthenticationCommand().login()
         try:
-            DeployCommand().create_crew(confirm=True, skip_validate=True)
+            DeployCommand().create_crew(confirm=True, skip_validate=True, source="tui")
         except AuthenticationRequiredError:
             console.print(
                 "\nDeploy failed: authentication is still required.\n",

lib/cli/tests/test_crew_run_tui.py

@@ -104,8 +104,8 @@ def test_chain_deploy_skips_validation_after_auth_retry(monkeypatch) -> None:
     run_crew._chain_deploy()
 
     assert create_calls == [
-        {"confirm": True, "skip_validate": True},
-        {"confirm": True, "skip_validate": True},
+        {"confirm": True, "skip_validate": True, "source": "tui"},
+        {"confirm": True, "skip_validate": True, "source": "tui"},
     ]
     assert login_calls == [True]
 
@@ -131,7 +131,7 @@ def test_chain_deploy_does_not_login_for_deploy_exit(monkeypatch, capsys) -> Non
 
     run_crew._chain_deploy()
 
-    assert create_calls == [{"confirm": True, "skip_validate": True}]
+    assert create_calls == [{"confirm": True, "skip_validate": True, "source": "tui"}]
     assert login_calls == []
     assert "Deploy failed with exit code 42" in capsys.readouterr().out
 
@@ -1721,3 +1721,86 @@ async def test_declarative_flow_runs_on_tui() -> None:
     assert app._final_output == "flow result"
     assert app._crew_result == "flow result"
     assert app._flow_steps[0]["status"] == "done"
+
+
+def test_view_traces_keybinding_records_telemetry(monkeypatch) -> None:
+    """The `t` binding reaches the action directly, never on_button_pressed."""
+    app = CrewRunApp()
+    app._status = "completed"
+    app._trace_url = "https://app.crewai.com/traces/test"
+    app._telemetry = Mock()
+    opened_urls: list[str] = []
+
+    monkeypatch.setattr("webbrowser.open", lambda url: opened_urls.append(url))
+
+    app.action_view_traces()
+
+    app._telemetry.feature_usage_span.assert_called_once_with("cli_usage:view_traces")
+    assert opened_urls == ["https://app.crewai.com/traces/test"]
+
+
+def test_deploy_keybinding_records_telemetry() -> None:
+    """The `d` binding reaches the action directly, never on_button_pressed."""
+    app = CrewRunApp()
+    app._status = "completed"
+    app._crew_result = object()
+    app._telemetry = Mock()
+    app._unsubscribe = lambda: None  # type: ignore[method-assign]
+    exits: list[object] = []
+    app.exit = lambda result: exits.append(result)  # type: ignore[method-assign]
+
+    app.action_deploy_crew()
+
+    app._telemetry.feature_usage_span.assert_called_once_with("cli_usage:deploy")
+    assert app._want_deploy is True
+    assert exits == [app._crew_result]
+
+
+def test_view_traces_before_completion_records_nothing() -> None:
+    """A keypress mid-run is a no-op, so it must not be counted as usage."""
+    app = CrewRunApp()
+    app._status = "running"
+    app._telemetry = Mock()
+
+    app.action_view_traces()
+
+    app._telemetry.feature_usage_span.assert_not_called()
+
+
+def test_deploy_before_completion_records_nothing() -> None:
+    app = CrewRunApp()
+    app._status = "running"
+    app._telemetry = Mock()
+
+    app.action_deploy_crew()
+
+    app._telemetry.feature_usage_span.assert_not_called()
+    assert app._want_deploy is False
+
+
+def test_button_press_records_exactly_once(monkeypatch) -> None:
+    """Recording moved into the action; the button must not double-count."""
+    app = CrewRunApp()
+    app._status = "completed"
+    app._trace_url = "https://app.crewai.com/traces/test"
+    app._telemetry = Mock()
+
+    monkeypatch.setattr("webbrowser.open", lambda url: None)
+
+    app.on_button_pressed(SimpleNamespace(button=SimpleNamespace(id="btn-traces")))
+
+    assert app._telemetry.feature_usage_span.call_count == 1
+
+
+def test_finished_traces_button_still_records(monkeypatch) -> None:
+    """The button's id is swapped to btn-traces-done once a trace URL exists."""
+    app = CrewRunApp()
+    app._status = "completed"
+    app._trace_url = "https://app.crewai.com/traces/test"
+    app._telemetry = Mock()
+
+    monkeypatch.setattr("webbrowser.open", lambda url: None)
+
+    app.on_button_pressed(SimpleNamespace(button=SimpleNamespace(id="btn-traces-done")))
+
+    app._telemetry.feature_usage_span.assert_called_once_with("cli_usage:view_traces")

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

@@ -18,7 +18,7 @@ from functools import cache
 import logging
 import os
 import threading
-from typing import Any, ClassVar, Final
+from typing import Any, ClassVar, Final, Literal
 
 from opentelemetry.context import Context
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
@@ -43,6 +43,9 @@ CREWAI_TELEMETRY_SERVICE_NAME: Final[str] = "crewAI-telemetry"
 
 TRACER_NAME: Final[str] = "crewai.telemetry"
 
+DeploySource = Literal["cli", "tui"]
+"""Where a deployment was initiated from: a direct CLI command, or the run TUI."""
+
 
 def close_span(span: Span) -> None:
     """Set span status to OK and end it."""
@@ -297,27 +300,48 @@ class Telemetry:
 
         self._safe_telemetry_procedure(_operation)
 
-    def start_deployment_span(self, uuid: str | None = None) -> None:
-        """Records the start of a deployment process."""
+    def start_deployment_span(
+        self, uuid: str | None = None, source: DeploySource = "cli"
+    ) -> None:
+        """Records redeploying an existing crew (``crewai deploy push``).
+
+        Also emits ``deploy:pushed`` so that deployments are countable from the
+        feature-usage aggregation regardless of where they were started from.
+
+        Args:
+            uuid: The deployment being pushed to.
+            source: Where the deployment was initiated from.
+        """
 
         def _operation() -> None:
             tracer = self.provider.get_tracer(TRACER_NAME)
             span = tracer.start_span("Start Deployment")
             if uuid:
                 self._add_attribute(span, "uuid", uuid)
+            self._add_attribute(span, "source", source)
             close_span(span)
 
         self._safe_telemetry_procedure(_operation)
+        self.feature_usage_span("deploy:pushed")
 
-    def create_crew_deployment_span(self) -> None:
-        """Records the creation of a new crew deployment."""
+    def create_crew_deployment_span(self, source: DeploySource = "cli") -> None:
+        """Records creating a new crew deployment (``crewai deploy create``).
+
+        Also emits ``deploy:created`` so that deployments are countable from the
+        feature-usage aggregation regardless of where they were started from.
+
+        Args:
+            source: Where the deployment was initiated from.
+        """
 
         def _operation() -> None:
             tracer = self.provider.get_tracer(TRACER_NAME)
             span = tracer.start_span("Create Crew Deployment")
+            self._add_attribute(span, "source", source)
             close_span(span)
 
         self._safe_telemetry_procedure(_operation)
+        self.feature_usage_span("deploy:created")
 
     def get_crew_logs_span(
         self, uuid: str | None, log_type: str = "deployment"

lib/crewai-core/tests/test_telemetry_deploy.py

@@ -0,0 +1,138 @@
+"""Deployment telemetry: attribution by origin, and an origin-independent count.
+
+``Create Crew Deployment`` and ``Start Deployment`` answer "which deployment,
+from where"; ``deploy:created`` / ``deploy:pushed`` answer "how many
+deployments", from the feature-usage aggregation, regardless of origin.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from crewai_core.telemetry import Telemetry
+import pytest
+
+
+@pytest.fixture
+def telemetry() -> Iterator[tuple[Telemetry, MagicMock]]:
+    """A Telemetry whose spans are captured instead of exported.
+
+    The singleton is disabled in tests and never builds a provider, so both the
+    gate and the provider are supplied here.
+    """
+    instance = Telemetry()
+    span = MagicMock()
+    provider = MagicMock()
+    provider.get_tracer.return_value.start_span.return_value = span
+
+    with (
+        patch.object(instance, "provider", provider, create=True),
+        patch.object(instance, "_should_execute_telemetry", return_value=True),
+        patch("crewai_core.telemetry.close_span"),
+    ):
+        yield instance, span
+
+
+def _attributes(span: MagicMock) -> dict[str, Any]:
+    return {call.args[0]: call.args[1] for call in span.set_attribute.call_args_list}
+
+
+def _span_names(provider: MagicMock) -> list[str]:
+    tracer = provider.get_tracer.return_value
+    return [call.args[0] for call in tracer.start_span.call_args_list]
+
+
+class TestCreateDeployment:
+    def test_defaults_to_cli(self, telemetry: tuple[Telemetry, MagicMock]) -> None:
+        instance, span = telemetry
+        instance.create_crew_deployment_span()
+        assert _attributes(span)["source"] == "cli"
+
+    def test_records_tui_when_started_from_the_run_ui(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        instance, span = telemetry
+        instance.create_crew_deployment_span(source="tui")
+        assert _attributes(span)["source"] == "tui"
+
+    def test_also_counts_the_deployment_as_a_feature(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        instance, _ = telemetry
+        with patch.object(instance, "feature_usage_span") as feature:
+            instance.create_crew_deployment_span()
+        feature.assert_called_once_with("deploy:created")
+
+    def test_feature_count_is_the_same_from_either_origin(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        """The whole point: one number for deployments, whatever started them."""
+        instance, _ = telemetry
+        with patch.object(instance, "feature_usage_span") as feature:
+            instance.create_crew_deployment_span(source="cli")
+            instance.create_crew_deployment_span(source="tui")
+        assert [call.args[0] for call in feature.call_args_list] == [
+            "deploy:created",
+            "deploy:created",
+        ]
+
+
+class TestStartDeployment:
+    def test_defaults_to_cli_and_keeps_the_uuid(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        instance, span = telemetry
+        instance.start_deployment_span("dep-123")
+        attributes = _attributes(span)
+        assert attributes["source"] == "cli"
+        assert attributes["uuid"] == "dep-123"
+
+    def test_records_tui_when_started_from_the_run_ui(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        instance, span = telemetry
+        instance.start_deployment_span("dep-123", source="tui")
+        assert _attributes(span)["source"] == "tui"
+
+    def test_source_is_recorded_even_without_a_uuid(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        """uuid is optional; source must not be conditional on it."""
+        instance, span = telemetry
+        instance.start_deployment_span(None, source="tui")
+        attributes = _attributes(span)
+        assert attributes["source"] == "tui"
+        assert "uuid" not in attributes
+
+    def test_also_counts_the_deployment_as_a_feature(
+        self, telemetry: tuple[Telemetry, MagicMock]
+    ) -> None:
+        instance, _ = telemetry
+        with patch.object(instance, "feature_usage_span") as feature:
+            instance.start_deployment_span("dep-123")
+        feature.assert_called_once_with("deploy:pushed")
+
+
+class TestDisabledTelemetry:
+    def test_opted_out_users_emit_nothing(self) -> None:
+        """No span and no feature count when telemetry is off."""
+        instance = Telemetry()
+        provider = MagicMock()
+
+        with (
+            patch.object(instance, "provider", provider, create=True),
+            patch.object(instance, "_should_execute_telemetry", return_value=False),
+            patch.object(instance, "feature_usage_span") as feature,
+        ):
+            instance.create_crew_deployment_span(source="tui")
+            instance.start_deployment_span("dep-123", source="tui")
+
+        assert _span_names(provider) == []
+        # feature_usage_span is itself gated, so it is still called; it is the
+        # export that must not happen. Assert it was not bypassed some other way.
+        assert [call.args[0] for call in feature.call_args_list] == [
+            "deploy:created",
+            "deploy:pushed",
+        ]