diff --git a/.changes/client-telemetry b/.changes/client-telemetry
new file mode 100644
index 000000000..eaa7de8d0
--- /dev/null
+++ b/.changes/client-telemetry
@@ -0,0 +1 @@
+minor type="added" "Client telemetry through the shared Rust core: each Room reports its connect, reconnect, publish and subscribe spans, RTC statistics, SDK warnings and errors and device state to its LiveKit Cloud project when the token carries the observability grant; apps can add Room.emitTelemetryEvent(_:attributes:) and Room.setTelemetryAttribute(_:value:), and opt out with LiveKitSDK.disableTelemetry()"
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index c5e5b29c8..c5984b4e5 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -111,6 +111,16 @@ jobs:
             # Data track schema definitions are stored as participant data blobs.
             enable_participant_data_blob: true
 
+      # The telemetry end-to-end test posts OTLP to 127.0.0.1:4319 and reads back what the
+      # collector wrote (Tests/LiveKitCoreTests/Telemetry/otelcol.yaml). A plain binary, like the
+      # dev server above: macOS runners have no container runtime.
+      - name: Run OpenTelemetry Collector
+        run: |
+          arch="$(uname -m | sed 's/x86_64/amd64/')"
+          curl -sSfL -o otelcol.tar.gz "https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.160.0/otelcol-contrib_0.160.0_darwin_${arch}.tar.gz"
+          tar -xzf otelcol.tar.gz otelcol-contrib
+          nohup ./otelcol-contrib --config Tests/LiveKitCoreTests/Telemetry/otelcol.yaml > "$RUNNER_TEMP/otelcol.log" 2>&1 &
+
       - uses: maxim-lobanov/setup-xcode@v1
         with:
           xcode-version: ${{ matrix.xcode }}
diff --git a/Benchmarks/LiveKitBenchmark/BenchmarkSupport/TokenGenerator.swift b/Benchmarks/LiveKitBenchmark/BenchmarkSupport/TokenGenerator.swift
index 2833cd193..a45123847 100644
--- a/Benchmarks/LiveKitBenchmark/BenchmarkSupport/TokenGenerator.swift
+++ b/Benchmarks/LiveKitBenchmark/BenchmarkSupport/TokenGenerator.swift
@@ -56,6 +56,7 @@ struct TokenGenerator {
             ingressAdmin: false,
             hidden: false,
             recorder: false,
+            agent: false,
         )
 
         let options = TokenOptions(
diff --git a/Benchmarks/Package.swift b/Benchmarks/Package.swift
index a6333f431..df4f0f4b8 100644
--- a/Benchmarks/Package.swift
+++ b/Benchmarks/Package.swift
@@ -2,6 +2,13 @@
 
 import PackageDescription
 
+// Same switch as the SDK's manifests: `LIVEKIT_UNIFFI_PATH` points at a local livekit-uniffi build.
+let uniFFI: Package.Dependency = if let path = Context.environment["LIVEKIT_UNIFFI_PATH"] {
+    .package(name: "livekit-uniffi-xcframework", path: path)
+} else {
+    .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", from: "0.0.1")
+}
+
 let package = Package(
     name: "LiveKitBenchmark",
     platforms: [
@@ -9,7 +16,7 @@ let package = Package(
     ],
     dependencies: [
         .package(name: "client-sdk-swift", path: "../"),
-        .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", from: "0.0.1"),
+        uniFFI,
         // No Jemalloc trait: jemalloc's malloc-zone hooks crash on macOS 26 (malloc metrics read 0 without it)
         .package(url: "https://github.com/ordo-one/benchmark.git", from: "1.29.0", traits: []),
     ],
diff --git a/Package.swift b/Package.swift
index 731e5d3d7..b336e977f 100644
--- a/Package.swift
+++ b/Package.swift
@@ -3,6 +3,15 @@
 
 import PackageDescription
 
+// Development wiring for unreleased Rust bindings: set `LIVEKIT_UNIFFI_PATH` to a local
+// livekit-uniffi Swift package (`cargo make swift-package-debug` in rust-sdks/livekit-uniffi builds
+// it into `packages/swift/LiveKitUniFFI`). Unset, the released package is used.
+let uniFFI: Package.Dependency = if let path = Context.environment["LIVEKIT_UNIFFI_PATH"] {
+    .package(name: "livekit-uniffi-xcframework", path: path)
+} else {
+    .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9")
+}
+
 let package = Package(
     name: "LiveKit",
     platforms: [
@@ -20,7 +29,7 @@ let package = Package(
     dependencies: [
         // LK-Prefixed Dynamic WebRTC XCFramework
         .package(url: "https://github.com/livekit/webrtc-xcframework.git", exact: "150.7871.02"),
-        .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9"),
+        uniFFI,
         // Test-only: conformance oracle for the nanopb facades.
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.31.0"),
         // Only used for DocC generation
diff --git a/Package@swift-6.2.swift b/Package@swift-6.2.swift
index 14c75afd4..14156e4cc 100644
--- a/Package@swift-6.2.swift
+++ b/Package@swift-6.2.swift
@@ -3,6 +3,15 @@
 
 import PackageDescription
 
+// Development wiring for unreleased Rust bindings: set `LIVEKIT_UNIFFI_PATH` to a local
+// livekit-uniffi Swift package (`cargo make swift-package-debug` in rust-sdks/livekit-uniffi builds
+// it into `packages/swift/LiveKitUniFFI`). Unset, the released package is used.
+let uniFFI: Package.Dependency = if let path = Context.environment["LIVEKIT_UNIFFI_PATH"] {
+    .package(name: "livekit-uniffi-xcframework", path: path)
+} else {
+    .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9")
+}
+
 let package = Package(
     name: "LiveKit",
     platforms: [
@@ -21,7 +30,7 @@ let package = Package(
     dependencies: [
         // LK-Prefixed Dynamic WebRTC XCFramework
         .package(url: "https://github.com/livekit/webrtc-xcframework.git", exact: "150.7871.02"),
-        .package(url: "https://github.com/livekit/livekit-uniffi-xcframework.git", exact: "0.1.9"),
+        uniFFI,
         // Test-only: conformance oracle for the nanopb facades.
         .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.31.0"),
         // Only used for DocC generation
diff --git a/Sources/LiveKit/Core/Room+Engine.swift b/Sources/LiveKit/Core/Room+Engine.swift
index d8c5c1df4..5a99110eb 100644
--- a/Sources/LiveKit/Core/Room+Engine.swift
+++ b/Sources/LiveKit/Core/Room+Engine.swift
@@ -16,6 +16,7 @@
 
 // swiftlint:disable file_length
 
+internal import LiveKitUniFFI
 import Foundation
 
 #if canImport(Network)
@@ -355,7 +356,9 @@ extension Room {
     }
 
     // swiftlint:disable:next cyclomatic_complexity function_body_length
-    func startReconnect(reason: StartReconnectReason, nextReconnectMode: ReconnectMode? = nil) async throws {
+    func startReconnect(reason: StartReconnectReason, nextReconnectMode: ReconnectMode? = nil,
+                        telemetryReason: ReconnectReason? = nil) async throws
+    {
         log("[Connect] Starting, reason: \(reason)")
 
         guard case .connected = _state.connectionState else {
@@ -385,6 +388,9 @@ extension Room {
             $0.nextReconnectMode = nextReconnectMode
         }
 
+        // One reconnect cycle = one span; attempts are its checkpoints.
+        let reconnectSpan = telemetryScope?.start(name: .reconnect(reason: telemetryReason ?? reason.telemetry), parent: nil)
+
         // quick connect sequence, does not update connection state
         @Sendable func quickReconnectSequence() async throws {
             log("[Connect] Starting .quick reconnect sequence...")
@@ -475,83 +481,90 @@ extension Room {
             _state.mutate { $0.connectedUrl = finalUrl }
 
             dataTracks?.handleReconnect(fullReconnect: true)
+            rtcTelemetry?.joined(self) // tracks re-announced while reconnecting raised no didPublish
         }
 
-        do {
-            let reconnectTask = Task.retrying(totalAttempts: _state.connectOptions.reconnectAttempts,
-                                              retryDelay: { @Sendable attempt in
-                                                  let delay = TimeInterval.computeReconnectDelay(forAttempt: attempt,
-                                                                                                 baseDelay: self._state.connectOptions.reconnectAttemptDelay,
-                                                                                                 maxDelay: self._state.connectOptions.reconnectMaxDelay,
-                                                                                                 totalAttempts: self._state.connectOptions.reconnectAttempts,
-                                                                                                 addJitter: true)
-                                                  self.log("[Connect] Retry cycle waiting for \(String(format: "%.2f", delay)) seconds before attempt \(attempt + 1)")
-                                                  return delay
-                                              }) { currentAttempt, totalAttempts in
-                // Not reconnecting state anymore
-                guard let currentMode = self._state.isReconnectingWithMode else {
-                    self.log("[Connect] Not in reconnect state anymore, exiting retry cycle.")
-                    return
-                }
+        await TelemetrySpan.$current.withValue(reconnectSpan) {
+            do {
+                let reconnectTask = Task.retrying(totalAttempts: _state.connectOptions.reconnectAttempts,
+                                                  retryDelay: { @Sendable attempt in
+                                                      let delay = TimeInterval.computeReconnectDelay(forAttempt: attempt,
+                                                                                                     baseDelay: self._state.connectOptions.reconnectAttemptDelay,
+                                                                                                     maxDelay: self._state.connectOptions.reconnectMaxDelay,
+                                                                                                     totalAttempts: self._state.connectOptions.reconnectAttempts,
+                                                                                                     addJitter: true)
+                                                      self.log("[Connect] Retry cycle waiting for \(String(format: "%.2f", delay)) seconds before attempt \(attempt + 1)")
+                                                      return delay
+                                                  }) { currentAttempt, totalAttempts in
+                    // Not reconnecting state anymore
+                    guard let currentMode = self._state.isReconnectingWithMode else {
+                        self.log("[Connect] Not in reconnect state anymore, exiting retry cycle.")
+                        return
+                    }
 
-                // Full reconnect failed, give up
-                guard currentMode != .full else { return }
+                    // Full reconnect failed, give up
+                    guard currentMode != .full else { return }
 
-                self.log("[Connect] Starting retry attempt \(currentAttempt)/\(totalAttempts) with mode: \(currentMode)")
+                    self.log("[Connect] Starting retry attempt \(currentAttempt)/\(totalAttempts) with mode: \(currentMode)")
 
-                // Try full reconnect for the final attempt
-                if totalAttempts == currentAttempt, self._state.nextReconnectMode == nil {
-                    self._state.mutate { $0.nextReconnectMode = .full }
-                }
+                    // Try full reconnect for the final attempt
+                    if totalAttempts == currentAttempt, self._state.nextReconnectMode == nil {
+                        self._state.mutate { $0.nextReconnectMode = .full }
+                    }
 
-                let mode: ReconnectMode = self._state.mutate {
-                    let mode: ReconnectMode = ($0.nextReconnectMode == .full || $0.isReconnectingWithMode == .full) ? .full : .quick
-                    $0.isReconnectingWithMode = mode
-                    $0.nextReconnectMode = nil
-                    return mode
-                }
+                    let mode: ReconnectMode = self._state.mutate {
+                        let mode: ReconnectMode = ($0.nextReconnectMode == .full || $0.isReconnectingWithMode == .full) ? .full : .quick
+                        $0.isReconnectingWithMode = mode
+                        $0.nextReconnectMode = nil
+                        return mode
+                    }
 
-                do {
-                    if case .quick = mode {
-                        try await quickReconnectSequence()
-                        self.log("[Connect] Quick reconnect succeeded for attempt \(currentAttempt)")
-                    } else if case .full = mode {
-                        try await fullReconnectSequence()
-                        self.log("[Connect] Full reconnect succeeded for attempt \(currentAttempt)")
+                    reconnectSpan?.step(step: .attempt(number: UInt32(currentAttempt), full: mode == .full))
+
+                    do {
+                        if case .quick = mode {
+                            try await quickReconnectSequence()
+                            self.log("[Connect] Quick reconnect succeeded for attempt \(currentAttempt)")
+                        } else if case .full = mode {
+                            try await fullReconnectSequence()
+                            self.log("[Connect] Full reconnect succeeded for attempt \(currentAttempt)")
+                        }
+                    } catch {
+                        self.log("[Connect] Reconnect mode: \(mode) failed with error: \(error)", .error)
+                        // Re-throw
+                        throw error
                     }
-                } catch {
-                    self.log("[Connect] Reconnect mode: \(mode) failed with error: \(error)", .error)
-                    // Re-throw
-                    throw error
                 }
-            }
 
-            _state.mutate {
-                $0.reconnectTask = reconnectTask.cancellable()
-            }
+                _state.mutate {
+                    $0.reconnectTask = reconnectTask.cancellable()
+                }
 
-            try await reconnectTask.value
+                try await reconnectTask.value
 
-            // Re-connect sequence successful
-            log("[Connect] Sequence completed")
-            _state.mutate {
-                $0.connectionState = .connected
-                $0.reconnectTask = nil
-                $0.isReconnectingWithMode = nil
-                $0.nextReconnectMode = nil
-            }
+                // Re-connect sequence successful
+                log("[Connect] Sequence completed")
+                reconnectSpan?.end(outcome: .ok, error: nil)
+                _state.mutate {
+                    $0.connectionState = .connected
+                    $0.reconnectTask = nil
+                    $0.isReconnectingWithMode = nil
+                    $0.nextReconnectMode = nil
+                }
 
-            if let providedUrl = _state.providedUrl, providedUrl.isCloud, let regionManager = await regionManager(for: providedUrl) {
-                // Clear failed region attempts after a successful reconnect.
-                await regionManager.resetAttempts()
-            }
-        } catch {
-            log("[Connect] Sequence failed with error: \(error)")
+                if let providedUrl = _state.providedUrl, providedUrl.isCloud, let regionManager = await regionManager(for: providedUrl) {
+                    // Clear failed region attempts after a successful reconnect.
+                    await regionManager.resetAttempts()
+                }
+            } catch {
+                log("[Connect] Sequence failed with error: \(error)")
+                if Task.isCancelled { reconnectSpan?.cancel() } else { reconnectSpan?.end(with: error) }
 
-            // Only clean up if the reconnect task wasn't cancelled — when cancelled,
-            // the caller (disconnect() or a new reconnect) handles cleanup separately.
-            if !Task.isCancelled {
-                await cleanUp(withError: error)
+                // Only clean up if the reconnect task wasn't cancelled — when cancelled,
+                // the caller (disconnect() or a new reconnect) handles cleanup separately.
+                if !Task.isCancelled {
+                    await cleanUp(withError: error, telemetryReason: .reconnectFailed) // the cause is on the reconnect span
+                }
             }
         }
     }
diff --git a/Sources/LiveKit/Core/Room+SignalClientDelegate.swift b/Sources/LiveKit/Core/Room+SignalClientDelegate.swift
index 60865544c..997b30ff2 100644
--- a/Sources/LiveKit/Core/Room+SignalClientDelegate.swift
+++ b/Sources/LiveKit/Core/Room+SignalClientDelegate.swift
@@ -19,6 +19,7 @@
 #if !COCOAPODS && !LK_XCFRAMEWORK
 import LiveKitNanopb
 #endif
+internal import LiveKitUniFFI
 import Foundation
 
 internal import LiveKitWebRTC
@@ -64,7 +65,7 @@ extension Room: SignalClientDelegate {
             // Abort current attempt
             await signalClient.cleanUp(withError: error)
         case .disconnect:
-            await cleanUp(withError: error)
+            await cleanUp(withError: error, telemetryReason: telemetryDisconnectReason(proto: Int32(reason.rawValue)))
         default:
             log("Unknown leave action: \(action), ignoring", .warning)
         }
@@ -127,6 +128,8 @@ extension Room: SignalClientDelegate {
                     }
                 }
             }
+            updateTelemetryRoom() // a full reconnect keeps the Room's sid but not the participant's
+            rtcTelemetry?.joined(self)
         }
     }
 
@@ -146,6 +149,7 @@ extension Room: SignalClientDelegate {
 
         if response.hasParticipant {
             localParticipant.set(info: response.participant, connectionState: _state.connectionState)
+            updateTelemetryRoom()
         }
 
         _republishLocalTracks()
diff --git a/Sources/LiveKit/Core/Room+TransportDelegate.swift b/Sources/LiveKit/Core/Room+TransportDelegate.swift
index 06abca9cd..4d33d2ff0 100644
--- a/Sources/LiveKit/Core/Room+TransportDelegate.swift
+++ b/Sources/LiveKit/Core/Room+TransportDelegate.swift
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+internal import LiveKitUniFFI
 import Foundation
 
 internal import LiveKitWebRTC
@@ -59,7 +60,8 @@ extension Room: TransportDelegate {
             if transport.isPrimary || (_state.hasPublished && transport.target == .publisher), pcState.isDisconnected {
                 Task {
                     do {
-                        try await startReconnect(reason: .transport)
+                        try await startReconnect(reason: .transport,
+                                                 telemetryReason: transport.target == .publisher ? .publisherFailed : .subscriberFailed)
                     } catch {
                         log("Failed calling startReconnect, error: \(error)", .error)
                     }
diff --git a/Sources/LiveKit/Core/Room.swift b/Sources/LiveKit/Core/Room.swift
index d44946eeb..f9f763452 100644
--- a/Sources/LiveKit/Core/Room.swift
+++ b/Sources/LiveKit/Core/Room.swift
@@ -147,6 +147,48 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
 
     var dataTracks: DataTracks? { _state.stage.connection?.dataTracks }
 
+    // MARK: - Telemetry
+
+    /// This Room's session in the telemetry pipeline, one trace for the Room's lifetime; `nil`
+    /// after ``LiveKitSDK/disableTelemetry()``.
+    let telemetryScope: TelemetryScope?
+    private(set) var rtcTelemetry: RTCTelemetry?
+
+    /// Records an app event in this Room's telemetry, exported as `custom.<name>` next to the
+    /// SDK's own records, with this Room's correlation attributes.
+    ///
+    /// ```swift
+    /// room.emitTelemetryEvent("checkout.started", attributes: ["cart.items": "3"])
+    /// ```
+    ///
+    /// - Note: Names and keys up to 128 bytes, values up to 1024 bytes, at most 64 attributes and
+    ///   no `lk.` keys; anything else is dropped, never truncated.
+    public func emitTelemetryEvent(_ name: String, attributes: [String: String] = [:]) {
+        telemetryScope?.emitCustom(name: name, attributes: attributes)
+    }
+
+    /// Sets a correlation attribute on every telemetry record this Room captures from now on, to
+    /// match them with your own data (an order id, a tenant); `nil` removes it.
+    ///
+    /// ```swift
+    /// room.setTelemetryAttribute("app.order_id", value: order.id)
+    /// ```
+    ///
+    /// - Note: Same limits as ``emitTelemetryEvent(_:attributes:)``, at most 64 per Room.
+    public func setTelemetryAttribute(_ key: String, value: String?) {
+        telemetryScope?.setAttribute(key: key, value: value)
+    }
+
+    /// The Room and local participant on every record from now on: at join, full reconnect, move
+    /// and room update. Takes the Room's state explicitly so `onDidMutate` can call it.
+    func updateTelemetryRoom(_ state: State? = nil) {
+        let state = state ?? _state.copy()
+        telemetryScope?.setRoom(room: RoomIdentity(sid: state.sid?.stringValue,
+                                                   name: state.name,
+                                                   participantSid: localParticipant.sid?.stringValue,
+                                                   participantIdentity: localParticipant.identity?.stringValue))
+    }
+
     // MARK: - PreConnect
 
     lazy var preConnectBuffer = PreConnectAudioBuffer(room: self)
@@ -288,7 +330,9 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
         _state = StateSync(State(connectOptions: connectOptions ?? ConnectOptions(),
                                  roomOptions: roomOptions ?? RoomOptions()))
 
+        telemetryScope = Telemetry.scope()
         super.init()
+        rtcTelemetry = telemetryScope.map { RTCTelemetry(room: self, scope: $0) }
         // log sdk & os versions
         log("sdk: \(LiveKitSDK.version), ffi: \(LiveKitSDK.ffiVersion), os: \(String(describing: Utils.os()))(\(Utils.osVersionString())), modelId: \(String(describing: Utils.modelIdentifier() ?? "unknown"))")
 
@@ -314,6 +358,15 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
         _state.onDidMutate = { [weak self] newState, oldState in
             guard let self else { return }
 
+            if newState.sid != oldState.sid || newState.name != oldState.name {
+                updateTelemetryRoom(newState)
+            }
+
+            // Telemetry uploads with this Room's latest token: at connect and on every refresh.
+            if let url = newState.providedUrl, let token = newState.token, token != oldState.token || url != oldState.providedUrl {
+                telemetryScope?.setServer(url: url.absoluteString, token: token)
+            }
+
             // sid updated
             if let sid = newState.sid, sid != oldState.sid {
                 // Resolve sid
@@ -417,13 +470,19 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
         // options): carried across full reconnects, released on disconnect.
         let dependencies = ConnectionDependencies(room: self, roomOptions: state.roomOptions)
 
+        // One connect() = one attempt; reconnect cycles get their own spans.
+        let span = sharedTracing.beginSpan("connect")
+
         try _state.mutate {
             try $0.stage.begin(dependencies)
-            $0.connectSpan = sharedTracing.beginSpan("connect")
+            $0.connectSpan = span
             $0.providedUrl = providedUrl
             $0.token = token
             $0.connectionState = .connecting
         }
+        // Started once the attempt exists: the catch below is what ends it.
+        let attempt = telemetryScope?.start(name: .connect, parent: nil)
+        span.telemetry = attempt
 
         // E2EE wiring: delegate registration and cryptors, then the pairs' manager — installed
         // for every options branch: sending consults the manager's data-channel toggle, but
@@ -433,89 +492,96 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
         subscriberDataChannel.set(e2eeManager: e2eeManager)
         publisherDataChannel.set(e2eeManager: e2eeManager)
 
-        var nextUrl = providedUrl
-        var nextRegion: RegionInfo?
-        let regionManager = await regionManager(for: providedUrl)
+        // Everything below runs inside the connect span: child spans nest under it and
+        // warn/error records emitted meanwhile point at it.
+        try await TelemetrySpan.$current.withValue(attempt) {
+            var nextUrl = providedUrl
+            var nextRegion: RegionInfo?
+            let regionManager = await regionManager(for: providedUrl)
 
-        if providedUrl.isCloud {
-            if let regionManager {
-                await regionManager.resetAttempts(onlyIfExhausted: true)
-
-                if let preparedRegion {
-                    nextUrl = preparedRegion.url
-                    nextRegion = preparedRegion
-                } else if await regionManager.shouldRequestSettings() {
-                    await regionManager.prepareSettingsFetch(token: token)
+            if providedUrl.isCloud {
+                if let regionManager {
+                    await regionManager.resetAttempts(onlyIfExhausted: true)
+
+                    if let preparedRegion {
+                        nextUrl = preparedRegion.url
+                        nextRegion = preparedRegion
+                    } else if await regionManager.shouldRequestSettings() {
+                        await regionManager.prepareSettingsFetch(token: token)
+                    }
                 }
             }
-        }
 
-        // Concurrent mic publish mode
-        let enableMicrophone = _state.connectOptions.enableMicrophone
-        log("Concurrent enable microphone mode: \(enableMicrophone)")
+            // Concurrent mic publish mode
+            let enableMicrophone = _state.connectOptions.enableMicrophone
+            log("Concurrent enable microphone mode: \(enableMicrophone)")
 
-        let createMicrophoneTrackTask: Task<LocalTrack, any Error>? = if let recorder = preConnectBuffer.recorder, recorder.isRecording {
-            Task {
-                recorder.track
-            }
-        } else if enableMicrophone {
-            Task {
-                let localTrack = await LocalAudioTrack.createTrack(options: _state.roomOptions.defaultAudioCaptureOptions,
-                                                                   reportStatistics: _state.roomOptions.reportRemoteTrackStatistics)
-                // Initializes AudioDeviceModule's recording
-                try await localTrack.start()
-                return localTrack
+            let createMicrophoneTrackTask: Task<LocalTrack, any Error>? = if let recorder = preConnectBuffer.recorder, recorder.isRecording {
+                Task {
+                    recorder.track
+                }
+            } else if enableMicrophone {
+                Task {
+                    let localTrack = await LocalAudioTrack.createTrack(options: _state.roomOptions.defaultAudioCaptureOptions,
+                                                                       reportStatistics: _state.roomOptions.reportRemoteTrackStatistics)
+                    // Initializes AudioDeviceModule's recording
+                    try await localTrack.start()
+                    return localTrack
+                }
+            } else {
+                nil
             }
-        } else {
-            nil
-        }
 
-        do {
-            let finalUrl: URL
-            if providedUrl.isCloud {
-                guard let regionManager else {
-                    throw LiveKitError(.onlyForCloud)
+            do {
+                let finalUrl: URL
+                if providedUrl.isCloud {
+                    guard let regionManager else {
+                        throw LiveKitError(.onlyForCloud)
+                    }
+
+                    finalUrl = try await connectWithCloudRegionFailover(regionManager: regionManager,
+                                                                        initialUrl: nextUrl,
+                                                                        initialRegion: nextRegion,
+                                                                        token: token)
+                } else {
+                    try await fullConnectSequence(nextUrl, token)
+                    finalUrl = nextUrl
                 }
 
-                finalUrl = try await connectWithCloudRegionFailover(regionManager: regionManager,
-                                                                    initialUrl: nextUrl,
-                                                                    initialRegion: nextRegion,
-                                                                    token: token)
-            } else {
-                try await fullConnectSequence(nextUrl, token)
-                finalUrl = nextUrl
-            }
+                // Connect sequence successful
+                log("Connect sequence completed")
+                // Final check if cancelled, don't fire connected events
+                try Task.checkCancellation()
 
-            // Connect sequence successful
-            log("Connect sequence completed")
-            // Final check if cancelled, don't fire connected events
-            try Task.checkCancellation()
+                connectSpan?.record("room_connected")
 
-            connectSpan?.record("room_connected")
+                _state.mutate {
+                    $0.connectedUrl = finalUrl
+                    $0.connectionState = .connected
+                }
 
-            _state.mutate {
-                $0.connectedUrl = finalUrl
-                $0.connectionState = .connected
-            }
+                connectSpan?.end()
 
-            connectSpan?.end()
+                // Publish mic if mic task was created (its own span: `lk.connect` has ended)
+                if let createMicrophoneTrackTask, !createMicrophoneTrackTask.isCancelled {
+                    let track = try await createMicrophoneTrackTask.value
+                    try await TelemetrySpan.$current.withValue(nil) {
+                        try await localParticipant._publish(track: track, options: _state.roomOptions.defaultAudioPublishOptions.withPreconnect(preConnectBuffer.recorder?.isRecording ?? false))
+                    }
+                }
+            } catch {
+                log("Failed to resolve a region or connect: \(error)")
+                attempt?.end(with: error)
+                // Stop the track if it was created but not published
+                if let createMicrophoneTrackTask, !createMicrophoneTrackTask.isCancelled,
+                   case let .success(track) = await createMicrophoneTrackTask.result
+                {
+                    try? await track.stop()
+                }
 
-            // Publish mic if mic task was created
-            if let createMicrophoneTrackTask, !createMicrophoneTrackTask.isCancelled {
-                let track = try await createMicrophoneTrackTask.value
-                try await localParticipant._publish(track: track, options: _state.roomOptions.defaultAudioPublishOptions.withPreconnect(preConnectBuffer.recorder?.isRecording ?? false))
-            }
-        } catch {
-            log("Failed to resolve a region or connect: \(error)")
-            // Stop the track if it was created but not published
-            if let createMicrophoneTrackTask, !createMicrophoneTrackTask.isCancelled,
-               case let .success(track) = await createMicrophoneTrackTask.result
-            {
-                try? await track.stop()
+                await cleanUp(withError: error)
+                throw error // Re-throw the original error
             }
-
-            await cleanUp(withError: error)
-            throw error // Re-throw the original error
         }
 
         log("Connected to \(String(describing: self))", .info)
@@ -592,9 +658,14 @@ extension Room {
     // calls to create a new task context — silently breaking Task.isCancelled
     // propagation. This method is internal and never called from ObjC.
     @nonobjc func cleanUp(withError disconnectError: Error? = nil,
-                          isFullReconnect: Bool = false) async
+                          isFullReconnect: Bool = false,
+                          telemetryReason: DisconnectReason? = nil) async
     {
         log("withError: \(String(describing: disconnectError)), isFullReconnect: \(isFullReconnect)")
+        // connect() cleans up first too; only a session that existed has ended.
+        if !isFullReconnect, _state.connectionState != .disconnected {
+            telemetryScope?.disconnected(reason: telemetryReason ?? DisconnectReason(disconnectError))
+        }
 
         // Reap all in-flight RPCs with `recipientDisconnected` (1503). Runs before the
         // participant-state wipe so callers don't hang on the response timeout during
diff --git a/Sources/LiveKit/Core/Transport.swift b/Sources/LiveKit/Core/Transport.swift
index bbaf65f54..b7cc5c2c1 100644
--- a/Sources/LiveKit/Core/Transport.swift
+++ b/Sources/LiveKit/Core/Transport.swift
@@ -451,6 +451,24 @@ extension Transport {
             }
         }
     }
+
+    /// The whole peer connection's report for telemetry: every sender and receiver in one call.
+    /// `nil`, and no request, once the app has opted out: admitted here, on the RTC executor.
+    /// `nil` too when no answer comes within `timeout`; a late answer is dropped.
+    func telemetryStatistics(timeout: TimeInterval = 5) async -> LKRTCStatisticsReport? {
+        await Telemetry.beforeGate?(.request)
+        return await withCheckedContinuation { (continuation: CheckedContinuation<LKRTCStatisticsReport?, Never>) in
+            let pending = StateSync<CheckedContinuation<LKRTCStatisticsReport?, Never>?>(continuation)
+            let resume: @Sendable (LKRTCStatisticsReport?) -> Void = { report in
+                pending.mutate { let first = $0; $0 = nil; return first }?.resume(returning: report)
+            }
+            let started = Telemetry.ifCollecting {
+                _pc.statistics { @Sendable sd in resume(sd) }
+            }
+            guard started else { return resume(nil) }
+            DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + timeout) { resume(nil) }
+        }
+    }
 }
 
 // MARK: - RTCPeerConnectionDelegate
diff --git a/Sources/LiveKit/E2EE/DataTrackE2EE.swift b/Sources/LiveKit/E2EE/DataTrackE2EE.swift
index a11240061..2e27ab642 100644
--- a/Sources/LiveKit/E2EE/DataTrackE2EE.swift
+++ b/Sources/LiveKit/E2EE/DataTrackE2EE.swift
@@ -52,7 +52,7 @@ final class DataTrackCryptor: EncryptionProvider, DecryptionProvider, @unchecked
                 keyIndex: UInt8(truncatingIfNeeded: packet.keyIndex),
             )
         } catch {
-            throw EncryptionError.Failed(message: String(describing: error))
+            throw EncryptionError.Failed(reason: String(describing: error))
         }
     }
 
@@ -65,7 +65,7 @@ final class DataTrackCryptor: EncryptionProvider, DecryptionProvider, @unchecked
         do {
             return try requireManager().handle(encryptedData: packet, participantIdentity: senderIdentity)
         } catch {
-            throw DecryptionError.Failed(message: String(describing: error))
+            throw DecryptionError.Failed(reason: String(describing: error))
         }
     }
 }
diff --git a/Sources/LiveKit/LiveKit+DeviceHelpers.swift b/Sources/LiveKit/LiveKit+DeviceHelpers.swift
index 0f57e1958..7a17faafa 100644
--- a/Sources/LiveKit/LiveKit+DeviceHelpers.swift
+++ b/Sources/LiveKit/LiveKit+DeviceHelpers.swift
@@ -28,20 +28,25 @@ private let kIsAppExtension = Bundle.main.bundleURL.pathExtension == "appex"
 #if canImport(UIKit) && (os(iOS) || os(visionOS) || os(tvOS)) && !targetEnvironment(macCatalyst)
 // Resolves `UIApplication.shared` through the Objective-C runtime because referencing it directly
 // does not compile with APPLICATION_EXTENSION_API_ONLY=YES, and consumers build this module into
-// broadcast upload extensions. An extension process is prohibited from calling `sharedApplication`,
-// so callers must be guarded by `kIsAppExtension`. The selector itself would resolve there too.
+// broadcast upload extensions. An extension process is prohibited from calling `sharedApplication`
+// (the selector itself would resolve there too), so this returns `nil` in one.
 @MainActor
-private func isApplicationForegrounded() -> Bool {
+func applicationState() -> UIApplication.State? {
     let selector = NSSelectorFromString("sharedApplication")
-    guard UIApplication.responds(to: selector),
+    guard !kIsAppExtension, UIApplication.responds(to: selector),
           let shared = UIApplication.perform(selector),
           let application = shared.takeUnretainedValue() as? UIApplication
-    else { return false }
+    else { return nil }
+    return application.applicationState
+}
+
+@MainActor
+private func isApplicationForegrounded() -> Bool {
     // Only .active can actually present the alert. While .inactive (locked screen, call banner, app
     // switcher, or launch before the scene activates) the system defers it, and requestAccess has no
     // cancellation-aware continuation, so waiting there would suspend the caller for as long as the
     // app stays inactive. Failing fast instead lets the next attempt prompt normally.
-    return application.applicationState == .active
+    applicationState() == .active
 }
 #endif
 
diff --git a/Sources/LiveKit/LiveKit.swift b/Sources/LiveKit/LiveKit.swift
index c5d661276..42e9c8378 100644
--- a/Sources/LiveKit/LiveKit.swift
+++ b/Sources/LiveKit/LiveKit.swift
@@ -55,6 +55,16 @@ public class LiveKitSDK: NSObject, Loggable {
         state.mutate { $0.tracing = tracing }
     }
 
+    /// Opts this process out of client telemetry. Collection stops when this returns, in existing
+    /// Rooms too, and Rooms created afterwards collect nothing; everything not yet sent (queued,
+    /// open, or cached on disk by this launch or an earlier one) is deleted in the background.
+    /// Not remembered across launches: call it at every launch, before creating a Room.
+    ///
+    /// - Note: TODO: final shape pending the token/consent discussion.
+    public static func disableTelemetry() {
+        Telemetry.disable()
+    }
+
     /// Set a custom logger for the SDK
     /// - Note: This method must be called before any other logging is done
     /// e.g. in the `App.init()` or `AppDelegate/SceneDelegate`
diff --git a/Sources/LiveKit/Participant/LocalParticipant.swift b/Sources/LiveKit/Participant/LocalParticipant.swift
index cffc392b2..c9161c925 100644
--- a/Sources/LiveKit/Participant/LocalParticipant.swift
+++ b/Sources/LiveKit/Participant/LocalParticipant.swift
@@ -16,6 +16,7 @@
 
 // swiftlint:disable file_length
 
+internal import LiveKitUniFFI
 import Combine
 import Foundation
 
@@ -629,9 +630,32 @@ extension [Livekit_SubscribedQuality] {
 // MARK: - Private
 
 extension LocalParticipant {
+    /// One publish attempt = one `lk.publish` span, nested under whatever span the caller runs in
+    /// (the connect span for the pre-connect microphone).
     @discardableResult
-    // swiftlint:disable:next cyclomatic_complexity function_body_length
     func _publish(track: LocalTrack, options: TrackPublishOptions? = nil) async throws -> LocalTrackPublication {
+        let span = telemetryScope?.start(name: .publish, parent: TelemetrySpan.current.flatMap { $0.isEnded() ? nil : $0 })
+        if let spanTrack = SpanTrack(track) { span?.setTrack(track: spanTrack) }
+        do {
+            let publication = try await TelemetrySpan.$current.withValue(span) {
+                try await _publishTrack(track: track, options: options)
+            }
+            if let spanTrack = SpanTrack(publication, remoteIdentity: nil) { span?.setTrack(track: spanTrack) }
+            span?.end(outcome: .ok, error: nil)
+            // The core now waits for the track's first outbound reading: take its shorter interval.
+            if let room = _room { room.rtcTelemetry?.poll(room) }
+            return publication
+        } catch {
+            span?.end(with: error)
+            if let device = CaptureDevice(track.source), let reason = CaptureFailure(error) {
+                telemetryDeviceEvent(event: .captureFailed(device: device, reason: reason))
+            }
+            throw error
+        }
+    }
+
+    // swiftlint:disable:next cyclomatic_complexity function_body_length
+    private func _publishTrack(track: LocalTrack, options: TrackPublishOptions? = nil) async throws -> LocalTrackPublication {
         log("[publish] \(track) options: \(String(describing: options ?? nil))...", .info)
 
         try checkPermissions(toPublish: track)
diff --git a/Sources/LiveKit/Support/Logger.swift b/Sources/LiveKit/Support/Logger.swift
index 4f110b8c7..4fae7863e 100644
--- a/Sources/LiveKit/Support/Logger.swift
+++ b/Sources/LiveKit/Support/Logger.swift
@@ -218,14 +218,16 @@ extension Loggable {
                  level,
                  file: file,
                  function: function,
-                 line: line)
+                 line: line,
+                 scope: level >= .warning ? (self as? TelemetryScoped)?.telemetryScope : nil)
     }
 
     static func log(_ message: CustomStringConvertible? = nil,
                     _ level: LogLevel = .debug,
                     file: StaticString = #fileID,
                     function: StaticString = #function,
-                    line: UInt = #line)
+                    line: UInt = #line,
+                    scope: TelemetryScope? = nil)
     {
         sharedLogger.log(message ?? "",
                          level,
@@ -233,6 +235,13 @@ extension Loggable {
                          type: Self.self,
                          function: function,
                          line: line)
+        // The core's default floor, checked here so nothing below it crosses the FFI.
+        if level >= .warning {
+            Telemetry.log(LogRecord(severity: level.severity, source: .sdk, body: message?.description ?? "",
+                                    logger: String(describing: Self.self), function: "\(function)", file: "\(file)",
+                                    line: UInt32(clamping: line)),
+                          scope: scope)
+        }
     }
 }
 
diff --git a/Sources/LiveKit/Support/MulticastDelegate.swift b/Sources/LiveKit/Support/MulticastDelegate.swift
index 3c0979ff6..a29e1fd95 100644
--- a/Sources/LiveKit/Support/MulticastDelegate.swift
+++ b/Sources/LiveKit/Support/MulticastDelegate.swift
@@ -48,6 +48,8 @@ public class MulticastDelegate<T: Sendable>: NSObject, @unchecked Sendable, Logg
 
     private struct State {
         let delegates = NSHashTable<AnyObject>.weakObjects()
+        /// The SDK's own observers: notified like the others, never listed or removed with them.
+        let internals = NSHashTable<AnyObject>.weakObjects()
     }
 
     private let _queue: DispatchQueue
@@ -80,6 +82,12 @@ public class MulticastDelegate<T: Sendable>: NSObject, @unchecked Sendable, Logg
         _state.mutate { $0.delegates.remove(delegate) }
     }
 
+    /// Add an SDK-internal observer that ``removeAllDelegates()`` leaves in place.
+    func add(internalDelegate delegate: T) {
+        guard let delegate = delegate as AnyObject? else { return }
+        _state.mutate { $0.internals.add(delegate) }
+    }
+
     /// Remove all delegates.
     public func removeAllDelegates() {
         _state.mutate { $0.delegates.removeAllObjects() }
@@ -87,7 +95,7 @@ public class MulticastDelegate<T: Sendable>: NSObject, @unchecked Sendable, Logg
 
     /// Notify delegates inside the queue.
     func notify(label _: (() -> String)? = nil, _ fnc: @Sendable @escaping (T) -> Void) {
-        let delegates = _state.read { $0.delegates.allObjects.compactMap { $0 as? T } }
+        let delegates = _state.read { ($0.delegates.allObjects + $0.internals.allObjects).compactMap { $0 as? T } }
 
         _queue.async {
             for delegate in delegates {
@@ -99,7 +107,7 @@ public class MulticastDelegate<T: Sendable>: NSObject, @unchecked Sendable, Logg
     /// Awaitable version of notify
     func notifyAsync(_ fnc: @Sendable @escaping (T) -> Void) async {
         // Read a copy of delegates
-        let delegates = _state.read { $0.delegates.allObjects.compactMap { $0 as? T } }
+        let delegates = _state.read { ($0.delegates.allObjects + $0.internals.allObjects).compactMap { $0 as? T } }
 
         // Convert to async
         await withCheckedContinuation { continuation in
diff --git a/Sources/LiveKit/Support/Tracing.swift b/Sources/LiveKit/Support/Tracing.swift
index 3cd39dd94..07c3e5ad1 100644
--- a/Sources/LiveKit/Support/Tracing.swift
+++ b/Sources/LiveKit/Support/Tracing.swift
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+internal import LiveKitUniFFI
 import Foundation
 
 // MARK: - Span
@@ -35,6 +36,9 @@ public final class Span: @unchecked Sendable, Equatable, CustomStringConvertible
     private var _ended = false
     private let _entries = StateSync<[Entry]>([])
 
+    /// The telemetry core's span for the same operation: every entry is also its checkpoint.
+    var telemetry: TelemetrySpan?
+
     public init(label: String) {
         self.label = label
         start = ProcessInfo.processInfo.systemUptime
@@ -44,6 +48,7 @@ public final class Span: @unchecked Sendable, Equatable, CustomStringConvertible
     public func end() {
         guard !_ended else { return }
         _ended = true
+        telemetry?.end(outcome: .ok, error: nil)
         onEnd?(self)
         onEnd = nil
     }
@@ -51,6 +56,7 @@ public final class Span: @unchecked Sendable, Equatable, CustomStringConvertible
     /// Record a named event. Timestamp defaults to now if not provided.
     public func record(_ event: String, at time: TimeInterval = ProcessInfo.processInfo.systemUptime) {
         _entries.mutate { $0.append(Entry(label: event, time: time)) }
+        telemetry?.step(step: SpanStep(event))
     }
 
     @available(*, deprecated, renamed: "record(_:at:)")
@@ -96,6 +102,14 @@ public final class Span: @unchecked Sendable, Equatable, CustomStringConvertible
     }
 }
 
+extension Span {
+    /// End on an error: the telemetry span as cancelled or failed, then the tracer's handler.
+    func end(with error: Error) {
+        telemetry?.end(with: error)
+        end()
+    }
+}
+
 // MARK: - Stopwatch typealias
 
 @available(*, deprecated, renamed: "Span")
diff --git a/Sources/LiveKit/Telemetry/DeviceTelemetry.swift b/Sources/LiveKit/Telemetry/DeviceTelemetry.swift
new file mode 100644
index 000000000..74ac3d1e7
--- /dev/null
+++ b/Sources/LiveKit/Telemetry/DeviceTelemetry.swift
@@ -0,0 +1,305 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+internal import LiveKitUniFFI
+import AVFoundation
+import Foundation
+import Network
+#if canImport(UIKit)
+import UIKit
+#endif
+
+/// The device instrument: thermal state, low-power mode, memory pressure, network path, battery
+/// and app lifecycle as `DeviceState`, audio-session changes as events. Every OS callback yields
+/// into one stream that this actor drains in order; the core turns state into `lk.device.*`
+/// records and upload holds. Nothing polls.
+actor DeviceTelemetry: TelemetryInstrument {
+    /// One OS observation.
+    enum Change: Sendable {
+        /// Thermal state or low-power mode: read fresh from `ProcessInfo`.
+        case power
+        case app(AppState)
+        case memory(MemoryPressure)
+        case network(NetworkType, expensive: Bool, constrained: Bool)
+        case battery(level: UInt32?, charging: Bool)
+        case event(DeviceEvent)
+    }
+
+    private static let queue = DispatchQueue(label: "LiveKitSDK.telemetry.device", qos: .utility)
+    private static let executor = DispatchQueueExecutor(queue: queue)
+
+    nonisolated var unownedExecutor: UnownedSerialExecutor {
+        Self.executor.asUnownedSerialExecutor()
+    }
+
+    nonisolated let changes: AsyncStream<Change>.Continuation
+    private let stream: AsyncStream<Change>
+
+    private var state = DeviceState(thermal: .unknown, appState: .foreground, memory: .normal, network: .unknown)
+    private var loop: AnyTaskCancellable?
+    /// Set by `stop()` itself, so nothing registers once stopped, whichever hop runs first.
+    private nonisolated let stopped = StateSync(false)
+    private var observers: [NSObjectProtocol] = []
+    private var pathMonitor: NWPathMonitor?
+    private var memorySource: DispatchSourceMemoryPressure?
+
+    init() {
+        // Bounded: the actor drains faster than the OS notifies; a stall keeps the latest changes.
+        (stream, changes) = AsyncStream.makeStream(of: Change.self, bufferingPolicy: .bufferingNewest(64))
+    }
+
+    nonisolated func start() {
+        Task { await self.observe() }
+    }
+
+    /// Called by the core on the caller's thread under its lifecycle lock: ends the stream at once
+    /// (nothing more reaches the core) and unregisters on the actor later; never blocks.
+    nonisolated func stop() {
+        stopped.mutate { $0 = true }
+        changes.finish()
+        Task { await self.cancel() }
+    }
+}
+
+// MARK: - Lifecycle
+
+private extension DeviceTelemetry {
+    func observe() {
+        guard !stopped.copy() else { return }
+        loop = Task { [weak self, stream] in
+            for await change in stream {
+                await self?.apply(change)
+            }
+        }.cancellable()
+        observeProcessInfo()
+        observeMemory()
+        observeNetwork() // delivers the current path at once: the first state push
+        observeBattery()
+        observeAudioSession()
+        observeAppState()
+    }
+
+    func cancel() {
+        for observer in observers {
+            NotificationCenter.default.removeObserver(observer)
+        }
+        observers.removeAll()
+        pathMonitor?.cancel()
+        memorySource?.cancel()
+        loop = nil
+        changes.finish()
+        Task { @MainActor in AppStateListener.shared.delegates.remove(delegate: self) }
+    }
+
+    func apply(_ change: Change) {
+        switch change {
+        case .power: break
+        case let .app(appState): state.appState = appState
+        case let .memory(pressure): state.memory = pressure
+        case let .network(type, expensive, constrained):
+            (state.network, state.networkExpensive, state.networkConstrained) = (type, expensive, constrained)
+        case let .battery(level, charging): (state.batteryLevel, state.batteryCharging) = (level, charging)
+        case let .event(event):
+            telemetryDeviceEvent(event: event)
+            return
+        }
+        let info = ProcessInfo.processInfo
+        state.thermal = ThermalState(info.thermalState)
+        if #available(macOS 12.0, *) { state.lowPowerMode = info.isLowPowerModeEnabled }
+        telemetrySetDeviceState(state: state)
+    }
+
+    func observe(_ name: Notification.Name, _ change: @escaping @Sendable (Notification) -> Change?) {
+        let changes = changes
+        observers.append(NotificationCenter.default.addObserver(forName: name, object: nil, queue: nil) { note in
+            if let change = change(note) { changes.yield(change) }
+        })
+    }
+}
+
+// MARK: - Power & thermal
+
+private extension DeviceTelemetry {
+    func observeProcessInfo() {
+        observe(ProcessInfo.thermalStateDidChangeNotification) { _ in .power }
+        if #available(macOS 12.0, *) {
+            observe(.NSProcessInfoPowerStateDidChange) { _ in .power }
+        }
+    }
+}
+
+// MARK: - Memory
+
+private extension DeviceTelemetry {
+    /// `DISPATCH_MEMORYPRESSURE_*`: the signal jetsam acts on, including the return to normal.
+    func observeMemory() {
+        let source = DispatchSource.makeMemoryPressureSource(eventMask: [.normal, .warning, .critical], queue: Self.queue)
+        let changes = changes
+        source.setEventHandler { [weak source] in
+            guard let event = source?.data else { return }
+            changes.yield(.memory(event.contains(.critical) ? .critical : event.contains(.warning) ? .warning : .normal))
+        }
+        source.resume()
+        memorySource = source
+    }
+}
+
+// MARK: - Network
+
+private extension DeviceTelemetry {
+    /// Path type, expensive (cellular, hotspot) and constrained (Low Data Mode).
+    func observeNetwork() {
+        let monitor = NWPathMonitor()
+        let changes = changes
+        monitor.pathUpdateHandler = { path in
+            changes.yield(.network(NetworkType(path), expensive: path.isExpensive, constrained: path.isConstrained))
+        }
+        monitor.start(queue: Self.queue)
+        pathMonitor = monitor
+    }
+}
+
+// MARK: - Battery
+
+private extension DeviceTelemetry {
+    func observeBattery() {
+        #if os(iOS) || os(visionOS)
+        let changes = changes
+        let read: @Sendable () -> Void = {
+            Task { @MainActor in
+                let device = UIDevice.current
+                device.isBatteryMonitoringEnabled = true // ponytail: stays on for the process
+                let level = device.batteryLevel // -1 while unknown
+                changes.yield(.battery(level: level < 0 ? nil : UInt32((level * 100).rounded()),
+                                       charging: device.batteryState == .charging || device.batteryState == .full))
+            }
+        }
+        read()
+        observe(UIDevice.batteryLevelDidChangeNotification) { _ in read(); return nil }
+        observe(UIDevice.batteryStateDidChangeNotification) { _ in read(); return nil }
+        #endif
+    }
+}
+
+// MARK: - Audio session
+
+private extension DeviceTelemetry {
+    /// Route changes and interruptions are events, not state: they explain audio glitches.
+    func observeAudioSession() {
+        #if os(iOS) || os(tvOS) || os(visionOS)
+        observe(AVAudioSession.routeChangeNotification) { note in
+            let reason = (note.userInfo?[AVAudioSessionRouteChangeReasonKey] as? UInt)
+                .flatMap(AVAudioSession.RouteChangeReason.init(rawValue:)) ?? .unknown
+            let outputs = AVAudioSession.sharedInstance().currentRoute.outputs.map { AudioOutput($0.portType) }
+            return .event(.audioRouteChanged(outputs: outputs, reason: AudioRouteReason(reason)))
+        }
+        observe(AVAudioSession.interruptionNotification) { note in
+            let began = (note.userInfo?[AVAudioSessionInterruptionTypeKey] as? UInt) == AVAudioSession.InterruptionType.began.rawValue
+            return .event(.audioInterruption(began: began))
+        }
+        #endif
+    }
+}
+
+// MARK: - App state
+
+private extension DeviceTelemetry {
+    func observeAppState() {
+        let changes = changes
+        let stopped = stopped
+        Task { @MainActor in
+            // Checked here too: the remove, queued by a later stop(), must not run before this add.
+            guard !stopped.copy() else { return }
+            AppStateListener.shared.delegates.add(delegate: self)
+            #if canImport(UIKit) && (os(iOS) || os(visionOS) || os(tvOS)) && !targetEnvironment(macCatalyst)
+            if let state = applicationState() { changes.yield(.app(state == .background ? .background : .foreground)) }
+            #endif
+        }
+    }
+}
+
+extension DeviceTelemetry: AppStateDelegate {
+    nonisolated func appDidEnterBackground() { changes.yield(.app(.background)) }
+    nonisolated func appWillEnterForeground() { changes.yield(.app(.foreground)) }
+    nonisolated func appWillTerminate() { changes.yield(.app(.background)) }
+    nonisolated func appWillSleep() { changes.yield(.app(.background)) }
+    nonisolated func appDidWake() { changes.yield(.app(.foreground)) }
+}
+
+// MARK: - Vocabulary
+
+extension ThermalState {
+    init(_ state: ProcessInfo.ThermalState) {
+        self = switch state {
+        case .nominal: .nominal
+        case .fair: .fair
+        case .serious: .serious
+        case .critical: .critical
+        @unknown default: .unknown
+        }
+    }
+}
+
+extension NetworkType {
+    init(_ path: NWPath) {
+        self = if path.status != .satisfied {
+            .unavailable
+        } else if path.usesInterfaceType(.wifi) {
+            .wifi
+        } else if path.usesInterfaceType(.cellular) {
+            .cell
+        } else if path.usesInterfaceType(.wiredEthernet) {
+            .wired
+        } else if path.usesInterfaceType(.other) {
+            .other
+        } else {
+            .unknown
+        }
+    }
+}
+
+#if os(iOS) || os(tvOS) || os(visionOS)
+extension AudioRouteReason {
+    init(_ reason: AVAudioSession.RouteChangeReason) {
+        self = switch reason {
+        case .newDeviceAvailable: .newDevice
+        case .oldDeviceUnavailable: .oldDeviceUnavailable
+        case .categoryChange: .categoryChange
+        case .override: .override
+        case .wakeFromSleep: .wakeFromSleep
+        case .noSuitableRouteForCategory: .noSuitableRoute
+        case .routeConfigurationChange: .routeConfigurationChange
+        default: .unknown
+        }
+    }
+}
+
+extension AudioOutput {
+    init(_ port: AVAudioSession.Port) {
+        self = switch port {
+        case .builtInSpeaker: .speaker
+        case .builtInReceiver: .receiver
+        case .headphones, .lineOut: .wiredHeadset
+        case .bluetoothA2DP, .bluetoothHFP, .bluetoothLE: .bluetooth
+        case .carAudio: .carAudio
+        case .airPlay: .airPlay
+        case .HDMI: .hdmi
+        case .usbAudio: .usb
+        default: .other
+        }
+    }
+}
+#endif
diff --git a/Sources/LiveKit/Telemetry/RTCTelemetry.swift b/Sources/LiveKit/Telemetry/RTCTelemetry.swift
new file mode 100644
index 000000000..9dbe3e433
--- /dev/null
+++ b/Sources/LiveKit/Telemetry/RTCTelemetry.swift
@@ -0,0 +1,200 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+internal import LiveKitUniFFI
+internal import LiveKitWebRTC
+import Foundation
+
+/// A Room's RTC instrument. Reports the tracks' lifecycle, from which the core runs the
+/// `lk.subscribe` span (intent → first media), and hands it one raw `getStats()` report per peer
+/// connection as often as it asks; the core maps every RTP stream to its track and windows it.
+/// Owned by the Room, registered as its (weakly held) delegate.
+final class RTCTelemetry: NSObject, Sendable {
+    private struct Polling {
+        var loop: AnyTaskCancellable?
+        /// When the running loop reads next (`DispatchTime` uptime nanoseconds).
+        var next: UInt64 = .max
+    }
+
+    private let scope: TelemetryScope
+    private let polling = StateSync(Polling())
+
+    init(room: Room, scope: TelemetryScope) {
+        self.scope = scope
+        super.init()
+        room.delegates.add(internalDelegate: self) // out of reach of the app's removeAllDelegates()
+        Telemetry.rtcInstruments.mutate { $0.add(self) }
+        poll(room)
+    }
+
+    /// Whether a polling loop is running.
+    var isPolling: Bool { polling.read { $0.loop != nil } }
+
+    /// The opt-out: no reading from now on. Never blocks: cancelling only flags the loop.
+    func stop() {
+        polling.mutate { $0 = Polling(loop: nil, next: 0) }
+    }
+
+    /// At join and after a full reconnect (whose join announces no participants yet), under
+    /// autoSubscribe: the tracks announced and not yet subscribed are wanted from now on.
+    func joined(_ room: Room) {
+        guard room._state.connectOptions.autoSubscribe else { return }
+        for participant in room.remoteParticipants.values {
+            for case let publication as RemoteTrackPublication in participant.trackPublications.values where publication.track == nil {
+                guard let track = SpanTrack(publication, remoteIdentity: participant.identity?.stringValue) else { continue }
+                scope.subscribeStarted(track: track)
+            }
+        }
+        poll(room)
+    }
+
+    /// Intent to subscribe: the core polls faster until first media, so start a fresh wait.
+    func subscribeStarted(_ publication: RemoteTrackPublication, of participant: RemoteParticipant, in room: Room) {
+        guard let track = SpanTrack(publication, remoteIdentity: participant.identity?.stringValue) else { return }
+        scope.subscribeStarted(track: track)
+        poll(room)
+    }
+
+    /// Makes the next reading come no later than the core's current interval from now: a wait
+    /// it just shortened (a subscribe awaiting first media, a track awaiting its first outbound
+    /// reading) starts now, a sooner reading already due is kept. The loop lives as long as the
+    /// Room and this instrument; while disconnected it wakes at the idle interval (30 s by
+    /// default) and reads nothing, which costs less than tracking the connection state here.
+    func poll(_ room: Room) {
+        let wait = scope.statsPollIntervalMs() * 1_000_000
+        let next = DispatchTime.now().uptimeNanoseconds + wait
+        polling.mutate { polling in
+            guard next < polling.next, !Telemetry.disabled.copy() else { return }
+            polling.next = next
+            polling.loop = loop(room, firstWait: wait)
+        }
+    }
+
+    private func loop(_ room: Room, firstWait: UInt64) -> AnyTaskCancellable {
+        let scope = scope
+        return Task { [weak self, weak room] in
+            var wait = firstWait
+            while !Task.isCancelled {
+                try? await Task.sleep(nanoseconds: wait)
+                guard !Task.isCancelled, !Telemetry.disabled.copy(), let room else { return }
+                await room.recordPeerStats(into: scope)
+                wait = scope.statsPollIntervalMs() * 1_000_000
+                let next = DispatchTime.now().uptimeNanoseconds + wait
+                self?.polling.mutate { if !Task.isCancelled { $0.next = next } }
+            }
+        }.cancellable()
+    }
+}
+
+extension RTCTelemetry: RoomDelegate {
+    func room(_ room: Room, participant: RemoteParticipant, didPublishTrack publication: RemoteTrackPublication) {
+        // With autoSubscribe the intent exists the moment the track is known.
+        guard room._state.connectOptions.autoSubscribe else { return }
+        subscribeStarted(publication, of: participant, in: room)
+    }
+
+    func room(_ room: Room, participant: RemoteParticipant, didSubscribeTrack publication: RemoteTrackPublication) {
+        guard let track = SpanTrack(publication, remoteIdentity: participant.identity?.stringValue) else { return }
+        scope.subscribed(track: track)
+        poll(room) // the core may poll faster now (a subscribe it had not seen opens here)
+    }
+
+    func room(_: Room, participant _: RemoteParticipant, didFailToSubscribeTrackWithSid trackSid: Track.Sid, error: LiveKitError) {
+        scope.subscribeFailed(sid: trackSid.stringValue, errorType: TelemetrySpan.errorType(error))
+    }
+
+    func room(_: Room, participant _: RemoteParticipant, didUnsubscribeTrack publication: RemoteTrackPublication) {
+        scope.trackEnded(sid: publication.sid.stringValue)
+    }
+
+    func room(_: Room, participant _: RemoteParticipant, didUnpublishTrack publication: RemoteTrackPublication) {
+        scope.trackEnded(sid: publication.sid.stringValue)
+    }
+
+    func room(_: Room, participant _: LocalParticipant, didUnpublishTrack publication: LocalTrackPublication) {
+        scope.trackEnded(sid: publication.sid.stringValue)
+    }
+}
+
+// MARK: - Stats
+
+extension Room {
+    /// One `getStats()` report per peer connection, with every track this Room sends or receives.
+    /// Each request and each submission is admitted under the opt-out lock, so none starts once
+    /// ``LiveKitSDK/disableTelemetry()`` has returned; a report already requested is dropped, and
+    /// so is a connection that does not answer within 5 s. Returns the reports answered and submitted.
+    @discardableResult
+    func recordPeerStats(into scope: TelemetryScope) async -> (answered: Int, submitted: Int) {
+        guard let transports = _state.transport?.allTransports else { return (0, 0) }
+        var tracks: [String: String] = [:]
+        for participant in [localParticipant as Participant] + remoteParticipants.values.map(\.self) {
+            for publication in participant.trackPublications.values {
+                if let track = publication.track { tracks[track.mediaTrack.trackId] = publication.sid.stringValue }
+            }
+        }
+        var counts = (answered: 0, submitted: 0)
+        for transport in transports {
+            guard let report = await transport.telemetryStatistics() else { continue } // opted out, or no answer
+            counts.answered += 1
+            let stats = report.telemetryStats // flattened outside the lock
+            await Telemetry.beforeGate?(.submit)
+            guard Telemetry.ifCollecting({
+                scope.recordPeerStats(report: stats, tracks: tracks, timestampNs: report.telemetryTimestampNs)
+            }) else { break }
+            counts.submitted += 1
+        }
+        return counts
+    }
+}
+
+extension LKRTCStatisticsReport {
+    /// Every entry with its standard members as the core takes them, nested maps flattened with a
+    /// dot (`qualityLimitationDurations.cpu`). No member names are known here.
+    var telemetryStats: [RtcStat] {
+        statistics.values.map { stat in
+            var members: [String: AttributeValue] = [:]
+            Self.flatten(stat.values, prefix: "", into: &members)
+            return RtcStat(kind: stat.type, id: stat.id, members: members)
+        }
+    }
+
+    var telemetryTimestampNs: UInt64 {
+        UInt64(max(0, timestamp_us)) * 1000
+    }
+
+    private static func flatten(_ values: [String: NSObject], prefix: String, into members: inout [String: AttributeValue]) {
+        for (key, value) in values {
+            let name = prefix.isEmpty ? key : "\(prefix).\(key)"
+            switch value {
+            case let number as NSNumber:
+                // A boxed Bool is an NSNumber too, and must stay one.
+                if CFGetTypeID(number) == CFBooleanGetTypeID() {
+                    members[name] = .bool(number.boolValue)
+                } else if number.doubleValue == number.doubleValue.rounded(), abs(number.doubleValue) < 9e15 {
+                    members[name] = .int(number.int64Value)
+                } else {
+                    members[name] = .double(number.doubleValue)
+                }
+            case let string as NSString:
+                members[name] = .str(string as String)
+            case let nested as [String: NSObject]:
+                flatten(nested, prefix: name, into: &members)
+            default:
+                break // sequences carry nothing the core reads
+            }
+        }
+    }
+}
diff --git a/Sources/LiveKit/Telemetry/Telemetry.swift b/Sources/LiveKit/Telemetry/Telemetry.swift
new file mode 100644
index 000000000..0a6475660
--- /dev/null
+++ b/Sources/LiveKit/Telemetry/Telemetry.swift
@@ -0,0 +1,333 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+internal import LiveKitUniFFI
+internal import LiveKitWebRTC
+import Foundation
+
+/// Client telemetry. The pipeline — destination, token, batching, retries, cache, holds, stats
+/// mapping, span state — lives in the Rust core, one per process; Swift installs it, feeds it OS
+/// signals and moves its bytes. Every tuning value is the core's default.
+enum Telemetry {
+    /// A new Room's scope, installing the pipeline first if this is the first Room; none after the opt-out.
+    static func scope() -> TelemetryScope? {
+        _ = installed
+        return telemetryScope()
+    }
+
+    /// Opt-out, in effect when this returns: no scope, no capture, configures refused; the core
+    /// purges what it holds in the background. A cache left by an earlier launch belongs to no
+    /// pipeline yet: the refused configure deletes it.
+    static func disable() {
+        disabled.mutate { $0 = true } // released before calling the core: never held across it
+        telemetryDisable()
+        configure()
+        for rtc in rtcInstruments.read({ $0.allObjects }) {
+            rtc.stop()
+        }
+    }
+
+    /// Set by ``disable()``: the Rooms' RTC instruments, which the core does not own, check it too.
+    static let disabled = StateSync(false)
+
+    /// Runs `start` only while collecting, holding the opt-out lock across the check and `start`,
+    /// so nothing starts once ``disable()`` has returned. `start` must be synchronous and short
+    /// (it only initiates) and must not call back into this lock.
+    static func ifCollecting(_ start: () -> Void) -> Bool {
+        disabled.read { optedOut in
+            guard !optedOut else { return false }
+            start()
+            return true
+        }
+    }
+
+    /// The stats poll's steps that ``ifCollecting(_:)`` admits.
+    enum GatedStep { case request, submit }
+
+    /// Test seam: awaited right before each gated step, in tasks that bind it.
+    @TaskLocal static var beforeGate: (@Sendable (GatedStep) async -> Void)?
+
+    /// Every live Room's RTC instrument, so the opt-out stops their polling at once.
+    static let rtcInstruments = StateSync(NSHashTable<RTCTelemetry>.weakObjects())
+
+    /// Installed synchronously on first use, so a Room never misses its scope.
+    private static let installed: Void = configure()
+
+    /// Install (or replace) the process pipeline; refused by the core after an opt-out.
+    static func configure() {
+        let sdk = TelemetryResource(sdk: .swift,
+                                    sdkVersion: LiveKitSDK.version,
+                                    osName: String(describing: Utils.os()),
+                                    osVersion: Utils.osVersionString(),
+                                    deviceModel: Utils.modelIdentifier())
+        let config = TelemetryConfig(sdk: sdk, storageDir: storageDirectory?.path)
+        // Fail-open: the app runs without telemetry rather than not at all.
+        try? telemetryConfigure(config: config, transport: URLSessionTelemetryTransport(), instruments: [DeviceTelemetry(), WebRTCLogCapture()])
+    }
+
+    /// `Caches/livekit-telemetry/<app>`: purgeable and never backed up; per app, since the Caches
+    /// directory of an unsandboxed macOS process is shared by every app of that user.
+    static var storageDirectory: URL? {
+        FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask).first?
+            .appendingPathComponent("livekit-telemetry", isDirectory: true)
+            .appendingPathComponent(Bundle.main.bundleIdentifier ?? ProcessInfo.processInfo.processName, isDirectory: true)
+    }
+
+    /// An SDK warning or error, filed under the ambient span, else the emitter's Room, else the process.
+    static func log(_ record: LogRecord, scope: TelemetryScope?) {
+        var record = record
+        record.spanId = TelemetrySpan.current.flatMap { $0.isEnded() ? nil : $0.context()?.spanId }
+        if record.spanId == nil, let scope {
+            scope.log(record: record)
+        } else {
+            telemetryLog(record: record)
+        }
+    }
+}
+
+// MARK: - Ambient span
+
+extension TelemetrySpan {
+    /// The span the current task works inside: child spans nest under it and warn/error records
+    /// point at it. Bound around connect, a reconnect cycle and publish.
+    @TaskLocal static var current: TelemetrySpan?
+
+    /// End on a Swift error: `cancelled` for a cancellation, `error` otherwise.
+    func end(with error: Error) {
+        if error is CancellationError || (error as? LiveKitError)?.type == .cancelled {
+            cancel()
+        } else {
+            fail(error: Self.errorType(error))
+        }
+    }
+
+    /// `error.type`: `LiveKitError.<code>` (its `description` is prose, not the case), else the
+    /// Swift type name.
+    static func errorType(_ error: Error) -> String {
+        if let error = error as? LiveKitError { return "LiveKitError.\(error.type.rawValue)" }
+        return String(describing: type(of: error))
+    }
+}
+
+// MARK: - Log capture
+
+/// WebRTC's own errors: one sink of its own, next to (never instead of) the app's console logger.
+/// Unchecked: `LKRTCCallbackLogger` is not `Sendable`, but the core calls `start` once and `stop`
+/// once per pipeline, and nothing else touches the logger.
+final class WebRTCLogCapture: TelemetryInstrument, @unchecked Sendable {
+    private let logger = LKRTCCallbackLogger()
+    private let capturing = StateSync(true)
+
+    func start() {
+        logger.severity = .error
+        logger.start { [capturing] message in
+            guard capturing.copy() else { return }
+            telemetryLog(record: LogRecord(severity: .error, source: .webRtc,
+                                           body: message.trimmingCharacters(in: .whitespacesAndNewlines),
+                                           logger: "WebRTC"))
+        }
+    }
+
+    /// Called by the core on the caller's thread under its lifecycle lock: flag and return. The
+    /// sink goes later, off that thread, since removing it waits on WebRTC's log lock, which a
+    /// WebRTC thread may hold while its callback waits on the core.
+    func stop() {
+        capturing.mutate { $0 = false }
+        DispatchQueue.global(qos: .utility).async { self.logger.stop() }
+    }
+}
+
+// MARK: - Transport
+
+/// Moves the core's requests: status, headers and body go back untouched and the core decides
+/// what they mean. Only a missing answer is an error.
+final class URLSessionTelemetryTransport: TelemetryTransport {
+    /// Ephemeral (no cookies, no cache), and never follows a redirect to another origin: the
+    /// request carries the participant token.
+    static let defaultSession = URLSession(configuration: .ephemeral, delegate: SameOriginRedirects(), delegateQueue: nil)
+
+    private let session: URLSession
+
+    init(session: URLSession = URLSessionTelemetryTransport.defaultSession) {
+        self.session = session
+    }
+
+    func send(request: ExportRequest) async throws -> ExportResponse {
+        guard let url = URL(string: request.url) else {
+            throw ExportError.Rejected(reason: "invalid url \(request.url)")
+        }
+        var urlRequest = URLRequest(url: url)
+        urlRequest.httpMethod = "POST"
+        urlRequest.httpBody = request.body
+        for (name, value) in request.headers {
+            urlRequest.setValue(value, forHTTPHeaderField: name)
+        }
+        let data: Data, response: URLResponse
+        do {
+            (data, response) = try await session.data(for: urlRequest)
+        } catch {
+            throw ExportError.Retryable(reason: error.localizedDescription, retryAfterMs: nil)
+        }
+        guard let http = response as? HTTPURLResponse else {
+            throw ExportError.Retryable(reason: "not an HTTP response", retryAfterMs: nil)
+        }
+        var headers: [String: String] = [:]
+        for case let (name as String, value as String) in http.allHeaderFields {
+            headers[name] = value
+        }
+        return ExportResponse(status: UInt16(clamping: http.statusCode), headers: headers, body: data)
+    }
+}
+
+/// A redirect to another scheme, host or port is not followed: the 3xx comes back as the answer.
+final class SameOriginRedirects: NSObject, URLSessionTaskDelegate, Sendable {
+    func urlSession(_: URLSession, task: URLSessionTask, willPerformHTTPRedirection _: HTTPURLResponse,
+                    newRequest request: URLRequest, completionHandler: @escaping @Sendable (URLRequest?) -> Void)
+    {
+        let from = task.originalRequest?.url, to = request.url
+        let sameOrigin = from?.scheme == to?.scheme && from?.host == to?.host && from?.port == to?.port
+        completionHandler(sameOrigin ? request : nil)
+    }
+}
+
+// MARK: - Shared vocabulary
+
+extension LogLevel {
+    var severity: Severity {
+        switch self {
+        case .debug: .debug
+        case .info: .info
+        case .warning: .warn
+        case .error: .error
+        }
+    }
+}
+
+extension SpanStep {
+    /// A ``Span`` entry label as the core's checkpoint; labels it has no type for stay custom.
+    init(_ label: String) {
+        self = switch label {
+        case "ws_open": .wsOpen
+        case "signal": .signal
+        case "join_recv": .joinRecv
+        case "pc_created": .pcCreated
+        case "offer_sent": .offerSent
+        case "answer_sent": .answerSent
+        case "engine": .engine
+        case "pc_connected": .pcConnected
+        case "room_connected": .roomConnected
+        default: .custom(name: label)
+        }
+    }
+}
+
+extension StartReconnectReason {
+    var telemetry: ReconnectReason {
+        switch self {
+        case .websocket: .signalDisconnected
+        case .transport: .transportFailed
+        case .networkSwitch: .networkChanged
+        case .debug: .debug
+        }
+    }
+}
+
+extension SpanTrack {
+    init?(_ track: Track) {
+        guard let kind = TrackKind(track.kind) else { return nil }
+        self.init(sid: track.sid?.stringValue, kind: kind, source: TrackSource(track.source))
+    }
+
+    init?(_ publication: TrackPublication, remoteIdentity: String?) {
+        guard let kind = TrackKind(publication.kind) else { return nil }
+        self.init(sid: publication.sid.stringValue, kind: kind, source: TrackSource(publication.source), remoteIdentity: remoteIdentity)
+    }
+}
+
+extension TrackKind {
+    init?(_ kind: Track.Kind) {
+        switch kind {
+        case .audio: self = .audio
+        case .video: self = .video
+        default: return nil
+        }
+    }
+}
+
+extension TrackSource {
+    init(_ source: Track.Source) {
+        self = switch source {
+        case .camera: .camera
+        case .microphone: .microphone
+        case .screenShareVideo: .screenShare
+        case .screenShareAudio: .screenShareAudio
+        case .unknown: .unknown
+        }
+    }
+}
+
+extension CaptureFailure {
+    /// A publish error that says the capture device failed, if it does.
+    init?(_ error: Error) {
+        guard let error = error as? LiveKitError else { return nil }
+        switch error.type {
+        case .deviceAccessDenied: self = .permissionDenied
+        case .deviceNotFound: self = .notFound
+        case .captureFormatNotFound, .unableToResolveFPSRange, .capturerDimensionsNotResolved, .audioEngine: self = .other
+        default: return nil
+        }
+    }
+}
+
+extension CaptureDevice {
+    init?(_ source: Track.Source) {
+        switch source {
+        case .camera: self = .camera
+        case .microphone: self = .microphone
+        case .screenShareVideo: self = .screenShare
+        default: return nil
+        }
+    }
+}
+
+extension DisconnectReason {
+    /// The error a Room's clean-up was given; none is the app's own `disconnect()`.
+    init(_ error: Error?) {
+        guard let error else { self = .clientInitiated; return }
+        self = switch (error as? LiveKitError)?.type {
+        case .cancelled: .clientInitiated
+        case .duplicateIdentity: .duplicateIdentity
+        case .serverShutdown: .serverShutdown
+        case .participantRemoved: .participantRemoved
+        case .roomDeleted: .roomDeleted
+        case .stateMismatch: .stateMismatch
+        case .joinFailure: .joinFailure
+        case .timedOut, .serverPingTimedOut: .connectionTimeout
+        default: .unknown
+        }
+    }
+}
+
+/// Something that belongs to a Room: its warnings and errors land in that Room's trace.
+protocol TelemetryScoped {
+    var telemetryScope: TelemetryScope? { get }
+}
+
+extension Room: TelemetryScoped {}
+
+extension Participant: TelemetryScoped {
+    var telemetryScope: TelemetryScope? { _room?.telemetryScope }
+}
diff --git a/Sources/LiveKit/TrackPublications/RemoteTrackPublication.swift b/Sources/LiveKit/TrackPublications/RemoteTrackPublication.swift
index f44bc6dce..2562b4fba 100644
--- a/Sources/LiveKit/TrackPublications/RemoteTrackPublication.swift
+++ b/Sources/LiveKit/TrackPublications/RemoteTrackPublication.swift
@@ -70,6 +70,10 @@ public class RemoteTrackPublication: TrackPublication, @unchecked Sendable {
 
         _state.mutate { $0.isSubscribePreferred = newValue }
 
+        if newValue, let participant = participant as? RemoteParticipant {
+            room.rtcTelemetry?.subscribeStarted(self, of: participant, in: room)
+        }
+
         if !newValue {
             // Proactively clear the track. In single PC mode the transceiver is
             // reused (direction changes) rather than removed, so the WebRTC
diff --git a/Tests/LiveKitCoreTests/Telemetry/TelemetrySupport.swift b/Tests/LiveKitCoreTests/Telemetry/TelemetrySupport.swift
new file mode 100644
index 000000000..df521533a
--- /dev/null
+++ b/Tests/LiveKitCoreTests/Telemetry/TelemetrySupport.swift
@@ -0,0 +1,192 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#if canImport(AVFAudio) && !os(macOS)
+import AVFAudio
+#endif
+import CoreVideo
+import Foundation
+@testable import LiveKit
+import LiveKitUniFFI
+import Testing
+#if canImport(LiveKitTestSupport)
+import LiveKitTestSupport
+#endif
+#if canImport(UIKit)
+import UIKit
+#endif
+
+/// Holds the first task that reaches `step` until resumed: a race made deterministic.
+final class Pause: Sendable {
+    private let step: Telemetry.GatedStep
+    private let armed = StateSync(true)
+    private let arrival = AsyncStream<Void>.makeStream()
+    private let release = AsyncStream<Void>.makeStream()
+
+    init(at step: Telemetry.GatedStep) {
+        self.step = step
+    }
+
+    var hook: @Sendable (Telemetry.GatedStep) async -> Void {
+        { [self] reached in
+            guard reached == step, armed.mutate({ let first = $0; $0 = false; return first }) else { return }
+            arrival.continuation.yield()
+            for await _ in release.stream {
+                return
+            }
+        }
+    }
+
+    func reached() async {
+        for await _ in arrival.stream {
+            return
+        }
+    }
+
+    func resume() {
+        release.continuation.yield()
+    }
+}
+
+/// This suite's own stand-in collector: no state shared with other suites' URL protocol mocks.
+final class CollectorStub: URLProtocol, @unchecked Sendable {
+    struct Answer {
+        let status: Int
+        let headers: [String: String]
+        let body: Data
+    }
+
+    static let answer = StateSync<Result<Answer, URLError>>(.failure(URLError(.badServerResponse)))
+
+    override static func canInit(with request: URLRequest) -> Bool { request.url?.host == "collector.test" }
+    override static func canonicalRequest(for request: URLRequest) -> URLRequest { request }
+    override func stopLoading() {}
+
+    override func startLoading() {
+        switch Self.answer.copy() {
+        case let .success(answer):
+            let response = HTTPURLResponse(url: request.url!, statusCode: answer.status, httpVersion: "HTTP/1.1", headerFields: answer.headers)!
+            client?.urlProtocol(self, didReceive: response, cacheStoragePolicy: .notAllowed)
+            client?.urlProtocol(self, didLoad: answer.body)
+            client?.urlProtocolDidFinishLoading(self)
+        case let .failure(error):
+            client?.urlProtocol(self, didFailWithError: error)
+        }
+    }
+}
+
+/// A microphone the user never allowed: capture fails the way a denied one does.
+final class DeniedMicrophone: LocalAudioTrack, @unchecked Sendable {
+    convenience init() {
+        let track = RTC.createAudioTrack(source: RTC.createAudioSource(nil))
+        self.init(name: Track.microphoneName, source: .microphone, track: RTCMediaTrack(track), reportStatistics: false, captureOptions: AudioCaptureOptions())
+    }
+
+    override func startCapture() async throws {
+        throw LiveKitError(.deviceAccessDenied)
+    }
+}
+
+extension BufferCapturer {
+    /// 10 fps of the same 320×240 frame: enough for encoders, stats and simulcast layers to exist.
+    func feedSyntheticFrames() -> Task<Void, Never> {
+        Task {
+            var pixelBuffer: CVPixelBuffer?
+            guard CVPixelBufferCreate(kCFAllocatorDefault, 320, 240, kCVPixelFormatType_32BGRA, nil, &pixelBuffer) == kCVReturnSuccess,
+                  let pixelBuffer else { return }
+            while !Task.isCancelled {
+                capture(pixelBuffer)
+                try? await Task.sleep(nanoseconds: 100_000_000)
+            }
+        }
+    }
+}
+
+/// What the collector wrote: OTLP/JSON, one export request per line.
+struct OTLPFile {
+    struct Log {
+        let eventName: String
+        let body: String?
+        let traceId: String
+        let spanId: String
+        let severity: Int
+        let attributes: [String: String]
+    }
+
+    struct Span {
+        let name: String
+        let traceId: String
+        let start: UInt64
+        let end: UInt64
+        var seconds: Double { Double(end - start) / 1e9 }
+        let attributes: [String: String]
+        /// Span event names: the checkpoints (`ws_open`, `first_media`, `attempt 1 quick`, …).
+        let events: [String]
+        /// When each checkpoint was stamped.
+        let eventTimes: [String: UInt64]
+    }
+
+    private(set) var logs: [Log] = []
+    private(set) var spans: [Span] = []
+
+    init(url: URL, since: UInt64) throws {
+        let text = try String(contentsOf: url, encoding: .utf8)
+        for line in text.split(separator: "\n") {
+            guard let request = try? JSONSerialization.jsonObject(with: Data(line.utf8)) as? [String: Any] else { continue }
+            for scope in Self.children(request, "resourceLogs", "scopeLogs") {
+                for record in scope["logRecords"] as? [[String: Any]] ?? [] where Self.nanos(record["timeUnixNano"]) >= since {
+                    logs.append(Log(eventName: record["eventName"] as? String ?? "",
+                                    body: (record["body"] as? [String: Any])?["stringValue"] as? String,
+                                    traceId: record["traceId"] as? String ?? "",
+                                    spanId: record["spanId"] as? String ?? "",
+                                    severity: record["severityNumber"] as? Int ?? 0,
+                                    attributes: Self.attributes(record["attributes"])))
+                }
+            }
+            for scope in Self.children(request, "resourceSpans", "scopeSpans") {
+                for span in scope["spans"] as? [[String: Any]] ?? [] where Self.nanos(span["startTimeUnixNano"]) >= since {
+                    spans.append(Span(name: span["name"] as? String ?? "",
+                                      traceId: span["traceId"] as? String ?? "",
+                                      start: Self.nanos(span["startTimeUnixNano"]),
+                                      end: Self.nanos(span["endTimeUnixNano"]),
+                                      attributes: Self.attributes(span["attributes"]),
+                                      events: (span["events"] as? [[String: Any]] ?? []).compactMap { $0["name"] as? String },
+                                      eventTimes: Dictionary((span["events"] as? [[String: Any]] ?? []).compactMap { event in
+                                          (event["name"] as? String).map { ($0, Self.nanos(event["timeUnixNano"])) }
+                                      }, uniquingKeysWith: { first, _ in first })))
+                }
+            }
+        }
+    }
+
+    private static func children(_ request: [String: Any], _ resources: String, _ scopes: String) -> [[String: Any]] {
+        (request[resources] as? [[String: Any]] ?? []).flatMap { $0[scopes] as? [[String: Any]] ?? [] }
+    }
+
+    private static func nanos(_ value: Any?) -> UInt64 {
+        (value as? String).flatMap(UInt64.init) ?? (value as? UInt64) ?? 0
+    }
+
+    /// OTLP/JSON attributes (`[{key, value: {stringValue | intValue | boolValue | doubleValue}}]`) as strings.
+    private static func attributes(_ value: Any?) -> [String: String] {
+        var result: [String: String] = [:]
+        for pair in value as? [[String: Any]] ?? [] {
+            guard let key = pair["key"] as? String, let any = pair["value"] as? [String: Any] else { continue }
+            result[key] = any["stringValue"] as? String ?? (any["intValue"] ?? any["boolValue"] ?? any["doubleValue"]).map { "\($0)" }
+        }
+        return result
+    }
+}
diff --git a/Tests/LiveKitCoreTests/Telemetry/TelemetryTests.swift b/Tests/LiveKitCoreTests/Telemetry/TelemetryTests.swift
new file mode 100644
index 000000000..6e765d040
--- /dev/null
+++ b/Tests/LiveKitCoreTests/Telemetry/TelemetryTests.swift
@@ -0,0 +1,377 @@
+/*
+ * Copyright 2026 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#if canImport(AVFAudio) && !os(macOS)
+import AVFAudio
+#endif
+import CoreVideo
+import Foundation
+@testable import LiveKit
+import LiveKitUniFFI
+import Testing
+#if canImport(LiveKitTestSupport)
+import LiveKitTestSupport
+#endif
+#if canImport(UIKit)
+import UIKit
+#endif
+
+/// End to end through the Rust core into a local OpenTelemetry collector: `livekit-server --dev`
+/// and `otelcol-contrib --config Tests/LiveKitCoreTests/Telemetry/otelcol.yaml`, which writes every
+/// OTLP request as a JSON line (CI starts both; see ci.yaml). The pipeline is process-wide, hence
+/// one serialized story.
+@Suite(.serialized, .tags(.e2e))
+struct TelemetryTests {
+    static let fileCollector = "http://127.0.0.1:4319"
+    static let collectorOutput = URL(fileURLWithPath: "/tmp/livekit-telemetry-otlp.jsonl")
+
+    /// One call and its aftermath: connect → publish (one publish fails on a denied microphone) →
+    /// subscribe to first media → quick and full reconnect → app data → disconnect → opt-out.
+    @Test(.enabled(if: collectorAvailable, "no LK_TELEMETRY_ENDPOINT and no collector on 127.0.0.1:4319"))
+    func aCallFromConnectToOptOut() async throws {
+        // The core reads the override when the pipeline starts: restart it pointed at the collector,
+        // or at a backend of your own when `LK_TELEMETRY_ENDPOINT` is already set (see the PR).
+        setenv("LK_TELEMETRY_ENDPOINT", Self.fileCollector, 0)
+        Telemetry.configure()
+        let start = UInt64(Date().timeIntervalSince1970 * 1e9)
+        let marker = UUID().uuidString
+
+        let call = try await makeCall(marker: marker)
+        guard ProcessInfo.processInfo.environment["LK_TELEMETRY_ENDPOINT"] == Self.fileCollector else {
+            await telemetryFlush() // nothing to read back: browse the traces instead
+            print("telemetry e2e: publisher trace \(call.publisher), subscriber trace \(call.subscriber), " +
+                "late joiner trace \(call.lateJoiner), manual subscriber trace \(call.manualSubscriber)")
+            return
+        }
+        let otlp = try await Self.flushedRecords(since: start) { otlp in
+            call.rooms.allSatisfy { trace in
+                otlp.spans.contains { $0.traceId == trace && $0.name == "lk.connect" }
+                    && otlp.logs.contains { $0.traceId == trace && $0.eventName == "lk.room.disconnected" }
+            }
+        }
+        try expectSpans(otlp, of: call)
+        try expectRejoin(otlp, of: call)
+        try expectRecords(otlp, of: call, marker: marker)
+        try await expectOptOut(marker: marker, since: start)
+    }
+
+    /// The Swift transport moves bytes: the collector's answer comes back untouched, and a
+    /// credential-bearing request never follows a redirect to another origin.
+    @Test func transportPassesAnswersThroughAndStaysOnItsOrigin() async throws {
+        let configuration = URLSessionConfiguration.ephemeral
+        configuration.protocolClasses = [CollectorStub.self]
+        let session = URLSession(configuration: configuration, delegate: SameOriginRedirects(), delegateQueue: nil)
+        let transport = URLSessionTelemetryTransport(session: session)
+        let request = ExportRequest(url: "http://collector.test/v1/logs", headers: ["Authorization": "Bearer token"], body: Data([1, 2, 3]))
+
+        CollectorStub.answer.mutate { $0 = .success(.init(status: 429, headers: ["Retry-After": "7"], body: Data("quota".utf8))) }
+        let throttled = try await transport.send(request: request)
+        #expect(throttled.status == 429 && throttled.headers["Retry-After"] == "7" && throttled.body == Data("quota".utf8))
+
+        CollectorStub.answer.mutate { $0 = .failure(URLError(.notConnectedToInternet)) }
+        await #expect(throws: ExportError.self, "no answer is the transport's only error") {
+            try await transport.send(request: request)
+        }
+
+        let task = try session.dataTask(with: URLRequest(url: #require(URL(string: "https://project.livekit.cloud/observability/client/logs/otlp/v0"))))
+        for (target, follows) in [("https://project.livekit.cloud/other", true), ("https://elsewhere.test/v0", false),
+                                  ("http://project.livekit.cloud/v0", false), ("https://project.livekit.cloud:8443/v0", false)]
+        {
+            let redirect = try URLRequest(url: #require(URL(string: target)))
+            let followed = await withCheckedContinuation { continuation in
+                SameOriginRedirects().urlSession(session, task: task, willPerformHTTPRedirection: HTTPURLResponse(),
+                                                 newRequest: redirect) { continuation.resume(returning: $0 != nil) }
+            }
+            #expect(followed == follows, "\(target)")
+        }
+    }
+}
+
+// MARK: - The story
+
+extension TelemetryTests {
+    /// The two Rooms' trace ids.
+    struct Call {
+        var publisher = "", subscriber = "", lateJoiner = "", manualSubscriber = ""
+        var rooms: [String] { [publisher, subscriber, lateJoiner, manualSubscriber] }
+    }
+
+    func makeCall(marker: String) async throws -> Call {
+        var call = Call()
+        try await TestEnvironment.withRooms([
+            RoomTestingOptions(canPublish: true),
+            RoomTestingOptions(canSubscribe: true),
+        ]) { rooms in
+            let publisher = rooms[0], subscriber = rooms[1]
+            call.publisher = try #require(publisher.telemetryScope?.traceId())
+            call.subscriber = try #require(subscriber.telemetryScope?.traceId())
+            // The app clearing its delegates leaves telemetry's own observer in place (subscribe asserts below).
+            subscriber.removeAllDelegates()
+            #expect(call.publisher.count == 32 && call.publisher != call.subscriber, "one trace per Room")
+
+            // The public tracing API is untouched: the connect span still times the steps.
+            let connect = try #require(publisher.connectSpan)
+            #expect(Set(connect.entries.map(\.label)).isSuperset(of: ["ws_open", "signal", "join_recv", "pc_created", "room_connected"]))
+
+            // App data: a correlation attribute on everything from now on (one set, one removed).
+            publisher.setTelemetryAttribute("app.call_id", value: marker)
+            subscriber.setTelemetryAttribute("app.call_id", value: marker)
+            publisher.setTelemetryAttribute("app.removed", value: marker)
+            publisher.setTelemetryAttribute("app.removed", value: nil)
+
+            // Publish synthetic media: no capture device or permission in a headless run.
+            let video = await LocalVideoTrack.createBufferTrack(name: "telemetry")
+            let frames = try #require(video.capturer as? BufferCapturer).feedSyntheticFrames()
+            defer { frames.cancel() }
+            try await publisher.localParticipant.publish(videoTrack: video)
+            let audio = try await publisher.localParticipant.publish(audioTrack: TestAudioTrack())
+            // …and a microphone the user never allowed: a failed publish and a capture failure.
+            await #expect(throws: LiveKitError.self) {
+                try await publisher.localParticipant.publish(audioTrack: DeniedMicrophone())
+            }
+
+            let identity = try #require(publisher.localParticipant.identity)
+            let remote = try #require(subscriber.remoteParticipants[identity])
+            try await poll(timeout: 15, interval: 0.2, for: "subscriber has audio and video") {
+                remote.audioTracks.first?.track != nil && remote.videoTracks.first?.track != nil
+            }
+            try await Task.sleep(nanoseconds: 3_000_000_000) // first media, at the core's 1 s polls
+
+            // Late joiners: one finds the tracks already there (autoSubscribe), one subscribes by hand.
+            (call.lateJoiner, call.manualSubscriber) = try await joinLate(publisher, publisherIdentity: identity)
+
+            // The test microphone never sends a packet: unpublishing it ends its subscribe before first media.
+            try await publisher.localParticipant.unpublish(publication: audio)
+
+            publisher.log("\(marker) error", .error) // no ambient span: the Room's session
+            publisher.emitTelemetryEvent("e2e.checkpoint", attributes: ["e2e.marker": marker])
+            postDeviceChanges()
+
+            try await publisher.debug_simulate(scenario: .quickReconnect)
+            try await poll(timeout: 30, interval: 0.2, for: "quick reconnect settled") { publisher.connectionState == .connected }
+            try await Task.sleep(nanoseconds: 1_000_000_000)
+            try await publisher.debug_simulate(scenario: .fullReconnect)
+            try await poll(timeout: 30, interval: 0.2, for: "full reconnect settled") { publisher.connectionState == .connected }
+            // The subscriber rejoins too: the server announces the video again at that join.
+            try await subscriber.debug_simulate(scenario: .fullReconnect)
+            try await poll(timeout: 30, interval: 0.2, for: "subscriber rejoined") { subscriber.connectionState == .connected }
+            try await Task.sleep(nanoseconds: 2_000_000_000)
+        }
+        return call
+    }
+
+    /// Two more subscribers join after the publish: one auto-subscribes to the tracks already
+    /// there, one subscribes to the video by hand. Returns their trace ids.
+    func joinLate(_ publisher: Room, publisherIdentity identity: Participant.Identity) async throws -> (String, String) {
+        let late = try await Self.join(publisher, as: "late-joiner", autoSubscribe: true)
+        let manual = try await Self.join(publisher, as: "manual-subscriber", autoSubscribe: false)
+        do {
+            let traces = try (#require(late.telemetryScope?.traceId()), #require(manual.telemetryScope?.traceId()))
+            try await poll(timeout: 10, interval: 0.2, for: "the manual subscriber sees the video") {
+                manual.remoteParticipants[identity]?.videoTracks.first != nil
+            }
+            let video = try #require(manual.remoteParticipants[identity]?.videoTracks.first as? RemoteTrackPublication)
+            try await video.set(subscribed: true)
+            try await Task.sleep(nanoseconds: 4_000_000_000) // first media, at the core's 1 s polls
+            await late.disconnect()
+            await manual.disconnect()
+            return traces
+        } catch {
+            await late.disconnect()
+            await manual.disconnect()
+            throw error
+        }
+    }
+
+    func expectSpans(_ otlp: OTLPFile, of call: Call) throws {
+        let publisherSpans = otlp.spans.filter { $0.traceId == call.publisher }
+        let subscriberSpans = otlp.spans.filter { $0.traceId == call.subscriber }
+
+        // Connect: one span per Room with the required checkpoints; reconnects are their own spans.
+        for spans in [publisherSpans, subscriberSpans] {
+            let connects = spans.filter { $0.name == "lk.connect" }
+            #expect(connects.count == 1, "one lk.connect per Room: \(connects.count)")
+            let connect = try #require(connects.first)
+            #expect(Set(connect.events).isSuperset(of: ["ws_open", "signal", "join_recv", "pc_created"]), "\(connect.events)")
+            #expect(connect.attributes["lk.outcome"] == "ok" && connect.attributes["lk.connect.attempt"] == "1")
+        }
+        let reconnects = publisherSpans.filter { $0.name == "lk.reconnect" }
+        #expect(reconnects.count == 2 && reconnects.allSatisfy { $0.attributes["lk.outcome"] == "ok" }, "\(reconnects.map(\.attributes))")
+        #expect(reconnects.allSatisfy { $0.attributes["lk.reconnect.reason"] == "debug" && $0.events.contains { $0.hasPrefix("attempt 1 ") } },
+                "\(reconnects.map(\.events))")
+
+        // Publish: two tracks, one denied microphone; subscribe: intent → first media.
+        let publishes = publisherSpans.filter { $0.name == "lk.publish" }
+        #expect(publishes.filter { $0.attributes["lk.outcome"] == "ok" && $0.attributes["lk.track.sid"] != nil }.count >= 2, "\(publishes.map(\.attributes))")
+        #expect(publishes.contains { $0.attributes["lk.outcome"] == "error" && $0.attributes["error.type"] == "LiveKitError.\(LiveKitErrorType.deviceAccessDenied.rawValue)" })
+        #expect(otlp.logs.contains { $0.eventName == "lk.device.capture.failed" && $0.attributes["lk.device.capture.device"] == "microphone"
+                && $0.attributes["lk.device.capture.reason"] == "permission_denied"
+        })
+        // The synthetic video reaches first media; the silent microphone, unpublished first, is cancelled.
+        let subscribes = subscriberSpans.filter { $0.name == "lk.subscribe" }
+        #expect(subscribes.contains { $0.attributes["lk.track.kind"] == "video" && $0.attributes["lk.outcome"] == "ok" && $0.events.contains("first_media") },
+                "\(subscribes.map(\.events))")
+        #expect(subscribes.contains { $0.attributes["lk.track.kind"] == "audio" && $0.attributes["lk.outcome"] == "cancelled" && !$0.events.contains("first_media") },
+                "a track ended before first media: \(subscribes.map { ($0.attributes["lk.outcome"], $0.events) })")
+
+        // Joining after the publish, or subscribing by hand: first media within the core's 1 s polls.
+        for trace in [call.lateJoiner, call.manualSubscriber] {
+            let video = otlp.spans.filter { $0.traceId == trace && $0.name == "lk.subscribe" && $0.attributes["lk.track.kind"] == "video" }
+            #expect(video.contains { $0.attributes["lk.outcome"] == "ok" && $0.events.contains("first_media") && $0.seconds < 10 },
+                    "\(video.map { ($0.events, $0.seconds) })")
+        }
+        // The late joiner wanted the tracks already in the Room from the moment it connected.
+        let lateConnect = try #require(otlp.spans.first { $0.traceId == call.lateJoiner && $0.name == "lk.connect" })
+        #expect(otlp.spans.contains { $0.traceId == call.lateJoiner && $0.name == "lk.subscribe" && $0.start <= lateConnect.end },
+                "the subscribe intent starts at join")
+    }
+
+    /// After a full reconnect the intent starts at the rejoin, not at the later subscription.
+    func expectRejoin(_ otlp: OTLPFile, of call: Call) throws {
+        let rejoin = try #require(otlp.spans.first { $0.traceId == call.subscriber && $0.name == "lk.reconnect" })
+        let resubscribed = otlp.spans.filter { $0.traceId == call.subscriber && $0.name == "lk.subscribe" && $0.start >= rejoin.start }
+        #expect(resubscribed.contains { span in span.eventTimes["subscribed"].map { $0 > span.start + 1_000_000 } ?? false },
+                "\(resubscribed.map { ($0.start, $0.eventTimes) })")
+    }
+
+    func expectRecords(_ otlp: OTLPFile, of call: Call, marker: String) throws {
+        let mine = otlp.logs.filter { [call.publisher, call.subscriber].contains($0.traceId) }
+
+        // RTC windows from one report per peer connection, each track in its direction.
+        // (No inbound audio: the test microphone never sends a packet.)
+        let windows = mine.filter { $0.eventName == "lk.rtc.stats.sample" }
+        for (trace, kind, direction) in [(call.publisher, "audio", "outbound"), (call.publisher, "video", "outbound"), (call.subscriber, "video", "inbound")] {
+            #expect(windows.contains { $0.traceId == trace && $0.attributes["lk.track.kind"] == kind && $0.attributes["lk.track.direction"] == direction },
+                    "\(direction) \(kind) window")
+        }
+
+        // App data: the event, the attribute on the Room's records after it was set, never the removed one.
+        #expect(mine.contains { $0.eventName == "custom.e2e.checkpoint" && $0.attributes["e2e.marker"] == marker && $0.attributes["app.call_id"] == marker })
+        #expect(mine.contains { $0.body == "\(marker) error" && $0.traceId == call.publisher && $0.spanId.isEmpty && $0.severity >= 17 },
+                "an SDK error lands in its Room's trace")
+        #expect(windows.allSatisfy { $0.attributes["app.call_id"] == marker }, "windows carry the correlation attribute")
+        #expect(!otlp.logs.contains { $0.attributes["app.removed"] != nil })
+
+        // The session ends once per Room, never on a reconnect.
+        let ended = mine.filter { $0.eventName == "lk.room.disconnected" }
+        #expect(ended.count == 2 && ended.allSatisfy { $0.attributes["lk.disconnect.reason"] == "client_initiated" }, "\(ended.map(\.attributes))")
+
+        // Device: the state's initial values, and what this platform can post.
+        var device = ["lk.device.thermal.changed", "lk.device.memory.changed", "lk.device.network.changed", "lk.device.low_power.changed"]
+        #if os(iOS) || os(tvOS) || os(visionOS)
+        device += ["lk.device.audio.interruption", "lk.device.audio_route.changed", "lk.device.app_state.changed"]
+        #endif
+        for event in device {
+            #expect(otlp.logs.contains { $0.eventName == event }, "\(event)")
+        }
+        #expect(otlp.logs.filter(\.eventName.isEmpty).allSatisfy { $0.severity >= 13 }, "log records are warnings and errors only")
+        let stats = try #require(telemetryStats())
+        #expect(stats.dropped == 0 && stats.cachedBatches == 0, "the whole call shipped: \(telemetryDiagnostics())")
+    }
+
+    func expectOptOut(marker: String, since start: UInt64) async throws {
+        // Opt-out: what was not yet sent is deleted, nothing is collected afterwards.
+        let pending = "\(marker) pending"
+        let room = Room()
+        room.emitTelemetryEvent(pending)
+        // A connected Room, kept alive, publishing: its stats polling stops when the call returns.
+        try await TestEnvironment.withRooms([RoomTestingOptions(canPublish: true)]) { rooms in
+            let connected = rooms[0]
+            let video = await LocalVideoTrack.createBufferTrack(name: "opt-out")
+            let frames = try #require(video.capturer as? BufferCapturer).feedSyntheticFrames()
+            defer { frames.cancel() }
+            try await connected.localParticipant.publish(videoTrack: video)
+            let rtc = try #require(connected.rtcTelemetry)
+            let scope = try #require(connected.telemetryScope)
+            #expect(rtc.isPolling)
+            // Two polls caught at the boundaries: one about to request getStats(), one about to submit.
+            let beforeRequest = Pause(at: .request), beforeSubmit = Pause(at: .submit)
+            let requesting = Task { await Telemetry.$beforeGate.withValue(beforeRequest.hook) { await connected.recordPeerStats(into: scope) } }
+            let submitting = Task { await Telemetry.$beforeGate.withValue(beforeSubmit.hook) { await connected.recordPeerStats(into: scope) } }
+            await beforeRequest.reached()
+            await beforeSubmit.reached()
+            LiveKitSDK.disableTelemetry()
+            beforeRequest.resume()
+            beforeSubmit.resume()
+            let caughtRequesting = await requesting.value, caughtSubmitting = await submitting.value
+            #expect(caughtRequesting.answered == 0 && caughtRequesting.submitted == 0, "no getStats() starts after the opt-out")
+            #expect(caughtSubmitting.answered == 1 && caughtSubmitting.submitted == 0, "no report is submitted after the opt-out")
+            #expect(!rtc.isPolling, "the poll loop is gone")
+            rtc.poll(connected)
+            #expect(!rtc.isPolling, "and nothing restarts it")
+        }
+        #expect(Room().telemetryScope == nil, "a Room created after the opt-out collects nothing")
+        try await poll(timeout: 5, interval: 0.1, for: "the pipeline is gone") { telemetryScope() == nil }
+        try await Task.sleep(nanoseconds: 2_000_000_000) // anything left would have arrived by now
+        let after = try await Self.flushedRecords(since: start)
+        #expect(!after.logs.contains { $0.eventName == "custom.\(pending)" }, "an unsent event is deleted, never uploaded")
+        let cached = try FileManager.default.contentsOfDirectory(atPath: #require(Telemetry.storageDirectory).path)
+        #expect(cached.isEmpty, "the on-disk cache is purged: \(cached)")
+    }
+}
+
+// MARK: - Helpers
+
+extension TelemetryTests {
+    /// A collector to talk to: a backend named by `LK_TELEMETRY_ENDPOINT`, or the local one.
+    static var collectorAvailable: Bool {
+        if ProcessInfo.processInfo.environment["LK_TELEMETRY_ENDPOINT"] != nil { return true }
+        let socket = Darwin.socket(AF_INET, SOCK_STREAM, 0)
+        guard socket >= 0 else { return false }
+        defer { close(socket) }
+        var address = sockaddr_in()
+        address.sin_family = sa_family_t(AF_INET)
+        address.sin_port = in_port_t(4319).bigEndian
+        address.sin_addr.s_addr = inet_addr("127.0.0.1")
+        return withUnsafePointer(to: &address) {
+            $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { connect(socket, $0, socklen_t(MemoryLayout<sockaddr_in>.size)) == 0 }
+        }
+    }
+
+    /// A subscriber that joins `publisher`'s Room once its tracks are there.
+    static func join(_ publisher: Room, as identity: String, autoSubscribe: Bool) async throws -> Room {
+        let room = Room(connectOptions: ConnectOptions(autoSubscribe: autoSubscribe))
+        let token = try TestEnvironment.liveKitServerToken(for: #require(publisher.name), identity: identity, canPublish: false,
+                                                           canPublishData: false, canPublishSources: [], canSubscribe: true)
+        try await room.connect(url: TestEnvironment.liveKitServerUrl(), token: token)
+        return room
+    }
+
+    /// Ships what the core holds and reads back what the collector wrote since `start`, waiting
+    /// (up to 15 s) until `complete` holds: the collector writes each request as it arrives.
+    static func flushedRecords(since start: UInt64, until complete: (OTLPFile) -> Bool = { _ in true }) async throws -> OTLPFile {
+        await telemetryFlush()
+        var otlp = try OTLPFile(url: collectorOutput, since: start)
+        for _ in 0 ..< 30 where !complete(otlp) {
+            try await Task.sleep(nanoseconds: 500_000_000)
+            otlp = try OTLPFile(url: collectorOutput, since: start)
+        }
+        return otlp
+    }
+
+    /// Posts the OS notifications a simulator or host never sends on its own.
+    func postDeviceChanges() {
+        #if os(iOS) || os(tvOS) || os(visionOS)
+        let center = NotificationCenter.default
+        center.post(name: AVAudioSession.routeChangeNotification, object: nil,
+                    userInfo: [AVAudioSessionRouteChangeReasonKey: AVAudioSession.RouteChangeReason.override.rawValue])
+        center.post(name: AVAudioSession.interruptionNotification, object: nil,
+                    userInfo: [AVAudioSessionInterruptionTypeKey: AVAudioSession.InterruptionType.began.rawValue])
+        center.post(name: UIApplication.didEnterBackgroundNotification, object: nil)
+        center.post(name: UIApplication.willEnterForegroundNotification, object: nil)
+        #endif
+    }
+}
diff --git a/Tests/LiveKitCoreTests/Telemetry/otelcol.yaml b/Tests/LiveKitCoreTests/Telemetry/otelcol.yaml
new file mode 100644
index 000000000..3ea5f51e9
--- /dev/null
+++ b/Tests/LiveKitCoreTests/Telemetry/otelcol.yaml
@@ -0,0 +1,22 @@
+# Collector for the telemetry end-to-end test: accepts OTLP/HTTP on :4319 and writes every batch
+# as one JSON line the test reads back. Run: otelcol-contrib --config Tests/LiveKitCoreTests/Telemetry/otelcol.yaml
+# (CI downloads the binary in ci.yaml: macOS runners have no container runtime.)
+receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: 127.0.0.1:4319
+exporters:
+  file:
+    path: /tmp/livekit-telemetry-otlp.jsonl
+service:
+  telemetry:
+    metrics:
+      level: none
+  pipelines:
+    logs:
+      receivers: [otlp]
+      exporters: [file]
+    traces:
+      receivers: [otlp]
+      exporters: [file]
diff --git a/Tests/LiveKitTestSupport/ObjCRoomHelper.swift b/Tests/LiveKitTestSupport/ObjCRoomHelper.swift
index 7b12be289..6ba615b33 100644
--- a/Tests/LiveKitTestSupport/ObjCRoomHelper.swift
+++ b/Tests/LiveKitTestSupport/ObjCRoomHelper.swift
@@ -92,6 +92,7 @@ public class LKObjCRoomHelper: NSObject {
             ingressAdmin: false,
             hidden: false,
             recorder: false,
+            agent: false,
         )
 
         return try tokenGenerator.sign()
diff --git a/Tests/LiveKitTestSupport/Room.swift b/Tests/LiveKitTestSupport/Room.swift
index b8d2f84aa..7cc66eb3c 100644
--- a/Tests/LiveKitTestSupport/Room.swift
+++ b/Tests/LiveKitTestSupport/Room.swift
@@ -105,7 +105,7 @@ public final class RoomWatcher<T: Decodable & Sendable>: RoomDelegate, Sendable
 
     // MARK: - Delegates
 
-    public func room(_: Room, participant _: RemoteParticipant?, didReceiveData data: Data, forTopic topic: String, encryptionType _: EncryptionType) {
+    public func room(_: Room, participant _: RemoteParticipant?, didReceiveData data: Data, forTopic topic: String, encryptionType _: LiveKit.EncryptionType) {
         // print("didReceiveData: \(data) for topic: \(topic)")
         Task {
             do {
diff --git a/Tests/LiveKitTestSupport/TestEnvironment.swift b/Tests/LiveKitTestSupport/TestEnvironment.swift
index 86dbaa62c..f4e212c9a 100644
--- a/Tests/LiveKitTestSupport/TestEnvironment.swift
+++ b/Tests/LiveKitTestSupport/TestEnvironment.swift
@@ -62,6 +62,7 @@ public enum TestEnvironment {
             ingressAdmin: false,
             hidden: false,
             recorder: false,
+            agent: false,
         )
 
         return try tokenGenerator.sign()