fix(subtitles): a landing line has to stand on ground somebody read (#416)

A PGS display set has no end of its own, so the reconstruction pass at a seek
landing decides which set is still on screen by the store holding nothing
between it and the playhead. Over a stretch nobody read, that silence proves
nothing. A run re-aimed just after it harvested a set leaves that set's own
clear on the far side of the skipped ground: the set decodes at the landing
looking unclosed, becomes the landing's active line, and takes its end from the
next stored packet, which is the far side of the authored silence rather than
its own successor. The reporter's asset shows a two-second sound-effect caption
standing over a scene ten seconds later, on an Apple TV 4K and on a Mac.

#362 round 2 established that the packets alone cannot show this and named the
ledger as the precise signal: a reader restarted BEHIND leaves a descending
harvest sequence at the boundary, but one re-anchored FORWARD hangs its packets
in ascending order behind the stretch it skipped, so the pair is
indistinguishable from an authored silence. It was left unbuilt for want of a
reproducible defect. This is that defect.

So the readers state what they read. `SubtitleHarvestCoverage` keeps one span
per run in the packet store, anchored where a reader positioned and extended as
it goes: the forward prefetcher reports its own read position and every
re-anchor, the pump's run begins where the producer opens or restarts (both
paths) and reaches at least the playhead, since playback is rendering there. A
set whose ground up to the playhead is not covered cannot be the landing's
active line, and the same rule closes the #100 hold's door onto the same claim.
A store nobody reports to answers every span with yes, so a harvest path without
notes behaves exactly as before.

The cost is the landing line in the case where a set really is still up and the
proof is missing, which takes an authored dwell long enough to span the whole
unread stretch; the alternative was paying it for every normally authored set
that ends inside one. A refusal says so as `landingWithheld=` on the delivery
line, beside `endsWithheld=`.

Measured with `aetherctl play` on a 500 s H.264 + PGS fixture whose sets are
scheduled (set at 184, authored clear at 190, then silence to 240), over a
loopback origin, seeking from 100 to 197 so the initial run's park edge falls
between the set and its clear and the restarted pump opens at 192, above it:
before, 5 of 5 runs published `CUE 184.00-240.00` at the landing; after, 0 of 5,
with `landingWithheld=1` on the runs that met the condition and the authored set
at 240 published in every run. Controls unchanged in both arms: a far seek into
fresh ground landing inside an authored dwell still publishes its line
(292.00-295.50), and a plain playthrough delivers the same four cues.

2053 tests green, tvOS + iOS Simulator builds green.

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

Author: superuser404notfound

Sources/AetherEngine/AetherEngine+Subtitles.swift

@@ -270,6 +270,12 @@ extension AetherEngine {
         guard !subtitleDrainTargets.isEmpty, let store = activeSubtitlePacketStore else { return }
         store.setProtectedStreams(Set(subtitleDrainTargets.values))   // #166: re-assert protection
         let playhead = sourceTime
+        // #416: the pump is rendering the frame at the playhead, so it has necessarily read from
+        // wherever it opened up to here. That is what the pump can state without a hook in its read
+        // loop, and it is exactly the stretch a landing claim rests on: the ground between a set
+        // decoded behind the playhead and the playhead itself. Its forward lookahead is left
+        // unclaimed, which costs a landing line at worst and can never invent coverage.
+        store.noteHarvestProgress(.pump, through: playhead)
         // #271: wall time since the previous tick, so a tick that itself ran long cannot be read as
         // a seek by the next one. See SubtitleOverlayDrainer.drainPlan.
         let tickUptime = Double(DispatchTime.now().uptimeNanoseconds) / 1_000_000_000
@@ -425,6 +431,7 @@ extension AetherEngine {
                         let applied = applySubtitleEvent(event, to: &cues, channel: channel)
                         tally.admitted += applied.admitted
                         tally.published += applied.published
+                        tally.landingWithheld += applied.landingWithheld   // #416
                         if applied.changed { didMutate = true }
                     }
                 }
@@ -788,6 +795,10 @@ extension AetherEngine {
             demuxer: demuxer, to: startAt, anchorStreamIndex: seekAnchor,
             fallbackDuration: engineDisplayDuration,
             timeout: Self.sideReaderSeekBudgetSeconds)
+        // #416: this session reads forwards from here, and nothing below it is this reader's to
+        // claim. Stated even when the positioning fell back or failed: what the loop then reads is
+        // still forwards from wherever it sits, and the anchor is the earliest it can be.
+        store.noteHarvestAnchor(.prefetch, at: startAt)
         if landed != .seek {
             EngineLog.emit(
                 "[AetherEngine] #151 forward prefetch seek to \(String(format: "%.2f", startAt))s timed out "
@@ -963,9 +974,21 @@ extension AetherEngine {
         // #112/#143: during a reconstruction pass any decoded composition at/behind the playhead becomes the
         // held active-line candidate, emitted once when the decode reaches the playhead (see
         // PGSStaleArrivalGate.admitDuringReconstruction).
+        // #416: a bitmap set decoded behind the playhead claims to be the line still on screen
+        // there, and that claim rests on the store being empty between the two. Ask the harvest
+        // whether it ever read that stretch before reading its silence as an answer.
+        // Asked for bitmap events only: a text cue carries its own duration, so it claims nothing
+        // about the ground behind it, and a dense text track would pay the lookup per packet.
+        let groundIsRead = event.isPGS
+            ? subtitleLandingGroundIsRead(cues: event.cues, playhead: sourceTime)
+            : true
+        // A false answer here means an actual refusal: the helper returns true when the event
+        // carries no cue behind the playhead, so there is nothing that could have been refused.
+        if !groundIsRead { applied.landingWithheld += 1 }
         let admitted = pgsStaleArrivalGates[channel, default: PGSStaleArrivalGate()]
             .admit(cues: event.cues, isPGS: event.isPGS,
-                   isSelfContained: event.isSelfContainedPGS, playhead: sourceTime)
+                   isSelfContained: event.isSelfContainedPGS, playhead: sourceTime,
+                   groundIsRead: groundIsRead)
         applied.admitted += admitted.count
         for cue in admitted {
             if insertSorted(cue, into: &cues) {
@@ -983,6 +1006,22 @@ extension AetherEngine {
         Self.insertCueSorted(cue, into: &cues, nextID: &nextRetainedSubtitleCueID)
     }
 
+    /// #416: was the source between the newest of these cues that lies behind `playhead` and the
+    /// playhead itself actually read by some harvest run?
+    ///
+    /// Only that one cue matters: it is the one the gate would make the landing's active line, and
+    /// the only stretch its claim depends on is the one between it and the playhead. Cues at or
+    /// after the playhead need no ground, and an event carrying none behind it asks nothing here.
+    ///
+    /// Coverage is a property of the SOURCE, not of a stream: a reader demuxes every stream it
+    /// passes over, so a span one of them read is read for all of them.
+    private func subtitleLandingGroundIsRead(cues: [SubtitleCue], playhead: Double) -> Bool {
+        guard let store = activeSubtitlePacketStore else { return true }
+        guard let newestBehind = cues.lazy.map(\.startTime).filter({ $0 <= playhead }).max()
+        else { return true }
+        return store.hasReadSpan(from: newestBehind, through: playhead)
+    }
+
     /// #107: close every non-image cue (text or rich text) whose window covers `trimAt` (teletext
     /// page-state semantics: each page transmission or erase replaces what came before it). Image
     /// cues are untouched; they have their own PGS trim. Static and pure for unit tests.

Sources/AetherEngine/AetherEngine.swift

@@ -3863,6 +3863,7 @@ public final class AetherEngine: ObservableObject {
             // Live SW: drive the host's ring-backed DVR reseed directly; no AVPlayer-clock translation applies.
             if softwareHost != nil, nativeHost == nil {
                 EngineLog.emit("[AetherEngine] SW live seek target=\(target)", category: .engine)
+                softwareSubtitlePacketStore?.noteHarvestAnchor(.pump, at: target)   // #416
                 await softwareHost?.seek(to: target)
                 guard loadGeneration == loadGen, seekGeneration == seekGen else { return }
                 clock.currentTime = target
@@ -3924,6 +3925,11 @@ public final class AetherEngine: ObservableObject {
         } else if let host = audioHost {
             hostReposition = await host.seek(to: clockTarget)
         } else if let host = softwareHost {
+            // #416: the software pump is this path's subtitle harvest, and it reads forwards from
+            // wherever this reposition puts it. Everything between where it had got to and here is
+            // ground nobody read; the drain must not read an empty store there as an authored
+            // silence. Stated before the seek: the demuxer lands at or before the target.
+            softwareSubtitlePacketStore?.noteHarvestAnchor(.pump, at: clockTarget)
             hostReposition = await host.seek(to: clockTarget)
         } else {
             // #93 retest: remember the target as recovery intent BEFORE awaiting; a wedged seek

Sources/AetherEngine/Diagnostics/SubtitleDeliveryStatement.swift

@@ -99,6 +99,12 @@ enum SubtitleDeliveryStatement {
         /// store on a different horizon. A window carrying a wrong end with `gapAt` absent said
         /// nothing at all until this count existed (the reporter's observation, exactly).
         var endsWithheld = 0
+        /// #416: bitmap events this tick refused to let claim the landing, because the source
+        /// between the set and the playhead was never read and an empty store there is not
+        /// evidence of anything. Its own field for the same reason `endsWithheld` has one: a
+        /// refusal that only shows up as a missing cue is indistinguishable from a delivery that
+        /// never arrived, and those two have opposite next steps.
+        var landingWithheld = 0
 
         /// Cues an event carried that the gate did not pass. The subtraction is safe: `admitted`
         /// can exceed the tick's own `cues` when a finalized candidate seeded by an earlier tick is
@@ -131,6 +137,9 @@ enum SubtitleDeliveryStatement {
         var admitted = 0
         /// Cues the insert took. `admitted - published` is the re-decode the store deduped.
         var published = 0
+        /// #416: events whose behind-playhead cues were refused the landing for want of read
+        /// ground. Counted here so the tick can total it into the delivery line.
+        var landingWithheld = 0
     }
 
     struct Statement: Equatable, Sendable {
@@ -178,6 +187,9 @@ enum SubtitleDeliveryStatement {
         if tally.endsWithheld > 0 {
             trailing.append("endsWithheld=\(tally.endsWithheld)")
         }
+        if tally.landingWithheld > 0 {
+            trailing.append("landingWithheld=\(tally.landingWithheld)")
+        }
         return (fields + trailing).joined(separator: " ")
     }
 

Sources/AetherEngine/Subtitles/Issue100PGSStaleArrival.swift

@@ -79,13 +79,26 @@ struct PGSStaleArrivalGate {
     /// delegated to `admitDuringReconstruction`, which holds the lead-in's compositions and emits only the line
     /// group active at the playhead. Outside reconstruction the #100 stale hold governs: a catch-up backlog of
     /// arrivals behind the playhead is held for successor resolution and cannot flash through the overlay.
-    mutating func admit(cues: [SubtitleCue], isPGS: Bool, isSelfContained: Bool = false, playhead: Double) -> [SubtitleCue] {
+    /// #416: `groundIsRead` says whether the source between these cues and the playhead was
+    /// actually read by some harvest run. Only a cue BEHIND the playhead needs it, and it needs it
+    /// because that is the whole of its claim: a PGS set has no end of its own, so "still on screen
+    /// at the playhead" means "nothing on this stream happened in between", and an empty store over
+    /// unread ground is not evidence of that. False makes such a cue ineligible, here and in the
+    /// reconstruction pass alike; a cue at or after the playhead is unaffected, it claims nothing
+    /// about ground behind it. Defaults to true so every caller that cannot answer keeps the
+    /// behaviour it had, which is also what an unreporting harvest path gets from the store.
+    mutating func admit(cues: [SubtitleCue], isPGS: Bool, isSelfContained: Bool = false,
+                        playhead: Double, groundIsRead: Bool = true) -> [SubtitleCue] {
         guard isPGS, !cues.isEmpty else { return cues }
         if reconstructing {
-            return admitDuringReconstruction(cues: cues, isSelfContained: isSelfContained, playhead: playhead)
+            return admitDuringReconstruction(cues: cues, isSelfContained: isSelfContained,
+                                             playhead: playhead, groundIsRead: groundIsRead)
         }
         let stale = cues.allSatisfy { $0.startTime < playhead - staleEpsilonSeconds }
         guard stale else { return cues }
+        // #416: a stale arrival publishes only if its window turns out to cover the playhead, which
+        // is the same claim over the same unread ground. Nothing to hold it for.
+        guard groundIsRead else { return [] }
         heldCues = cues
         return []
     }
@@ -104,11 +117,21 @@ struct PGSStaleArrivalGate {
     /// references are missing fails decode and never gets here), and the steady-state path outside reconstruction
     /// already publishes Normal compositions unconditionally. `isSelfContained` stays on the signature: callers
     /// keep reporting the PCS classification, which the epoch-start-aware backscan direction would need.
-    private mutating func admitDuringReconstruction(cues: [SubtitleCue], isSelfContained: Bool, playhead: Double) -> [SubtitleCue] {
+    private mutating func admitDuringReconstruction(cues: [SubtitleCue], isSelfContained: Bool,
+                                                    playhead: Double,
+                                                    groundIsRead: Bool = true) -> [SubtitleCue] {
         // #146: all objects of one display set arrive in a single decode event sharing a start, so
         // the candidate is the newest same-start GROUP behind the playhead; a same-start re-seed is
         // a re-decode of the same set and replaces the group wholesale.
-        let behind = cues.filter { $0.startTime <= playhead }
+        // #416: only a set whose ground up to the playhead was read can be that candidate. The
+        // reporter's case is a set the harvest reached just before it was re-aimed somewhere else,
+        // whose own clear sits in the stretch that was then skipped: it decodes at the landing, no
+        // successor is stored between it and the playhead because nobody looked, and the pass emits
+        // half a minute of history over the new scene. Refusing costs the landing line in the case
+        // where the set really is still up and the proof is missing, which needs a dwell long
+        // enough to span the whole unread stretch; publishing costs it in every case where it is
+        // not, which is every normally authored set in that stretch.
+        let behind = groundIsRead ? cues.filter { $0.startTime <= playhead } : []
         if let newestStart = behind.map(\.startTime).max(),
            reconstructionCandidates.first.map({ newestStart >= $0.startTime }) ?? true {
             reconstructionCandidates = behind.filter { $0.startTime == newestStart }

Sources/AetherEngine/Subtitles/SubtitleForwardPrefetcher.swift

@@ -21,6 +21,11 @@ enum SubtitleForwardPrefetcher {
     /// not already behind the read head when the first packet arrives (#112 round 10).
     static let anchorBackscanSeconds: Double = 2.0
 
+    /// #416: step between two coverage notes, in seconds of source. Small against every authored
+    /// dwell (so a hole of subtitle size is always visible as one), large against a packet (so a
+    /// dense track does not take the store's lock per packet for it).
+    static let coverageNoteStepSeconds: Double = 0.5
+
     /// Where a positioning attempt landed. The byte estimate is a fallback, not a failure: the
     /// reader keeps working from wherever it put the cursor, it just cannot trust timestamps.
     enum Positioning: Equatable {
@@ -228,6 +233,10 @@ enum SubtitleForwardPrefetcher {
         var timeBaseCache: [Int32: AVRational] = [:]
         var timeBaseFailures = 0
         var exit = Exit.cancelled
+        /// #416: the last read position reported to the store's coverage ledger. Reported in steps
+        /// rather than per packet: the ledger only has to be able to tell a hole of authored size
+        /// from continuous reading, and a step keeps this off the store's lock on a dense track.
+        var lastCoverageNoted = -Double.infinity
         /// #240: the valve's grant window. Set when a yield hit the cap, checked before the next
         /// arbitration so the reader keeps the link for a while rather than for one packet.
         var valveGrantedUntil: DispatchTime? = nil
@@ -287,6 +296,12 @@ enum SubtitleForwardPrefetcher {
                 // Stamped here rather than at request time so the window between the two reads
                 // as unresolved instead of validating a position the seek has not reached yet.
                 SubtitlePrefetchTelemetry.recordReanchor(seekGeneration: target.seekGeneration)
+                // #416: the run that was reading up to here is over, and everything between where
+                // it got to and this anchor is ground nobody read. Stated as the anchor the reader
+                // ASKED for, not where the seek landed: a subtitle-axis seek lands at or before the
+                // request, so this under-claims by up to one authored gap and never over-claims.
+                store.noteHarvestAnchor(.prefetch, at: target.seconds)
+                lastCoverageNoted = -Double.infinity
                 anchorGraceUntil = DispatchTime.now()
                     + (link?.anchorGraceSeconds ?? SideReaderLinkPolicy.anchorGraceSeconds)
                 if let fresh = await playhead() { playheadSnapshot = fresh }
@@ -354,6 +369,12 @@ enum SubtitleForwardPrefetcher {
             // `prefetchLead` tracks the reader rather than the last cue, so on a sparse track it
             // now moves between cues instead of standing still.
             SubtitlePrefetchTelemetry.recordPacket(seconds: position, harvested: harvested)
+            // #416: this reader has now read to here. The pacing packets are what make the claim
+            // continuous over an authored silence, which is the stretch the ledger exists for.
+            if position >= lastCoverageNoted + coverageNoteStepSeconds {
+                store.noteHarvestProgress(.prefetch, through: position)
+                lastCoverageNoted = position
+            }
             var didPark = false
             while !Task.isCancelled, position > playheadSnapshot + leadSeconds {
                 if !didPark {