fix(live): measure a window slide against the consumer's next fetch

The AE#441 retest read `live window slid past the consumer` four times in the
48 s after a deep rewound landing, every one of them firstVisible ==
consumerTarget + 1, with no stall and no cache miss behind any of them.

The line was reading the LAST fetch when the cost is decided by the NEXT one.
`declareTarget` fires as a segment is served and the consumer walks indices
forward, so everything below consumerTarget is already in AVPlayer's buffer and
the index it asks for next is consumerTarget + 1. A viewer parked at the floor
therefore sits one segment below firstVisible for part of every slide, the same
one-segment sawtooth residentFloorOutputSeconds() has against the rendered
playhead. Its next fetch is firstVisible itself, which is resident.

Chasing that turned up the reason the tolerance is not merely cosmetic: at
exactly that off-by-one, eviction unlinks the segment currently being served.
A serve holds a URL rather than a file handle (mediaSegmentURL hands peekURL's
result to a response that stats and opens it afterwards), so the window is a 404
for an index the playlist offered when it was asked for. Eviction now stops at
the fetch point, which is the bound evictBelow already documented for itself,
and never trails firstVisible by more than that one segment so a consumer that
stopped fetching cannot pin retention behind it. Not observed to fire on
loopback, where a serve completes in well under a segment; closed by
construction.

The harness could not reach this regime at all, which is why the local
counter-run read 0 lines against the retest's 4. `live --rewind-hold` filled
`window * 0.6` before rewinding, so the floor was always the session's own start
rather than an eviction frontier. It now fills past full, warns when --seconds
leaves too little hold behind that, and reports the shape of each latched line
instead of only counting them.

Measured, 220 s parked at the floor, window 60 s, paced 1x:
  before  4 latched lines, all gap 1, 0 stalls, max inversion 3.80 s (TD 4)
  after   0 latched lines, 0 stalls, max inversion 3.60 s
Unpaced backlog burst, window 30 s, after the change: 1326 lines, gap up to 12,
1326 of them above one segment, 8 stalled ticks. The two classes stay separated.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WbciAqCSxpaiUpTumrCuA9

Author: superuser404notfound

Sources/AetherEngine/Video/SegmentCache.swift

@@ -346,8 +346,9 @@ final class SegmentCache: @unchecked Sendable {
         return currentTargetIndex >= target
     }
 
-    /// Evict segments strictly below cutoff (= live firstVisible). Bounded by firstVisible <= currentTargetIndex
-    /// so it only removes segments the playlist already dropped; pruneOutsideWindow handles the forward bound.
+    /// Evict segments strictly below cutoff, which the live caller bounds at the consumer's own fetch
+    /// point (`VideoSegmentProvider.liveEvictionFloor`) so this never unlinks the segment a response is
+    /// about to stat; pruneOutsideWindow handles the forward bound.
     func evictBelow(_ cutoff: Int) {
         condition.lock()
         var doomed: [URL] = []

Sources/AetherEngine/Video/VideoSegmentProvider.swift

@@ -590,7 +590,8 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable {
                     // its own lock; nesting the two here would invert the ordering evictBelow's async
                     // hop exists to avoid).
                     let consumerTarget = cacheRef.targetIndex
-                    cacheRef.evictBelow(cutoff)
+                    cacheRef.evictBelow(Self.liveEvictionFloor(firstVisible: cutoff,
+                                                               consumerTarget: consumerTarget))
                     self?.noteWindowSlideRelativeToConsumer(cutoff: cutoff, consumerTarget: consumerTarget)
                 }
             }
@@ -606,6 +607,36 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable {
         return (segments.count, 0, refreshCounter, false, 0)
     }
 
+    /// AE#441 round 3: what a window slide costs is decided by the consumer's NEXT fetch, not its last.
+    ///
+    /// `declareTarget` is called as a segment is served and the consumer walks indices forward one at a
+    /// time, so everything strictly below `consumerTarget` is already in AVPlayer's own buffer and the
+    /// index it will ask for next is `consumerTarget + 1`. A viewer parked at the floor therefore sits
+    /// at `firstVisible - 1` for part of every segment, by the same one-segment sawtooth
+    /// `residentFloorOutputSeconds()` has against the rendered playhead: the slide moves in whole
+    /// segments while the consumer moves continuously. Its next fetch is `firstVisible` itself, which is
+    /// resident, so nothing is missed. From `firstVisible - 2` down the index it is about to ask for has
+    /// been deleted, and that is the cache miss this line exists to name.
+    static func windowSlidPastConsumer(firstVisible: Int, consumerTarget: Int) -> Bool {
+        return consumerTarget + 1 < firstVisible
+    }
+
+    /// The lowest index a window slide may unlink, which is not always the playlist's new first visible.
+    ///
+    /// A serve holds a URL, not a file handle: `mediaSegmentURL` hands `peekURL`'s result to a response
+    /// that stats and opens the file afterwards, so unlinking the segment currently being served turns
+    /// into a 404 for an index the playlist offered when it was asked for. And a viewer riding the floor
+    /// puts the slide exactly one segment above the fetch point routinely, not rarely (measured: every
+    /// latched line of a 220 s parked run had `firstVisible == consumerTarget + 1`). So eviction stops at
+    /// the fetch point, which is the bound `evictBelow` already documents for itself.
+    ///
+    /// Bounded on the other side too: the floor never trails `firstVisible` by more than one segment, so
+    /// a consumer that has stopped fetching entirely cannot pin retention behind it.
+    static func liveEvictionFloor(firstVisible: Int, consumerTarget: Int) -> Int {
+        guard consumerTarget >= 0 else { return firstVisible }
+        return max(firstVisible - 1, min(firstVisible, consumerTarget))
+    }
+
     /// The sliding window overtaking the consumer's fetch point is the failure mode the removed advance
     /// park used to make impossible (it capped the producer 10 segments ahead of that point). Live is
     /// source-paced, so a real-time origin cannot get there; an origin that hands over more than one
@@ -616,7 +647,7 @@ final class VideoSegmentProvider: HLSSegmentProvider, @unchecked Sendable {
     private func noteWindowSlideRelativeToConsumer(cutoff: Int, consumerTarget: Int) {
         guard consumerTarget >= 0 else { return }
         stateLock.lock()
-        let outside = consumerTarget < cutoff
+        let outside = Self.windowSlidPastConsumer(firstVisible: cutoff, consumerTarget: consumerTarget)
         let shouldLog = outside && !_liveConsumerOutsideWindowLatched
         _liveConsumerOutsideWindowLatched = outside
         stateLock.unlock()

Sources/aetherctl/LiveCmd.swift

@@ -785,10 +785,18 @@ private func liveRewindHoldTest(url: URL, seconds playSeconds: Double, dvrWindow
         return 1
     }
 
-    // Fill the window before rewinding: parking just above the floor only means anything once the
-    // floor is retention rather than the session's own start.
-    let fillFor = min(max(10.0, dvrWindow * 0.6), playSeconds * 0.4)
-    print(String(format: "  FILL: %.0fs before the rewind (window=%.0fs)", fillFor, dvrWindow))
+    // Fill the window before rewinding, and fill it PAST full. A floor below a window that has not
+    // filled yet is the session's own start, which never moves, so a park against it measures a
+    // regime that has no eviction in it at all. The first version of this leg filled `window * 0.6`
+    // and therefore reported a clean bill of health for a regime it could not reach.
+    let fillFor = dvrWindow + 12.0
+    print(String(format: "  FILL: %.0fs before the rewind (window=%.0fs, filled past full)",
+                 fillFor, dvrWindow))
+    if playSeconds < fillFor + 30.0 {
+        print(String(format: "  WARNING: --seconds %.0f leaves %.0fs of hold after a %.0fs fill; "
+                     + "the window will not be sliding for long enough to read anything",
+                     playSeconds, playSeconds - fillFor, fillFor))
+    }
     let startTime = Date()
     while Date().timeIntervalSince(startTime) < fillFor {
         try? await Task.sleep(nanoseconds: 500_000_000)
@@ -830,7 +838,10 @@ private func liveRewindHoldTest(url: URL, seconds playSeconds: Double, dvrWindow
     print(String(format: "  ticks with floor above the playhead: %d/%d (max %.2fs)",
                  invertedTicks, holdTicks, maxInversion))
     print(String(format: "  ticks where the clock did not advance: %d/%d", stalledTicks, holdTicks))
-    print("  'live window slid past the consumer': \(slides.count)")
+    print("  'live window slid past the consumer': \(slides.count)"
+          + (slides.count > 0
+             ? " (max firstVisible-consumerTarget gap \(slides.maxGap), \(slides.gapsAboveOne) above one segment)"
+             : ""))
     print("  final state: \(finalState)")
 
     // The inversion alone is not the defect; the window passing the consumer's FETCH point is.
@@ -843,11 +854,27 @@ private func liveRewindHoldTest(url: URL, seconds playSeconds: Double, dvrWindow
     return 0
 }
 
+/// Counts the latched slide line AND reads its shape. The gap `firstVisible - consumerTarget` is the
+/// whole reading: a gap of 1 means the consumer's next fetch is the new first visible segment, which
+/// is still resident, while a gap above 1 means the segment it will ask for next is already deleted.
 private final class SlideCounter: @unchecked Sendable {
     private let lock = NSLock()
     private(set) var count = 0
+    private(set) var maxGap = 0
+    private(set) var gapsAboveOne = 0
     func note(_ line: String) {
         lock.lock(); defer { lock.unlock() }
-        if line.contains("live window slid past the consumer") { count += 1 }
+        guard line.contains("live window slid past the consumer") else { return }
+        count += 1
+        guard let first = Self.intField("firstVisible=", in: line),
+              let target = Self.intField("consumerTarget=", in: line) else { return }
+        let gap = first - target
+        maxGap = max(maxGap, gap)
+        if gap > 1 { gapsAboveOne += 1 }
+    }
+    private static func intField(_ key: String, in line: String) -> Int? {
+        guard let r = line.range(of: key) else { return nil }
+        let rest = line[r.upperBound...].prefix { $0 == "-" || $0.isNumber }
+        return Int(rest)
     }
 }

Tests/AetherEngineTests/Issue441ResidentLiveFloorTests.swift

@@ -112,3 +112,68 @@ struct Issue441ResidentLiveFloorTests {
         #expect(cache.highestResidentIndex == nil)
     }
 }
+
+/// AE#441 round 3: the retest read `live window slid past the consumer` four times in the 48 s after a
+/// deep rewound landing, every one of them `firstVisible == consumerTarget + 1`, with no stall and no
+/// cache miss behind any of them. Reproduced on the harness (`live --rewind-hold`, 220 s, window 60 s):
+/// four latched lines, gap 1 on all four. So the line was reading the LAST fetch when the cost is
+/// decided by the NEXT one, and the slide that reaches the fetch point was also unlinking the segment
+/// under the serve.
+@Suite("AE#441 round 3 a window slide is measured against the consumer's next fetch")
+struct Issue441ConsumerFetchPointTests {
+
+    // MARK: - The discriminator
+
+    /// The sawtooth on the fetch axis. A viewer parked at the floor sits one segment below the new
+    /// first-visible index for part of every slide; the index it asks for next is that very segment.
+    @Test("one segment behind the window is the parked viewer's ordinary position, not a defect")
+    func oneSegmentBehindIsNotADefect() {
+        #expect(VideoSegmentProvider.windowSlidPastConsumer(firstVisible: 5, consumerTarget: 4) == false)
+        #expect(VideoSegmentProvider.windowSlidPastConsumer(firstVisible: 33, consumerTarget: 32) == false)
+    }
+
+    /// Two segments behind is the real thing: `consumerTarget + 1` has already been deleted, so the
+    /// consumer's next request is a miss whatever the playlist says.
+    @Test("two segments behind means the next fetch is already deleted")
+    func twoSegmentsBehindIsTheDefect() {
+        #expect(VideoSegmentProvider.windowSlidPastConsumer(firstVisible: 6, consumerTarget: 4))
+        #expect(VideoSegmentProvider.windowSlidPastConsumer(firstVisible: 40, consumerTarget: 11))
+    }
+
+    /// A consumer at or ahead of the window is the healthy steady state and was never a defect.
+    @Test("a consumer inside the window stays silent")
+    func consumerInsideTheWindow() {
+        #expect(VideoSegmentProvider.windowSlidPastConsumer(firstVisible: 5, consumerTarget: 5) == false)
+        #expect(VideoSegmentProvider.windowSlidPastConsumer(firstVisible: 5, consumerTarget: 12) == false)
+    }
+
+    // MARK: - The eviction floor
+
+    /// The case the retest exposed: the slide reaching `consumerTarget + 1` used to unlink the segment
+    /// whose URL a response was about to stat, which is a 404 for an index the playlist offered.
+    @Test("eviction stops at the segment being served")
+    func evictionSpareTheServedSegment() {
+        #expect(VideoSegmentProvider.liveEvictionFloor(firstVisible: 5, consumerTarget: 4) == 4)
+    }
+
+    /// A consumer already inside the window costs nothing: the floor is the playlist's own.
+    @Test("a consumer inside the window does not hold eviction back")
+    func consumerInsideDoesNotHoldBack() {
+        #expect(VideoSegmentProvider.liveEvictionFloor(firstVisible: 5, consumerTarget: 5) == 5)
+        #expect(VideoSegmentProvider.liveEvictionFloor(firstVisible: 5, consumerTarget: 40) == 5)
+    }
+
+    /// The other side of the bound. A consumer that stopped fetching entirely must not pin retention
+    /// behind it, so the floor never trails the window by more than the one served segment.
+    @Test("a stalled consumer cannot pin retention behind the window")
+    func stalledConsumerCannotPinRetention() {
+        #expect(VideoSegmentProvider.liveEvictionFloor(firstVisible: 40, consumerTarget: 4) == 39)
+        #expect(VideoSegmentProvider.liveEvictionFloor(firstVisible: 900, consumerTarget: 0) == 899)
+    }
+
+    /// Before the first fetch there is no point to protect.
+    @Test("no declared fetch point evicts to the window")
+    func noConsumerYet() {
+        #expect(VideoSegmentProvider.liveEvictionFloor(firstVisible: 12, consumerTarget: -1) == 12)
+    }
+}