Harden experimental P7 fallback initialization identity

Labstream/Shared/Debug/DebugPlaybackScenario.swift

@@ -64,15 +64,19 @@ enum DebugPlaybackScenario {
                                    timeoutSeconds: Int,
                                    sessionIsCurrent: () -> Bool = { true },
                                    playbackFailed: () -> Bool = { false },
-                                   consentPending: () -> Bool = { false }) async throws {
-        let deadline = ContinuousClock.now.advanced(by: .seconds(timeoutSeconds))
-        while ContinuousClock.now < deadline {
+                                   consentPending: () -> Bool = { false },
+                                   now: () -> ContinuousClock.Instant = { .now },
+                                   poll: (Duration) async throws -> Void = { try await Task.sleep(for: $0) }) async throws {
+        // Injection is only for deterministic DEBUG tests. Live callers retain the
+        // continuous-clock deadline and 250ms cancellation-aware polling interval.
+        let deadline = now().advanced(by: .seconds(timeoutSeconds))
+        while now() < deadline {
             try Task.checkCancellation()
             guard sessionIsCurrent() else { throw Blocked(reason: .backendChanged) }
             if playbackFailed() { throw Blocked(reason: .playbackFailed) }
             if consentPending() { throw Blocked(reason: .consentRequired) }
             if let current = player.currentItem, current !== priorItem { return }
-            try await Task.sleep(for: .milliseconds(250))
+            try await poll(.milliseconds(250))
         }
         throw Blocked(reason: .deadline)
     }

LabstreamTests/PlaybackAgentEvidenceTests.swift

@@ -139,32 +139,122 @@ struct PlaybackAgentEvidenceTests {
         #expect(try FileManager.default.contentsOfDirectory(atPath: directory.path).isEmpty)
     }
 
+    /// Virtual polling separates the identity/deadline oracle from MainActor load.
     @Test func delayedTrackReplacementDoesNotAcceptThePredecessor() async throws {
         let old = AVPlayerItem(asset: AVMutableComposition())
         let replacement = AVPlayerItem(asset: AVMutableComposition())
         let player = AVPlayer(playerItem: old)
-        let task = Task { @MainActor in
-            try await Task.sleep(for: .milliseconds(20))
-            player.replaceCurrentItem(with: nil)
-            try await Task.sleep(for: .milliseconds(20))
-            player.replaceCurrentItem(with: replacement)
-        }
-        defer { task.cancel(); player.replaceCurrentItem(with: nil) }
-        try await DebugPlaybackScenario.waitForReplacement(of: old, player: player, timeoutSeconds: 2)
+        defer { player.replaceCurrentItem(with: nil) }
+        var now = ContinuousClock.now
+        var polls = 0
+        try await DebugPlaybackScenario.waitForReplacement(of: old, player: player, timeoutSeconds: 2,
+            now: { now }, poll: { delay in
+                #expect(delay == .milliseconds(250))
+                polls += 1
+                now = now.advanced(by: delay)
+                switch polls {
+                case 1:
+                    #expect(player.currentItem === old)
+                    // Keep the predecessor for another complete observation.
+                case 2:
+                    #expect(player.currentItem === old)
+                    player.replaceCurrentItem(with: nil)
+                case 3:
+                    #expect(player.currentItem == nil)
+                    player.replaceCurrentItem(with: replacement)
+                default:
+                    Issue.record("Replacement must finish without another poll")
+                }
+            })
+        #expect(polls == 3)
         #expect(player.currentItem === replacement)
-        try await task.value
     }
 
-    @Test func missingReplacementFailsClosed() async throws {
+    @Test(arguments: [false, true])
+    func missingReplacementFailsClosed(detached: Bool) async throws {
         let old = AVPlayerItem(asset: AVMutableComposition())
+        let player = AVPlayer(playerItem: detached ? nil : old)
+        defer { player.replaceCurrentItem(with: nil) }
+        let start = ContinuousClock.now
+        var now = start
+        var polls = 0
+        do {
+            try await DebugPlaybackScenario.waitForReplacement(of: old, player: player, timeoutSeconds: 2,
+                now: { now }, poll: { delay in
+                    #expect(delay == .milliseconds(250))
+                    polls += 1
+                    now = now.advanced(by: delay)
+                })
+            Issue.record("Neither predecessor nor detached item may satisfy replacement")
+        } catch let error as DebugPlaybackScenario.Blocked {
+            #expect(error.reason == .deadline)
+        }
+        #expect(polls == 8)
+        #expect(start.duration(to: now) == .seconds(2))
+    }
+
+    @Test func replacementAtExpiredDeadlineIsNotAccepted() async throws {
+        let old = AVPlayerItem(asset: AVMutableComposition())
+        let replacement = AVPlayerItem(asset: AVMutableComposition())
         let player = AVPlayer(playerItem: old)
         defer { player.replaceCurrentItem(with: nil) }
+        var now = ContinuousClock.now
         do {
-            try await DebugPlaybackScenario.waitForReplacement(of: old, player: player, timeoutSeconds: 0)
-            Issue.record("The predecessor must not satisfy replacement")
+            try await DebugPlaybackScenario.waitForReplacement(of: old, player: player, timeoutSeconds: 2,
+                now: { now }, poll: { _ in
+                    now = now.advanced(by: .seconds(2))
+                    player.replaceCurrentItem(with: replacement)
+                })
+            Issue.record("A late replacement must not bypass the deadline")
         } catch let error as DebugPlaybackScenario.Blocked {
             #expect(error.reason == .deadline)
         }
+        #expect(player.currentItem === replacement)
+    }
+
+    @Test(arguments: [DebugPlaybackScenario.Reason.backendChanged, .playbackFailed, .consentRequired])
+    func replacementCannotBypassSafetyGates(reason: DebugPlaybackScenario.Reason) async throws {
+        let old = AVPlayerItem(asset: AVMutableComposition())
+        let replacement = AVPlayerItem(asset: AVMutableComposition())
+        let player = AVPlayer(playerItem: old)
+        defer { player.replaceCurrentItem(with: nil) }
+        var now = ContinuousClock.now
+        var replaced = false
+        do {
+            try await DebugPlaybackScenario.waitForReplacement(of: old, player: player, timeoutSeconds: 2,
+                sessionIsCurrent: { !(replaced && reason == .backendChanged) },
+                playbackFailed: { replaced && reason == .playbackFailed },
+                consentPending: { replaced && reason == .consentRequired },
+                now: { now }, poll: { delay in
+                    now = now.advanced(by: delay)
+                    player.replaceCurrentItem(with: replacement)
+                    replaced = true
+                })
+            Issue.record("A replacement cannot supersede a safety gate")
+        } catch let error as DebugPlaybackScenario.Blocked {
+            #expect(error.reason == reason)
+        }
+        #expect(replaced)
+    }
+
+    @Test func cancelledReplacementWaitDoesNotPoll() async throws {
+        let old = AVPlayerItem(asset: AVMutableComposition())
+        let player = AVPlayer(playerItem: old)
+        defer { player.replaceCurrentItem(with: nil) }
+        let task = Task { @MainActor in
+            do {
+                try await DebugPlaybackScenario.waitForReplacement(of: old, player: player, timeoutSeconds: 2,
+                    poll: { _ in Issue.record("Cancelled wait must not poll") })
+                Issue.record("Cancelled wait must throw")
+            } catch is CancellationError {
+                // Expected before any item observation or suspension.
+            } catch {
+                Issue.record("Unexpected cancellation error: \(error)")
+            }
+        }
+        // Both run on MainActor: cancellation occurs before the task can begin.
+        task.cancel()
+        await task.value
     }
 
     @Test func onlyMetadataSubtitleMechanismsRequireReplacement() {

LabstreamTests/SeasonPlannerIntegrationTests.swift

@@ -60,6 +60,16 @@ struct SeasonPlannerIntegrationTests {
             try #require(await waitUntil {
                 store.record(for: deleted) == nil && !store.isDeletionPending(ratingKey: deleted)
             })
+            // Row removal precedes the artifact worker's terminal index commit. In-memory
+            // absence is not a durability barrier; await the already-submitted revision
+            // before asking a fresh store to prove deletion survived reopening.
+            let deletionTicket = store.currentPersistenceTicket()
+            let flush = await store.flushPersistence(through: deletionTicket, timeout: 10)
+            guard case .committed(let revision) = flush else {
+                Issue.record("Deletion snapshot did not commit: \(flush)")
+                return
+            }
+            #expect(revision >= deletionTicket.revision)
             // A newly opened index must agree before the suspended response is released.
             #expect(DownloadStore(baseDirectory: directory).record(for: deleted) == nil)
         }

PMSKit/Sources/PMSKit/MediaSession/MediaSessionProxy.swift

@@ -228,7 +228,8 @@ public actor MediaSessionProxy {
                     try await p7Session?.admit(data)
                 case .initialization:
                     guard resp.statusCode == 200 else { throw P7HDR10Playlist.Rejection.unsupported }
-                    let normalized = try P7HDR10Initialization.normalize(data)
+                    guard let p7Session else { throw P7HDR10Playlist.Rejection.unsupported }
+                    let normalized = try await p7Session.normalizeInitialization(data, at: upstreamURL)
                     let range: Range<Int>
                     do { range = try P7HDR10Playlist.range(head.value(for: "Range"), length: normalized.count) }
                     catch {

PMSKit/Sources/PMSKit/MediaSession/P7HDR10Session.swift

@@ -7,6 +7,7 @@ actor P7HDR10Session {
     private let playlist: URL
     private var initialization: URL?
     private var segments: Set<URL> = []
+    private var initializationBytes: Data?
 
     init(playlist: URL) { self.playlist = playlist }
 
@@ -17,6 +18,20 @@ actor P7HDR10Session {
         throw P7HDR10Playlist.Rejection.unsupported
     }
 
+    /// Pin the complete validated upstream representation before serving any range.
+    /// A stable MAP URI alone cannot prevent different range requests from receiving
+    /// different parameter sets or audio descriptions. This actor method has no await:
+    /// concurrent fetch completions cannot replace the first admitted representation.
+    func normalizeInitialization(_ data: Data, at url: URL) throws -> Data {
+        guard initialization == url else { throw P7HDR10Playlist.Rejection.unsupported }
+        if let initializationBytes, initializationBytes != data {
+            throw P7HDR10Playlist.Rejection.unsupported
+        }
+        let normalized = try P7HDR10Initialization.normalize(data)
+        initializationBytes = data
+        return normalized
+    }
+
     func admit(_ data: Data) throws {
         let parsed = try P7HDR10Playlist.parse(data, at: playlist)
         guard parsed.initialization != playlist,

PMSKit/Tests/PMSKitTests/P7HDR10InitializationTests.swift

@@ -105,7 +105,86 @@ final class P7HDR10InitializationTests: XCTestCase {
             XCTAssertThrowsError(try P7HDR10Initialization.normalize(input))
         }
     }
+    func testSessionPinsExactInitializationBytesUntilFreshOpen() async throws {
+        let root = URL(string: "https://plex.example.internal/media.m3u8")!
+        let initURL = root.deletingLastPathComponent().appendingPathComponent("init.mp4")
+        let playlist = Data("#EXTM3U\n#EXT-X-MAP:URI=\"init.mp4\"\n#EXTINF:10,\n0.m4s\n".utf8)
+        let session = P7HDR10Session(playlist: root)
+        let original = fixture()
+        // Both configurations independently pass admission. A stable URL is not
+        // sufficient to establish that their byte-range representations are identical.
+        let changed = fixture(audio: audioTrack())
+        let expected = try P7HDR10Initialization.normalize(original)
+        _ = try P7HDR10Initialization.normalize(changed)
+        do {
+            _ = try await session.normalizeInitialization(original, at: initURL)
+            XCTFail("unadmitted initialization accepted")
+        } catch {}
+        try await session.admit(playlist)
+        do {
+            _ = try await session.normalizeInitialization(Data(), at: initURL)
+            XCTFail("malformed initialization accepted")
+        } catch {}
+        let first = try await session.normalizeInitialization(original, at: initURL)
+        XCTAssertEqual(first, expected)
+        do {
+            _ = try await session.normalizeInitialization(changed, at: initURL)
+            XCTFail("changed representation accepted at the same URL")
+        } catch {}
+        try await session.admit(playlist)
+        let repeated = try await session.normalizeInitialization(original, at: initURL)
+        XCTAssertEqual(repeated, expected)
+        let reopened = P7HDR10Session(playlist: root)
+        try await reopened.admit(playlist)
+        let fresh = try await reopened.normalizeInitialization(changed, at: initURL)
+        XCTAssertEqual(fresh, try P7HDR10Initialization.normalize(changed))
+    }
+
     #if canImport(Network)
+    func testProxyRejectsChangedInitializationDuringRangeAndResetsOnReopen() async throws {
+        actor Responses {
+            var initialization: Data
+            init(_ data: Data) { initialization = data }
+            func replace(_ data: Data) { initialization = data }
+            func current() -> Data { initialization }
+        }
+        let original = fixture()
+        let changed = fixture(audio: audioTrack())
+        let responses = Responses(original)
+        let playlist = Data("#EXTM3U\n#EXT-X-MAP:URI=\"init.mp4\"\n#EXTINF:10,\n0.m4s\n".utf8)
+        let proxy = MediaSessionProxy(upstreamFetch: { request in
+            let url = request.url!
+            let body = url.path.hasSuffix("m3u8") ? playlist : await responses.current()
+            return (body, HTTPURLResponse(url: url, statusCode: 200, httpVersion: nil,
+                headerFields: ["Content-Type": url.path.hasSuffix("m3u8") ? "application/vnd.apple.mpegurl" : "video/mp4"])!)
+        }, p7HDR10Fallback: true)
+        let session = URLSession(configuration: .ephemeral)
+        defer { session.invalidateAndCancel() }
+        let root = URL(string: "https://plex.example.internal/media.m3u8")!
+        let first = try await proxy.standUpLoopback(forStream: root)
+        _ = try await session.data(from: first.localURL)
+        let initURL = first.localURL.deletingLastPathComponent().appendingPathComponent("init.mp4")
+        var range = URLRequest(url: initURL)
+        range.setValue("bytes=0-7", forHTTPHeaderField: "Range")
+        let (initial, initialResponse) = try await session.data(for: range)
+        XCTAssertEqual((initialResponse as? HTTPURLResponse)?.statusCode, 206)
+        XCTAssertEqual(initial, Data(try P7HDR10Initialization.normalize(original).prefix(8)))
+        await responses.replace(changed)
+        let (rejected, rejectedResponse) = try await session.data(for: range)
+        XCTAssertEqual((rejectedResponse as? HTTPURLResponse)?.statusCode, 502)
+        XCTAssertTrue(rejected.isEmpty)
+        let (_, fullResponse) = try await session.data(from: initURL)
+        XCTAssertEqual((fullResponse as? HTTPURLResponse)?.statusCode, 502)
+        await proxy.stop(generation: first.generation)
+        let reopened = try await proxy.standUpLoopback(forStream: root)
+        _ = try await session.data(from: reopened.localURL)
+        let freshURL = reopened.localURL.deletingLastPathComponent().appendingPathComponent("init.mp4")
+        let (fresh, freshResponse) = try await session.data(from: freshURL)
+        XCTAssertEqual((freshResponse as? HTTPURLResponse)?.statusCode, 200)
+        XCTAssertEqual(fresh, try P7HDR10Initialization.normalize(changed))
+        await proxy.stop(generation: reopened.generation)
+    }
+
     func testProxyNormalizesOnlyAdmittedInitializationAndServesRanges() async throws {
         let initData = fixture()
         let expected = try P7HDR10Initialization.normalize(initData)

docs/DEVELOPMENT.md

@@ -388,6 +388,16 @@ uv run python scripts/check-docs-mermaid.py
 Run the app suite for the platform affected by a change (all three hosts — iOS, macOS,
 and tvOS — for shared app infrastructure):
 
+On Xcode 27, a printed test-pass summary is not a completed hosted run: require the
+terminal `xcodebuild` result and a readable `.xcresult`. If the runner stops making
+progress, take a bounded process sample before termination. One observed iOS stall
+was inside `collectSimulatorDiagnostics` / `simCtlDiagnose` after the test host exited.
+A diagnostic repeat with `-collect-test-diagnostics never` can separate test execution
+from that ancillary verbose-diagnostics collection; retain the original non-green run
+and state the flag explicitly. It does not waive test failures, replace result bundles
+or prove every runner hang has the same cause. The [P7 review-preparation evidence](https://github.com/jlipworth/Labstream/blob/main/docs/research/dv-p7-decoder-boundary.md#review-preparation-deterministic-replacement-and-persistence-barriers)
+records the bounded observation and its limitations.
+
 For a deterministic smoke, affected-platform, or full plan, use the checked
 [`native-test-matrix.py`](TESTING-STRATEGY.md#native-apple-matrix-driver) driver. The commands below
 remain the direct forms for an individually leased platform lane.

docs/MEDIA-CORPUS-TESTING.md

@@ -179,8 +179,11 @@ server defect. Original/Maximum and Generic-profile production policies are unch
 Metadata-owned audio and subtitle changes restart asynchronously. Their probes wait for
 a nonnil replacement AVPlayerItem before testing readiness and sustained progress, just
 as quality transitions do. Soft AVFoundation/offline subtitle switches do not require
-replacement. The hold still fails if its item changes unexpectedly; it is not weakened
-to tolerate arbitrary restarts. See [issue #312](https://github.com/jlipworth/Labstream/issues/312).
+replacement. Replacement-wait unit tests inject a monotonic clock and poll step to
+exercise predecessor, detached, replacement, deadline and safety-gate states without
+racing independently scheduled fixture sleeps. Live probes retain the continuous-clock
+deadline and 250-millisecond polling interval. The hold still fails if its item changes
+unexpectedly; it is not weakened to tolerate arbitrary restarts. See [issue #312](https://github.com/jlipworth/Labstream/issues/312).
 
 The [historical bounded Plex audit](https://github.com/jlipworth/Labstream/blob/main/docs/evidence/audits/2026-09-13-plex-corpus.md) records the tested
 cohorts, first failures, independent fixes and hardware gates. For focused hosted