feat: EWMA throughput estimator + measured cellular ceiling (fixes #24)

StreamBroadcast/RTMPPublisher.swift

@@ -528,6 +528,7 @@ actor RTMPPublisher: Publisher {
         if previous == nil || snapshot != previous {
             await networkController.setPathProfile(
                 ceiling: snapshot.videoBitRateCeiling(configuredMaximum: settings.videoBitrate),
+                seed: snapshot.videoBitRateSeed(configuredMaximum: settings.videoBitrate),
                 interface: snapshot.interface,
                 isBaseline: previous == nil,
                 applyingTo: stream)
@@ -1013,8 +1014,13 @@ actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy {
             }
 
         case .status(let report):
+            // Pass live audio bitrate so the upward-probe's throughput gate compares
+            // video-vs-video (the estimate carries total socket egress). Paused → 0,
+            // mirroring the insufficient-bandwidth path's guard.
+            let audioBitRate = abr.capturePaused ? 0 : await stream.audioSettings.bitRate
             let decision = abr.onStatus(bytesOutPerSecond: report.currentBytesOutPerSecond,
-                                        queueBytesOut: report.currentQueueBytesOut)
+                                        queueBytesOut: report.currentQueueBytesOut,
+                                        audioBitRate: audioBitRate)
             if decision.shouldApply {
                 await applyDecision(decision, to: stream)
                 // Match the original: only the upward-PROBE apply logs "recovered".
@@ -1046,10 +1052,12 @@ actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy {
     /// seed. The baseline (first) emission only adopts the interface and the
     /// ceiling, so a broadcast still STARTS at the configured full bitrate.
     func setPathProfile(ceiling: Int,
+                        seed: Int,
                         interface: NetworkPathSnapshot.Interface,
                         isBaseline: Bool,
                         applyingTo stream: some StreamConvertible) async {
-        let decision = abr.onPathProfile(ceiling: ceiling, interface: interface, isBaseline: isBaseline)
+        let decision = abr.onPathProfile(ceiling: ceiling, seed: seed,
+                                         interface: interface, isBaseline: isBaseline)
         if decision.shouldApply { await applyDecision(decision, to: stream) }
     }
 

StreamBroadcast/SessionPublisher.swift

@@ -284,6 +284,7 @@ actor SessionPublisher: Publisher {
         if let path = currentPath {
             await networkController.setPathProfile(
                 ceiling: path.videoBitRateCeiling(configuredMaximum: settings.videoBitrate),
+                seed: path.videoBitRateSeed(configuredMaximum: settings.videoBitrate),
                 interface: path.interface,
                 isBaseline: false,
                 applyingTo: stream)
@@ -468,6 +469,7 @@ actor SessionPublisher: Publisher {
         if result.shouldUpdateCeiling, let stream {
             await networkController.setPathProfile(
                 ceiling: snapshot.videoBitRateCeiling(configuredMaximum: settings.videoBitrate),
+                seed: snapshot.videoBitRateSeed(configuredMaximum: settings.videoBitrate),
                 interface: snapshot.interface,
                 isBaseline: result.isBaseline,
                 applyingTo: stream)

StreamCore/AdaptiveBitRateState.swift

@@ -40,16 +40,30 @@ public struct AdaptiveBitRateState: Sendable, Equatable {
     public private(set) var congestionActive = false
     public private(set) var capturePaused = false
     public private(set) var pathCeiling: Int
+    /// The conservative opening bid for the current path (2.5 Mbps on cellular, the
+    /// hard ceiling otherwise). The upward probe climbs freely up to this seed to
+    /// RECOVER, but only exceeds it with measured-throughput evidence (issue #24).
+    public private(set) var pathSeed: Int
     public private(set) var currentInterface: NetworkPathSnapshot.Interface?
     public private(set) var lastGoodTarget: [NetworkPathSnapshot.Interface: Int] = [:]
     public private(set) var thermalBitRateScale: Double = 1.0
     public private(set) var thermalFrameRateCap: Int = .max
+    /// Smoothed measured egress throughput, fed the 1 Hz `bytesOutPerSecond` on
+    /// every status/insufficient event. Drives the smoothed reduction basis and the
+    /// probe's "is the link really carrying this?" gate.
+    public private(set) var throughput = ThroughputEstimator()
+
+    /// Above the seed, the probe requires measured throughput to be within this
+    /// fraction of the current target before it climbs further — evidence the link
+    /// is actually delivering the rate, not the encoder idling on low-motion frames.
+    static let probeThroughputFraction = 0.9
 
     public init(maximumBitRate: Int, frameRate: Int) {
         self.maximumBitRate = maximumBitRate
         minimumBitRate = max(300_000, maximumBitRate / 10)
         targetBitRate = maximumBitRate
         pathCeiling = maximumBitRate
+        pathSeed = maximumBitRate
         // Clamp only to a sane hardware maximum, NOT 30 — a 60 fps stream must keep
         // configuredFrameRate = 60 so the `> 30` congestion tier in frameInterval is
         // reachable.
@@ -94,6 +108,16 @@ public struct AdaptiveBitRateState: Sendable, Equatable {
         congestionActive = true
         if bytesOutPerSecond > 0 {
             zeroOutputSeconds = 0
+            throughput.record(bytesOutPerSecond: bytesOutPerSecond)
+            // Reduce against the RAW current egress. This event only fires after
+            // HaishinKit sees ~3 s of sustained queue growth, so the sample already
+            // reflects true link capacity (the queue is backed up, we're draining as
+            // fast as the link allows). The EWMA is deliberately NOT used here: it is
+            // also fed healthy status ticks where egress = encoder DEMAND, not
+            // capacity, so smoothing the cut against it either under-reduces (after a
+            // high-demand period) or over-reduces (after a static-screen low-demand
+            // period) — both failure modes were caught in review. The estimator's
+            // sound job is the upward-probe climb-above-seed gate below (issue #24).
             let available = max(0, bytesOutPerSecond * 8 - audioBitRate)
             // Leave substantial headroom for Wi-Fi variance, RTMP overhead, and
             // keyframes instead of targeting the measured ceiling.
@@ -112,13 +136,15 @@ public struct AdaptiveBitRateState: Sendable, Equatable {
     /// `else`, so the `healthySeconds >= 30` probe guard cannot pass the same tick.
     /// A single `ABRDecision` is therefore byte-exact — do NOT collapse the branches.
     public mutating func onStatus(bytesOutPerSecond: Int,
-                                  queueBytesOut: Int) -> ABRDecision {
+                                  queueBytesOut: Int,
+                                  audioBitRate: Int = 0) -> ABRDecision {
         queueBytes = queueBytesOut
         guard !capturePaused else {
             zeroOutputSeconds = 0
             healthySeconds = 0
             return ABRDecision(shouldApply: false, severe: false)
         }
+        throughput.record(bytesOutPerSecond: bytesOutPerSecond)
         var apply = false
         var severe = false
         // Zero output with an empty queue is normal for a paused/static capture; it
@@ -145,12 +171,38 @@ public struct AdaptiveBitRateState: Sendable, Equatable {
             return ABRDecision(shouldApply: apply, severe: severe)
         }
         healthySeconds = 0
-        if targetBitRate < effectiveMaximum {
-            targetBitRate = min(effectiveMaximum,
-                                targetBitRate + max(75_000, effectiveMaximum / 20))
-        } else {
-            congestionActive = false
+        // A sustained clean queue means we are no longer congested — restore full
+        // frame rate NOW, even when the target is deliberately held at the seed and
+        // never reaches `effectiveMaximum`. (Clearing this only at the ceiling would
+        // pin a >30 fps cellular stream at 30 fps forever once the seed gate holds.)
+        // `wasCongested` forces an apply on the CLEARING tick so the actor actually
+        // re-applies the restored frame interval to the encoder — clearing the pure
+        // flag without an apply signal would leave the encoder stuck at 30 fps.
+        let wasCongested = congestionActive
+        congestionActive = false
+        guard targetBitRate < effectiveMaximum else {
+            return ABRDecision(shouldApply: true, severe: false)
+        }
+        // Below the conservative per-path seed the probe climbs freely — this is
+        // recovery back toward a rate the link already sustained. To climb ABOVE the
+        // seed (e.g. past 2.5 Mbps on a strong 5G uplink), a clean queue alone isn't
+        // enough: require measured throughput within `probeThroughputFraction` of the
+        // current target, so a low-motion encoder idling below the seed doesn't push
+        // the target onto capacity the link never demonstrated (issue #24). The
+        // estimate carries total socket egress, so compare video-vs-video by netting
+        // out audio, mirroring the reduction path.
+        if targetBitRate >= pathSeed {
+            let measuredVideo = throughput.bitsPerSecond - audioBitRate
+            let justifiesClimb = throughput.hasSample
+                && measuredVideo >= Int(Double(targetBitRate) * Self.probeThroughputFraction)
+            guard justifiesClimb else {
+                // Hold the target at the seed, but still apply if this tick cleared
+                // congestion so the encoder's frame rate is restored.
+                return ABRDecision(shouldApply: wasCongested, severe: false)
+            }
         }
+        targetBitRate = min(effectiveMaximum,
+                            targetBitRate + max(75_000, effectiveMaximum / 20))
         return ABRDecision(shouldApply: true, severe: false)
     }
 
@@ -160,20 +212,35 @@ public struct AdaptiveBitRateState: Sendable, Equatable {
         zeroOutputSeconds = 0
     }
 
-    /// Per-path ceiling + interface seeding. ORDER IS LOAD-BEARING: `raised` is
-    /// computed against the OLD `pathCeiling`; `pathCeiling` is assigned BEFORE the
-    /// `effectiveMaximum` clamp so the clamp folds in the NEW ceiling + thermal cap.
+    /// Per-path ceiling + interface seeding. `ceiling` is the HARD cap; `seed` is the
+    /// conservative opening bid (2.5 Mbps on cellular). ORDER IS LOAD-BEARING:
+    /// `raised` is computed against the OLD `pathCeiling`; `pathCeiling` is assigned
+    /// BEFORE the `effectiveMaximum` clamp so the clamp folds in the NEW ceiling +
+    /// thermal cap. The conservative-seed clamp is applied ONLY on the baseline
+    /// emission and on a genuine interface handoff — never on a same-interface
+    /// ceiling re-emission, or every path churn would yank a probed-up target back
+    /// down to the seed.
     public mutating func onPathProfile(ceiling: Int,
+                                       seed: Int,
                                        interface: NetworkPathSnapshot.Interface,
                                        isBaseline: Bool) -> ABRDecision {
         let clamped = max(minimumBitRate, min(ceiling, maximumBitRate))
+        let clampedSeed = max(minimumBitRate, min(seed, clamped))
+        pathSeed = clampedSeed
         if isBaseline || currentInterface == nil {
             currentInterface = interface
+            // Open at the seed on cellular (start conservative), full otherwise.
+            targetBitRate = min(targetBitRate, clampedSeed)
         } else if let current = currentInterface, interface != current {
             lastGoodTarget[current] = targetBitRate
             currentInterface = interface
-            let seed = lastGoodTarget[interface] ?? Int(Double(clamped) * 0.6)
-            targetBitRate = max(minimumBitRate, min(seed, clamped))
+            // The old link's throughput average says nothing about this one.
+            throughput.reset()
+            // Resume a remembered rate for a known interface; else open at the
+            // conservative seed but never above the historical 0.6x cold-start.
+            let cold = min(clampedSeed, Int(Double(clamped) * 0.6))
+            let seedTarget = lastGoodTarget[interface] ?? cold
+            targetBitRate = max(minimumBitRate, min(seedTarget, clamped))
             healthySeconds = 0
             zeroOutputSeconds = 0
         }

StreamCore/NetworkPathSnapshot.swift

@@ -85,17 +85,31 @@ public struct NetworkPathSnapshot: Sendable, Equatable {
     /// What "a different link" means for reconnect purposes.
     public var linkIdentity: String { "\(interface.rawValue)/\(interfaceName)" }
 
-    /// Per-path video bitrate ceiling policy. Cellular caps the max so a 3 Mbps
-    /// Wi-Fi target is never re-attempted verbatim on a weak cell; Low Data
-    /// Mode and iOS 26 link-quality signals cap harder. Numbers are policy,
-    /// tunable on device.
+    /// Per-path video bitrate ceiling — the HARD cap the upward probe may never
+    /// exceed. Low Data Mode and the iOS 26 link-quality signals are genuine
+    /// hardware/policy limits and stay hard here. Cellular is deliberately NOT a
+    /// hard cap any more: a real 5G uplink often exceeds the old 2.5 Mbps clamp, so
+    /// that value moved to `videoBitRateSeed` as a conservative starting bid the
+    /// probe can climb above with measured evidence (issue #24). Numbers are
+    /// policy, tunable on device.
     public func videoBitRateCeiling(configuredMaximum maximum: Int) -> Int {
         guard isSatisfied else { return maximum }
         var ceiling = maximum
-        if interface == .cellular || isExpensive { ceiling = min(ceiling, 2_500_000) }
         if isConstrained { ceiling = min(ceiling, 1_200_000) }
         if isUltraConstrained { ceiling = min(ceiling, 800_000) }
         if linkQualityIsMinimal { ceiling = Swift.max(300_000, ceiling / 2) }
         return ceiling
     }
+
+    /// The conservative INITIAL bitrate for this path — where a fresh connection or
+    /// a post-handoff reseed opens before the EWMA probe measures real capacity.
+    /// Cellular/expensive links seed at 2.5 Mbps (a safe 5G/LTE opening bid); every
+    /// other link seeds at its hard ceiling (i.e. starts full). Never above the
+    /// hard ceiling, so Low Data Mode still wins.
+    public func videoBitRateSeed(configuredMaximum maximum: Int) -> Int {
+        let ceiling = videoBitRateCeiling(configuredMaximum: maximum)
+        guard isSatisfied else { return ceiling }
+        if interface == .cellular || isExpensive { return min(ceiling, 2_500_000) }
+        return ceiling
+    }
 }

StreamCore/ThroughputEstimator.swift

@@ -0,0 +1,50 @@
+/// A pure exponentially-weighted moving average of measured egress throughput,
+/// fed the 1 Hz `currentBytesOutPerSecond` deltas HaishinKit's `NetworkMonitor`
+/// already surfaces. It smooths the single-sample capacity reading the ABR used
+/// to react to, so a lone keyframe burst or a momentary radio stall no longer
+/// drives an over-deep cut, and it gives the upward probe measured evidence that
+/// a link is actually carrying the current rate before it climbs above the
+/// conservative per-path seed (issue #24).
+///
+/// Lives in StreamCore, `Sendable` + `Equatable`, so the arithmetic is unit-tested
+/// on CI alongside `AdaptiveBitRateState` rather than resting on device testing.
+public struct ThroughputEstimator: Sendable, Equatable {
+    /// EWMA weight on the newest sample. 0.4 tracks a genuine capacity change
+    /// (e.g. a 5G→LTE handoff) within a few ticks while still filtering the
+    /// per-second jitter of RTMP framing and keyframes.
+    public static let smoothingFactor = 0.4
+
+    /// The smoothed estimate in BITS per second (the sample is bytes/s × 8), or 0
+    /// before the first positive sample.
+    public private(set) var bitsPerSecond = 0
+
+    /// False until at least one positive sample has seeded the average, so the
+    /// first sample is adopted verbatim instead of being blended with 0.
+    public private(set) var hasSample = false
+
+    public init() {}
+
+    /// Fold one 1 Hz egress sample (bytes/s, as reported by the network monitor)
+    /// into the average. Non-positive samples are ignored: a zero-output tick is a
+    /// stall, not evidence of low capacity, and blending it in would poison the
+    /// estimate right when the ABR most needs a stable capacity read.
+    public mutating func record(bytesOutPerSecond: Int) {
+        guard bytesOutPerSecond > 0 else { return }
+        let sample = bytesOutPerSecond * 8
+        if hasSample {
+            bitsPerSecond = Int(Self.smoothingFactor * Double(sample)
+                                + (1 - Self.smoothingFactor) * Double(bitsPerSecond))
+        } else {
+            bitsPerSecond = sample
+            hasSample = true
+        }
+    }
+
+    /// Drop the learned estimate. Called on a handoff to a DIFFERENT physical link
+    /// whose capacity is unknown, so a stale Wi-Fi average can never justify
+    /// probing a fresh, weaker cellular link above its seed.
+    public mutating func reset() {
+        bitsPerSecond = 0
+        hasSample = false
+    }
+}