feat: dropped-frame / achieved-fps telemetry (fixes #23) (#34)

Instrument the four previously-silent frame-shed sites — PTS-deadline pacing,
bufferingNewest(1) backpressure, compositor pool exhaustion, and
VideoFrameAdmission — plus the encoded-frame rate, so the M3 HUD and the M1/M9
adaptive logic finally have real measured observability instead of only the
encoder's targets.

- FrameTelemetry (StreamCore): thread-safe cumulative counters
  (OSAllocatedUnfairLock, mirroring VideoFrameAdmission) and a pure, unit-tested
  achieved-fps / congestion-drops-per-second rate function.
- LiveStats gains achievedFrameRate + droppedFrames. Congestion drops =
  backpressure + admission (the frames the pipeline wanted to encode but couldn't);
  intentional pacing and overlay-only compositor drops are tracked but excluded.
- The capture output records captured/pacing/backpressure/compositor on the sample
  queue; both publishers record admission + encoded and hold the shared telemetry
  so the adaptive controller (M9) can consume the congestion signal.
- The controller folds measured fps + cumulative drops into liveStats each ~1s
  poll; the Settings HUD shows achieved fps and a "N frames dropped" notice that
  appears only under real congestion.

Tests (StreamCore, CI-testable): FrameTelemetry counters + rate math and the new
LiveStats formatting/equality.

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

Author: joeblau

Stream/ScreenCaptureController.swift

@@ -42,6 +42,15 @@ final class ScreenCaptureController: NSObject {
     /// ~1s telemetry poll, live only. Cancelled on every teardown path (they all
     /// funnel through `stopCapture`).
     @ObservationIgnored private var statsTask: Task<Void, Never>?
+    /// Shared frame-drop / achieved-fps counters (issue #23 / M8), created per
+    /// broadcast and handed to BOTH the capture output and the publisher so the four
+    /// shed sites and the encoded rate land in one place. Read once per stats poll to
+    /// fold measured fps + congestion drops into `liveStats`.
+    @ObservationIgnored private var frameTelemetry: FrameTelemetry?
+    /// The previous telemetry snapshot + its capture time, so each poll derives
+    /// achieved fps from the delta over the real elapsed window.
+    @ObservationIgnored private var lastTelemetrySnapshot: FrameTelemetrySnapshot?
+    @ObservationIgnored private var lastTelemetryAt: UInt64 = 0
 
     override init() {
         super.init()
@@ -121,24 +130,56 @@ final class ScreenCaptureController: NSObject {
                 // Write through INCLUDING nil: the publishers return nil during a
                 // reconnect, which must clear the card to its "—" placeholders rather
                 // than freezing on stale last-good telemetry while the stream is down.
-                self?.updateLiveStats(snapshot)
+                self?.applyPolledStats(snapshot)
                 try? await Task.sleep(for: .seconds(1))
             }
         }
     }
 
+    /// One stats tick: fold the measured frame telemetry (achieved fps + cumulative
+    /// congestion drops) into the publisher's target-based snapshot, then store it.
+    /// Advances the achieved-fps delta baseline as a side effect, so it must run
+    /// exactly once per poll.
+    private func applyPolledStats(_ base: LiveStats?) {
+        updateLiveStats(enrichWithTelemetry(base))
+    }
+
+    /// Merges the shared `FrameTelemetry` into the publisher's snapshot: the measured
+    /// achieved fps (from the encoded-frame delta over the real elapsed window) and
+    /// the cumulative congestion drops. Returns `base` untouched when telemetry isn't
+    /// available (pre-live / torn down). Mutates the delta baseline.
+    private func enrichWithTelemetry(_ base: LiveStats?) -> LiveStats? {
+        guard let telemetry = frameTelemetry else { return base }
+        let current = telemetry.snapshot()
+        let now = DispatchTime.now().uptimeNanoseconds
+        var achieved = 0
+        if let previous = lastTelemetrySnapshot, lastTelemetryAt > 0, now > lastTelemetryAt {
+            let elapsed = Double(now &- lastTelemetryAt) / 1_000_000_000
+            achieved = FrameTelemetry.rate(from: previous, to: current, elapsed: elapsed).achievedFrameRate
+        }
+        // Advance the baseline every tick — even when `base` is nil (reconnecting) —
+        // so the next window measures against a fresh, ~1s-old sample.
+        lastTelemetrySnapshot = current
+        lastTelemetryAt = now
+        guard var stats = base else { return nil }
+        stats.achievedFrameRate = achieved
+        stats.droppedFrames = current.congestionDrops
+        return stats
+    }
+
     /// Assigns the latest telemetry, skipping a redundant write (and the 1 Hz leaf
     /// re-render it would otherwise trigger) when the snapshot is unchanged.
     private func updateLiveStats(_ stats: LiveStats?) {
         if liveStats != stats { liveStats = stats }
     }
 
-    private func makePublisher(for transport: StreamCore.StreamProtocol) -> any Publisher {
+    private func makePublisher(for transport: StreamCore.StreamProtocol,
+                               telemetry: FrameTelemetry) -> any Publisher {
         switch transport {
         case .rtmp, .rtmps:
-            RTMPPublisher()
+            RTMPPublisher(telemetry: telemetry)
         case .srt, .whip:
-            SessionPublisher(protocol: transport)
+            SessionPublisher(protocol: transport, telemetry: telemetry)
         }
     }
 
@@ -152,8 +193,15 @@ final class ScreenCaptureController: NSObject {
             return
         }
 
-        let publisher = makePublisher(for: settings.selectedProtocol)
-        let output = ScreenCaptureOutput(publisher: publisher, settings: settings) { [weak self] error in
+        // One telemetry instance per broadcast, shared by the capture output (the
+        // four shed sites) and the publisher (admission + encoded rate).
+        let telemetry = FrameTelemetry()
+        frameTelemetry = telemetry
+        lastTelemetrySnapshot = nil
+        lastTelemetryAt = 0
+        let publisher = makePublisher(for: settings.selectedProtocol, telemetry: telemetry)
+        let output = ScreenCaptureOutput(publisher: publisher, settings: settings,
+                                         telemetry: telemetry) { [weak self] error in
             Task { @MainActor [weak self] in await self?.captureDidStop(error: error) }
         }
 
@@ -228,6 +276,9 @@ final class ScreenCaptureController: NSObject {
         heartbeatTask = nil
         statsTask?.cancel()
         statsTask = nil
+        frameTelemetry = nil
+        lastTelemetrySnapshot = nil
+        lastTelemetryAt = 0
         isLive = false
         thermalNotice = nil
         liveStats = nil
@@ -408,6 +459,10 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
     /// the capture pacing so a 1080p60 pick never asks a device for more than it
     /// can sustain. The thermal governor tightens this further at runtime.
     private let capability = StreamCapability.current
+    /// Shared frame telemetry (issue #23 / M8): this output records the capture,
+    /// pacing (PTS-deadline), backpressure (`bufferingNewest(1)`), and compositor
+    /// (pool-exhaustion) sites on the serial sample queue.
+    private let telemetry: FrameTelemetry
     private let onStopped: @Sendable (Error) -> Void
     private let micLevelMeter: ScreenCaptureMicrophoneMeter
     private let micLevelChannel = MicrophoneLevelChannel()
@@ -427,9 +482,11 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
 
     init(publisher: any Publisher,
          settings: StreamSettings,
+         telemetry: FrameTelemetry,
          onStopped: @escaping @Sendable (Error) -> Void) {
         self.publisher = publisher
         self.settings = settings
+        self.telemetry = telemetry
         self.onStopped = onStopped
         micLevelMeter = ScreenCaptureMicrophoneMeter(gain: settings.micVolume)
         targetFrameInterval = CMTime(value: 1,
@@ -454,22 +511,28 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
                     await publisher.setOutputSize(target, nativeShortEdge: min(width, height))
                     targetSize = target
                 }
-                if settings.pipEnabled, let targetSize,
-                   let camera = self.facecam.latest.take(),
-                   let composited = self.compositor.composite(
-                       screen: image,
-                       camera: camera,
-                       targetSize: targetSize,
-                       orientation: .up,
-                       corner: settings.pipCorner,
-                       scale: settings.pipScale,
-                       cameraPosition: settings.cameraPosition
-                   ),
-                   let output = self.compositor.makeSampleBuffer(
-                       from: composited,
-                       timingSource: sampleBuffer
-                   ) {
-                    await publisher.appendVideo(output)
+                if settings.pipEnabled, let targetSize, let camera = self.facecam.latest.take() {
+                    if let composited = self.compositor.composite(
+                           screen: image,
+                           camera: camera,
+                           targetSize: targetSize,
+                           orientation: .up,
+                           corner: settings.pipCorner,
+                           scale: settings.pipScale,
+                           cameraPosition: settings.cameraPosition
+                       ),
+                       let output = self.compositor.makeSampleBuffer(
+                           from: composited,
+                           timingSource: sampleBuffer
+                       ) {
+                        await publisher.appendVideo(output)
+                    } else {
+                        // Pool exhausted (a slow encoder holding surfaces) or the wrap
+                        // failed: skip the overlay for this frame and send the raw
+                        // screen. The frame itself is NOT lost — the overlay is.
+                        self.telemetry.recordDrop(.compositor)
+                        await publisher.appendVideo(sampleBuffer)
+                    }
                 } else {
                     await publisher.appendVideo(sampleBuffer)
                 }
@@ -491,12 +554,16 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
         guard sampleBuffer.isValid, sampleBuffer.dataReadiness == .ready else { return }
         switch type {
         case .screen:
+            telemetry.recordCaptured()
             // Downsample the native-refresh feed to the target frame rate by PTS
             // deadline: emit the first frame at/after each deadline, then advance
             // by one interval; re-anchor after a stall so a gap doesn't burst.
             let pts = sampleBuffer.presentationTimeStamp
             if pts.isValid {
                 if nextVideoDeadline.isValid, CMTimeCompare(pts, nextVideoDeadline) < 0 {
+                    // Intentional pacing shed (e.g. 120 Hz → 30 fps), not a fault —
+                    // counted separately from the congestion drops the HUD surfaces.
+                    telemetry.recordDrop(.pacing)
                     return   // arrived before the next target-fps slot — drop it
                 }
                 let advanced = CMTimeAdd(nextVideoDeadline.isValid ? nextVideoDeadline : pts,
@@ -505,7 +572,12 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
                     ? advanced
                     : CMTimeAdd(pts, targetFrameInterval)
             }
-            videoContinuation.yield(sampleBuffer)
+            // `bufferingNewest(1)`: if the consumer (compositor/append) hasn't drained
+            // the previous frame, this yield evicts it — a backpressure shed the
+            // pipeline couldn't keep up with. `.dropped` carries that evicted frame.
+            if case .dropped = videoContinuation.yield(sampleBuffer) {
+                telemetry.recordDrop(.backpressure)
+            }
         case .audio:
             if settings.includeAppAudio { publisher.enqueueApp(sampleBuffer) }
         case .microphone:

Stream/SettingsView.swift

@@ -1022,7 +1022,8 @@ private final class MicrophoneLevelMonitor {
 /// A leaf view so its ~1s telemetry ticks invalidate only this subtree (not the
 /// whole launcher List), and every value is monospaced + single-line so the card's
 /// height is invariant per tick — the content-sized drawer detent never springs on
-/// a number change (only an occasional thermal-notice appearance resizes it).
+/// a number change (only an occasional thermal- or dropped-frames notice appearing
+/// resizes it).
 private struct LiveStatsCard: View {
     var capture: ScreenCaptureController
 
@@ -1037,6 +1038,7 @@ private struct LiveStatsCard: View {
                     .lineLimit(1)
                     .minimumScaleFactor(0.8)
             }
+            droppedNotice
         }
         .padding(14)
         .frame(maxWidth: .infinity, alignment: .leading)
@@ -1062,18 +1064,32 @@ private struct LiveStatsCard: View {
         }
     }
 
-    /// Bitrate (ABR target) · effective fps · uplink health. Values are the
-    /// encoder's applied targets, not measured throughput (measured fps needs M8).
+    /// Bitrate (ABR target) · achieved fps · uplink health. Bitrate is the encoder's
+    /// applied target; FPS is the MEASURED achieved rate (M8), falling back to the
+    /// target for the first second before a rate is available.
     private var metricsRow: some View {
         HStack(spacing: 0) {
             metric("Bitrate", value: capture.liveStats?.bitRateLabel ?? "—")
             divider
-            metric("FPS", value: capture.liveStats.map { "\($0.frameRate)" } ?? "—")
+            metric("FPS", value: capture.liveStats.map { "\($0.displayFrameRate)" } ?? "—")
             divider
             metric("Uplink", value: capture.liveStats?.linkHealth.label ?? "—", tint: uplinkTint)
         }
     }
 
+    /// A subtle count of the frames congestion has shed this session (backpressure +
+    /// admission). Hidden until the first drop, so a healthy stream shows nothing;
+    /// the coloured Uplink metric already carries the live severity signal.
+    @ViewBuilder private var droppedNotice: some View {
+        if let stats = capture.liveStats, stats.droppedFrames > 0 {
+            Label("\(stats.droppedLabel) frames dropped", systemImage: "square.stack.3d.up.slash")
+                .font(.caption2)
+                .foregroundStyle(.secondary)
+                .lineLimit(1)
+                .minimumScaleFactor(0.8)
+        }
+    }
+
     private var divider: some View { Divider().frame(height: 26) }
 
     private var uplinkTint: Color {

StreamBroadcast/RTMPPublisher.swift

@@ -59,6 +59,12 @@ actor RTMPPublisher: Publisher {
     /// Sheds input video frames when the outbound queue is deep, to bound latency
     /// (HaishinKit's send queue is otherwise unbounded). Refreshed by checkNetworkHealth.
     private let videoAdmission = VideoFrameAdmission()
+    /// Shared frame-drop / achieved-fps counters (issue #23 / M8). This actor records
+    /// the admission shed + every encoded append; the capture side records the
+    /// capture/pacing/backpressure/compositor sites into the same instance, and the
+    /// controller reads one snapshot per telemetry poll. Reachable here so the
+    /// adaptive logic (M9) can consume the congestion-drop signal.
+    private let telemetry: FrameTelemetry
 
     // Ordered, lossless audio ingress: ScreenCaptureKit yields synchronously, a
     // single consumer per track awaits each append — preserving PTS order so
@@ -80,7 +86,8 @@ actor RTMPPublisher: Publisher {
     /// Shared outbound-queue stall watchdog state (stalledTicks + lastQueueBytes).
     private var watchdog = WatchdogState()
 
-    init() {
+    init(telemetry: FrameTelemetry = FrameTelemetry()) {
+        self.telemetry = telemetry
         (micStream, micCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
         (appStream, appCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
     }
@@ -770,10 +777,14 @@ actor RTMPPublisher: Publisher {
         lastVideoBuffer = sb
         // Under outbound congestion, drop a proportion of video frames (audio is
         // never dropped) so latency stays bounded instead of the queue growing.
-        guard videoAdmission.admit() else { return }
+        guard videoAdmission.admit() else {
+            telemetry.recordDrop(.admission)
+            return
+        }
         let duration = CMTime(value: 1,
                               timescale: CMTimeScale(encodeFrameRate))
         let normalized = timeline.normalize(sb, kind: .video, fallbackDuration: duration)
+        telemetry.recordEncoded()
         await mixer.append(enforceMonotonicVideo(normalized, minStep: duration))
     }
 

StreamBroadcast/SessionPublisher.swift

@@ -69,6 +69,10 @@ actor SessionPublisher: Publisher {
 
     private var timeline = MediaTimelineNormalizer()
     private let videoAdmission = VideoFrameAdmission()
+    /// Shared frame-drop / achieved-fps counters (issue #23 / M8). See RTMPPublisher:
+    /// records the admission shed + every encoded append into the same instance the
+    /// capture side and controller share.
+    private let telemetry: FrameTelemetry
 
     // Ordered, lossless audio ingress: ScreenCaptureKit yields synchronously, a
     // single consumer per track awaits each append — preserving PTS order.
@@ -123,8 +127,10 @@ actor SessionPublisher: Publisher {
     /// The reconnect loop's currently-sleeping backoff; cancelling = "retry now".
     private var backoffSleepTask: Task<Void, any Error>?
 
-    init(protocol streamProtocol: StreamCore.StreamProtocol) {
+    init(protocol streamProtocol: StreamCore.StreamProtocol,
+         telemetry: FrameTelemetry = FrameTelemetry()) {
         self.transport = streamProtocol
+        self.telemetry = telemetry
         (micStream, micCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
         (appStream, appCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
     }
@@ -644,9 +650,13 @@ actor SessionPublisher: Publisher {
         lastMediaAt = now
         lastVideoAppendAt = now
         lastVideoBuffer = sb
-        guard videoAdmission.admit() else { return }
+        guard videoAdmission.admit() else {
+            telemetry.recordDrop(.admission)
+            return
+        }
         let duration = CMTime(value: 1, timescale: CMTimeScale(encodeFrameRate))
         let normalized = timeline.normalize(sb, kind: .video, fallbackDuration: duration)
+        telemetry.recordEncoded()
         await mixer.append(enforceMonotonicVideo(normalized, minStep: duration))
     }