feat(cache): expose session byte budgets and lifecycle status

CHANGELOG.md

@@ -22,6 +22,8 @@ the public-API contract.
 
 ### Added
 
+- Optional session-cache byte budgets reuse existing volume and multi-session safeguards. A separate diagnostics snapshot reports effective bytes, protected playback-window overflow and typed cleanup results. Cache state is sampled off-main and old-session updates are rejected; defaults and live DVR policy remain unchanged.
+
 - **`LiveTelemetry.objectAudio`: how many Atmos objects play, and how many overhead.** While
   `objectAudioRendering` plays TrueHD Atmos or DTS:X (a build with the `TrueHDObjectAudio` /
   `DTSXObjectAudio` traits; without them it stays nil), the 1 Hz snapshot carries an

Sources/AetherEngine/AetherEngine+Diagnostics.swift

@@ -488,6 +488,25 @@ extension AetherEngine {
 
     // MARK: - Live telemetry bridge
 
+    /// Lifecycle callbacks can arrive after stop or a new load. Only cleanup may publish after
+    /// the owning session has detached; the token still prevents it crossing into its successor.
+    func acceptSessionCacheStatus(_ status: SessionCacheStatus, token: UInt64) {
+        guard token == sessionCacheStatusToken else { return }
+        guard status.cleanupResult != .notRun || nativeVideoSession != nil else { return }
+        diagnostics.sessionCacheStatus = status
+    }
+
+    /// Read the cache ledger off-main. An old tick must never overwrite a new session or cleanup.
+    func refreshSessionCacheStatus() async {
+        guard let session = nativeVideoSession else { return }
+        let token = sessionCacheStatusToken
+        let generation = loadGeneration
+        let status = await BlockingWork.detached(priority: .utility) { session.sessionCacheStatus() }.value
+        guard !Task.isCancelled, sessionCacheStatusToken == token, loadGeneration == generation,
+              nativeVideoSession === session else { return }
+        diagnostics.sessionCacheStatus = status
+    }
+
     /// Single write-through point: sampler never reaches into `EngineDiagnostics` directly.
     func applyLiveTelemetry(_ snapshot: LiveTelemetry) {
         diagnostics.liveTelemetry = snapshot

Sources/AetherEngine/AetherEngine+Loading.swift

@@ -373,6 +373,9 @@ extension AetherEngine {
         // stale caller cannot adopt its successor's generation.
         if let generation { try checkLoadCurrent(generation) }
         playbackBackend = .native
+        sessionCacheStatusToken &+= 1
+        diagnostics.sessionCacheStatus = .nativeRemoteHLS(
+            requestedBudgetBytes: options.sessionCacheByteBudget.map { max(0, $0) })
         // #168 follow-up: detect a superseding load()/stop() between the carriage verdict and the reroute.
         let bypassGeneration = generation ?? loadGeneration
 
@@ -888,8 +891,17 @@ extension AetherEngine {
             // the host asked for; without it the flag lasts exactly as long as the pre-opened demuxer.
             heldSourceConnection: loadedOptions.heldSourceConnection,
             declaredDurationSeconds: loadedOptions.declaredDurationSeconds,
-            forwardBufferSegments: loadedOptions.forwardBufferSegments
+            forwardBufferSegments: loadedOptions.forwardBufferSegments,
+            sessionCacheByteBudget: loadedOptions.sessionCacheByteBudget
         )
+        sessionCacheStatusToken &+= 1
+        let cacheStatusToken = sessionCacheStatusToken
+        session.onSessionCacheStatusChanged = { [weak self] status in
+            Task { @MainActor [weak self] in
+                self?.acceptSessionCacheStatus(status, token: cacheStatusToken)
+            }
+        }
+
         // AE#792: the session's live header store, shared with the playback demuxer it was handed, so
         // `updateRequestHeaders(_:)` reaches that reader and every reopen the producer makes. Set before start().
         session.requestHeaders = sessionRequestHeaders
@@ -1324,7 +1336,8 @@ extension AetherEngine {
 
         // session.start() opens its own Demuxer + prewarm seek (~1-3 s on slow CDN); detach so @MainActor doesn't block.
         var playbackURL = try await BlockingWork.detached(priority: .userInitiated) { [session] in
-            try session.start()
+            do { return try session.start() }
+            catch { session.stop(reason: .loadFailed); throw error }
         }.value
         // AirPlay (#86): while external playback is active, serve the loopback over the device's LAN IP so
         // the receiver reaches the engine-processed stream (DV/Atmos/subtitles preserved). An HDR/DV master
@@ -1334,7 +1347,7 @@ extension AetherEngine {
         playbackURL = served.url
         // Superseded while starting: stop and unwind before touching shared state.
         if loadGeneration != generation {
-            session.stop()
+            session.stop(reason: .sourceChanged)
             try checkLoadCurrent(generation)
         }
         self.nativeVideoSession = session
@@ -1961,6 +1974,9 @@ extension AetherEngine {
         generation: UInt64,
         elType: String? = nil
     ) async throws {
+        try checkLoadCurrent(generation)
+        sessionCacheStatusToken &+= 1
+        diagnostics.sessionCacheStatus = .inactive
         let deinterlaceMode = loadedOptions.deinterlaceMode
         let waitStarted = DispatchTime.now()
         if let outcome = await DeinterlaceHardwareWarmup.shared.waitIfNeeded(
@@ -2123,6 +2139,7 @@ extension AetherEngine {
         }
         if loadGeneration == generation { recordStartupCheckpoint(.sessionConstructed) }   // #361
         let forwardBufferSegments = loadedOptions.forwardBufferSegments
+        let sessionCacheByteBudget = loadedOptions.sessionCacheByteBudget
         let dvrRetention = loadedOptions.softwareDVRRetention
         try await BlockingWork.detached(priority: .userInitiated) {
             [host, preopenedDemuxer, url, sourceHTTPHeaders, requestHeaders = sessionRequestHeaders, isLive, dvrWindowSeconds, probesize, maxAnalyzeDuration, sequentialOrigin, heldSourceConnection, declaredDuration, networkPhaseSink] in
@@ -2141,7 +2158,8 @@ extension AetherEngine {
                 isLive: isLive,
                 dvrWindowSeconds: dvrWindowSeconds,
                 dvrRetention: dvrRetention,
-                forwardBufferSegments: forwardBufferSegments
+                forwardBufferSegments: forwardBufferSegments,
+                sessionCacheByteBudget: sessionCacheByteBudget
             )
         }.value
         // Superseded: stop idempotently to tear down the demuxer the detached closure opened, then unwind.
@@ -2476,6 +2494,8 @@ extension AetherEngine {
         stopInternal(resetDisplayCriteria: false,
                      keepNativeHost: !targetSoftwarePath && !mediaServicesWereReset,
                      keepCustomReader: true, keepCurrentItem: keepAudioSwitchItem)
+        sessionCacheStatusToken &+= 1
+        diagnostics.sessionCacheStatus = .inactive
         pendingInPlaceItemHandover = keepAudioSwitchItem
         if mediaServicesWereReset { dropAudioPlayerHostAfterMediaServicesReset() }
         EngineLog.emit("[AetherEngine] reload: stopInternal done (\(elapsedMs(since: reloadStart))ms)", category: .engine)

Sources/AetherEngine/AetherEngine+ReloadWithOptions.swift

@@ -454,7 +454,7 @@ enum SessionOptionCorrection {
         "preserveASSMarkup", "prepareNativeSubtitles", "eagerNativeSubtitleReaders", "serveIFramePlaylist", "confirmAtmos",
         "nativeSubtitlePreferredLanguages", "sequentialOrigin", "maxConcurrentSourceRequests", "heldSourceConnection",
         "declaredDurationSeconds", "probesize", "maxAnalyzeDuration", "preferredAudioLanguages",
-        "preferredSubtitleLanguages", "externalSubtitles", "forwardBufferSegments", "progressiveSegmentDelivery", "autoplay",
+        "preferredSubtitleLanguages", "externalSubtitles", "forwardBufferSegments", "sessionCacheByteBudget", "progressiveSegmentDelivery", "autoplay",
         "audioDelaySeconds", "teletextPage", "deinterlaceMode", "deinterlaceFieldRate", "preferredDecodePath",
         "escalatesToSoftwarePath", "sharedOutputRole", "isLiveRejoin", "subtitleSessionCarryover",
     ]

Sources/AetherEngine/AetherEngine.swift

@@ -2123,6 +2123,9 @@ public final class AetherEngine: ObservableObject {
     /// dismissal. A superseded load throws CancellationError at the first checkpoint.
     var loadGeneration: UInt64 = 0
 
+    /// Separates cache lifecycle callbacks from superseding sessions.
+    var sessionCacheStatusToken: UInt64 = 0
+
     /// Distinct across engine instances and refreshed whenever teardown invalidates preview work.
     public private(set) var scrubPreviewSourceGeneration = UUID()
 
@@ -4163,6 +4166,8 @@ public final class AetherEngine: ObservableObject {
         // (audio-only fast path, suppressed hosts); a load() that throws before routing leaves it for stop().
         claimSoftwarePathTakeover()   // AE#629
         stopInternal(resetDisplayCriteria: false, keepNativeHost: preserveNativeHost, keepCurrentItem: handOverInPlace)
+        sessionCacheStatusToken &+= 1
+        diagnostics.sessionCacheStatus = .inactive
         if mediaServicesWereReset { dropAudioPlayerHostAfterMediaServicesReset() }
         // #35/#93: a genuinely new item has not rendered yet; re-arm the cold-startup wedge suspension.
         // Scrub/seek/producer-restart never route through load(), so mid-stream #93 detection stays armed.
@@ -6593,7 +6598,8 @@ public final class AetherEngine: ObservableObject {
         releaseHeldPicture(reason: "stop")
         nextLoadRequestsInPlaceItemHandover = false
         stopInternal(resetDisplayCriteria: resetDisplayCriteria,
-                     finalTeardown: finalTeardown ?? resetDisplayCriteria)
+                     finalTeardown: finalTeardown ?? resetDisplayCriteria,
+                     cacheCleanupReason: .sessionStopped)
         state = .idle
         // #361: the sequence ends here without ever reaching its last checkpoint, which is the whole
         // point: nothing about a stop should read as a finished startup. Cleared in stop() and not in
@@ -7587,7 +7593,8 @@ public final class AetherEngine: ObservableObject {
     ///   never settles and burns the full settle timeout (~12 s of
     ///   black-screen latency per audio switch on the old fixed 5 s
     ///   poll; capped at ~2 s since #117, but still worth skipping).
-    func stopInternal(resetDisplayCriteria: Bool = true, keepNativeHost: Bool = false, keepCustomReader: Bool = false, keepCurrentItem: Bool = false, finalTeardown: Bool = false) {
+    func stopInternal(resetDisplayCriteria: Bool = true, keepNativeHost: Bool = false, keepCustomReader: Bool = false, keepCurrentItem: Bool = false, finalTeardown: Bool = false,
+                      cacheCleanupReason: SessionCacheCleanupReason = .sourceChanged) {
         // AE#560: a recording never outlives its session. This covers stop(), a new load() and
         // every reload, all of which pass through here before the demuxer goes away.
         endRecordingIfRunning(reason: .sessionEnded)
@@ -7664,7 +7671,10 @@ public final class AetherEngine: ObservableObject {
         // host has already retired for the next item. Only the session's slot is cleared; the engine
         // keeps the host's observer and re-arms the next session with it in load().
         nativeVideoSession?.setNativeVideoFrameTimeObserver(nil)
-        nativeVideoSession?.stop()
+        if diagnostics.sessionCacheStatus.route == .nativeRemoteHLS {
+            diagnostics.sessionCacheStatus = .inactive
+        }
+        nativeVideoSession?.stop(reason: cacheCleanupReason)
         nativeVideoSession = nil
         nativeSubtitleRenditionsServed = false
         dolbyVisionConversion = nil

Sources/AetherEngine/Diagnostics/EngineDiagnostics.swift

@@ -6,6 +6,9 @@ import Combine
 /// Stats overlays observe this object; everything else observes the engine and is unaffected by telemetry samples.
 @MainActor
 public final class EngineDiagnostics: ObservableObject {
+    /// Session-only fMP4 cache state, sampled off-main and cleared on a new source.
+    @Published public internal(set) var sessionCacheStatus: SessionCacheStatus = .inactive
+
 
     /// 1 Hz snapshot while playing/paused; nil while idle. Cleared in stopInternal so sessions don't inherit stale numbers.
     @Published public internal(set) var liveTelemetry: LiveTelemetry?

Sources/AetherEngine/Diagnostics/LiveTelemetrySampler.swift

@@ -456,6 +456,7 @@ final class LiveTelemetrySampler {
             rssMb: AetherEngine.residentMemoryMB()
         )
         engine.applyLiveTelemetry(snapshot)
+        await engine.refreshSessionCacheStatus()
     }
 
     /// One line per tick on the native path (#93 post-recovery lag diagnosis). Discriminates

Sources/AetherEngine/Native/SoftwarePlaybackHost.swift

@@ -45,6 +45,15 @@ final class SoftwarePlaybackHost {
     }()
 
 
+    /// Zero still needs a working spool; every choice retains the volume and shared-claim clamp.
+    nonisolated static func packetCacheBudgetBytes(requestedBytes: Int?, availableBytes: Int64?,
+                                                   capRelaxed: Bool) -> Int {
+        let budget = HLSVideoEngine.resolveSessionCacheBudget(
+            requestedBytes: requestedBytes, volumeAvailableBytes: availableBytes, capRelaxed: capRelaxed)
+        return min(budget.volumeSafetyLimitBytes,
+                   budget.baseEffectiveBytes > 0 ? budget.baseEffectiveBytes : 32 << 20)
+    }
+
     // MARK: - Published state (mirrors NativeAVPlayerHost surface)
 
     /// Frames enqueued on the display layer; read by LiveTelemetrySampler at 1 Hz for observed FPS. Lock-guarded (demux thread writes, main-actor reads).
@@ -998,7 +1007,8 @@ final class SoftwarePlaybackHost {
         isLive: Bool = false,
         dvrWindowSeconds: Double? = nil,
         dvrRetention: SoftwareDVRRetentionOptions? = nil,
-        forwardBufferSegments: Int? = nil
+        forwardBufferSegments: Int? = nil,
+        sessionCacheByteBudget: Int? = nil
     ) async throws {
         self.demuxer = dem
         self.duration = dem.duration
@@ -1334,7 +1344,8 @@ final class SoftwarePlaybackHost {
                         .volumeAvailableCapacity.map(Int64.init)
                     let capRelaxed = HLSVideoEngine.retentionCapRelaxed(forwardWindowSegments: segments)
                     let diskClaim = RetentionClaims.shared.claim(volumeAvailableBytes: available) {
-                        HLSVideoEngine.sessionRetentionBudgetBytes(volumeAvailableBytes: $0, capRelaxed: capRelaxed)
+                        Self.packetCacheBudgetBytes(requestedBytes: sessionCacheByteBudget,
+                                                    availableBytes: $0, capRelaxed: capRelaxed)
                     }
                     bytes = diskClaim.bytes
                     guard bytes > 0 else { return nil }

Sources/AetherEngine/PlayerState.swift

@@ -882,6 +882,9 @@ public struct LoadOptions: Sendable, Equatable {
     /// remote server directly.
     public var forwardBufferSegments: Int?
 
+    /// Optional temporary session-cache budget. Nil retains automatic sizing; zero keeps only the working window.
+    public var sessionCacheByteBudget: Int? = nil
+
     /// Serve each VOD loopback segment while it is being written instead of after its cut. The
     /// muxer flushes a fragment about every half second and the loopback server sends each one as
     /// it lands, so AVPlayer can show and start on the first fragments of a segment rather than
@@ -1023,6 +1026,7 @@ public struct LoadOptions: Sendable, Equatable {
         preferredSubtitleLanguages: [String] = [],
         externalSubtitles: [ExternalSubtitleTrack] = [],
         forwardBufferSegments: Int? = nil,
+        sessionCacheByteBudget: Int? = nil,
         progressiveSegmentDelivery: Bool = false,
         autoplay: Bool = true,
         teletextPage: Int? = nil,
@@ -1077,6 +1081,7 @@ public struct LoadOptions: Sendable, Equatable {
         self.preferredSubtitleLanguages = preferredSubtitleLanguages
         self.externalSubtitles = externalSubtitles
         self.forwardBufferSegments = forwardBufferSegments
+        self.sessionCacheByteBudget = sessionCacheByteBudget
         self.progressiveSegmentDelivery = progressiveSegmentDelivery
         self.autoplay = autoplay
         self.teletextPage = teletextPage