feat(telemetry): emit v1 schema fields [skip-runtime-e2e]

CHANGELOG.md

@@ -61,6 +61,11 @@ telemetry contract — see `Removed` at the bottom of this entry for that.
   internals (they're underscore-prefixed), but the change is recorded
   here for completeness.
 
+### Telemetry payload (v1 schema, axonflow-enterprise#2008)
+
+- New heartbeat fields: `telemetry_type: "sdk"`, `profile` (from `AXONFLOW_PROFILE`, `unknown` when unset), `deployment_mode` aligned to `self_hosted | community_saas | unknown` via the new `_classify_deployment_mode` (host + `AXONFLOW_TRY=1` override).
+- `_classify_endpoint` no longer returns `community-saas` — that value moved off endpoint_type onto deployment_mode; analytics queries on the legacy value must update.
+
 ## [7.1.0] - 2026-05-06 — X-Axonflow-Client header + scope-aware license validation
 
 **Companion release to platform v7.7.0.** The Python SDK now sends an

axonflow/telemetry.py

@@ -118,17 +118,18 @@ def _classify_endpoint(url: str | None) -> str:  # noqa: PLR0911
     """Classify the configured AxonFlow endpoint for analytics (#1525).
 
     Returns one of:
-        ``"community-saas"``    — try.getaxonflow.com shared evaluation server
         ``"localhost"``         — localhost, 127.0.0.1, ::1, 0.0.0.0, ``*.localhost``
         ``"private_network"``   — RFC1918 ranges, link-local, ``*.local``,
                                   ``*.internal``, ``*.lan``, ``*.intranet``
         ``"remote"``            — everything else
         ``"unknown"``           — on any parse failure
 
     The raw URL is never sent — only the classification. See issue #1525.
+
+    As of v8.0 the legacy ``"community-saas"`` return value is removed —
+    deployment topology lives on ``deployment_mode`` (see
+    ``_classify_deployment_mode``) per the v1 schema (axonflow-enterprise#2008).
     """
-    if os.environ.get("AXONFLOW_TRY") == "1":
-        return "community-saas"
     if not url:
         return "unknown"
     try:
@@ -158,6 +159,36 @@ def _classify_endpoint(url: str | None) -> str:  # noqa: PLR0911
     return "remote"
 
 
+def _classify_deployment_mode(url: str | None) -> str:
+    """Classify deployment topology for the v1 telemetry schema (#2008).
+
+    Returns one of:
+        ``"community_saas"``    — try.getaxonflow.com host or AXONFLOW_TRY=1
+        ``"self_hosted"``       — any other reachable endpoint
+        ``"unknown"``           — empty / unparseable endpoint
+
+    The classifier deliberately resolves empty/unparseable to ``"unknown"``
+    rather than ``"self_hosted"`` to keep the self-hosted bucket clean of
+    config gaps. ``AXONFLOW_TRY=1`` is the explicit override path for
+    tenants whose endpoint resolves to a custom hostname proxying
+    ``try.getaxonflow.com``.
+    """
+    if os.environ.get("AXONFLOW_TRY") == "1":
+        return "community_saas"
+    if not url:
+        return "unknown"
+    try:
+        host = urlparse(url).hostname
+    except (ValueError, AttributeError):
+        return "unknown"
+    if not host:
+        return "unknown"
+    host = host.lower()
+    if host == "try.getaxonflow.com" or host.endswith(".try.getaxonflow.com"):
+        return "community_saas"
+    return "self_hosted"
+
+
 def _normalize_arch(arch: str) -> str:
     """Normalize architecture names to match other SDKs."""
     if arch == "aarch64":
@@ -171,9 +202,21 @@ def _build_payload(
     mode: str,
     platform_version: str | None = None,
     endpoint_type: str = "unknown",
+    deployment_mode: str = "unknown",
 ) -> dict[str, object]:
     """Build the JSON payload for the checkpoint ping.
 
+    v1 telemetry-schema fields (axonflow-enterprise#2008):
+
+    * ``telemetry_type`` — always ``"sdk"`` (discriminator for the
+      receiver to route SDK pings vs plugin / platform / synthetic).
+    * ``deployment_mode`` — ``self_hosted | community_saas | unknown``,
+      derived from the endpoint host plus ``AXONFLOW_TRY=1`` override
+      (see ``_classify_deployment_mode``). The ``mode`` parameter is
+      kept for legacy callers but no longer drives this dimension.
+    * ``profile`` — sourced from ``AXONFLOW_PROFILE``; ``"unknown"``
+      when unset. Free-form deployment classifier; analytics only.
+
     The ``stream`` field classifies the heartbeat sub-stream. Sandbox-mode
     clients emit ``"sandbox"`` so analytics can distinguish dev/test pings
     from production heartbeat without conflating them; production-mode and
@@ -182,17 +225,22 @@ def _build_payload(
     wire-allowlist is enforced server-side — see checkpoint-service
     ``IsValidIncomingStream``.
     """
+    profile_env = os.environ.get("AXONFLOW_PROFILE")
+    profile_stripped = profile_env.strip() if profile_env is not None else ""
+    profile = profile_stripped if profile_stripped else "unknown"
     payload: dict[str, object] = {
+        "telemetry_type": "sdk",
         "sdk": "python",
         "sdk_version": _SDK_VERSION,
         "platform_version": platform_version,
         "os": platform.system().lower(),
         "arch": _normalize_arch(platform.machine()),
         "runtime_version": platform.python_version(),
-        "deployment_mode": mode,
+        "deployment_mode": deployment_mode,
         "endpoint_type": endpoint_type,
         "features": [],
         "instance_id": str(uuid.uuid4()),
+        "profile": profile,
     }
     if mode == "sandbox":
         payload["stream"] = "sandbox"
@@ -228,7 +276,8 @@ def _send_telemetry_ping_now(url: str, mode: str, endpoint: str, debug: bool) ->
         if endpoint and health_budget > _MIN_BUDGET_SECONDS:
             platform_version = _detect_platform_version(endpoint, timeout=health_budget)
         endpoint_type = _classify_endpoint(endpoint)
-        payload = _build_payload(mode, platform_version, endpoint_type)
+        deployment_mode = _classify_deployment_mode(endpoint)
+        payload = _build_payload(mode, platform_version, endpoint_type, deployment_mode)
 
         # POST uses all remaining budget.
         post_budget = max(0.0, deadline - time.monotonic())

tests/test_telemetry.py

@@ -93,24 +93,38 @@ class TestBuildPayload:
 
     def test_payload_format(self) -> None:
         """Verify all expected fields are present and correctly typed."""
-        payload = _build_payload("production")
+        payload = _build_payload("production", deployment_mode="self_hosted")
 
+        assert payload["telemetry_type"] == "sdk"
         assert payload["sdk"] == "python"
         assert isinstance(payload["sdk_version"], str)
         assert payload["platform_version"] is None
         assert isinstance(payload["os"], str)
         assert isinstance(payload["arch"], str)
         assert isinstance(payload["runtime_version"], str)
-        assert payload["deployment_mode"] == "production"
+        assert payload["deployment_mode"] == "self_hosted"
         assert payload["features"] == []
         assert isinstance(payload["instance_id"], str)
         # Should be a valid UUID
         assert len(payload["instance_id"]) == 36  # UUID v4 string length
-
-    def test_payload_mode_propagated(self) -> None:
-        """deployment_mode reflects the supplied mode."""
-        assert _build_payload("sandbox")["deployment_mode"] == "sandbox"
-        assert _build_payload("production")["deployment_mode"] == "production"
+        # v1 telemetry-schema profile field
+        assert payload["profile"] == "unknown"
+
+    def test_payload_deployment_mode_propagated(self) -> None:
+        """deployment_mode reflects the supplied v1 schema value."""
+        sh = _build_payload("sandbox", deployment_mode="self_hosted")
+        cs = _build_payload("production", deployment_mode="community_saas")
+        un = _build_payload("production", deployment_mode="unknown")
+        assert sh["deployment_mode"] == "self_hosted"
+        assert cs["deployment_mode"] == "community_saas"
+        assert un["deployment_mode"] == "unknown"
+
+    def test_payload_profile_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """profile sourced from AXONFLOW_PROFILE; unknown when unset."""
+        monkeypatch.setenv("AXONFLOW_PROFILE", "production")
+        assert _build_payload("production")["profile"] == "production"
+        monkeypatch.delenv("AXONFLOW_PROFILE", raising=False)
+        assert _build_payload("production")["profile"] == "unknown"
 
     def test_payload_instance_id_unique(self) -> None:
         """Each call generates a new instance_id."""
@@ -173,8 +187,11 @@ def test_payload_posted_correctly(
         assert url == _DEFAULT_CHECKPOINT_URL
 
         payload = call_args[1].get("json") or call_args[0][1]
+        assert payload["telemetry_type"] == "sdk"
         assert payload["sdk"] == "python"
-        assert payload["deployment_mode"] == "production"
+        # v1 schema: deployment_mode classifies from endpoint host.
+        assert payload["deployment_mode"] == "self_hosted"
+        assert payload["profile"] == "unknown"
         assert "instance_id" in payload
         # Production-mode payload omits stream (server defaults to heartbeat).
         assert "stream" not in payload
@@ -214,7 +231,9 @@ def test_sandbox_mode_fires_with_stream_tag(
 
         mock_httpx.post.assert_called_once()
         payload = mock_httpx.post.call_args[1].get("json") or mock_httpx.post.call_args[0][1]
-        assert payload["deployment_mode"] == "sandbox"
+        # v1 schema: deployment_mode classifies from endpoint host (self_hosted),
+        # NOT from config.Mode. The sandbox marker lives on `stream`.
+        assert payload["deployment_mode"] == "self_hosted"
         assert payload.get("stream") == "sandbox"
 
     @patch("axonflow.telemetry.httpx")