fix(sw): stop repositioning a forward-only source on the software path

CHANGELOG.md

@@ -27,6 +27,7 @@ the public-API contract.
 
 ### Fixed
 
+- A forward-only source (one chunked response with no ranges, a sequential origin, a one-shot custom reader) is no longer repositioned on the software video path, which ended the session each time: the reposition flushed the demuxer's queued packets and the reader could not rewind to read them again (`partial file`). A start position within 2 s is now reached by decoding forward from the stream origin, and a later one is dropped so playback begins at the origin; `seek(to:)` is rejected with the new `SeekEvent.Rejection.sourceNotSeekable` and playback continues (a new public case: a host that switches exhaustively over `SeekEvent.Rejection` must add an arm for it); and `play()` no longer rewinds such a source to zero, which it did whenever the container reported a duration shorter than the stream, as a fragmented MP4 served as one response does. Seekable and live sources are unchanged. The audio-only host still repositions a forward-only source.
 - The software video decoder no longer runs more than 16 frame threads. It used one per core, and each frame thread holds back one decoded frame, so a 32-core Mac waited for 31 frames before showing the first one after a load or seek (about 3 s at 10 fps). FFmpeg also warns above 16 threads. Hosts with 16 or fewer cores keep their current thread count.
 - A paused video no longer starts playing by itself. When the player item died while paused (`failedToPlayToEndTime`), the recovery reload bypassed the pause guard and called `play()` on the fresh item. The reload now keeps a pause made before the item died, whether it came through the engine, AVKit, Control Center or PiP, and mounts the item paused at the same position.
 - A dead item's recovery no longer restarts the title from where the session was first opened. When AVPlayer refused the recovery item's master (`-11868`), the media fallback reloaded at the first mount's start position, so a title opened from its beginning restarted at 0:00. The fallback now reloads where the refused item was placed. Upstream #621.

Sources/AetherEngine/AetherEngine.swift

@@ -3210,13 +3210,25 @@ public final class AetherEngine: ObservableObject {
     /// `AVPlayer.play()` at end-of-media is a no-op, freezing the button. `.ended` is deliberately
     /// excluded: it is terminal (#63), the host revives it by reloading, and a play press racing the end
     /// card must not silently restart the finished session.
+    ///
+    /// A source that cannot be repositioned never rewinds. Its container can report a duration far
+    /// short of the stream (a fragmented MP4 served as one response reports its first fragment, a
+    /// few seconds), which reads as "parked at the end" for the rest of the session; the rewind is a
+    /// seek the source cannot serve, and it ended playback on the first play after a pause.
     nonisolated static func shouldRewindBeforePlay(
-        state: PlaybackState, currentTime: Double, duration: Double, isLive: Bool
+        state: PlaybackState, currentTime: Double, duration: Double, isLive: Bool,
+        sourceCanReposition: Bool = true
     ) -> Bool {
-        guard state != .ended else { return false }
+        guard state != .ended, sourceCanReposition else { return false }
         return isAtEndOfMedia(currentTime: currentTime, duration: duration, isLive: isLive)
     }
 
+    /// Whether a seek is refused because the loaded source cannot be repositioned. Live sources
+    /// keep their own (DVR window) rules.
+    nonisolated static func seekRefusedForForwardOnlySource(isLive: Bool, sourceCanReposition: Bool) -> Bool {
+        !isLive && !sourceCanReposition
+    }
+
     #if DEBUG
     /// Test-only override for the session's restart-in-flight signal (#93 residual deferral tests).
     var testHookRestartInFlightOverride: Bool? = nil
@@ -4828,7 +4840,8 @@ public final class AetherEngine: ObservableObject {
         // first, then resume. `.ended` is excluded (see shouldRewindBeforePlay): it stays terminal so a
         // play press racing the host end card does not silently revive a finished session (#63).
         if Self.shouldRewindBeforePlay(
-            state: state, currentTime: currentTime, duration: duration, isLive: isLive
+            state: state, currentTime: currentTime, duration: duration, isLive: isLive,
+            sourceCanReposition: softwareHost?.sourceCanReposition ?? true
         ) {
             Task { @MainActor in
                 await self.seek(to: 0)
@@ -5039,6 +5052,24 @@ public final class AetherEngine: ObservableObject {
                 return
             }
         }
+        // A forward-only source on the software path has no other position to offer, and its
+        // container can report a duration far short of the stream. Forwarding the seek would
+        // clamp the target to that duration, publish it as the position and park the state, all
+        // for a reposition the host has to refuse. Reject it here and leave the session playing.
+        if Self.seekRefusedForForwardOnlySource(
+            isLive: isLive, sourceCanReposition: softwareHost?.sourceCanReposition ?? true
+        ) {
+            EngineLog.emit("[AetherEngine] seek(to:\(seconds)) ignored: the source is forward-only", category: .engine)
+            // A seek stashed during load is replayed here, and its ticket is still open. Close
+            // that ticket with the rejection, or `isSeeking` stays latched over a session that
+            // plays on. A seek with no stash gets the standalone event instead, never both.
+            if deferredSeekInFlight || deferredSeekTicket != nil {
+                endDeferredSeek(.rejected(.sourceNotSeekable))
+            } else {
+                emitSeekRejected(.sourceNotSeekable, target: seconds)
+            }
+            return
+        }
         // #127: pre-ready native item (background-teardown reload, cold start): forwarding the seek now
         // would clamp to 0 against empty seekable ranges and replace load()'s pending startPosition seek.
         // Stash the latest target (publishing it optimistically so scrub UI follows) and replay at readiness.

Sources/AetherEngine/Native/SoftwarePlaybackHost.swift

@@ -706,6 +706,42 @@ final class SoftwarePlaybackHost {
         ) == .stop
     }
 
+    // MARK: - Start position
+
+    /// How `load` honours a start position.
+    enum StartPlan: Equatable {
+        /// No start to honour: play from the stream origin.
+        case fromOrigin
+        /// Reposition the source to the start.
+        case reposition
+        /// Leave the source where it is and decode forward to the start, dropping what precedes it.
+        case decodeForward
+        /// The start cannot be reached; play from the stream origin.
+        case dropStart
+    }
+
+    /// Furthest start a forward-only source is decoded forward to: the clock anchor's own tolerance.
+    ///
+    /// With no reposition the first audio sample is at the stream origin, and the clock is armed by
+    /// `SWClockAnchorPolicy` from that sample. Inside its tolerance the clock keeps the load anchor,
+    /// so it starts at `start`, the audio before it is late and discarded, and the video before it
+    /// is dropped by the skip threshold. Past the tolerance the policy re-anchors at the first
+    /// sample instead: the pre-roll audio would play with no picture and the position would snap
+    /// back to the origin. Such a start is dropped, and the pre-roll plays with its picture.
+    nonisolated static let forwardOnlyStartSkipLimitSeconds: Double = SWClockAnchorPolicy.toleranceSeconds
+
+    /// A forward-only source (one chunked response with no ranges, a sequential origin, a one-shot
+    /// custom reader) cannot be repositioned, so its start is reached by decoding or not at all.
+    nonisolated static func startPlan(startPosition: Double?, isSourceSeekable: Bool, isLive: Bool) -> StartPlan {
+        guard let start = startPosition, start.isFinite, start > 0 else { return .fromOrigin }
+        // Live keeps what it had: this policy is about finite forward-only sources.
+        if isSourceSeekable || isLive { return .reposition }
+        return start <= forwardOnlyStartSkipLimitSeconds ? .decodeForward : .dropStart
+    }
+
+    /// False when the loaded source cannot be repositioned (see `startPlan`). True with no source.
+    var sourceCanReposition: Bool { demuxer?.isSourceSeekable ?? true }
+
     // MARK: - Load
 
     func load(
@@ -914,14 +950,37 @@ final class SoftwarePlaybackHost {
         // Reset the live feeder state for the new session.
         resetFeederState()
 
-        if let start = startPosition, start > 0 {
-            // #254: same off-main, deadline-bounded reposition the transport seek uses. A resume into a
-            // remote source that has to scan for its landing would otherwise block the main thread here.
-            _ = await dem.seekBounded(to: start, timeout: Self.seekBudgetSeconds, on: seekQueue)
-            // This is load()'s only suspension point, so it is also the only place a stop() can land
-            // mid-load. Arming the clock and publishing isReady on a session already torn down would
-            // hand the engine a host it has stopped.
-            guard !stopRequested else { return }
+        let startPlan = Self.startPlan(startPosition: startPosition,
+                                       isSourceSeekable: dem.isSourceSeekable, isLive: isLive)
+        if startPlan == .dropStart, let start = startPosition {
+            EngineLog.emit(
+                "[SWHost] start at \(String(format: "%.2f", start))s dropped: the source is forward-only "
+                + "and the start is past the \(String(format: "%.0f", Self.forwardOnlyStartSkipLimitSeconds))s "
+                + "the clock anchor holds without a reposition; playing from the stream origin",
+                category: .swPlayback
+            )
+        }
+        if let start = startPosition, startPlan == .reposition || startPlan == .decodeForward {
+            if startPlan == .reposition {
+                // #254: same off-main, deadline-bounded reposition the transport seek uses. A resume into a
+                // remote source that has to scan for its landing would otherwise block the main thread here.
+                _ = await dem.seekBounded(to: start, timeout: Self.seekBudgetSeconds, on: seekQueue)
+                // This is load()'s only suspension point, so it is also the only place a stop() can land
+                // mid-load. Arming the clock and publishing isReady on a session already torn down would
+                // hand the engine a host it has stopped.
+                guard !stopRequested else { return }
+            } else {
+                // The read position already stands at the stream origin and the packets the probe read
+                // are still queued. A seek here flushes them, and a forward-only reader cannot rewind to
+                // read them again (`partial file`, then the read loop ends). The thresholds armed below
+                // drop the video that precedes `start`, and the clock arms at `start` because it is
+                // inside the anchor tolerance (see `forwardOnlyStartSkipLimitSeconds`).
+                EngineLog.emit(
+                    "[SWHost] start at \(String(format: "%.2f", start))s reached by decoding from the stream "
+                    + "origin: the source is forward-only, so it is not repositioned",
+                    category: .swPlayback
+                )
+            }
             // Mirror seek() skip-PTS + clock alignment so demux drops pre-keyframe frames and synchronizer starts at the resume offset.
             let startTime = CMTime(seconds: start, preferredTimescale: 90000)
             videoDecoder.skipUntilPTS = startTime
@@ -1229,6 +1288,17 @@ final class SoftwarePlaybackHost {
     func seek(to seconds: Double) async -> Demuxer.RepositionOutcome {
         guard !stopRequested else { return .superseded }
         guard let dem = demuxer else { return .stalled }
+        // A forward-only source has no other position to offer. Attempting the reposition flushes
+        // both decoders and the demuxer's queued packets, and the reader cannot rewind to read them
+        // again, so the session dies on the next read. Refuse before anything is disturbed; the
+        // session keeps playing where it is. Live seeks stay on their own (DVR ring) path below.
+        if !isLive, !dem.isSourceSeekable {
+            EngineLog.emit(
+                "[SWHost] seek to \(String(format: "%.2f", seconds))s refused: the source is forward-only",
+                category: .swPlayback
+            )
+            return .stalled
+        }
         // Stop loop + bump generation to invalidate in-flight packets. Captured right after, so the
         // reposition can tell on `seekQueue` whether a newer seek has already taken over.
         bumpSeekGeneration()

Sources/AetherEngine/SeekEvent.swift

@@ -41,6 +41,9 @@ public struct SeekEvent: Sendable, Equatable {
         case noActiveSession
         /// Live source without a DVR window; there is no seekable range to land in.
         case liveWithoutDVR
+        /// The source is forward-only (one response with no ranges, or a one-shot reader): it has
+        /// no other position to offer. Playback continues where it is.
+        case sourceNotSeekable
     }
 
     public enum Outcome: Sendable, Equatable {

Tests/AetherEngineTests/SeekToEndOfMediaTests.swift

@@ -81,6 +81,20 @@ struct SeekToEndOfMediaTests {
         #expect(!AetherEngine.shouldRewindBeforePlay(state: .paused, currentTime: 100, duration: 100, isLive: true))
     }
 
+    @Test("A source that cannot be repositioned never rewinds, whatever its duration claims")
+    func forwardOnlySourceNeverRewinds() {
+        // A fragmented MP4 served as one response reports its first fragment as the duration:
+        // 3.6 s for a film with over an hour left. Twenty minutes in, that reads as parked at
+        // the end, and the rewind is a seek this source cannot serve.
+        #expect(!AetherEngine.shouldRewindBeforePlay(
+            state: .paused, currentTime: 1_196.7, duration: 3.6, isLive: false, sourceCanReposition: false))
+        #expect(!AetherEngine.shouldRewindBeforePlay(
+            state: .paused, currentTime: 100, duration: 100, isLive: false, sourceCanReposition: false))
+        // The same playhead on a source that can rewind still does.
+        #expect(AetherEngine.shouldRewindBeforePlay(
+            state: .paused, currentTime: 100, duration: 100, isLive: false, sourceCanReposition: true))
+    }
+
     // MARK: - Integration: real seek() / play() paths (no hosts)
 
     @MainActor

Tests/AetherEngineTests/SoftwareHostForwardOnlyStartTests.swift

@@ -0,0 +1,68 @@
+import Testing
+@testable import AetherEngine
+
+/// How the software host honours a start position on a source it cannot reposition.
+///
+/// A forward-only source (one chunked response with no ranges, a sequential origin, a one-shot
+/// custom reader) has exactly one read position. Seeking it on load flushed the packets the probe
+/// had read; the reader could not rewind to read them again, libavformat reported `partial file`
+/// for the first sample, and the session reached `ready` and then never played. A host resuming a
+/// server-seeked progressive stream hit this on every resume, because the server starts the
+/// stream at the preceding keyframe and asks the player to skip the pre-roll.
+@Suite("Software host start position on a forward-only source")
+struct SoftwareHostForwardOnlyStartTests {
+    private typealias Host = SoftwarePlaybackHost
+
+    @Test("No start, a zero start and a non-finite start all play from the origin")
+    func noStartPlaysFromOrigin() {
+        for start in [nil, 0, -3, Double.nan, Double.infinity] as [Double?] {
+            for seekable in [true, false] {
+                #expect(Host.startPlan(startPosition: start, isSourceSeekable: seekable, isLive: false) == .fromOrigin)
+            }
+        }
+    }
+
+    @Test("A seekable source is repositioned to its start, however far")
+    func seekableSourceRepositions() {
+        #expect(Host.startPlan(startPosition: 2.8, isSourceSeekable: true, isLive: false) == .reposition)
+        #expect(Host.startPlan(startPosition: 5_000, isSourceSeekable: true, isLive: false) == .reposition)
+    }
+
+    @Test("A forward-only source decodes forward to a start inside the limit and is never repositioned")
+    func forwardOnlySourceDecodesForward() {
+        #expect(Host.startPlan(startPosition: 0.8, isSourceSeekable: false, isLive: false) == .decodeForward)
+        #expect(Host.startPlan(startPosition: Host.forwardOnlyStartSkipLimitSeconds,
+                               isSourceSeekable: false, isLive: false) == .decodeForward)
+    }
+
+    @Test("The decode-forward limit is the clock anchor's tolerance")
+    func limitIsTheClockAnchorTolerance() {
+        // Past the tolerance the clock re-anchors at the first sample, which without a reposition
+        // is the stream origin: pre-roll audio with no picture. The two must move together.
+        let limit = Host.forwardOnlyStartSkipLimitSeconds
+        #expect(limit == SWClockAnchorPolicy.toleranceSeconds)
+        #expect(SWClockAnchorPolicy.resolve(initialSeconds: limit, firstSampleSeconds: 0).anchorSeconds == limit)
+        #expect(SWClockAnchorPolicy.resolve(initialSeconds: limit + 0.1, firstSampleSeconds: 0).anchorSeconds == 0)
+    }
+
+    @Test("A forward-only source drops a start the clock anchor would not hold")
+    func forwardOnlySourceDropsAFarStart() {
+        #expect(Host.startPlan(startPosition: Host.forwardOnlyStartSkipLimitSeconds + 0.1,
+                               isSourceSeekable: false, isLive: false) == .dropStart)
+        #expect(Host.startPlan(startPosition: 1_004.87, isSourceSeekable: false, isLive: false) == .dropStart)
+    }
+
+    @Test("A seek is refused on a forward-only source, and never on a live or seekable one")
+    func seekRefusal() {
+        #expect(AetherEngine.seekRefusedForForwardOnlySource(isLive: false, sourceCanReposition: false))
+        #expect(!AetherEngine.seekRefusedForForwardOnlySource(isLive: false, sourceCanReposition: true))
+        #expect(!AetherEngine.seekRefusedForForwardOnlySource(isLive: true, sourceCanReposition: false))
+        #expect(!AetherEngine.seekRefusedForForwardOnlySource(isLive: true, sourceCanReposition: true))
+    }
+
+    @Test("A live source keeps its existing start handling")
+    func liveKeepsItsStartHandling() {
+        #expect(Host.startPlan(startPosition: 40, isSourceSeekable: false, isLive: true) == .reposition)
+        #expect(Host.startPlan(startPosition: 40, isSourceSeekable: true, isLive: true) == .reposition)
+    }
+}