fix(hls): keep empty FIFO writes from signaling EOF

CHANGELOG.md

@@ -168,6 +168,11 @@ 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.
 
+- **The internal HLS byte FIFO defensively rejects operations that could look like EOF.**
+  Empty writes complete without occupying capacity, and nonpositive read lengths fail without
+  consuming bytes. Current ingest callers already filter both cases, so this hardens the internal
+  contract for future callers rather than changing existing playback.
+
 - **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/IO/HLSIngest/ByteFIFO.swift

@@ -43,6 +43,8 @@ final class ByteFIFO: @unchecked Sendable {
     func write(_ data: Data) -> Bool {
         condition.lock()
         defer { condition.unlock() }
+        // Empty writes neither occupy capacity nor make a reader's EOF condition true.
+        if data.isEmpty { return !finished && !cancelled }
         parked += 1
         while pendingCount >= capacity && !finished && !cancelled {
             condition.wait()
@@ -55,8 +57,9 @@ final class ByteFIFO: @unchecked Sendable {
         return true
     }
 
-    /// Blocking read. Returns: >0 bytes copied; 0 = EOF (finished + drained); -1 = cancelled.
+    /// Blocking read. Returns: >0 bytes copied; 0 = EOF (finished + drained); -1 = cancelled or invalid length.
     func read(into buffer: UnsafeMutablePointer<UInt8>, maxLength: Int) -> Int {
+        guard maxLength > 0 else { return -1 }
         condition.lock()
         defer { condition.unlock() }
         parked += 1

Tests/AetherEngineTests/ByteFIFOTests.swift

@@ -1,4 +1,6 @@
 import XCTest
+import Testing
+import Synchronization
 @testable import AetherEngine
 
 final class ByteFIFOTests: XCTestCase {
@@ -108,3 +110,65 @@ final class ByteFIFOTests: XCTestCase {
         wait(for: [expectation], timeout: 300)
     }
 }
+
+// Empty writes must not masquerade as the EOF that the ingest reader returns to FFmpeg.
+// Run potentially blocking calls on real threads so cancellation can still release them.
+@Suite(.timeLimit(.minutes(1)))
+struct ByteFIFOEmptyOperationsTests {
+    @Test("an empty write leaves a reader waiting for bytes")
+    func emptyWriteDoesNotSignalEOF() async throws {
+        let fifo = ByteFIFO(capacity: 4)
+        defer { fifo.cancel() }
+        #expect(fifo.write(Data()))
+        let result = Mutex<Int?>(nil)
+        Thread.detachNewThread {
+            var byte: UInt8 = 0
+            let count = fifo.read(into: &byte, maxLength: 1)
+            result.withLock { $0 = count }
+        }
+        try await waitFor { result.withLock { $0 != nil } || fifo.parkedWaiterCount > 0 }
+        #expect(result.withLock { $0 == nil }, "an empty write must not return zero (EOF)")
+        #expect(fifo.isEmptyWithReaderParked)
+        #expect(fifo.write(Data([42])))
+        try await waitFor { result.withLock { $0 != nil } }
+        #expect(result.withLock { $0 == 1 })
+    }
+
+    @Test("an empty write completes even when capacity is occupied")
+    func emptyWriteDoesNotWaitForCapacity() async throws {
+        let fifo = ByteFIFO(capacity: 1)
+        defer { fifo.cancel() }
+        #expect(fifo.write(Data([42])))
+        let result = Mutex<Bool?>(nil)
+        Thread.detachNewThread {
+            let accepted = fifo.write(Data())
+            result.withLock { $0 = accepted }
+        }
+        try await waitFor { fifo.parkedWaiterCount > 0 || result.withLock { $0 != nil } }
+        #expect(fifo.parkedWaiterCount == 0)
+        fifo.cancel()
+        try await waitFor { result.withLock { $0 != nil } }
+        #expect(result.withLock { $0 == true })
+    }
+
+    @Test("empty writes still reject a finished or cancelled FIFO")
+    func emptyWriteHonorsTerminalState() {
+        let finished = ByteFIFO(capacity: 1)
+        finished.finish()
+        #expect(!finished.write(Data()))
+        let cancelled = ByteFIFO(capacity: 1)
+        cancelled.cancel()
+        #expect(!cancelled.write(Data()))
+    }
+
+    @Test("a nonpositive read length is an error and preserves queued bytes", arguments: [0, -1, Int.min])
+    func invalidLengthPreservesBytes(length: Int) {
+        let fifo = ByteFIFO(capacity: 1)
+        #expect(fifo.write(Data([42])))
+        var byte: UInt8 = 99
+        #expect(fifo.read(into: &byte, maxLength: length) == -1)
+        #expect(byte == 99)
+        #expect(fifo.read(into: &byte, maxLength: 1) == 1)
+        #expect(byte == 42)
+    }
+}

docs/architecture.md

@@ -542,7 +542,7 @@ Sources/AetherEngine/
 │       ├── HLSPlaylistTracker.swift         Pure segment cursor: duration-capped edge join, window-slide rejoin, stall budget
 │       ├── HLSSegmentDecryptor.swift        AES-128-CBC clear-key segment decryption (key fetch + memoise, PKCS7)
 │       ├── PackedAudioSegments.swift        Packed-audio rendition support: LiveSegmentFormat classification + ID3 PRIV timestamp parser (raw ADTS segments)
-│       ├── ByteFIFO.swift                   Bounded blocking byte queue between the fetch loop and the demux thread
+│       ├── ByteFIFO.swift                   Bounded blocking byte queue between the fetch loop and the demux thread; empty writes are no-ops, invalid read lengths return an error, and only a finished, drained queue returns EOF
 │       ├── HLSIngestError.swift             Typed terminal errors (encrypted, fMP4, unreachable, invalid, stalled)
 │       ├── LiveIngestSourceInfo.swift       Internal seam: upstream segment cadence (shapes TARGETDURATION + blocking-reload eligibility) and DualSourceMergeOrder for the dual-source DTS merge
 │       ├── HLSVODIngestReader.swift         Ingest reader for an immutable ENDLIST playlist: seeks restart the source at the segment preceding the requested playlist time, on an elapsed-media-time axis (the EXTINF sum) rather than the container's PTS