Let kernel own default telemetry enabled state

src/databricks/sql/backend/kernel/client.py

@@ -191,11 +191,9 @@ def _kernel_telemetry_kwargs(options: Dict[str, Any]) -> Dict[str, Any]:
         # The Python telemetry model does not currently track process
         # name; omit it and let the kernel fill what it can derive.
         "process_name": None,
-        # Defaults to False by design: the kernel path deliberately diverges
-        # from the connector-wide True default (see session.py and client.py).
-        # Telemetry is off unless explicitly enabled on this backend.
-        "telemetry_enabled": bool(options.get("enable_telemetry", False)),
     }
+    if options.get("enable_telemetry") is not None:
+        out["telemetry_enabled"] = bool(options["enable_telemetry"])
     if options.get("telemetry_batch_size") is not None:
         out["telemetry_batch_size"] = options["telemetry_batch_size"]
     if options.get("telemetry_circuit_breaker_enabled") is not None:

src/databricks/sql/session.py

@@ -258,13 +258,9 @@ def _create_backend(
             # identity at Session construction time so kernel-owned
             # telemetry can populate its system configuration.
             kernel_telemetry_options = {
-                # Intentionally defaults to False, diverging from the
-                # connector-wide True default on the Thrift/SEA path
-                # (client.py). The kernel path opts out of telemetry unless
-                # explicitly enabled; this is asserted by
-                # test_telemetry_enabled_defaults_false_for_kernel_client.
-                # Do not "fix" this back to True to match the other backends.
-                "enable_telemetry": kwargs.get("enable_telemetry", False),
+                # Preserve the caller's explicit telemetry choice. When unset,
+                # leave it as None so the kernel applies its own default.
+                "enable_telemetry": kwargs.get("enable_telemetry"),
                 # Match the connector's default batch size (client.py forwards
                 # the same TelemetryClientFactory.DEFAULT_BATCH_SIZE fallback)
                 # so an unset telemetry_batch_size resolves to the same value

tests/unit/test_session.py

@@ -619,7 +619,7 @@ def test_telemetry_kwargs_threaded_into_kernel_client(self):
             finally:
                 conn.close()
 
-    def test_telemetry_enabled_defaults_false_for_kernel_client(self):
+    def test_telemetry_enabled_defaults_none_for_kernel_client(self):
         import sys
         import types
 
@@ -651,7 +651,7 @@ def test_telemetry_enabled_defaults_false_for_kernel_client(self):
             try:
                 _, kwargs = mock_kernel_client.call_args
                 opts = kwargs["telemetry_options"]
-                assert opts["enable_telemetry"] is False
+                assert opts["enable_telemetry"] is None
             finally:
                 conn.close()
 

tests/unit/test_telemetry.py

@@ -332,10 +332,17 @@ def test_is_telemetry_enabled_returns_false_for_kernel(
             is expected_kernel_telemetry_enabled
         )
 
-    def test_kernel_telemetry_enabled_defaults_false(self):
+    def test_kernel_telemetry_enabled_omitted_when_unset(self):
         kernel_kwargs = self._kernel_telemetry_kwargs_for_test({})
 
-        assert kernel_kwargs["telemetry_enabled"] is False
+        assert "telemetry_enabled" not in kernel_kwargs
+
+    def test_kernel_telemetry_enabled_omitted_when_none(self):
+        kernel_kwargs = self._kernel_telemetry_kwargs_for_test(
+            {"enable_telemetry": None}
+        )
+
+        assert "telemetry_enabled" not in kernel_kwargs
 
 
 class TestTelemetryFactory: