fix(audio): preserve explicit pause authority across clock writes

CHANGELOG.md

@@ -152,6 +152,12 @@ the public-API contract.
 
 ### Fixed
 
+- **Explicit software-host pause wins over late decoder, rebuffer and preroll clock starts.** The output
+  latches transport authority under the same lock as every clock write, including before the first frame
+  arms the clock. A paused seek still updates its anchor at zero rate, and explicit play restores normal
+  preroll. A play that arrives before the feeder publishes its armed latch preserves the pending initial
+  clock start. Stopping an already-paused output retires and flushes it without resetting its clock.
+
 - **A native VOD loaded with a start position no longer reports an earlier time once it plays (AE#881, thanks
   @martijnklomp).** AVPlayer finishes the mount seek the moment it is issued, before the item is ready, and that
   completion released the clock hold. The periodic observer then published the keyframe or segment start the
@@ -180,6 +186,13 @@ the public-API contract.
   the bound port while `stop()` reset it under the lock. Host-only redirects now take the same lock,
   and the port stays private to the server. Explicit `--redirect-port` targets are unchanged.
 
+- **A software playback clock stays stopped when first-frame arming overlaps a pause.**
+  The feeder's post-arm rate correction could restart the synchronizer after `pause()` stopped it
+  but before the host published its paused intent, leaving the clock running behind parked decode
+  loops and causing video to catch up rapidly on resume. The correction now runs on the main actor
+  with transport changes, ignores retired outputs and seek generations, and yields to newer clock
+  commands, including rebuffer pauses. Pending pauses and end-of-media tail playback are preserved.
+
 - **Source failures keep their cause and request ownership.** Initial live probes preserve HTTP,
   URL and TLS failures without changing the live-start `liveSourceUnavailable` kind (HTTP and
   certificate classifications stay specific). A transport failure at open (a reset or a timeout from

Sources/AetherEngine/Audio/AudioOutput.swift

@@ -16,9 +16,22 @@ final class AudioOutput: @unchecked Sendable {
     let timeline = RendererTimeline()
 
     private let lock = NSLock()
+    /// Explicit host pause owns every clock write, including stale feeder work that carries no
+    /// command revision. Buffering pauses leave this intent alone. Guarded by `lock`.
+    private var playbackEnabled = true
     @MainActor private var attachedVideoLayer: AVSampleBufferDisplayLayer?
     @MainActor private var videoLayerRemovalTask: Task<Void, Never>?
 
+    /// External clock commands retire queued arm corrections. Compared under the same lock as
+    /// the rate change so a later feeder rebuffer pause cannot be overwritten by a stale correction.
+    /// Preroll completion belongs to its initiating command and does not advance this revision.
+    private var _clockCommandRevision: UInt64 = 0
+    var clockCommandRevision: UInt64 {
+        lock.lock()
+        defer { lock.unlock() }
+        return _clockCommandRevision
+    }
+
     /// AE#464: the host's audio presentation offset, applied to every buffer on its way into the
     /// renderer. Guarded because the host writes it from the main actor while the demux thread reads
     /// it in `enqueue`.
@@ -196,42 +209,78 @@ final class AudioOutput: @unchecked Sendable {
         set { renderer.volume = newValue }
     }
 
-    /// Set playback speed (0.5-2.0). Hosts own rate state (lastRate/pausedByHost); this object is stateless about it.
-    func setRate(_ rate: Float) {
+    /// Update transport authority without starting an unanchored clock. A disabled output can
+    /// still seek at rate zero; only an explicit play enables positive rates again.
+    func setPlaybackEnabled(_ enabled: Bool) {
+        lock.lock()
+        playbackEnabled = enabled
+        // Enabling owns no clock write. A feeder may already hold the revision for an arm
+        // whose latch play() has not seen yet; that correction still owes the initial start.
+        guard !enabled else { lock.unlock(); return }
+        _clockCommandRevision &+= 1
+        endPreroll()
+        let at = synchronizer.currentTime()
+        let sourceSeconds = timeline.sourceSeconds(forRenderer: at.seconds)
+        if synchronizer.rate != 0 {
+            synchronizer.setRate(0, time: at)
+        }
+        lock.unlock()
+        EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", sourceSeconds)) (host transport disabled)",
+                       category: .swPlayback)
+    }
+
+    /// Set playback speed (0.5-2.0). Hosts retain their preferred speed; explicit pause wins.
+    /// A queued correction supplies `ifRevision`; normal transport commands omit it.
+    func setRate(_ rate: Float, ifRevision revision: UInt64? = nil) {
         // A host resumes right after it arms the clock; while a preroll waits, that rate is the one
         // the preroll starts at, not a start of its own.
-        if rate > 0, updatePrerollRate(rate) { return }
+        if rate > 0, updatePrerollRate(rate, ifRevision: revision) { return }
         // A clock that stands still starts through the preroll on a session that has one: a seek
         // landed while paused anchors at rate 0 and never asks the gate, so the play that follows
         // would otherwise start it with nothing queued.
         if rate > 0, synchronizer.rate == 0, videoPrerollSatisfied?(currentTime.seconds) != nil {
-            seekClock(to: currentTime, rate: rate)
+            seekClock(to: currentTime, rate: rate, ifRevision: revision)
+            return
+        }
+        lock.lock()
+        if (rate != 0 && !playbackEnabled) || (revision != nil && revision != _clockCommandRevision) {
+            lock.unlock()
             return
         }
-        cancelPreroll()
+        _clockCommandRevision &+= 1
+        endPreroll()
         let at = synchronizer.currentTime()
-        EngineLog.emit("[AudioOutput] setRate \(rate) at t=\(String(format: "%.3f", timeline.sourceSeconds(forRenderer: at.seconds)))", category: .swPlayback)
+        let sourceSeconds = timeline.sourceSeconds(forRenderer: at.seconds)
         synchronizer.setRate(rate, time: at)
+        lock.unlock()
+        EngineLog.emit("[AudioOutput] setRate \(rate) at t=\(String(format: "%.3f", sourceSeconds))", category: .swPlayback)
     }
 
     /// Pause audio (and the master clock). Hosts resume via setRate (pausedByHost pattern); deliberately no resume() here.
     func pause() {
-        cancelPreroll()
+        lock.lock()
+        _clockCommandRevision &+= 1
+        endPreroll()
         let at = synchronizer.currentTime()
-        EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", timeline.sourceSeconds(forRenderer: at.seconds)))", category: .swPlayback)
+        let sourceSeconds = timeline.sourceSeconds(forRenderer: at.seconds)
         synchronizer.setRate(0.0, time: at)
+        lock.unlock()
+        EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", sourceSeconds))", category: .swPlayback)
     }
 
     /// Pause the master clock, at `latest` if it has already run past it. A park deferred to the last
     /// sample runs whenever its task is scheduled, and on a starved main actor that is after the clock
     /// has walked on (#694: 1.149 s on a 1.0 s source on a CI runner).
     func pause(notAfter latest: Double) {
-        cancelPreroll()
+        lock.lock()
+        _clockCommandRevision &+= 1
+        endPreroll()
         let now = timeline.sourceTime(forRenderer: synchronizer.currentTime())
         let seconds = SoftwareEndOfMediaClock.parkSeconds(clockSeconds: CMTimeGetSeconds(now), notAfter: latest)
         let at = seconds.map { CMTime(seconds: $0, preferredTimescale: 90000) } ?? now
-        EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", at.seconds))", category: .swPlayback)
         synchronizer.setRate(0.0, time: timeline.rendererTime(forSource: at))
+        lock.unlock()
+        EngineLog.emit("[AudioOutput] pause at t=\(String(format: "%.3f", at.seconds))", category: .swPlayback)
     }
 
     /// AE#464: set the audio presentation offset. Positive presents audio later than video, which on
@@ -318,6 +367,23 @@ final class AudioOutput: @unchecked Sendable {
     /// Test-only: runs inside `enqueue`'s locked section, right after the renderer took the buffer, which
     /// is where a renderer that posts its notifications from inside `enqueue` would post them.
     var afterRendererEnqueueForTesting: (@Sendable () -> Void)?
+
+    /// Holds a pending start without its timeout, so transport tests exercise cancellation
+    /// independently of how long the main actor takes to run on a loaded test runner.
+    func holdPrerollForTesting(rate: Float) {
+        lock.lock()
+        defer { lock.unlock() }
+        _clockCommandRevision &+= 1
+        endPreroll()
+        prerollRate = rate
+    }
+
+    var prerollRateForTesting: Float? {
+        lock.lock()
+        defer { lock.unlock() }
+        return prerollRate
+    }
+
     private var _loggedFirstEnqueue = false
     #endif
     private var _loggedRendererError = false
@@ -400,31 +466,41 @@ final class AudioOutput: @unchecked Sendable {
     func flush() {
         lock.lock()
         defer { lock.unlock() }
+        _clockCommandRevision &+= 1
         _epoch &+= 1
         renderer.flush()
     }
 
     func stop() {
         lock.lock()
         defer { lock.unlock() }
+        _clockCommandRevision &+= 1
         _epoch &+= 1
+        playbackEnabled = false
         endPreroll()
-        synchronizer.setRate(0.0, time: .zero)
+        // Hosts discard the output after stop. An already-paused clock needs no time reset;
+        // its native rate is updated synchronously even while the timebase settles.
+        if synchronizer.rate != 0 {
+            synchronizer.setRate(0.0, time: .zero)
+        }
         renderer.flush()
     }
 
     /// Atomically jump the master clock to a time and resume at a rate. The ONLY way the clock is (re)anchored:
     /// demux loops call it once on the first decoded packet, seek paths call it directly. Avoids the
     /// falling-through-time races that pause -> flush -> setRate would expose.
-    func seekClock(to time: CMTime, rate: Float) {
+    func seekClock(to time: CMTime, rate: Float, ifRevision revision: UInt64? = nil) {
         // Asked before taking `lock`: the gate reads the video renderer's state under its own lock.
         let preroll = rate > 0 ? videoPrerollSatisfied?(time.seconds) : nil
         let progress = preroll == false ? videoPrerollProgress?() : nil
         lock.lock()
         defer { lock.unlock() }
+        if let revision, revision != _clockCommandRevision { return }
+        let rate = playbackEnabled ? rate : 0
+        _clockCommandRevision &+= 1
         EngineLog.emit("[AudioOutput] seekClock to=\(String(format: "%.3f", time.seconds)) rate=\(rate)", category: .swPlayback)
         endPreroll()
-        switch preroll {
+        switch rate > 0 ? preroll : nil {
         case nil:
             synchronizer.setRate(rate, time: timeline.rendererTime(forSource: time))
         case true?:
@@ -552,7 +628,7 @@ final class AudioOutput: @unchecked Sendable {
         }
         lock.lock()
         defer { lock.unlock() }
-        guard prerollGeneration == wait.gen, let rate = prerollRate else { return }
+        guard playbackEnabled, prerollGeneration == wait.gen, let rate = prerollRate else { return }
         prerollRate = nil
         EngineLog.emit("[AudioOutput] preroll \(ready ? "ready" : "timed out") after "
                        + "\(String(format: "%.0f", waited * 1000)) ms; starting at t=\(String(format: "%.3f", wait.time.seconds)) rate=\(rate)",
@@ -569,12 +645,6 @@ final class AudioOutput: @unchecked Sendable {
         synchronizer.setRate(rate, time: timeline.rendererTime(forSource: time), atHostTime: host)
     }
 
-    private func cancelPreroll() {
-        lock.lock()
-        endPreroll()
-        lock.unlock()
-    }
-
     /// Retires the current preroll chain, and reports a hold that was still waiting as cancelled.
     /// Called with `lock` held.
     private func endPreroll() {
@@ -585,10 +655,15 @@ final class AudioOutput: @unchecked Sendable {
         prerollRate = nil
     }
 
-    private func updatePrerollRate(_ rate: Float) -> Bool {
+    private func updatePrerollRate(_ rate: Float, ifRevision revision: UInt64?) -> Bool {
         lock.lock()
         defer { lock.unlock() }
+        if let revision, revision != _clockCommandRevision { return false }
+        // A stale feeder resume is handled even without a pending preroll. The final clock
+        // write checks again in case an explicit pause lands after this preflight.
+        guard playbackEnabled else { return true }
         guard prerollRate != nil else { return false }
+        _clockCommandRevision &+= 1
         prerollRate = rate
         return true
     }

Sources/AetherEngine/Native/SoftwarePlaybackHost.swift

@@ -1594,6 +1594,7 @@ final class SoftwarePlaybackHost {
     func play() {
         guard !stopRequested, failure == nil else { return }
         renderer.recoverAfterSuspensionIfNeeded()
+        audioOutput?.setPlaybackEnabled(true)
         // Resume after pause(): gate on clockArmed, not demuxLoopStarted. A rate change on the
         // un-anchored synchronizer (no media at its clock time yet) wedges the delayed-rate-change
         // machinery permanently frozen; the arming seekClock applies the current lastRate (#107).
@@ -1681,8 +1682,46 @@ final class SoftwarePlaybackHost {
     }
 
     var clockRateForTesting: Float? { audioOutput?.rate }
+
+    func installClockOutputForTesting(_ output: AudioOutput, armed: Bool = false) {
+        audioOutput = output
+        clockArmed = armed
+    }
+
+    func finishClockOutputForTesting() {
+        didReachEnd = true
+        isPlaying = false
+    }
     #endif
 
+    /// The latch is synchronous for the feeder; transport reconciliation shares play/pause's actor.
+    func makeClockArmCallback(audioOutput aOut: AudioOutput?) -> @Sendable () -> Void {
+        { [weak self] in
+            guard let self, let aOut else { return }
+            let generation = self.seekGeneration
+            let revision = aOut.clockCommandRevision
+            self.clockArmed = true
+            // A transport change before arming still needs correction, but the feeder must not
+            // restart the clock between pause() stopping it and publishing isPlaying=false.
+            DispatchQueue.main.async { [weak self] in
+                guard let self, self.audioOutput === aOut,
+                      self.admitsRead(generation: generation) else { return }
+                // EOF may drain its queued audio tail, but an explicit pre-arm pause still wins.
+                guard self.pausedByHost || (!self.didParkClockAtEnd && !self.didReachEnd) else { return }
+                if !self.isPlaying || self.pausedBeforeFirstFrame {
+                    // A pending preroll can own a future start even while the clock reads zero.
+                    aOut.setRate(0, ifRevision: revision)
+                } else if let rate = PausedFirstFrame.rateCorrectionAtArm(
+                    transportPlaying: true, pausedBeforeFirstFrame: false,
+                    synchronizerRate: aOut.synchronizer.rate, lastRate: self.lastRate) {
+                    // A later rebuffer pause, seek or rate command owns the output now. The
+                    // comparison and correction are atomic with those commands inside AudioOutput.
+                    aOut.setRate(rate, ifRevision: revision)
+                }
+            }
+        }
+    }
+
     /// AE#549: re-anchor a master clock that stopped without a pause of ours, where it stands.
     ///
     /// The one re-anchor this path can make: the demuxer is an audio lead ahead of the clock by now
@@ -1702,10 +1741,9 @@ final class SoftwarePlaybackHost {
     }
 
     func pause() {
-        // Un-anchored clock: only latch the pause; the loops park on isPlaying (#107).
-        if clockArmed {
-            audioOutput?.pause()
-        }
+        // Latch even before arming: stale decoder and rebuffer work cannot restart the clock
+        // after this pause while the host publishes its transport state.
+        audioOutput?.setPlaybackEnabled(false)
         pausedByHost = true
         rate = 0
         if PausedFirstFrame.holdsForFirstFrame(loopsStarted: demuxLoopStarted, framesEnqueued: framesEnqueued),
@@ -2028,7 +2066,7 @@ final class SoftwarePlaybackHost {
             // Keep the last observed clock; the engine must not finalize this request as a landing.
             isPlaying = false
             pausedByHost = true
-            audioOutput?.setRate(0)
+            audioOutput?.setPlaybackEnabled(false)
             currentTime = clockSessionZero > 0
                 ? max(0, sourceClockSeconds - clockSessionZero) : sourceClockSeconds
             packetSource?.close()
@@ -2431,16 +2469,7 @@ final class SoftwarePlaybackHost {
         let getClockArmed: @Sendable () -> Bool = { [weak self] in
             self?.clockArmed ?? true
         }
-        let setClockArmed: @Sendable () -> Void = { [weak self] in
-            guard let self else { return }
-            self.clockArmed = true
-            // Sodalite#104 round 4: a pause() or play() between this arm reading its rate and the line
-            // above found no clock to act on.
-            guard let aOut, let rate = PausedFirstFrame.rateCorrectionAtArm(
-                transportPlaying: self.isPlaying, pausedBeforeFirstFrame: self.pausedBeforeFirstFrame,
-                synchronizerRate: aOut.synchronizer.rate, lastRate: self.lastRate) else { return }
-            if rate == 0 { aOut.pause() } else { aOut.setRate(rate) }
-        }
+        let setClockArmed = makeClockArmCallback(audioOutput: aOut)
         let getSeekGeneration: @Sendable () -> UInt64 = { [weak self] in
             self?.seekGeneration ?? 0
         }