fix(probe): evidence in hand outranks the budget that bounded fetching it

CHANGELOG.md

@@ -23,7 +23,20 @@ the public-API contract.
 - HDR10+ confirmation validates codec metadata structures and the registered ST 2094-40 payload
   instead of matching a marker anywhere in compressed bytes. Playback and probing share the validator;
   Dolby Vision stays primary. Per-pass byte limits now reject oversized packets before inspection or
-  decode, and a result arriving after the pass deadline cannot confirm metadata.
+  decode.
+- A message the HDR10+ validator has parsed in full is not withdrawn by damage elsewhere in the same
+  packet. Malformed framing after a confirmed ST 2094-40 payload ends the walk and reports the
+  confirmation, where it previously discarded it, which cost a real badge whenever a vendor SEI, a
+  trailing byte or a second unreadable NAL sat next to the metadata.
+- The HDR10+ and Atmos detail passes no longer retract a detection they already made because their
+  soft wall-clock budget expired. The budgets bound what a pass spends; a withheld confirmation is
+  indistinguishable to the caller from a source that carries none.
+- A controlled probe opens with the playback analysis budget, clamped by the caller's own
+  `maxInputBytes`, instead of the still extractor's smaller one, so passing `limits` no longer
+  reports fewer streams than the same call without it. The recordless Dolby Vision audit remains
+  unavailable to it: that audit opens the source a second time by URL, outside the probe's budget.
+- A controlled HTTP probe waits for an origin request slot until its own deadline rather than
+  failing with `sourceBusy` the moment another request holds the origin.
 
 ## [7.9.0] - 2026-09-20
 

Sources/AetherEngine/AetherEngine+Probe.swift

@@ -252,9 +252,24 @@ extension AetherEngine {
         return probe
     }
 
+    /// The caller's own limits are the only thing allowed to bind a controlled probe.
+    ///
+    /// Built from `.playback` rather than `.stillExtraction`: a host that asks for limits is asking to
+    /// bound the COST, not to be answered from a shallower read, and a second hidden budget (the still
+    /// extractor's 2 MiB probe and 2 s analysis) would make `probe(url:limits:)` report fewer streams
+    /// than `probe(url:)` on exactly the sparse sources where that matters. `maxInputBytes` clamps the
+    /// probe size, and the deadline plus the FFmpeg interrupt callback bound the analysis instead.
+    /// The profile's AVIO tuning is not read on this path at all: a controlled probe always opens
+    /// through a reader, whose own transport settings live in `ProbeHTTPReader`.
+    ///
+    /// The recordless Dolby Vision audit stays off, and that is a real gap rather than a tuning choice:
+    /// it opens the source a SECOND time by URL (`DolbyVisionRecordAudit.rpuProfileOfSource`), traffic
+    /// this probe's budget and cancellation do not reach, and `open(reader:)` clears `auditSource`
+    /// regardless. A controlled probe of an untagged 10-bit HEVC source therefore reports no Dolby
+    /// Vision record where an uncontrolled one does (AE#567).
     private nonisolated static func probeOpenProfile(limits: ProbeLimits?) -> DemuxerOpenProfile {
         guard let limits else { return .playback }
-        var profile = DemuxerOpenProfile.stillExtraction
+        var profile = DemuxerOpenProfile.playback
         // FFmpeg has a minimum probe size. The input seam still enforces smaller caller limits.
         profile.probesize = max(32, min(profile.probesize, limits.maxInputBytes))
         profile.auditsRecordlessDolbyVision = false

Sources/AetherEngine/Audio/AtmosDetectionProbe.swift

@@ -283,11 +283,9 @@ extension AetherEngine {
             av_packet_free_safe(pkt)
 
             if confirmed {
-                let elapsed = Double(DispatchTime.now().uptimeNanoseconds - start.uptimeNanoseconds) / 1_000_000_000
-                guard elapsed < options.timeBudget else {
-                    return AtmosDetectionOutcome(stopReason: .timeCap, packetsRead: packetsRead,
-                                                 bytesRead: bytesRead, decodedProfile: nil)
-                }
+                // A decoded JOC frame is evidence the pass already paid for. The wall-clock budget bounds
+                // what this pass SPENDS, so an overrun retires the pass, it does not retract its answer:
+                // a caller cannot tell a withheld confirmation apart from a source that carries no Atmos.
                 return AtmosDetectionOutcome(
                     stopReason: .frameDecoded, packetsRead: packetsRead, bytesRead: bytesRead,
                     decodedProfile: lastProfile

Sources/AetherEngine/Demuxer/AVIOReader.swift

@@ -3989,7 +3989,13 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
         }
         guard !isClosed else { throw CancellationError() }
         try probeControl.check()
-        guard let ticket = OriginRequestBudget.shared.tryAcquire(for: url, label: label) else {
+        // Wait for the slot, but never past the probe's own deadline. A slot wait is the one wait the
+        // watchdog cannot interrupt (it fires at reads), so the deadline has to bound it here instead.
+        // Not waiting at all would fail a probe that merely arrived while one other request held the
+        // origin, which is the ordinary shape when a host probes several items off one server.
+        let slotWait = probeControl.remainingTime.map { min(timeout, $0) } ?? timeout
+        guard let ticket = OriginRequestBudget.shared.acquire(
+            for: url, label: label, timeout: slotWait) else {
             probeControl.stop(ProbeError.sourceBusy)
             throw ProbeError.sourceBusy
         }

Sources/AetherEngine/ProbeControl.swift

@@ -191,6 +191,12 @@ final class ProbeControl: @unchecked Sendable {
         catch { return true }
     }
 
+    /// Seconds left on the deadline, or nil when no numeric limits were installed. For the one wait the
+    /// watchdog cannot interrupt (an origin request slot), so it can be bounded by the deadline instead.
+    var remainingTime: TimeInterval? {
+        deadline.map { max(0, $0 - now()) }
+    }
+
     func inputAllowance(_ requested: Int32) throws -> Int32 {
         try check()
         guard let limits else { return requested }

Sources/AetherEngine/Video/HDR10PlusDetectionProbe.swift

@@ -203,14 +203,17 @@ extension AetherEngine {
                 bytesRead += packetBytes
                 found = HDR10PlusMetadataScan.packetCarriesHDR10Plus(pkt, codecParameters: codecpar)
             }
-            guard elapsed() < options.timeBudget else {
+            // Evidence in hand outranks the soft budget. The caps exist to bound what this pass SPENDS,
+            // and a confirmation is already paid for; dropping it would turn an overrun into a false
+            // negative on a slow origin, which the caller cannot tell apart from "this source has none".
+            if found {
                 return HDR10PlusDetectionOutcome(
-                    stopReason: .timeCap, packetsRead: packetsRead, bytesRead: bytesRead)
+                    stopReason: .found, packetsRead: packetsRead, bytesRead: bytesRead)
             }
 
-            if found {
+            guard elapsed() < options.timeBudget else {
                 return HDR10PlusDetectionOutcome(
-                    stopReason: .found, packetsRead: packetsRead, bytesRead: bytesRead)
+                    stopReason: .timeCap, packetsRead: packetsRead, bytesRead: bytesRead)
             }
 
             // AVDISCARD_ALL is advisory, so a container that keeps handing back foreign packets still ends.

Sources/AetherEngine/Video/HDR10PlusMetadataScan.swift

@@ -59,6 +59,10 @@ enum HDR10PlusMetadataScan {
 
     // MARK: - H.264 / HEVC
 
+    /// A walk that runs into malformed framing stops and reports what it had already VALIDATED, rather
+    /// than discarding it. Only `validT35` ever sets the answer, so an aborted walk cannot invent one;
+    /// abandoning a confirmed payload because a later NAL in the same packet is malformed would be a
+    /// false negative on the one carriage this scan exists to find.
     private static func scanNALs(
         _ bytes: UnsafeBufferPointer<UInt8>, codecID: AVCodecID, framing: VideoNALFraming
     ) -> Bool {
@@ -86,13 +90,13 @@ enum HDR10PlusMetadataScan {
         case .lengthPrefixed(let width):
             guard (1...4).contains(width) else { return false }
             var offset = 0
-            while offset < bytes.count {
-                guard bytes.count - offset >= width else { return false }
+            while offset < bytes.count, !found {
+                guard bytes.count - offset >= width else { return found }
                 var count = 0
                 for index in offset..<(offset + width) { count = (count << 8) | Int(bytes[index]) }
                 offset += width
                 guard count > 0, count <= bytes.count - offset,
-                      visit(offset, offset + count) else { return false }
+                      visit(offset, offset + count) else { return found }
                 offset += count
             }
         case .annexB:
@@ -103,15 +107,16 @@ enum HDR10PlusMetadataScan {
                 if byte == 1, zeros >= 2 {
                     let end = index - zeros
                     if let start = nalStart {
-                        guard visit(start, end) else { return false }
+                        guard visit(start, end) else { return found }
+                        if found { return true }
                     } else if end != 0 {
-                        return false
+                        return found
                     }
                     nalStart = index + 1
                 }
                 zeros = byte == 0 ? zeros + 1 : 0
             }
-            guard let start = nalStart, visit(start, bytes.count - zeros) else { return false }
+            guard let start = nalStart, visit(start, bytes.count - zeros) else { return found }
         }
         return found
     }
@@ -138,7 +143,9 @@ enum HDR10PlusMetadataScan {
         return rbsp
     }
 
-    /// Nil denotes malformed SEI framing; false is a well-formed SEI without HDR10+.
+    /// Nil denotes malformed SEI framing; false is a well-formed SEI without HDR10+. A message this
+    /// walk has already VALIDATED outranks framing it cannot finish reading, so nil is only ever the
+    /// answer when nothing was confirmed.
     private static func scanSEI(_ bytes: [UInt8]) -> Bool? {
         var offset = 0
         var found = false
@@ -157,7 +164,7 @@ enum HDR10PlusMetadataScan {
         while offset < bytes.count {
             if offset == bytes.count - 1, bytes[offset] == 0x80 { return found }
             guard let type = extendedValue(), let size = extendedValue(),
-                  size <= bytes.count - offset else { return nil }
+                  size <= bytes.count - offset else { return found ? true : nil }
             if type == 4 {
                 let valid = bytes.withUnsafeBufferPointer {
                     validT35($0.baseAddress! + offset, size: size)
@@ -166,40 +173,41 @@ enum HDR10PlusMetadataScan {
             }
             offset += size
         }
-        return nil // rbsp_trailing_bits is mandatory.
+        return found ? true : nil // rbsp_trailing_bits is mandatory.
     }
 
     // MARK: - AV1 low-overhead OBU stream (demuxed packet framing)
 
+    /// Same contract as `scanNALs`: an aborted walk reports what it validated, never less.
     private static func scanOBUs(_ bytes: UnsafeBufferPointer<UInt8>) -> Bool {
         var offset = 0
         var found = false
-        while offset < bytes.count {
+        while offset < bytes.count, !found {
             let header = bytes[offset]
             offset += 1
-            guard header & 0x81 == 0 else { return false }
+            guard header & 0x81 == 0 else { return found }
             let type = (header >> 3) & 15
             if header & 4 != 0 {
-                guard offset < bytes.count, bytes[offset] & 7 == 0 else { return false }
+                guard offset < bytes.count, bytes[offset] & 7 == 0 else { return found }
                 offset += 1
             }
             let size: Int
             if header & 2 != 0 {
                 guard let declared = leb128(bytes, offset: &offset, end: bytes.count),
-                      declared <= bytes.count - offset else { return false }
+                      declared <= bytes.count - offset else { return found }
                 size = declared
             } else {
                 size = bytes.count - offset
             }
             let end = offset + size
             if type == 5 {
-                guard let metadataType = leb128(bytes, offset: &offset, end: end) else { return false }
+                guard let metadataType = leb128(bytes, offset: &offset, end: end) else { return found }
                 if metadataType == 4 {
                     // AV1 permits arbitrarily many trailing zero bytes after the stop bit.
                     // They belong to the OBU, not to the byte-aligned T.35 payload.
                     var trailing = end
                     while trailing > offset, bytes[trailing - 1] == 0 { trailing -= 1 }
-                    guard trailing > offset, bytes[trailing - 1] == 0x80 else { return false }
+                    guard trailing > offset, bytes[trailing - 1] == 0x80 else { return found }
                     found = validT35(bytes.baseAddress! + offset, size: trailing - offset - 1) || found
                 }
             }

Tests/AetherEngineTests/HDR10PlusMetadataScanTests.swift

@@ -216,9 +216,6 @@ struct HDR10PlusMetadataScanTests {
         for cut in 0..<packet.count {
             #expect(!scan(Array(packet.prefix(cut)), framing: .lengthPrefixed(size: 4)))
         }
-        #expect(!scan(packet + [0], framing: .lengthPrefixed(size: 4)))
-        #expect(!scan(packet + [0, 0, 0, 0], framing: .lengthPrefixed(size: 4)))
-        #expect(!scan(packet + [0xFF, 0xFF, 0xFF, 0xFF], framing: .lengthPrefixed(size: 4)))
         for width in [0, -1, 5, Int.max] {
             #expect(!scan(packet, framing: .lengthPrefixed(size: width)))
         }
@@ -227,12 +224,9 @@ struct HDR10PlusMetadataScanTests {
         #expect(!scan(Self.annexB([0x4E, 0x00] + nal.dropFirst(2))))
         #expect(!scan(Self.annexB([0x66] + Self.sei(hevc: false).dropFirst()), codecID: AV_CODEC_ID_H264))
         #expect(!scan([0xAA] + Self.annexB(nal)))
-        #expect(!scan(Self.annexB(nal) + [0, 0, 1]))
         for badRBSP in [[4, 255], [255], [4, 100, 0xB5, 0x80], [0, 0, 3], [0, 0, 3, 4]] as [[UInt8]] {
             #expect(!scan(Self.annexB([0x4E, 0x01] + badRBSP)))
         }
-        let truncatedAfterMetadata = Self.message(Self.t35()) + [4, 100, 0x80]
-        #expect(!scan(Self.annexB([0x4E, 0x01] + Self.escaped(truncatedAfterMetadata))))
         #expect(!scan(Self.annexB([0x4E, 0x01] + Self.message(Self.t35(windows: 2)) + [0x80])))
         #expect(!HDR10PlusMetadataScan.bytesCarryHDR10Plus(nil, size: 10, codecID: AV_CODEC_ID_HEVC))
         #expect(!scan([]))
@@ -276,7 +270,34 @@ struct HDR10PlusMetadataScanTests {
         #expect(!scan([0x2A] + Array(repeating: 0x80, count: 8), codecID: AV_CODEC_ID_AV1))
         #expect(!scan([0x2A, 0xFF, 0xFF, 0xFF, 0xFF, 0x1F], codecID: AV_CODEC_ID_AV1))
         #expect(!scan(Self.obu([0x80]), codecID: AV_CODEC_ID_AV1))
-        #expect(!scan(packet + [0x2A], codecID: AV_CODEC_ID_AV1))
+    }
+
+    /// The scan answers ONE question: is validated ST 2094-40 metadata present. Damage further along a
+    /// packet says nothing about a message already parsed in full, and treating it as a retraction is a
+    /// false negative on real media, where a vendor SEI or a trailing byte next to the HDR10+ message is
+    /// ordinary. Nothing here can invent a positive: only `validT35` ever sets one.
+    @Test("A message validated in full outranks malformed framing that follows it")
+    func damageAfterAValidatedMessage() {
+        let nal = Self.sei()
+        let lengthPrefixed = Self.lengthPrefixed(nal, width: 4)
+        for junk in [[0], [0, 0, 0, 0], [0xFF, 0xFF, 0xFF, 0xFF]] as [[UInt8]] {
+            #expect(scan(lengthPrefixed + junk, framing: .lengthPrefixed(size: 4)))
+        }
+        // A second NAL the walk cannot read: forbidden_zero_bit set, then temporal_id zero.
+        for bad in [[0x82, 0x01, 0xAA], [0x4E, 0x00, 0xAA]] as [[UInt8]] {
+            #expect(scan(Self.annexB(nal) + Self.annexB(bad)))
+        }
+        // rbsp_trailing_bits missing. Length-prefixed framing, because Annex B strips the zero bytes
+        // ahead of a start code and the payload's own zero padding goes with them, which truncates the
+        // message rather than damaging what follows it.
+        #expect(scan(Self.lengthPrefixed(Array(nal.dropLast()), width: 4), framing: .lengthPrefixed(size: 4)))
+        // A start code with nothing behind it.
+        #expect(scan(Self.annexB(nal) + [0, 0, 1]))
+        // A second SEI message whose declared size runs off the end of the same NAL.
+        let overrunAfterMetadata = Self.message(Self.t35()) + [4, 100, 0x80]
+        #expect(scan(Self.annexB([0x4E, 0x01] + Self.escaped(overrunAfterMetadata))))
+        // AV1: a metadata OBU that validated, followed by an OBU with an unreadable size field.
+        #expect(scan(Self.obu([4] + Self.t35() + [0x80]) + [0x2A], codecID: AV_CODEC_ID_AV1))
     }
 
     private func withPacket(

Tests/AetherEngineTests/HDR10PlusProbeIntegrationTests.swift

@@ -215,7 +215,10 @@ struct HDR10PlusProbeIntegrationTests {
         #expect(clockReads == 3)
     }
 
-    @Test("Metadata found at the deadline is not published as a positive")
+    /// The budget bounds what the pass SPENDS, not what it may report. A confirmation is already paid
+    /// for by the time the clock is read again, and withholding it hands the caller a `false` it cannot
+    /// tell apart from a source that carries no HDR10+ at all.
+    @Test("Metadata found as the deadline passes is still published")
     func deadlineAfterScan() throws {
         let url = try Self.writeFixture(Self.hdr10PlusBase64, name: "scan-deadline")
         defer { try? FileManager.default.removeItem(at: url) }
@@ -228,13 +231,17 @@ struct HDR10PlusProbeIntegrationTests {
             options: HDR10PlusDetectionOptions(timeBudget: 1),
             now: {
                 defer { clockReads += 1 }
+                // The cap check and the post-read check still fall inside the budget; the read that
+                // used to follow the scan, and retract the finding, is the one that would be over it.
                 return clockReads < 3 ? 0 : 1_000_000_000
             })
-        #expect(outcome.stopReason == .timeCap)
-        #expect(!outcome.carriesHDR10Plus)
+        #expect(outcome.stopReason == .found)
+        #expect(outcome.carriesHDR10Plus)
         #expect(outcome.packetsRead == 1)
         #expect(outcome.bytesRead == 91)
-        #expect(clockReads == 4)
+        // The start stamp, the cap check and the post-read check. The fourth read, the one that was
+        // over budget and used to retract the finding, is never taken.
+        #expect(clockReads == 3)
     }
 
     @Test("A read returning EOF after the deadline still reports the time cap")

Tests/AetherEngineTests/ProbeControlTests.swift

@@ -163,6 +163,30 @@ struct ProbeControlTests {
         #expect(interrupts.value == 1)
     }
 
+    /// The origin request slot is the one wait the watchdog cannot interrupt: it fires at reads, and an
+    /// `acquire` is already blocked when it does. So the slot wait is bounded by what is left of the
+    /// deadline instead, and a probe that merely arrived while another request held the origin waits for
+    /// it rather than failing on the spot.
+    @Test("Remaining time bounds the one wait the watchdog cannot interrupt")
+    func remainingTimeTracksTheDeadline() throws {
+        let clock = ProbeTestBox<TimeInterval>(10)
+        let control = try ProbeControl(
+            limits: .init(timeBudget: 5), cancellation: nil,
+            now: { clock.value }, scheduleDeadline: false)
+        defer { control.finish() }
+        #expect(control.remainingTime == 5)
+        clock.update { $0 = 13 }
+        #expect(control.remainingTime == 2)
+        clock.update { $0 = 99 }
+        #expect(control.remainingTime == 0)
+
+        let uncapped = try ProbeControl(
+            limits: nil, cancellation: ProbeCancellation(),
+            now: { clock.value }, scheduleDeadline: false)
+        defer { uncapped.finish() }
+        #expect(uncapped.remainingTime == nil)
+    }
+
     @Test("Completion checks time even without another read or watchdog tick")
     func completionChecksDeadline() throws {
         let clock = ProbeTestBox<TimeInterval>(10)