Merge branch 'fix/ae446-rejoin-is-not-a-host-scrub'

Author: Vincent Herbst

Sources/AetherEngine/AetherEngine+Live.swift

@@ -148,6 +148,97 @@ extension AetherEngine {
             ?? (outputFloor + playlistShiftSeconds)
     }
 
+    /// AE#446 round 4: measure how far the current item's own timeline sits below the session's.
+    ///
+    /// A live item's zero is the first segment ITS playlist listed. The producer's window floor and
+    /// the item's own floor slide together, because one rule sizes both (`LiveWindowSizing` is the
+    /// single source of truth for the playlist's first visible segment and the cache's eviction), so
+    /// their difference is the offset and it holds still while both ends move. Measured on the harness
+    /// across twelve seconds of sliding: 50.00 s at every sample, while the item's floor walked from
+    /// 0.00 to 15.00 and the producer's from 51.40 to 66.40.
+    ///
+    /// Latched per item, because it is a property of the playlist that item loaded, and re-measured
+    /// when the item under the host changes. It reads 0 for the item a session starts with, which is
+    /// why nothing needed it until a swap attached a second one.
+    @MainActor
+    func measureLiveItemAxisOffset() {
+        guard isLive, let host = nativeHost else { return }
+        guard host.itemGeneration != liveItemAxisOffsetGeneration else { return }
+        // A range of zero width is an item that has not reported yet, not an item at the origin.
+        guard host.seekableEnd > host.seekableStart,
+              let producerFloor = residentLiveFloorSessionSeconds() else { return }
+        let offset = Self.liveItemAxisOffset(producerFloorSession: producerFloor,
+                                             itemSeekableStart: host.seekableStart,
+                                             shift: playlistShiftSeconds)
+        guard offset.isFinite else { return }
+        liveItemAxisOffsetGeneration = host.itemGeneration
+        liveItemAxisOffsetSeconds = offset
+        guard liveItemAxisOffsetSeconds > 0.01 else { return }
+        EngineLog.emit(
+            "[AetherEngine] #446 this item's playlist began \(String(format: "%.2f", liveItemAxisOffsetSeconds))s "
+            + "into the session, so its own clock reads that much below the session's; folding it into "
+            + "every conversion for as long as this item is the one playing",
+            category: .engine)
+    }
+
+    /// AE#446 round 4: the arithmetic behind `measureLiveItemAxisOffset`, on its own so the case can
+    /// be stated without a session.
+    ///
+    /// The producer's floor is on the session axis; the item's floor is on the item's own. One rule
+    /// sizes both, so their difference is what separates the axes, and it is a difference rather than
+    /// an assumption. Negative is not a case that can be acted on (an item claiming to hold content
+    /// older than the producer does), and it folds to 0, which is the pre-swap behaviour.
+    nonisolated static func liveItemAxisOffset(
+        producerFloorSession: Double, itemSeekableStart: Double, shift: Double
+    ) -> Double {
+        Swift.max(0, producerFloorSession - (itemSeekableStart + shift))
+    }
+
+    /// AE#446 round 4: the three readings a rejoin's placement is argued from, on one line.
+    ///
+    /// They were only ever available separately, which is why an item's clock and an item's seekable
+    /// range could disagree for a whole investigation without anyone being able to say so. Bounded to
+    /// the seconds after a swap, so a live session does not pay for it.
+    @MainActor
+    func auditLiveRejoinPlacement() {
+        guard let until = liveRejoinAuditUntil, let host = nativeHost else { return }
+        let now = Date()
+        guard now < until else { liveRejoinAuditUntil = nil; return }
+        if let last = liveRejoinAuditLastEmit, now.timeIntervalSince(last) < 1.0 { return }
+        liveRejoinAuditLastEmit = now
+        let producer = residentLiveRangeSessionSeconds()
+        EngineLog.emit(
+            "[AetherEngine] #446 placement audit: item clock \(String(format: "%.2f", nativeClockSeconds))s "
+            + "in item range \(String(format: "%.2f", host.seekableStart))..\(String(format: "%.2f", host.seekableEnd))s, "
+            + "shift \(String(format: "%.2f", playlistShiftSeconds))s + item offset "
+            + "\(String(format: "%.2f", liveItemAxisOffsetSeconds))s -> session \(String(format: "%.2f", currentTime))s; "
+            + "the producer holds "
+            + (producer.map { "\(String(format: "%.2f", $0.lowerBound))..\(String(format: "%.2f", $0.upperBound))s" } ?? "nothing it can state"),
+            category: .engine)
+    }
+
+    /// AE#446 round 4: what the producer holds right now, on the session axis, both ends.
+    ///
+    /// This is the range a rejoin is measured against, and it is deliberately not either of the two
+    /// the engine publishes. `LiveWindow.edgeTime` is a running maximum an outage freezes BELOW the
+    /// playhead that legitimately ran past it, and a freshly swapped item's `seekableEnd` is a range
+    /// it has not finished reporting at the readiness instant the rejoin replays in (measured on the
+    /// harness: 43.4 s while the place held was 71.4 s and the producer was cutting past 100 s). The
+    /// cache is the only party that is neither ahead of nor behind itself.
+    ///
+    /// nil where there is no cache to ask, which leaves the rejoin exactly where it was.
+    func residentLiveRangeSessionSeconds() -> ClosedRange<Double>? {
+        guard let session = nativeVideoSession,
+              let floorOutput = session.residentFloorOutputSeconds(),
+              let ceilingOutput = session.residentCeilingOutputSeconds() else { return nil }
+        let floor = presentationAxis.sourceSeconds(forItemSeconds: floorOutput)
+            ?? (floorOutput + playlistShiftSeconds)
+        let ceiling = presentationAxis.sourceSeconds(forItemSeconds: ceilingOutput)
+            ?? (ceilingOutput + playlistShiftSeconds)
+        guard ceiling >= floor else { return nil }
+        return floor...ceiling
+    }
+
     /// AE#442: the TARGETDURATION the live playlist is serving, nil on every path that serves none
     /// (remote HLS live, the software live path, and before the first playlist build).
     var liveTargetDurationSeconds: Double? {
@@ -161,6 +252,7 @@ extension AetherEngine {
         w.noteEdge(edgeSessionTime)
         w.notePlayhead(currentTime)
         w.noteResidentFloor(residentLiveFloorSessionSeconds())
+        auditLiveRejoinPlacement()
         liveWindow = w
         // AE#442: tick-to-tick advancement, not a running maximum: a backward DVR seek drops the
         // playhead, and the next advancing publish has to be able to record the new, larger distance.
@@ -174,6 +266,32 @@ extension AetherEngine {
         clock.behindLiveSeconds = w.behindLiveSeconds
     }
 
+    /// AE#446 round 4: who asked for a seek. The two differ in exactly two places, both about a live
+    /// session that advertises no DVR window: whether the seek is refused outright, and whether its
+    /// landing is measured against what the session offers or against what the item holds.
+    ///
+    /// A host that draws no scrubber can still have a place to come back to. Reported from a device:
+    /// the outage swap carried the held position, the replay went out through the public `seek(to:)`
+    /// like any host scrub, and the live-only guard refused it before it could land.
+    enum SeekOrigin: Sendable {
+        /// A scrub the host asked for, bound by the contract `seekableLiveRange` states.
+        case host
+        /// The engine coming back to a position it decided itself (the AE#446 outage swap, AE#442's
+        /// in-place recovery reload). Not a scrub, and not bound by the scrubber's contract.
+        case liveRejoin
+    }
+
+    /// AE#446 round 4: whether a live seek is refused for having no DVR window to land in.
+    ///
+    /// The refusal is the host contract's defence-in-depth: hosts hide the scrubber when
+    /// `seekableLiveRange` is nil, and one that does not must not put the item somewhere it cannot
+    /// play from. It says nothing about the engine's own rejoin, which picked its position out of
+    /// content the session itself served.
+    nonisolated static func liveSeekRefusedWithoutDVR(origin: SeekOrigin, windowSeconds: Double?) -> Bool {
+        guard origin == .host else { return false }
+        return windowSeconds == nil
+    }
+
     /// AE#446 round 3: where a live seek lands, decided from ONE sample of the item's own clock.
     ///
     /// The two halves used to read different clocks. The target was clamped against
@@ -199,14 +317,31 @@ extension AetherEngine {
         window: LiveWindow,
         itemEnd: Double,
         shift: Double,
-        axis: PresentationAxisMap
+        axis: PresentationAxisMap,
+        origin: SeekOrigin = .host,
+        residentRange: ClosedRange<Double>? = nil,
+        itemAxisOffset: Double = 0
     ) -> (sessionTarget: Double, clockTarget: Double) {
         // An item with no seekable range of its own yet has nothing to sample; the window's own edge
         // is then the only edge there is, and clamping against `shift` alone would collapse the range.
-        let edge = itemEnd > 0 ? itemEnd + shift : window.edgeTime
-        let sessionTarget = window.clamp(requested, edge: edge)
+        let edge = itemEnd > 0 ? itemEnd + shift + itemAxisOffset : window.edgeTime
+        // AE#446 round 4: a host scrub is bound by what the session ADVERTISES, and the engine's own
+        // rejoin by what the producer HOLDS. They are different questions, and at the moment a rejoin
+        // runs they have different answers: the advertised range is measured against an edge that is
+        // stale in one direction or the other (see `residentLiveRangeSessionSeconds`), while the
+        // carried position is content this same session cut and served, so the only thing that can
+        // disqualify it is eviction.
+        let sessionTarget: Double
+        if origin == .liveRejoin, let resident = residentRange {
+            sessionTarget = Swift.min(Swift.max(requested, resident.lowerBound), resident.upperBound)
+        } else {
+            sessionTarget = window.clamp(requested, edge: edge)
+        }
+        // AE#446 round 4: and then down onto the item's own axis, which for an item attached after
+        // the window slid begins above the session's zero. See `measureLiveItemAxisOffset`.
         let clockTarget = Swift.max(
-            0, axis.itemSeconds(forSourceSeconds: sessionTarget) ?? (sessionTarget - shift))
+            0, (axis.itemSeconds(forSourceSeconds: sessionTarget) ?? (sessionTarget - shift))
+               - itemAxisOffset)
         return (sessionTarget, clockTarget)
     }
 

Sources/AetherEngine/AetherEngine+Loading.swift

@@ -20,6 +20,8 @@ extension AetherEngine {
     func applyNativeHostClockTick(_ value: Double) {
         // nativeClockSeconds preserves the raw AVPlayer clock for onPlaylistShiftChanged to re-derive against.
         nativeClockSeconds = value
+        // AE#446 round 4: before anything folds, establish which axis this item's clock is even on.
+        measureLiveItemAxisOffset()
         // Newest seam at or before the raw clock wins: activates seams on forward play, re-applies pre-seam shift on backward DVR seeks.
         if let active = presentationAxis.shiftSeconds(atItemSeconds: value) {
             playlistShiftSeconds = active
@@ -28,12 +30,13 @@ extension AetherEngine {
             // AE#105: fold the disc's clip-0 STC base back out so the published playhead sits on the same
             // 0-based axis as the MPLS duration (origin 0 for normal/live -> no-op).
             clock.currentTime = PresentationAxis.display(
-                sourcePTS: value + playlistShiftSeconds,
+                sourcePTS: value + playlistShiftSeconds + liveItemAxisOffsetSeconds,
                 origin: displayOrigin(forShift: playlistShiftSeconds))
         }
         // Live edge must fold with the same playlistShiftSeconds as the playhead; opposite sign would make behindLiveSeconds meaningless.
         if isLive {
-            publishLiveWindow(edgeSessionTime: (nativeHost?.seekableEnd ?? 0) + playlistShiftSeconds)
+            publishLiveWindow(edgeSessionTime: (nativeHost?.seekableEnd ?? 0) + playlistShiftSeconds
+                              + liveItemAxisOffsetSeconds)
         }
     }
 
@@ -221,10 +224,10 @@ extension AetherEngine {
                 // #127: replay the latest host seek that arrived while the item was pre-ready.
                 // #178: not while still .loading (autostart paths hold .loading past readiness);
                 // replaying now would just re-stash. The state didSet resolves that case.
-                if ready, self.state != .loading, let pending = self.pendingPreReadySeekSeconds {
-                    self.pendingPreReadySeekSeconds = nil
-                    EngineLog.emit("[AetherEngine] replaying deferred pre-ready seek to \(String(format: "%.2f", pending))s (#127)", category: .engine)
-                    Task { @MainActor in await self.seek(to: pending) }
+                if ready, self.state != .loading, let pending = self.pendingPreReadySeek {
+                    self.pendingPreReadySeek = nil
+                    EngineLog.emit("[AetherEngine] replaying deferred pre-ready seek to \(String(format: "%.2f", pending.seconds))s (#127)", category: .engine)
+                    Task { @MainActor in await self.seek(to: pending.seconds, origin: pending.origin) }
                 }
             }
             .store(in: &cancellables)
@@ -1413,7 +1416,7 @@ extension AetherEngine {
         // way (HLSVideoEngine drops the resume anchor for a sequential origin), so leaving the
         // item on the EVENT edge default would start it mid-archive with no way back.
         if !isLive, loadedOptions.sequentialOrigin {
-            pendingPreReadySeekSeconds = 0.0
+            pendingPreReadySeek = PendingPreReadySeek(seconds: 0.0, origin: .host)
         }
         // AE#158: consume-and-reset so only the load() that armed the handover swaps in place; audio-switch
         // and recovery reloads keep their own contracts.