fix(video): prime rotated muxers with the last muxed audio frame

Sources/AetherEngine/Video/HLSSegmentProducer.swift

@@ -249,6 +249,15 @@ final class HLSSegmentProducer: @unchecked Sendable {
     /// AE#222: real audio frame handed to every muxer this producer builds, so moov (with its packet-derived
     /// dec3/dac3/dmlp) is written at init instead of at the first cut. nil on a normal session.
     private let audioMoovPrimeFrame: [UInt8]?
+    /// Most recent audio frame a muxer accepted, retained so every LATER allocation (same-PID parameter-set
+    /// rotation, SSAI program switch) starts primed too: video leads audio across a seam, so a rotated
+    /// muxer's first cut can arrive before any post-seam audio packet exists — unprimed, that cut defers
+    /// for a sample entry the bridge's encoder latency will never deliver in time and the pump dies with
+    /// `muxerFailed`. Pump-thread confined; seeded from `audioMoovPrimeFrame`.
+    private var lastMuxedAudioPrimeFrame: [UInt8]?
+    /// True when the session's audio codec derives its mp4 sample entry from a parsed packet
+    /// (AC-3/E-AC-3/TrueHD); only then is a per-frame prime copy worth the memcpy. AAC never copies.
+    private let capturesAudioPrimeFrames: Bool
     /// Latched when a cut deferred for want of an audio sample entry; the pump then scans forward for one
     /// real audio frame and exits with `.needsAudioSampleEntryPrime` so the host can rebuild primed.
     private var cutDeferredAwaitingAudioSampleEntry: Bool = false
@@ -790,6 +799,9 @@ final class HLSSegmentProducer: @unchecked Sendable {
     ) throws {
         self.epoch = epoch
         self.audioMoovPrimeFrame = audioMoovPrimeFrame
+        self.lastMuxedAudioPrimeFrame = audioMoovPrimeFrame
+        self.capturesAudioPrimeFrames =
+            audio.map { MP4SegmentMuxer.audioNeedsParsedPacketForMoov($0.codecpar.pointee.codec_id) } ?? false
         self.bufferAheadSegments = bufferAheadSegments
         self.prefetchDiskBudgetBytes = prefetchDiskBudgetBytes
         self.demuxer = demuxer
@@ -1320,8 +1332,10 @@ final class HLSSegmentProducer: @unchecked Sendable {
                 // Floored at 8s (the historical 2 x 4s value): a sub-second fastZap cut target (AE#195)
                 // must not shrink the cap below typical TS A/V interleave skew.
                 maxBufferedFragmentSeconds: max(8.0, 2 * targetSegmentDurationSeconds),
-                // AE#222: only set on a session that already proved its first segment carries no audio.
-                audioMoovPrimeFrame: audioMoovPrimeFrame,
+                // AE#222 + mid-session rotation: prefer the last frame a muxer accepted, so a rotation
+                // stays primed even when video leads audio across the seam; the construction-time prime
+                // covers only the first allocation of a session whose first segment carries no audio.
+                audioMoovPrimeFrame: lastMuxedAudioPrimeFrame ?? audioMoovPrimeFrame,
                 onInitCaptured: { [weak self] initBytes in
                     guard let self = self else { return }
                     if versionedInit {
@@ -2831,7 +2845,10 @@ final class HLSSegmentProducer: @unchecked Sendable {
                             }
                             fp.pointee.stream_index = muxer.audioOutputStreamIndex
                             av_packet_rescale_ts(fp, audio.inputTimeBase, muxer.muxerAudioTimeBase)
-                            _ = muxer.writePacket(fp)
+                            let prime = copyAudioPrimeCandidate(fp)
+                            if muxer.writePacket(fp).rc >= 0, let prime {
+                                lastMuxedAudioPrimeFrame = prime
+                            }
                             trackedPacketFree(&fpVar)
                         }
                         if bridgedMuxerGone {
@@ -3120,6 +3137,16 @@ final class HLSSegmentProducer: @unchecked Sendable {
         packet.pointee.size -= headerLen
     }
 
+    /// Copies the payload of an audio packet about to be muxed, BEFORE the write: movenc consumes the
+    /// packet's data reference, so afterwards there is nothing left to copy. The caller commits the copy
+    /// into `lastMuxedAudioPrimeFrame` only when the write succeeded.
+    private func copyAudioPrimeCandidate(_ packet: UnsafeMutablePointer<AVPacket>) -> [UInt8]? {
+        guard capturesAudioPrimeFrames, let data = packet.pointee.data, packet.pointee.size > 0 else {
+            return nil
+        }
+        return [UInt8](UnsafeBufferPointer(start: data, count: Int(packet.pointee.size)))
+    }
+
     /// Stream-copy audio only; bridge audio bypasses this (FLAC encoder sets durations correctly).
     private func finalizeAndWriteAudio(
         _ packet: UnsafeMutablePointer<AVPacket>,
@@ -3138,7 +3165,10 @@ final class HLSSegmentProducer: @unchecked Sendable {
 
         packet.pointee.stream_index = muxer.audioOutputStreamIndex
         av_packet_rescale_ts(packet, audio.inputTimeBase, muxer.muxerAudioTimeBase)
-        _ = muxer.writePacket(packet)
+        let prime = copyAudioPrimeCandidate(packet)
+        if muxer.writePacket(packet).rc >= 0, let prime {
+            lastMuxedAudioPrimeFrame = prime
+        }
 
         var pkt: UnsafeMutablePointer<AVPacket>? = packet
         trackedPacketFree(&pkt)

Sources/AetherEngine/Video/MP4SegmentMuxer.swift

@@ -140,6 +140,15 @@ final class MP4SegmentMuxer {
     /// from codecpar alone, so they never wedge, and gating the #64 RAM-cap flush on them would needlessly
     /// weaken that memory bound. Latched at init from the audio codec_id.
     private let audioNeedsParsedPacketForMoov: Bool
+
+    /// Only AC-3 / E-AC-3 / TrueHD build their mp4 sample entry from a parsed packet (dac3/dec3/dmlp),
+    /// so only they can hit the "moov before audio parsed" wedge and need the #64-flush guard. Shared with
+    /// the producer, which uses it to decide whether retaining a moov-prime frame copy buys anything.
+    static func audioNeedsParsedPacketForMoov(_ codecID: AVCodecID) -> Bool {
+        codecID == AV_CODEC_ID_AC3 ||
+        codecID == AV_CODEC_ID_EAC3 ||
+        codecID == AV_CODEC_ID_TRUEHD
+    }
     /// Latched when the next staging file open fails; producer must stop the pump.
     private(set) var isWedged: Bool = false
     /// Latched after avformat_write_header; mp4 muxer rewrites time_base to its own pick
@@ -183,16 +192,8 @@ final class MP4SegmentMuxer {
         self.currentSegmentIndex = initialSegmentIndex
         self.sessionDir = sessionDir
         self.haveAudio = audio != nil
-        // Only AC-3 / E-AC-3 / TrueHD build their mp4 sample entry from a parsed packet (dac3/dec3/dmlp),
-        // so only they can hit the "moov before audio parsed" wedge and need the #64-flush guard.
-        if let audioCodecID = audio?.codecpar.pointee.codec_id {
-            self.audioNeedsParsedPacketForMoov =
-                audioCodecID == AV_CODEC_ID_AC3 ||
-                audioCodecID == AV_CODEC_ID_EAC3 ||
-                audioCodecID == AV_CODEC_ID_TRUEHD
-        } else {
-            self.audioNeedsParsedPacketForMoov = false
-        }
+        self.audioNeedsParsedPacketForMoov =
+            audio.map { Self.audioNeedsParsedPacketForMoov($0.codecpar.pointee.codec_id) } ?? false
 
         let firstPath = Self.stagingPath(forSegmentIndex: initialSegmentIndex,
                                          in: sessionDir)

Tests/AetherEngineTests/LiveRotationAudioPrimeTests.swift

@@ -0,0 +1,202 @@
+import Testing
+import Foundation
+import Libavformat
+import Libavcodec
+import Libavutil
+@testable import AetherEngine
+
+/// Live rotation wedge: a mid-session muxer rotation (same-PID parameter-set change after a live reconnect
+/// join, or an SSAI program switch) builds a brand-new muxer, and with BRIDGED E-AC-3 the rotated muxer's
+/// first cut can arrive before any post-seam audio packet exists — video leads audio across the seam and
+/// the bridge adds encoder latency on top. Unprimed, that cut defers for a sample entry (AE#222), the
+/// producer converts the deferral into a pump exit, teardown finalize fails on the same precondition, and
+/// the live session dies with `muxerFailed`. The exit-scan cannot help either: it rightly excludes bridged
+/// sessions (a raw source frame cannot prime a bridge-encoded track).
+///
+/// The fix has the producer retain the last audio frame a muxer accepted and prime every later allocation
+/// with it. These tests pin the claim that rests on, which AE#222's source-frame tests do not cover: a
+/// BRIDGE-OUTPUT E-AC-3 frame is a valid moov prime for a muxer whose audio track IS the bridge encoder.
+@Suite("Live rotation: bridge-output audio frame as moov prime")
+struct LiveRotationAudioPrimeTests {
+
+    // MARK: - Harness
+
+    /// Little-endian 16-bit PCM WAV with a 440 Hz sine (same shape as Issue99BridgeResumeTests).
+    private static func makeWAV(sampleRate: Int, channels: Int, seconds: Double) -> Data {
+        let frames = Int(Double(sampleRate) * seconds)
+        var pcm = Data(capacity: frames * channels * 2)
+        for n in 0..<frames {
+            let v = Int16(9000 * sin(2 * .pi * 440 * Double(n) / Double(sampleRate)))
+            for _ in 0..<channels {
+                withUnsafeBytes(of: v.littleEndian) { pcm.append(contentsOf: $0) }
+            }
+        }
+        var d = Data()
+        func str(_ s: String) { d.append(s.data(using: .ascii)!) }
+        func u32(_ v: UInt32) { withUnsafeBytes(of: v.littleEndian) { d.append(contentsOf: $0) } }
+        func u16(_ v: UInt16) { withUnsafeBytes(of: v.littleEndian) { d.append(contentsOf: $0) } }
+        str("RIFF"); u32(UInt32(36 + pcm.count)); str("WAVE")
+        str("fmt "); u32(16); u16(1); u16(UInt16(channels)); u32(UInt32(sampleRate))
+        u32(UInt32(sampleRate * channels * 2)); u16(UInt16(channels * 2)); u16(16)
+        str("data"); u32(UInt32(pcm.count)); d.append(pcm)
+        return d
+    }
+
+    /// A live bridged session in miniature: WAV source -> AudioBridge(.surroundCompat) -> E-AC-3 packets,
+    /// plus the H.264 video fixture, feeding a muxer whose audio track is the bridge's encoder.
+    private final class Rig {
+        let videoDemuxer = Demuxer()
+        let audioDemuxer = Demuxer()
+        let sessionDir: URL
+        var bridge: AudioBridge?
+        var initBytes: Data?
+        var muxer: MP4SegmentMuxer?
+        /// Payloads of every E-AC-3 packet the bridge emitted, in order.
+        var bridgeFrames: [[UInt8]] = []
+
+        init() throws {
+            sessionDir = FileManager.default.temporaryDirectory
+                .appendingPathComponent("live-rotation-prime-\(UUID().uuidString)")
+            try FileManager.default.createDirectory(at: sessionDir, withIntermediateDirectories: true)
+            try videoDemuxer.open(
+                reader: DataIOReader(data: Data(
+                    base64Encoded: AtmosDetectionProbeIntegrationTests.videoOnlyBase64,
+                    options: .ignoreUnknownCharacters) ?? Data()),
+                formatHint: "mp4"
+            )
+        }
+
+        deinit {
+            muxer = nil
+            bridge?.close()
+            videoDemuxer.close()
+            audioDemuxer.close()
+            try? FileManager.default.removeItem(at: sessionDir)
+        }
+
+        /// Opens the bridge over an in-memory WAV and runs every source packet through it.
+        func runBridge() throws {
+            try audioDemuxer.open(
+                reader: DataIOReader(data: LiveRotationAudioPrimeTests.makeWAV(
+                    sampleRate: 48_000, channels: 2, seconds: 1.0))
+            )
+            let audioIdx = audioDemuxer.audioStreamIndex
+            guard audioIdx >= 0, let stream = audioDemuxer.stream(at: audioIdx) else {
+                throw RigError.noStream
+            }
+            let bridge = try AudioBridge(
+                srcCodecpar: stream.pointee.codecpar,
+                srcTimeBase: stream.pointee.time_base,
+                mode: .surroundCompat
+            )
+            self.bridge = bridge
+            while let packet = try audioDemuxer.readPacket() {
+                var p: UnsafeMutablePointer<AVPacket>? = packet
+                defer { trackedPacketFree(&p) }
+                guard packet.pointee.stream_index == audioIdx else { continue }
+                for out in try bridge.feed(packet: packet) {
+                    var o: UnsafeMutablePointer<AVPacket>? = out
+                    defer { trackedPacketFree(&o) }
+                    if let data = out.pointee.data, out.pointee.size > 0 {
+                        bridgeFrames.append(
+                            [UInt8](UnsafeBufferPointer(start: data, count: Int(out.pointee.size))))
+                    }
+                }
+            }
+        }
+
+        /// The rotated muxer: fresh AVFormatContext whose audio track is the bridge's E-AC-3 encoder,
+        /// exactly what `rotateMuxerForProgramSwitch` allocates mid-session.
+        func makeRotatedMuxer(audioPrime: [UInt8]?) throws -> MP4SegmentMuxer {
+            guard let vStream = videoDemuxer.stream(at: videoDemuxer.videoStreamIndex),
+                  let encoderCodecpar = bridge?.encoderCodecpar, let bridge else {
+                throw RigError.noStream
+            }
+            let m = try MP4SegmentMuxer(
+                initialSegmentIndex: 414,
+                sessionDir: sessionDir,
+                video: MP4SegmentMuxer.VideoConfig(
+                    codecpar: UnsafePointer(vStream.pointee.codecpar),
+                    timeBase: vStream.pointee.time_base,
+                    codecTagOverride: nil
+                ),
+                audio: MP4SegmentMuxer.AudioConfig(
+                    codecpar: UnsafePointer(encoderCodecpar),
+                    timeBase: bridge.encoderTimeBase
+                ),
+                audioMoovPrimeFrame: audioPrime,
+                onInitCaptured: { [self] bytes in self.initBytes = bytes }
+            )
+            muxer = m
+            return m
+        }
+
+        /// Post-seam video: every fixture packet, rescaled like the producer does.
+        @discardableResult
+        func writeAllVideoPackets(into muxer: MP4SegmentMuxer) throws -> Int {
+            guard let vStream = videoDemuxer.stream(at: videoDemuxer.videoStreamIndex) else { return 0 }
+            let sourceTb = vStream.pointee.time_base
+            var written = 0
+            while let pkt = try videoDemuxer.readPacket() {
+                var p: UnsafeMutablePointer<AVPacket>? = pkt
+                defer { trackedPacketFree(&p) }
+                guard pkt.pointee.stream_index == videoDemuxer.videoStreamIndex else { continue }
+                pkt.pointee.stream_index = muxer.videoOutputStreamIndex
+                av_packet_rescale_ts(pkt, sourceTb, muxer.muxerVideoTimeBase)
+                _ = muxer.writePacket(pkt)
+                written += 1
+            }
+            return written
+        }
+    }
+
+    private enum RigError: Error { case noStream }
+
+    private static func containsBox(_ data: Data?, _ fourCC: String) -> Bool {
+        guard let data else { return false }
+        return data.range(of: Data(fourCC.utf8)) != nil
+    }
+
+    // MARK: - The incident, pinned
+
+    @Test("unprimed rotation with bridged EAC3: video-only cut defers and teardown finalize salvages nothing")
+    func unprimedRotationDiesLikeTheIncident() throws {
+        let rig = try Rig()
+        try rig.runBridge()
+        let muxer = try rig.makeRotatedMuxer(audioPrime: nil)
+
+        let written = try rig.writeAllVideoPackets(into: muxer)
+        #expect(written > 0, "fixture must contribute post-seam video packets")
+
+        #expect(muxer.cutFragmentForNextSegment(415) == .deferredAwaitingAudioSampleEntry,
+                "the rotated muxer's first cut arrives before any post-seam bridge output")
+        #expect(muxer.finalize() == nil,
+                "teardown finalize fails on the same precondition — the 'final finalize failed; not adopted' -> muxerFailed pump death")
+    }
+
+    // MARK: - The fix's load-bearing claim
+
+    @Test("a bridge-output EAC3 frame primes the rotated muxer: moov+dec3 at init, video-only cut completes")
+    func bridgeOutputFramePrimesRotatedMuxer() throws {
+        let rig = try Rig()
+        try rig.runBridge()
+        #expect(!rig.bridgeFrames.isEmpty, "bridge must emit E-AC-3 packets")
+
+        // What the producer's capture retains: the payload of the last frame a muxer accepted.
+        let muxer = try rig.makeRotatedMuxer(audioPrime: rig.bridgeFrames.last)
+
+        #expect(Self.containsBox(rig.initBytes, "moov"), "moov is written at init, before any packet")
+        #expect(Self.containsBox(rig.initBytes, "dec3"),
+                "dec3 must be derivable from a BRIDGE-OUTPUT frame, not just a source frame")
+
+        try rig.writeAllVideoPackets(into: muxer)
+        guard case .completed(let path, let bytes) = muxer.cutFragmentForNextSegment(415) else {
+            Issue.record("primed rotation cut must complete")
+            return
+        }
+        #expect(bytes > 0)
+        let segment = try Data(contentsOf: path)
+        #expect(Self.containsBox(segment, "moof"), "the delivered segment is a plain fragment")
+        #expect(!Self.containsBox(segment, "moov"), "moov already went to the versioned init")
+    }
+}