Sequential-origin sources: one unranged connection, native routing, and an append playlist with real durations

Sources/AetherEngine/AetherEngine+Loading.swift

@@ -126,7 +126,15 @@ extension AetherEngine {
         }
         duration
             .sink { [weak self] value in
-                if value > 0 { self?.duration = value }
+                guard let self else { return }
+                // A caller-declared duration outranks the host's: a sequential session's append
+                // playlist grows while it plays, so the item duration is the produced span, not
+                // the window length the host UI should scale its scrubber to.
+                if let declared = self.loadedOptions.declaredDurationSeconds, declared > 0 {
+                    self.duration = declared
+                } else if value > 0 {
+                    self.duration = value
+                }
             }
             .store(in: &cancellables)
         isReady
@@ -502,6 +510,8 @@ extension AetherEngine {
             // Caller-bounded probe budget (#68) for the fallback open / live reopen; the happy path reuses preopenedDemuxer.
             probesize: loadedOptions.probesize,
             maxAnalyzeDuration: loadedOptions.maxAnalyzeDuration,
+            sequentialOrigin: loadedOptions.sequentialOrigin,
+            declaredDurationSeconds: loadedOptions.declaredDurationSeconds,
             forwardBufferSegments: loadedOptions.forwardBufferSegments
         )
         // #240: the pump claims the source link through this gate while it is fetching, so the
@@ -1123,6 +1133,14 @@ extension AetherEngine {
         // forwardBufferDuration default (4 s): deep buffer lets AVPlayer race to the live edge and hit the transcode warm-up gap head-on (-12888); 4 s PACES consumption. Verified: 8 s worsened startup pause (8-10 s vs ~1 s).
         // Live REJOIN: skip initial seek so AVPlayer picks edge-minus-holdback instead; seek-to-0 against the re-served backlog wedged the reloaded item in waitingToPlay (device repro: tvOS 26, Jellyfin stream.ts). See LiveReloadPolicy.
         lastNativeVideoStartPosition = startPosition ?? 0
+        // Sequential append playlist: AVPlayer treats the growing playlist as an EVENT and
+        // defaults to edge-minus-holdback (~6 s in on a fresh session, more once the producer
+        // has raced ahead). The load-time seek to 0 fires before readyToPlay and the item
+        // re-anchors to the edge default afterwards, so queue a post-readiness seek through
+        // the #127 replay instead - every segment stays retained, so 0 is always reachable.
+        if !isLive, loadedOptions.sequentialOrigin, startPosition == nil {
+            pendingPreReadySeekSeconds = 0.0
+        }
         // 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.
         let inPlaceHandover = pendingInPlaceItemHandover
@@ -1292,18 +1310,20 @@ extension AetherEngine {
         // Capture the caller's probe budget (#68) before the detach: loadedOptions is @MainActor-isolated and unreachable inside the closure. Only used on the fallback open (probe absent).
         let probesize = loadedOptions.probesize
         let maxAnalyzeDuration = loadedOptions.maxAnalyzeDuration
+        let sequentialOrigin = loadedOptions.sequentialOrigin
+        let declaredDuration = loadedOptions.declaredDurationSeconds
         // Built on the main actor, captured into the detach: surfaces source stall/reconnect to playbackPhase (#85).
         let networkPhaseSink: @Sendable (ReaderNetworkPhase) -> Void = { [weak self] phase in
             Task { @MainActor in self?.setReaderNetworkPhase(phase) }
         }
         try await Task.detached(priority: .userInitiated) {
-            [host, preopenedDemuxer, url, sourceHTTPHeaders, isLive, dvrWindowSeconds, probesize, maxAnalyzeDuration, networkPhaseSink] in
+            [host, preopenedDemuxer, url, sourceHTTPHeaders, isLive, dvrWindowSeconds, probesize, maxAnalyzeDuration, sequentialOrigin, declaredDuration, networkPhaseSink] in
             let dem: Demuxer
             if let pre = preopenedDemuxer {
                 dem = pre
             } else {
                 dem = Demuxer()
-                try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration), isLive: isLive)
+                try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration).withSequentialOrigin(sequentialOrigin, declaredDuration: declaredDuration), isLive: isLive)
             }
             dem.onNetworkPhaseChanged = networkPhaseSink
             try await host.load(
@@ -1359,18 +1379,20 @@ extension AetherEngine {
         // Caller's probe budget (#68) captured before the detach; only used on the fallback open (probe absent).
         let probesize = loadedOptions.probesize
         let maxAnalyzeDuration = loadedOptions.maxAnalyzeDuration
+        let sequentialOrigin = loadedOptions.sequentialOrigin
+        let declaredDuration = loadedOptions.declaredDurationSeconds
         // Built on the main actor, captured into the detach: surfaces source stall/reconnect to playbackPhase (#85).
         let networkPhaseSink: @Sendable (ReaderNetworkPhase) -> Void = { [weak self] phase in
             Task { @MainActor in self?.setReaderNetworkPhase(phase) }
         }
         try await Task.detached(priority: .userInitiated) {
-            [host, preopenedDemuxer, url, sourceHTTPHeaders, probesize, maxAnalyzeDuration, networkPhaseSink] in
+            [host, preopenedDemuxer, url, sourceHTTPHeaders, probesize, maxAnalyzeDuration, sequentialOrigin, declaredDuration, networkPhaseSink] in
             let dem: Demuxer
             if let pre = preopenedDemuxer {
                 dem = pre
             } else {
                 dem = Demuxer()
-                try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration))
+                try dem.open(url: url, extraHeaders: sourceHTTPHeaders, profile: .playback.withProbeBudget(probesize: probesize, maxAnalyzeDuration: maxAnalyzeDuration).withSequentialOrigin(sequentialOrigin, declaredDuration: declaredDuration))
             }
             dem.onNetworkPhaseChanged = networkPhaseSink
             try await host.load(

Sources/AetherEngine/AetherEngine.swift

@@ -2568,6 +2568,8 @@ public final class AetherEngine: ObservableObject {
                 // demuxer is reused as the session demuxer, so the cap lands on the open that actually pays it.
                 let probeProfile = DemuxerOpenProfile.playback.withProbeBudget(
                     probesize: options.probesize, maxAnalyzeDuration: options.maxAnalyzeDuration)
+                    .withSequentialOrigin(options.sequentialOrigin,
+                                          declaredDuration: options.declaredDurationSeconds)
                 switch source {
                 case .url(let u):
                     // isLive configures the AVIOReader for endless-feed mode; must be set at open time because
@@ -2996,9 +2998,27 @@ public final class AetherEngine: ObservableObject {
         // the forward-only streaming reader (#126: unknown-length HTTP MP4 produced zero segments).
         // Live sources are exempt: the live producer never seeks backward, scrub previews come from
         // the DVR segment cache, and audio-switch is already no-op for forward-only sources.
+        // An explicit sequential-origin declaration is exempt for the same reasons a live session
+        // is: the producer reads the archive linearly from byte 0, the duration is caller-declared
+        // (no tail estimate), the segment plan is the uniform-stride fallback, restarts/revives are
+        // gated off, and the cue prewarm fails fast on the non-seekable pb. Forcing those archives
+        // onto the software path traded AVPlayer's buffering and hardware decode for nothing
+        // (device trace: a 720p50 timeshift archive played clean on the native path and visibly
+        // stuttered on the software one). Declared-interlaced archives still route software via
+        // the field-order policy above - the #232 refute probe cannot run without a rewind.
         if !probe.isSourceSeekable && !options.isLive {
-            useSoftwarePath = true
-            EngineLog.emit("[AetherEngine] source is forward-only, forcing software path", category: .engine)
+            if options.sequentialOrigin {
+                if !useSoftwarePath {
+                    EngineLog.emit(
+                        "[AetherEngine] sequential origin keeps the native path (linear read, "
+                        + "declared duration, seeks unavailable)",
+                        category: .engine
+                    )
+                }
+            } else {
+                useSoftwarePath = true
+                EngineLog.emit("[AetherEngine] source is forward-only, forcing software path", category: .engine)
+            }
         }
         // TEST-ONLY: forces SW path for aetherctl live --sw; unset in shipping builds.
         if Self.forceSoftwarePathForTesting {

Sources/AetherEngine/Demuxer/AVIOReader.swift

@@ -190,6 +190,12 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
     private var streamEnded = false
     private let streamLock = NSLock()
     private let streamDataReady = DispatchSemaphore(value: 0)
+    /// Advisory body length from the streaming response's Content-Length (-1 unknown). Only the
+    /// sequential-origin path reads it: a connection that ends short of it (or stalls out) is a
+    /// LOST source, not end-of-media, and must surface as a read error rather than EOF - the
+    /// consumer treats EOF as "played to the end" and deliberately never retries it. Guarded by
+    /// `streamLock`.
+    private var streamExpectedBytes: Int64 = -1
 
     // MARK: - Persistent Mode (single forward-streaming connection, playback path)
 
@@ -497,6 +503,13 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
     /// reports EIO (-5) instead of EOF when the reconnect cap is hit.
     let isLive: Bool
 
+    /// `LoadOptions.sequentialOrigin` (via `DemuxerOpenProfile.avioSequentialOnly`): the origin
+    /// fabricates range answers, so only byte 0 is addressable. `open()` routes straight onto the
+    /// forward-only streaming mode (one unranged GET) and never issues a ranged request - no tail
+    /// prefetch, no optimistic persistent open, no size probe, no detours. `fileSize` stays -1 by
+    /// construction, which keeps `isStreaming` true and the pb non-seekable (#126 block below).
+    let sequentialOnly: Bool
+
     /// Detour cache is VOD-only: live feeds have no meaningful random access and a
     /// non-authoritative size, so they stay on the unchanged reconnect path.
     private var detourEligible: Bool { !isLive && fileSize > 0 }
@@ -541,13 +554,14 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
     /// origin at once and the line used to name none of them.
     private let label: String
 
-    init(url: URL, extraHeaders: [String: String] = [:], label: String = "source", chunkSize: Int = 4 * 1024 * 1024, prefetchEnabled: Bool = true, isLive: Bool = false, chunkRequestTimeout: TimeInterval = 35, chunkMaxRetries: Int = 3, boundedInitialFetch: Int64? = nil, connStallTimeout: TimeInterval = AVIOReader.connStallTimeoutDefault, windowHighWater: Int? = nil) {
+    init(url: URL, extraHeaders: [String: String] = [:], label: String = "source", chunkSize: Int = 4 * 1024 * 1024, prefetchEnabled: Bool = true, isLive: Bool = false, chunkRequestTimeout: TimeInterval = 35, chunkMaxRetries: Int = 3, boundedInitialFetch: Int64? = nil, sequentialOnly: Bool = false, connStallTimeout: TimeInterval = AVIOReader.connStallTimeoutDefault, windowHighWater: Int? = nil) {
         self.url = url
         self.label = label
         self.extraHeaders = extraHeaders
         self.chunkSize = chunkSize
         self.prefetchEnabled = prefetchEnabled
         self.isLive = isLive
+        self.sequentialOnly = sequentialOnly
         self.chunkRequestTimeout = chunkRequestTimeout
         self.chunkMaxRetries = max(1, chunkMaxRetries)
         self.boundedInitialFetch = boundedInitialFetch.map { max(1, $0) }
@@ -603,7 +617,17 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
 
         context = ctx
 
-        if prefetchEnabled {
+        if sequentialOnly {
+            // Sequential origin: the only trustworthy request shape is one unranged GET from
+            // byte 0. Everything the branches below would issue on top - the suffix tail
+            // prefetch, the optimistic `Range: bytes=0-` persistent open, the dedicated size
+            // probe - is a ranged request the origin would answer with fabricated positions,
+            // so none of it runs. `fileSize` stays -1: `isStreaming` routes read()/seek()
+            // onto the forward-only streaming path and the #126 block below marks the pb
+            // non-seekable.
+            startStreamingDownload()
+            _ = streamDataReady.wait(timeout: .now() + .seconds(15))
+        } else if prefetchEnabled {
             // #281: the parse seeks that follow this open are what the retained head exists for.
             winCond.lock()
             openPhaseActive = true
@@ -1098,7 +1122,27 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
             }
         }
 
-        return totalRead > 0 ? Int32(totalRead) : FFmpegErr.eof
+        if totalRead > 0 { return Int32(totalRead) }
+        if sequentialOnly {
+            streamLock.lock()
+            let ended = streamEnded
+            let received = streamBytesRead + Int64(streamBuffer.count)
+            let expected = streamExpectedBytes
+            streamLock.unlock()
+            // A sequential origin cannot be resumed at an offset, so a stalled-out wait or a
+            // body that ended short of its advisory length is a LOST source: report EIO so the
+            // pump exits on a read error the session can surface. EOF here would read as
+            // end-of-media, which the consumer deliberately never retries.
+            if !ended || (expected > 0 && received < expected) {
+                EngineLog.emit(
+                    "[AVIOReader] sequential stream \(ended ? "ended short" : "stalled out") at "
+                    + "\(received)\(expected > 0 ? "/\(expected)" : "") bytes; reporting EIO",
+                    category: .demux
+                )
+                return AVERROR_EIO_VALUE
+            }
+        }
+        return FFmpegErr.eof
     }
 
     // MARK: - Persistent Read (single forward-streaming connection)
@@ -2382,7 +2426,19 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
 
         let semaphore = DispatchSemaphore(value: 0)
 
-        let delegate = StreamingDelegate { [weak self] data in
+        let delegate = StreamingDelegate(
+            extraHeaders: extraHeaders,
+            onResponse: { [weak self] response in
+                // Advisory length for the sequential-origin EOF/EIO distinction; -1 (chunked /
+                // unknown) leaves the clean-end path as the only EOF source.
+                guard let self, self.sequentialOnly else { return }
+                let expected = response.expectedContentLength
+                guard expected > 0 else { return }
+                self.streamLock.lock()
+                self.streamExpectedBytes = expected
+                self.streamLock.unlock()
+            }
+        ) { [weak self] data in
             guard let self, !self.isClosed else { return }
             self.streamLock.lock()
             self.streamBuffer.append(data)
@@ -3206,12 +3262,46 @@ private final class ChunkFetchDelegate: NSObject, URLSessionDataDelegate, @unche
 private final class StreamingDelegate: NSObject, URLSessionDataDelegate {
     let onData: @Sendable (Data) -> Void
     let onComplete: @Sendable () -> Void
+    /// Response hook (advisory Content-Length capture on the sequential-origin path).
+    let onResponse: (@Sendable (URLResponse) -> Void)?
+    /// Re-applied across cross-host redirects like every other delegate in this file;
+    /// IPTV origins routinely 302 twice (portal -> panel -> archive host) and the final
+    /// host must still see the caller's User-Agent / auth headers.
+    let extraHeaders: [String: String]
 
-    init(onData: @escaping @Sendable (Data) -> Void, onComplete: @escaping @Sendable () -> Void) {
+    init(
+        extraHeaders: [String: String] = [:],
+        onResponse: (@Sendable (URLResponse) -> Void)? = nil,
+        onData: @escaping @Sendable (Data) -> Void,
+        onComplete: @escaping @Sendable () -> Void
+    ) {
+        self.extraHeaders = extraHeaders
+        self.onResponse = onResponse
         self.onData = onData
         self.onComplete = onComplete
     }
 
+    func urlSession(
+        _ session: URLSession,
+        task: URLSessionTask,
+        willPerformHTTPRedirection response: HTTPURLResponse,
+        newRequest request: URLRequest,
+        completionHandler: @escaping (URLRequest?) -> Void
+    ) {
+        completionHandler(redirectPreservingHeaders(
+            task: task, newRequest: request, extraHeaders: extraHeaders))
+    }
+
+    func urlSession(
+        _ session: URLSession,
+        dataTask: URLSessionDataTask,
+        didReceive response: URLResponse,
+        completionHandler: @escaping (URLSession.ResponseDisposition) -> Void
+    ) {
+        onResponse?(response)
+        completionHandler(.allow)
+    }
+
     func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
         onData(data)
     }