fix(playback): make the stall watchdog level-triggered with a host escalation rung

Sources/AetherEngine/AetherEngine+Loading.swift

@@ -1038,17 +1038,43 @@ extension AetherEngine {
                 let fetchesAtStall = self.nativeVideoSession?.mediaFetchCountSnapshot ?? 0
                 self.stallReengageTask?.cancel()
                 self.stallReengageTask = Task { @MainActor [weak self, weak host] in
+                    // Level re-watch (#65): fetch activity inside the grace window used to disarm
+                    // this watchdog permanently — but a player that drains its remaining TAIL
+                    // segments and then parks on a frozen playlist (fwd buffer non-empty, so
+                    // playbackStalled never re-fires) was exactly that case, and nothing ever
+                    // re-armed. Re-baseline and keep watching instead, bounded so trickling
+                    // fetches on a merely slow session hand back to the producer-side arms.
+                    var baseline = fetchesAtStall
+                    var passes = 0
+                    watch: while true {
+                        try? await Task.sleep(
+                            nanoseconds: UInt64(Self.stallReengageGraceSeconds * 1_000_000_000))
+                        guard !Task.isCancelled, let self, let host,
+                              host.stallCount == count,
+                              let player = self.currentAVPlayer else { return }
+                        let fetchesNow = self.nativeVideoSession?.mediaFetchCountSnapshot ?? 0
+                        switch Self.stallWatchVerdict(
+                            fetchesNow: fetchesNow,
+                            baseline: baseline,
+                            isWaitingToPlay:
+                                player.timeControlStatus == .waitingToPlayAtSpecifiedRate,
+                            itemFailed: player.currentItem?.status == .failed,
+                            passesSoFar: passes,
+                            cap: Self.maxStallWatchPasses
+                        ) {
+                        case .disarm:
+                            return
+                        case .escalate:
+                            break watch
+                        case .rewatch:
+                            passes += 1
+                            baseline = fetchesNow
+                        }
+                    }
+                    guard let self, let host,
+                          let player = self.currentAVPlayer else { return }
                     // Stage 1: nudge seek. Device-proven to reach AVPlayer (rate re-asserts)
                     // but NOT always to revive its loader; stage 2 covers that.
-                    try? await Task.sleep(
-                        nanoseconds: UInt64(Self.stallReengageGraceSeconds * 1_000_000_000))
-                    guard !Task.isCancelled, let self, let host,
-                          host.stallCount == count else { return }
-                    let fetchesNow = self.nativeVideoSession?.mediaFetchCountSnapshot ?? 0
-                    guard fetchesNow == fetchesAtStall,
-                          let player = self.currentAVPlayer,
-                          player.timeControlStatus == .waitingToPlayAtSpecifiedRate,
-                          player.currentItem?.status != .failed else { return }
                     self.reengageStalledConsumer(
                         position: player.currentTime().seconds,
                         trigger: "stall + \(Int(Self.stallReengageGraceSeconds))s without fetches")
@@ -1063,7 +1089,42 @@ extension AetherEngine {
                           let player2 = self.currentAVPlayer,
                           player2.timeControlStatus == .waitingToPlayAtSpecifiedRate,
                           player2.currentItem?.status != .failed else { return }
-                    self.reloadStalledConsumerItem(position: player2.currentTime().seconds)
+                    // Storm shape of the final rung: on a frozen live playlist each reload replays
+                    // the tail and re-stalls within seconds, and the fresh stall supersedes this
+                    // task BEFORE the post-reload rung below can run. The persistent gate spans
+                    // stall events: reloads at the same frozen position exhaust it, then the only
+                    // remaining move is the host's (fresh session against the server route).
+                    let reloadPosition = player2.currentTime().seconds
+                    if self.isLive, !self.stallReloadReviveGate.admit(position: reloadPosition) {
+                        EngineLog.emit(
+                            "[AetherEngine] #65 stage-2 reload budget exhausted at frozen "
+                            + "\(String(format: "%.2f", reloadPosition))s; "
+                            + "publishing liveSourceReset to host",
+                            category: .engine)
+                        self.liveSourceReset.send()
+                        return
+                    }
+                    self.reloadStalledConsumerItem(position: reloadPosition)
+                    // Final rung (#65, live only): a reload against a FROZEN playlist refills the
+                    // same tail and parks again, with no notification left to re-fire. A rendered
+                    // clock that has not moved a whole post-reload window later means the local
+                    // session is unrecoverable consumer-side; only the host can retune.
+                    let clockAtReload = host.renderedTime
+                    try? await Task.sleep(
+                        nanoseconds: UInt64(2 * Self.stallReengageGraceSeconds * 1_000_000_000))
+                    guard !Task.isCancelled, host.stallCount == count,
+                          Self.shouldPublishLiveSourceReset(
+                              isLive: self.isLive,
+                              clockAtReload: clockAtReload,
+                              clockNow: host.renderedTime,
+                              isWaitingToPlay: self.currentAVPlayer?.timeControlStatus
+                                  == .waitingToPlayAtSpecifiedRate
+                          ) else { return }
+                    EngineLog.emit(
+                        "[AetherEngine] #65 stage-2 reload did not move a frozen live clock; "
+                        + "publishing liveSourceReset to host",
+                        category: .engine)
+                    self.liveSourceReset.send()
                 }
             }
             .store(in: &nativeCancellables)

Sources/AetherEngine/AetherEngine.swift

@@ -1394,12 +1394,59 @@ public final class AetherEngine: ObservableObject {
     /// of the playbackStalled notification. Cancelled on load reset; superseded by newer stalls.
     var stallReengageTask: Task<Void, Never>? = nil
     nonisolated static let stallReengageGraceSeconds: TimeInterval = 6.0
+    /// #65 level re-watch: fetch activity inside the grace window used to disarm the watchdog
+    /// PERMANENTLY (single instantaneous check), which parked a player that drained its tail
+    /// segments and then waited forever on a frozen playlist — playbackStalled never re-fires
+    /// while the buffer is non-empty, so nothing re-armed. The loop re-baselines instead, capped
+    /// so trickling fetches on a merely slow session hand back to the producer-side arms.
+    nonisolated static let maxStallWatchPasses = 10
+
+    /// #65 level re-watch verdict, one grace window at a time: silence escalates into the
+    /// nudge/reload ladder, fetch activity re-arms the watch (bounded by `cap`), a recovered,
+    /// paused, or failed player disarms it (recovery has other owners for those states).
+    enum StallWatchVerdict: Equatable {
+        case escalate
+        case rewatch
+        case disarm
+    }
+
+    nonisolated static func stallWatchVerdict(
+        fetchesNow: UInt64,
+        baseline: UInt64,
+        isWaitingToPlay: Bool,
+        itemFailed: Bool,
+        passesSoFar: Int,
+        cap: Int
+    ) -> StallWatchVerdict {
+        guard isWaitingToPlay, !itemFailed else { return .disarm }
+        if fetchesNow == baseline { return .escalate }
+        return passesSoFar + 1 < cap ? .rewatch : .disarm
+    }
+
+    /// #65 final rung: a stage-2 reload against a FROZEN live playlist refills nothing — AVPlayer
+    /// re-buffers the same tail and parks again, and no notification ever re-fires. If the rendered
+    /// clock has not moved a whole post-reload window later and the player still waits, the local
+    /// session is unrecoverable consumer-side and only the host can retune (liveSourceReset).
+    nonisolated static func shouldPublishLiveSourceReset(
+        isLive: Bool,
+        clockAtReload: Double,
+        clockNow: Double,
+        isWaitingToPlay: Bool
+    ) -> Bool {
+        isLive && clockNow == clockAtReload && isWaitingToPlay
+    }
 
     /// #93 round 3: item death (failedToPlayToEndTime after -12889 strikes) escalation.
     /// Deferred-confirm task (a transient that resumes within the window self-clears) plus the
     /// bounded reload budget. Cancelled on load reset; superseded by newer deaths.
     var itemDeathConfirmTask: Task<Void, Never>? = nil
     var itemDeathReviveGate = ItemDeathReviveGate(maxAttempts: 3)
+    /// #65 final rung, storm shape: on a frozen live playlist each stage-2 reload replays the tail,
+    /// re-stalls within seconds, and the fresh stall SUPERSEDES the ladder task before its
+    /// post-reload rung can run — so the reload cycle alone would loop forever. This gate persists
+    /// across stall events: stage-2 reloads at the same frozen position exhaust it (then the ladder
+    /// publishes liveSourceReset instead of reloading again); real progress restores the budget.
+    var stallReloadReviveGate = ItemDeathReviveGate(maxAttempts: 2)
 
     /// #199: masters whose #168 carriage verdict fired; consulted at the top of `load(source:)` to
     /// route known cases straight onto the live-ingest loopback. Engine-lifetime by design: it must
@@ -2482,6 +2529,7 @@ public final class AetherEngine: ObservableObject {
         itemDeathConfirmTask?.cancel()
         itemDeathConfirmTask = nil
         itemDeathReviveGate = ItemDeathReviveGate(maxAttempts: 3)
+        stallReloadReviveGate = ItemDeathReviveGate(maxAttempts: 2)
         masterFallbackUsed = false
         nativeSubtitleReanchorTask?.cancel()
         nativeSubtitleReanchorTask = nil

Tests/AetherEngineTests/StallWatchdogLevelRearmTests.swift

@@ -0,0 +1,93 @@
+import XCTest
+@testable import AetherEngine
+
+/// #65 follow-up: the stall re-engage watchdog was one-shot and edge-triggered — armed per
+/// playbackStalled notification, then a SINGLE instantaneous check 6 s later. Any fetch activity
+/// inside that grace window disarmed it permanently, which parked a live player that drained its
+/// remaining tail segments and then waited forever on a frozen playlist: with a non-empty forward
+/// buffer, playbackStalled never re-fires, failedToPlayToEndTime never fires while waiting, and the
+/// producer-side wedge detector died with the pump. Field trace: playlist frozen at its last
+/// segment, AVPlayer parked in waitingToMinimizeStalls with ~2 s buffered, no recovery layer ever
+/// re-examined the session.
+final class StallWatchdogLevelRearmTests: XCTestCase {
+
+    // MARK: - Level re-watch verdict
+
+    func testFetchActivityDuringGraceRewatchesInsteadOfDisarming() {
+        XCTAssertEqual(
+            AetherEngine.stallWatchVerdict(
+                fetchesNow: 7, baseline: 3, isWaitingToPlay: true, itemFailed: false,
+                passesSoFar: 0, cap: 10),
+            .rewatch,
+            "the incident: AVPlayer fetched tail segments inside the grace window; the old single check returned permanently and nothing ever re-armed")
+    }
+
+    func testSilentGraceEscalatesIntoTheLadder() {
+        XCTAssertEqual(
+            AetherEngine.stallWatchVerdict(
+                fetchesNow: 5, baseline: 5, isWaitingToPlay: true, itemFailed: false,
+                passesSoFar: 3, cap: 10),
+            .escalate)
+    }
+
+    func testRecoveredPlayerDisarms() {
+        XCTAssertEqual(
+            AetherEngine.stallWatchVerdict(
+                fetchesNow: 9, baseline: 5, isWaitingToPlay: false, itemFailed: false,
+                passesSoFar: 0, cap: 10),
+            .disarm,
+            "a playing or paused player is not this watchdog's business (user pause has its own guard; playback has recovered)")
+    }
+
+    func testFailedItemDisarms() {
+        XCTAssertEqual(
+            AetherEngine.stallWatchVerdict(
+                fetchesNow: 5, baseline: 5, isWaitingToPlay: true, itemFailed: true,
+                passesSoFar: 0, cap: 10),
+            .disarm,
+            "a failed item belongs to the item-death escalation, not the stall ladder")
+    }
+
+    func testTricklingFetchesExhaustTheWatchCap() {
+        XCTAssertEqual(
+            AetherEngine.stallWatchVerdict(
+                fetchesNow: 42, baseline: 40, isWaitingToPlay: true, itemFailed: false,
+                passesSoFar: 9, cap: 10),
+            .disarm,
+            "a merely slow session that keeps fetching hands back to the producer-side arms instead of watching forever")
+    }
+
+    // MARK: - Final rung: frozen live clock after the stage-2 reload
+
+    func testFrozenLiveClockAfterReloadPublishesReset() {
+        XCTAssertTrue(AetherEngine.shouldPublishLiveSourceReset(
+            isLive: true, clockAtReload: 62.89, clockNow: 62.89, isWaitingToPlay: true),
+            "a reload against a frozen playlist refills the same tail; only the host can retune")
+    }
+
+    func testMovedClockMeansTheReloadTookAndNothingFires() {
+        XCTAssertFalse(AetherEngine.shouldPublishLiveSourceReset(
+            isLive: true, clockAtReload: 62.89, clockNow: 68.11, isWaitingToPlay: false))
+    }
+
+    func testVODNeverPublishesLiveSourceReset() {
+        XCTAssertFalse(AetherEngine.shouldPublishLiveSourceReset(
+            isLive: false, clockAtReload: 100.0, clockNow: 100.0, isWaitingToPlay: true),
+            "VOD stalls have their own arms (#99/#126/#169); liveSourceReset is a live-retune contract")
+    }
+
+    // MARK: - Storm shape: reload budget across superseding stall events
+
+    func testStageTwoReloadsAtFrozenPositionExhaustThenProgressRestores() {
+        // On a frozen playlist each reload replays the tail and re-stalls within seconds; the fresh
+        // stall supersedes the ladder task before its post-reload rung can run. The persistent gate
+        // is what breaks that loop.
+        var gate = ItemDeathReviveGate(maxAttempts: 2)
+        XCTAssertTrue(gate.admit(position: 62.89), "first reload is always worth trying")
+        XCTAssertTrue(gate.admit(position: 62.88), "clock jitter below the epsilon is the same dead spot")
+        XCTAssertFalse(gate.admit(position: 62.89),
+            "third reload at the same frozen position is futile; the ladder must publish liveSourceReset instead")
+        XCTAssertTrue(gate.admit(position: 130.4),
+            "real progress (the reload took, or the user zapped/scrubbed) is a fresh episode with a fresh budget")
+    }
+}