feat: retain streaming media and add opt-in automatic quality

Checks/automatic-quality-playback.swift

@@ -0,0 +1,79 @@
+import AppKit
+
+@MainActor
+private final class FixedQualityClock { var now = 1000.0 }
+
+@main
+@MainActor
+struct AutomaticQualityPlaybackChecks {
+    static func require(_ condition: @autoclosure () -> Bool, _ message: String) {
+        if !condition() {
+            FileHandle.standardError.write(Data(("FAIL: " + message + "\n").utf8))
+            exit(1)
+        }
+    }
+    // Only the infrastructure watchdog uses elapsed time. Adaptation uses an injected clock;
+    // playback readiness and network stalls are observable media events, not sleeps.
+    static func until(_ model: PlayerModel, _ message: String, _ condition: () -> Bool) {
+        let deadline = ContinuousClock.now.advanced(by: .seconds(20))
+        while !condition(), ContinuousClock.now < deadline {
+            RunLoop.main.run(until: Date(timeIntervalSinceNow: 0.01))
+        }
+        require(condition(), message + " (height=\(model.youtubeHeight ?? 0), status=\(model.state.status), error=\(model.errorText ?? "none"))")
+    }
+    static func polled(_ model: PlayerModel) {
+        let revision = model.state.revision
+        until(model, "native model processes a fresh playback snapshot") { model.state.revision != revision }
+    }
+    static func main() {
+        _ = NSApplication.shared
+        NSApp.setActivationPolicy(.prohibited)
+        let suite = "terebi.automatic-quality-playback-check.\(UUID().uuidString)"
+        let defaults = UserDefaults(suiteName: suite)!
+        defer { defaults.removePersistentDomain(forName: suite) }
+        let clock = FixedQualityClock()
+        let model = PlayerModel(defaults: defaults, qualityClock: { clock.now })
+        defer { model.shutdown() }
+        model.setVolume(0)
+        require(!model.automaticYouTubeQuality, "automatic quality is opt-in")
+        model.openYouTube("https://www.youtube.com/watch?v=aqz-KE-bpKQ")
+        until(model, "default selects real 1080p and runs out of the gated network prefix") {
+            model.youtubeHeight == 1080 && model.canTogglePlayback &&
+                model.state.status == PLAYER_PLAYING && model.state.position > 0.5 && model.state.buffering
+        }
+        clock.now = 2000
+        polled(model)
+        require(model.youtubeHeight == 1080 && !model.resolvingYouTube,
+                "disabled automatic quality leaves a stalled manual/default stream alone")
+        model.setAutomaticYouTubeQuality(true)
+        polled(model)
+        let position = model.state.position
+        clock.now = 2003
+        until(model, "persistent stall automatically reopens actual 720p media") {
+            model.youtubeHeight == 720 && model.canTogglePlayback && model.state.status == PLAYER_PLAYING &&
+                !model.state.seek_pending && (model.bufferedRange.map { $0.upperBound - model.state.position >= 20 } ?? false)
+        }
+        require(model.state.position >= position && model.state.position < position + 3,
+                "automatic quality reopening preserves current playback progress")
+        polled(model)
+        clock.now = 2063
+        until(model, "sustained health automatically restores actual 1080p media") {
+            model.youtubeHeight == 1080 && model.canTogglePlayback && model.state.status == PLAYER_PLAYING && !model.state.seek_pending
+        }
+        model.setYouTubeQuality(model.youtubeHeight)
+        require(!model.automaticYouTubeQuality && !model.isBusy && model.state.status == PLAYER_PLAYING,
+                "reselecting the current manual resolution disables auto without disrupting playback")
+        model.setAutomaticYouTubeQuality(true)
+        let reopened = PlayerModel(defaults: defaults, qualityClock: { clock.now })
+        require(reopened.automaticYouTubeQuality, "explicit opt-in survives player restart")
+        reopened.shutdown()
+        model.setYouTubeQuality(nil)
+        require(!model.automaticYouTubeQuality && model.youtubeQuality == nil,
+                "default quality is also a manual override")
+        let manual = PlayerModel(defaults: defaults, qualityClock: { clock.now })
+        require(!manual.automaticYouTubeQuality, "manual override persists after restart")
+        manual.shutdown()
+        model.closeFile()
+        print("PASS native adaptive playback: default-off/stalled downshift/healthy upshift/progress preservation/current+default manual override/opt-in persistence")
+    }
+}

Checks/automatic-quality.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Run adaptive PlayerModel playback against real loopback media and offline yt-dlp metadata."""
+import argparse
+import json
+import os
+from pathlib import Path
+import shlex
+import shutil
+import subprocess
+import tempfile
+
+from streaming import StreamingServer
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("binary", type=Path)
+    parser.add_argument("fixtures", type=Path)
+    args = parser.parse_args()
+    extractor = shutil.which("yt-dlp")
+    if extractor is None:
+        extractor = next((str(path) for path in (Path("/opt/homebrew/bin/yt-dlp"), Path("/usr/local/bin/yt-dlp"))
+                          if os.access(path, os.X_OK)), None)
+    if extractor is None:
+        print("SKIP native adaptive playback: yt-dlp is not installed; controlled-clock policy checks still run")
+        return 0
+    required = ("quality-1080.mp4", "quality-720.mp4")
+    if any(not (args.fixtures / name).is_file() for name in required):
+        parser.error("Missing real quality playback fixtures")
+    with StreamingServer(args.fixtures) as server, tempfile.TemporaryDirectory(prefix="terebi-quality-check-") as directory:
+        root = Path(directory)
+        formats = []
+        for height, width in ((720, 1280), (1080, 1920)):
+            path = "/media/quality-720.mp4" if height == 720 else "/stall/quality-low/quality-1080.mp4?after_seconds=2"
+            formats.append({"format_id": str(height), "height": height, "width": width,
+                            "url": server.base_url + path, "protocol": "http", "ext": "mp4",
+                            "vcodec": "mpeg4", "acodec": "aac", "dynamic_range": "SDR",
+                            "http_headers": {"X-Terebi-Fixture": "combined"}})
+        info = {"id": "aqz-KE-bpKQ", "title": "Automatic quality playback fixture", "duration": 120,
+                "is_live": False, "webpage_url": "https://www.youtube.com/watch?v=aqz-KE-bpKQ",
+                "extractor": "youtube", "extractor_key": "Youtube", "formats": formats}
+        metadata = root / "info.json"
+        metadata.write_text(json.dumps(info))
+        wrapper = root / "yt-dlp"
+        wrapper.write_text("#!/bin/bash\nset -eu\nargs=(\"$@\")\nexec " + shlex.quote(extractor)
+                           + " --load-info-json " + shlex.quote(str(metadata))
+                           + ' "${args[@]:0:${#args[@]}-2}"\n')
+        wrapper.chmod(0o755)
+        environment = dict(os.environ, PATH=str(root) + os.pathsep + os.environ.get("PATH", ""))
+        try:
+            result = subprocess.run([str(args.binary.resolve())], env=environment, timeout=90, check=False)
+        except subprocess.TimeoutExpired:
+            raise SystemExit("FAIL: adaptive playback exceeded its infrastructure deadline")
+        return result.returncode
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

Checks/automatic-quality.swift

@@ -0,0 +1,145 @@
+import Foundation
+
+@main
+struct AutomaticQualityChecks {
+    static func require(_ condition: @autoclosure () -> Bool, _ message: String) {
+        if !condition() { fatalError("FAIL: " + message) }
+    }
+
+    static func sample(at now: Double, ahead: Double? = 30, height: Int? = 720) -> AutomaticQuality.Sample {
+        .init(position: 100 + now, duration: 1000, ahead: ahead, height: height,
+              heights: [240, 480, 720, 1080, 2160])
+    }
+
+    static func main() {
+        var policy = AutomaticQuality()
+        for second in 0..<6 {
+            require(policy.choose(sample(at: Double(second), ahead: 3), at: Double(second)) == nil,
+                    "brief low reserve does not change quality")
+        }
+        require(policy.choose(sample(at: 6, ahead: 3), at: 6) == 480, "persistent low reserve steps down one rung")
+        policy.resetHealth() // Automatic reopen resets evidence without losing its cooldown.
+        for second in 7..<36 {
+            require(policy.choose(sample(at: Double(second), ahead: 3, height: 480), at: Double(second)) == nil,
+                    "automatic reopen preserves the 30-second switch cooldown")
+        }
+        require(policy.choose(sample(at: 36, ahead: 3, height: 480), at: 36) == 240,
+                "continued shortage may downshift after cooldown")
+
+        policy = AutomaticQuality()
+        for second in 0..<60 {
+            require(policy.choose(sample(at: Double(second), height: 240), at: Double(second)) == nil,
+                    "upshift waits for sustained healthy recorded playback")
+        }
+        require(policy.choose(sample(at: 60, height: 240), at: 60) == 480,
+                "healthy playback increases exactly one rung")
+        for second in 61..<90 {
+            require(policy.choose(sample(at: Double(second), ahead: 1, height: 480), at: Double(second)) == nil,
+                    "a sudden shortage cannot immediately reverse an upshift")
+        }
+        require(policy.choose(sample(at: 90, ahead: 1, height: 480), at: 90) == 240,
+                "downshift becomes possible after upshift cooldown")
+
+        policy = AutomaticQuality()
+        for second in 0...5 {
+            require(policy.choose(sample(at: Double(second), ahead: 3), at: Double(second)) == nil, "low reserve warmup")
+        }
+        require(policy.choose(sample(at: 6, ahead: 8), at: 6) == nil, "middle reserve clears low-water evidence")
+        for second in 7..<13 {
+            require(policy.choose(sample(at: Double(second), ahead: 3), at: Double(second)) == nil,
+                    "hysteresis requires a fresh continuous shortage")
+        }
+        require(policy.choose(sample(at: 13, ahead: 3), at: 13) == 480, "new shortage has its own threshold")
+        policy = AutomaticQuality()
+        for second in 0...90 {
+            require(policy.choose(sample(at: Double(second), ahead: 19, height: 480), at: Double(second)) == nil,
+                    "middle reserve is not healthy enough for an upshift")
+        }
+
+        for suppression in ["disabled", "paused", "seeking", "opening", "local", "missing height", "empty heights", "EOF"] {
+            policy = AutomaticQuality()
+            for second in 0...150 {
+                let now = Double(second)
+                var input = sample(at: now, ahead: 1)
+                switch suppression {
+                case "disabled": input.enabled = false
+                case "paused": input.playing = false
+                case "seeking": input.seeking = true
+                case "opening": input.opening = true
+                case "local": input.network = false
+                case "missing height": input.height = nil
+                case "empty heights": input.heights = []
+                default: input.duration = input.position + 14
+                }
+                require(policy.choose(input, at: now) == nil, suppression + " suppresses quality switching")
+            }
+        }
+        for interruption in ["pause", "seek", "open"] {
+            policy = AutomaticQuality()
+            require(policy.choose(sample(at: 0), at: 0) == nil, "start healthy evidence")
+            var input = sample(at: 59)
+            input.playing = interruption != "pause"
+            input.seeking = interruption == "seek"
+            input.opening = interruption == "open"
+            require(policy.choose(input, at: 59) == nil, "interrupt healthy evidence")
+            require(policy.choose(sample(at: 60), at: 60) == nil, "resume starts fresh health")
+            require(policy.choose(sample(at: 119), at: 119) == nil, "no pre-interruption health is reused")
+            require(policy.choose(sample(at: 120), at: 120) == 1080, "fresh uninterrupted health can upshift")
+        }
+
+        policy = AutomaticQuality()
+        var sparse = sample(at: 0, ahead: 1)
+        sparse.heights = [2160, 1080, 480, 144, 480, 0, -1]
+        require(policy.choose(sparse, at: 0) == nil, "sparse heights warmup")
+        sparse.position += 6
+        require(policy.choose(sparse, at: 6) == 480, "nearest lower available rung handles sparse heights")
+        policy = AutomaticQuality()
+        sparse.height = 2160
+        require(policy.choose(sparse, at: 0) == 1080, "enabling auto caps an existing high-resolution selection")
+        policy = AutomaticQuality()
+        for second in 0...150 {
+            require(policy.choose(sample(at: Double(second), height: 1080), at: Double(second)) == nil,
+                    "automatic upshift never exceeds 1080p")
+        }
+
+        for live in [false, true] {
+            policy = AutomaticQuality()
+            for second in 0..<120 {
+                let now = Double(second)
+                var input = sample(at: now, ahead: live ? 0 : nil, height: 480)
+                input.live = live
+                input.duration = 0
+                require(policy.choose(input, at: now) == nil, "live/unknown reserve needs 120 seconds of stability")
+            }
+            var input = sample(at: 120, ahead: live ? 0 : nil, height: 480)
+            input.live = live
+            input.duration = 0
+            require(policy.choose(input, at: 120) == 720, "live/unknown reserve can cautiously upshift")
+            policy = AutomaticQuality()
+            input.buffering = true
+            for second in 0..<3 {
+                require(policy.choose(input, at: Double(second)) == nil, "brief live/unknown stall is tolerated")
+            }
+            require(policy.choose(input, at: 3) == 240, "persistent live/unknown stall downshifts")
+        }
+        policy = AutomaticQuality()
+        let frozen = sample(at: 0, ahead: nil)
+        for second in 0..<6 {
+            require(policy.choose(frozen, at: Double(second)) == nil, "position stall first waits for progress then hysteresis")
+        }
+        require(policy.choose(frozen, at: 6) == 480, "lack of playback progress also detects a stall")
+
+        policy = AutomaticQuality()
+        require(policy.choose(sample(at: 0, ahead: 1), at: 0) == nil, "failure scenario warmup")
+        require(policy.choose(sample(at: 6, ahead: 1), at: 6) == 480, "failure scenario requests a switch")
+        policy.extractionFailed(at: 7)
+        policy.resetHealth()
+        for second in 8..<127 {
+            require(policy.choose(sample(at: Double(second), ahead: 1), at: Double(second)) == nil,
+                    "extraction failure preserves old height and backs off for 120 seconds")
+        }
+        require(policy.choose(sample(at: 127, ahead: 1), at: 127) == 480,
+                "failure backoff permits a later attempt, not a retry every poll")
+        print("Automatic quality checks passed")
+    }
+}

Checks/network-playback.mm

@@ -33,8 +33,8 @@ static void validBuffer(PlayerState state) {
     require(std::isfinite(state.duration) && state.duration > 0 &&
             state.buffered_start >= 0 && buffered(state) && state.buffered_end <= state.duration,
             "retained range is nonempty and bounded by finite media duration");
-    require(std::abs(state.buffered_start - state.position) < .00001,
-            "retained range follows the authoritative playback position, not a loaded prefix");
+    require(state.buffered_start <= state.position && state.position <= state.buffered_end,
+            "the retained interval covers the current playhead");
 }
 
 // Only these deadlines use elapsed time: a stuck command/read must fail the check,
@@ -330,20 +330,46 @@ static void prefixBound(PlayerState state, GateState gate, bool splitAudio = fal
             "retained coverage cannot bridge withheld audio/video packets or reordered B-frames");
 }
 
-static void bufferCoverage(Player *player, const std::string &base) {
-    require(openNetwork(player, base, "/media/streaming.mp4", "", 30),
-            "open long network fixture for bounded retained coverage");
-    auto initial = until(player, [](auto s) { return s.status == PLAYER_PAUSED && buffered(s); },
-                         "paused network source exposes actual retained media");
-    require(initial.buffered_end < initial.duration,
-            "bounded queues do not report an entire downloaded movie as retained");
-    require(player_play(player), "play to advance retained network coverage");
-    until(player, [=](auto s) {
-        return s.status == PLAYER_PLAYING && s.position > initial.buffered_end && buffered(s) &&
-               s.buffered_end > initial.buffered_end;
-    }, "retained horizon advances with real playback and refills");
-    require(player_close(player), "close advancing buffered source");
+static void cachedWindow(Player *player, const std::string &base) {
+    const std::string token = "cached-window";
+    require(openNetwork(player, base, "/stall/" + token + "/cache.mp4?after_seconds=100", "", 120),
+            "open a two-minute source with a controlled future network stall");
+    auto initial = until(player, [](auto s) {
+        return s.status == PLAYER_PAUSED && buffered(s) && s.buffered_end >= 59;
+    }, "paused playback prefetches a minute without waiting to start");
+    require(initial.buffered_end < 65, "read-ahead is time-bounded rather than a full download");
+    seekCompleted(player, 50, PLAYER_PAUSED);
+    auto retained = until(player, [&](auto s) {
+        return s.status == PLAYER_PAUSED && buffered(s) && s.buffered_end > 95 &&
+               gateState(base, token).entered;
+    }, "cache follows a new playhead until the future network gate");
+    require(retained.position - retained.buffered_start >= 29,
+            "a reusable thirty-second rewind window survives forward playback");
+    require(retained.buffered_start > 15, "old media is evicted instead of caching the whole movie");
+    const auto before = gateState(base, token);
+    auto rewind = seekCompleted(player, 20.5, PLAYER_PAUSED);
+    require(buffered(rewind), "cached rewind preserves the retained interval");
+    auto after = gateState(base, token);
+    require(after.requests == before.requests && after.ranges == before.ranges,
+            "rewinding within history makes no new HTTP request");
+    stillBlocked(base, token);
+    seekCompleted(player, 45, PLAYER_PAUSED);
+    after = gateState(base, token);
+    require(after.requests == before.requests && after.ranges == before.ranges,
+            "forward seeking inside retained media makes no new HTTP request");
+    stillBlocked(base, token);
+    seekCompleted(player, 5, PLAYER_PAUSED);
+    until(player, [&](auto s) {
+        return s.status == PLAYER_PAUSED && buffered(s) && s.buffered_end > 6 &&
+               gateState(base, token).requests > before.requests;
+    }, "seeking before retained history refills missing media over HTTP");
+    require(player_close(player), "close cache while the original future response remains blocked");
     closed(player);
+    stillBlocked(base, token);
+    std::puts("PASS sliding cache: minute read-ahead/thirty-second history/keyframe replay/cached bidirectional seeks/miss refill/cancel");
+}
+
+static void bufferCoverage(Player *player, const std::string &base) {
 
     const std::string token = "buffer-release";
     require(openNetwork(player, base, "/stall/" + token + "/streaming.mp4", "", 30),
@@ -373,7 +399,7 @@ static void bufferCoverage(Player *player, const std::string &base) {
     }, "released server gate advances retained coverage beyond withheld timestamps");
     require(player_close(player), "close released buffered source");
     closed(player);
-    std::puts("PASS retained coverage: bounded queues/advancing clock/stalled timestamp bounds/B-frame dependencies/gate-release refill");
+    std::puts("PASS retained coverage: supplied timestamp bounds/B-frame dependencies/gate-release refill");
 }
 
 static void blockedPlayback(Player *player, const std::string &base, const std::string &token,
@@ -421,19 +447,17 @@ static void cancellation(Player *player, const std::string &base, const std::str
         until(player, [](auto s) { return s.status == PLAYER_PAUSED && buffered(s); },
               "retain high-timestamp media before invalidating it");
         seekCompleted(player, .5, PLAYER_PAUSED);
-        auto rewound = until(player, [](auto s) {
+        until(player, [](auto s) {
             return s.status == PLAYER_PAUSED && !s.seek_pending && buffered(s);
         }, "backward seek rebuilds retained coverage at the new target");
-        require(rewound.buffered_end < 28,
-                "backward seek drops the old high-water horizon instead of retaining distant media");
         stillBlocked(base, token);
         require(player_close(player), "close resumed formerly stalled source");
         closed(player);
         std::puts(splitAudio
-            ? "PASS stalled secondary audio: required-track bound/seek invalidation/new-target coverage/backward high-water clearing"
+            ? "PASS stalled secondary audio: required-track bound/seek acknowledgement/new-target coverage/backward refill"
             : splitVideo
-            ? "PASS stalled primary video: required-track bound/seek invalidation/new-target coverage/backward high-water clearing"
-            : "PASS stalled combined input: bounded prefix/pause/seek invalidation/new-target coverage/backward high-water clearing");
+            ? "PASS stalled primary video: required-track bound/seek acknowledgement/new-target coverage/backward refill"
+            : "PASS stalled combined input: bounded prefix/pause/seek acknowledgement/new-target coverage/backward refill");
     }
 
     blockedOpening(player, base, "opening-close");
@@ -515,6 +539,7 @@ int main(int argc, char **argv) {
         frameSteps(player.get(), view, base);
         nativeHLS(player.get(), view, base);
         liveAndReplacement(player.get(), base, fixtures);
+        cachedWindow(player.get(), base);
         bufferCoverage(player.get(), base);
         cancellation(player.get(), base, fixtures);
         failures(player.get(), base);

Checks/streaming.py

@@ -32,6 +32,9 @@ class StreamingServer:
     headers = {
         "av.mp4": "combined",
         "streaming.mp4": "combined",
+        "cache.mp4": "combined",
+        "quality-1080.mp4": "combined",
+        "quality-720.mp4": "combined",
         "frame-step.mp4": "combined",
         "video.mp4": "video",
         "frame-step-video.mp4": "video",
@@ -134,6 +137,19 @@ class StreamingServer:
             self.prefixes[key] = {kind: min(end, first_missing[kind]) for kind, end in ends.items()}
         return self.prefixes[key]
 
+    def _timestamp_offset(self, name: str, seconds: float) -> int:
+        probe = subprocess.run(
+            ["ffprobe", "-v", "error", "-show_packets", "-show_format",
+             "-show_entries", "format=start_time:packet=pts_time,pos", "-of", "json",
+             str(self.fixtures / name)],
+            capture_output=True, text=True, check=True, timeout=30,
+        )
+        metadata = json.loads(probe.stdout)
+        origin = float(metadata.get("format", {}).get("start_time", 0))
+        return min(int(packet["pos"]) for packet in metadata["packets"]
+                   if "pos" in packet and "pts_time" in packet
+                   and float(packet["pts_time"]) - origin >= seconds)
+
     def _get(self, request):
         parsed = urlsplit(request.path)
         parts = unquote(parsed.path).strip("/").split("/")
@@ -206,8 +222,10 @@ class StreamingServer:
         if token is not None:
             query = parse_qs(parsed.query)
             try:
-                gate_at = int(query.get("after", ["131072"])[0]) if mode == "stall" else start
-            except ValueError:
+                gate_at = (self._timestamp_offset(name, float(query["after_seconds"][0]))
+                           if mode == "stall" and "after_seconds" in query else
+                           int(query.get("after", ["131072"])[0]) if mode == "stall" else start)
+            except (ValueError, OverflowError):
                 self._reply(request, 400, b"Invalid gate offset", "text/plain")
                 return
             with self.lock:
@@ -270,7 +288,7 @@ def main():
     parser.add_argument("fixtures", type=Path)
     args = parser.parse_args()
     required = ("av.mp4", "video.mp4", "audio.wav", "frame-step-video.mp4",
-                "frame-step-audio.m4a", "streaming.mp4", "network-vod.m3u8")
+                "frame-step-audio.m4a", "streaming.mp4", "cache.mp4", "network-vod.m3u8")
     missing = [name for name in required if not (args.fixtures / name).is_file()]
     if missing:
         parser.error("Missing streaming fixtures: " + ", ".join(missing))