import json
import os
import math
from pathlib import Path
import plistlib
import selectors
import shutil
import subprocess
import sys
import tempfile
import time
import uuid

from urllib.parse import urlsplit

from streaming import StreamingServer, resolved_fixtures


class StartupStreamingServer(StreamingServer):
    def __init__(self, fixtures):
        super().__init__(fixtures)
        self.subtitle_requests = [0, 0]

    def _get(self, request):
        parsed = urlsplit(request.path)
        if parsed.path in ("/startup-0.vtt", "/startup-1.vtt"):
            index = int(parsed.path[len("/startup-")])
            if parsed.query != self.resolved_token or request.headers.get("X-Subtitle") != f"track-{index}":
                self._reply(request, 403, b"Missing subtitle credentials")
                return
            with self.lock:
                self.subtitle_requests[index] += 1
            body = f"WEBVTT\n\n00:00:00.000 --> 00:01:00.000\nIndependent subtitle {index}\n".encode()
            self._reply(request, 200, body, "text/vtt")
            return
        super()._get(request)


def check_cli_instances(root, app, fixtures):
    cli = root / "terebi"
    cli.symlink_to(app / "Contents/MacOS/terebi-cli")
    extractor = app / "Contents/MacOS/yt-dlp-offline"
    (app / "Contents/MacOS/yt-dlp").rename(extractor)
    runtime = app / "Contents/MacOS/deno"
    canonical = "https://www.youtube.com/watch?v=aqz-KE-bpKQ"
    reported_urls = [f"https://www.youtube.com/watch?v=progress00{index}" for index in range(1, 5)]

    config = root / "xdg/terebi/config.toml"
    environment = dict(os.environ, XDG_CONFIG_HOME=str(config.parent.parent))

    def invoke(arguments, expected_code=0):
        result = subprocess.run([str(cli), *arguments], capture_output=True, text=True, timeout=20,
                                env=environment)
        if result.returncode != expected_code:
            raise RuntimeError(f"CLI {arguments} returned {result.returncode}, expected {expected_code}; {result.stdout}{result.stderr}")
        return result

    def process_diagnostics(pid):
        try:
            return (root / f"stderr-{pid}.log").read_text(errors="replace")
        except FileNotFoundError:
            return "<not published>"

    def wait_states(count, ready, message, allow_errors=()):
        deadline = time.monotonic() + 30
        states = {}
        while time.monotonic() < deadline:
            states = {int(path.stem.split("-")[1]): json.loads(path.read_text())
                      for path in root.glob("state-*.json")}
            for pid in states:
                try:
                    os.kill(pid, 0)
                except ProcessLookupError:
                    raise RuntimeError(f"{message}: GUI PID {pid} exited unexpectedly; "
                                       f"last state: {states[pid]}; diagnostics: {process_diagnostics(pid)}")
            complete = len(states) == count and all(
                set(state["pids"]) == set(states) and state["windows"] == 1 for state in states.values())
            if any(state["error"] for pid, state in states.items() if pid not in allow_errors):
                raise RuntimeError(f"{message}: native playback error: {states}")
            if complete and ready(states):
                return states
            time.sleep(0.02)  # Infrastructure deadline; assertions use native events/PTS, not wall dates.
        raise RuntimeError(f"{message}: native state deadline: {states}; "
                           f"process diagnostics: { {pid: process_diagnostics(pid) for pid in states} }")
    def settings(state):
        return (state["showBufferInfo"], state["automaticQuality"], state["snapToEdges"])

    def write_request(pid, events=None, media=None, action=None, **arguments):
        request = {"id": uuid.uuid4().hex, "events": events or [], **arguments}
        if media is not None:
            request["media"] = str(media)
        if action is not None:
            request["action"] = action
        path = root / f"keys-{pid}.json"
        pending = path.with_suffix(".tmp")
        pending.write_text(json.dumps(request))
        pending.replace(path)
        return request["id"]

    def input_request(pid, count, ready, message, events=None, media=None, action=None,
                      allow_error=False, **arguments):
        request_id = write_request(pid, events=events, media=media, action=action, **arguments)
        return wait_states(count, lambda states: states[pid]["requestID"] == request_id and ready(states[pid]),
                           f"{message} (expected request {request_id})", allow_errors=(pid,) if allow_error else ())

    def read_reports(path, allowed):
        try:
            lines = path.read_bytes().split(b"\n")[:-1]  # The writer may still be filling the trailing line.
        except FileNotFoundError:
            return []
        records = [json.loads(line) for line in lines if line]
        miruro = isinstance(allowed, list)
        identity_field = "media_identity" if miruro else "media_url"
        fields = {"version", "event", identity_field, "position_seconds", "duration_seconds", "completion_eligible"}
        for record in records:
            duration = record.get("duration_seconds")
            position = record.get("position_seconds")
            valid = (set(record) == fields and record["version"] == (2 if miruro else 1)
                     and record["event"] in ("progress", "ended", "closed")
                     and record[identity_field] in allowed
                     and type(position) in (int, float) and math.isfinite(position) and position >= 0
                     and (duration is None or (type(duration) in (int, float)
                          and math.isfinite(duration) and duration > 0))
                     and type(record["completion_eligible"]) is bool
                     and (not record["completion_eligible"] or duration is not None))
            if not valid:
                raise RuntimeError(f"invalid or mixed native progress stream {path}: {record}")
        return records

    def wait_report(path, allowed, ready, message):
        deadline = time.monotonic() + 30
        records = []
        while time.monotonic() < deadline:
            records = read_reports(path, allowed)
            if ready(records):
                return records
            time.sleep(0.02)
        raise RuntimeError(f"{message}: native report deadline {path}: {records}")

    def at_position(state, position):
        return (state["status"] == 2 and not state["busy"] and not state["seekPending"]
                and abs(state["position"] - position) < 0.5)

    def progress_at(record, url, position):
        return (record["event"] == "progress" and record["media_url"] == url
                and abs(record["position_seconds"] - position) < 0.5)

    def watch_fact(record):
        return (record["completion_eligible"] and record["duration_seconds"] is not None
                and record["position_seconds"] / record["duration_seconds"] >= 0.95)

    def quit_instance(pid, count, events=None):
        if events is None:
            write_request(pid, action="quit")
        else:
            write_request(pid, events=events)
        deadline = time.monotonic() + 30
        while time.monotonic() < deadline:
            survivors = [json.loads(path.read_text()) for path in root.glob("state-*.json")
                         if path.stem != f"state-{pid}"]
            exited = False
            if count == 1:
                try:
                    os.kill(pid, 0)
                except ProcessLookupError:
                    exited = True
            if exited or (survivors and all(pid not in state["pids"] for state in survivors)):
                (root / f"state-{pid}.json").unlink()
                (root / f"keys-{pid}.json").unlink()
                return wait_states(count - 1, lambda _: True, "normal quit unregisters only the targeted GUI")
            time.sleep(0.02)
        raise RuntimeError(f"normal quit did not unregister GUI PID {pid}")

    def check_resolved_instances(server, count):
        originals = wait_states(count, lambda _: True, "existing players are ready before resolved launches")
        states = originals
        resolved_pids = []
        resolved_reports = []
        identities = []
        for index in range(2):
            source = root / f"resolved source {index}.json"
            report = root / f"resolved progress {index}.jsonl"
            identity = {"provider": "miruro", "media_id": "startup_opaque", "episode": index + 1}
            source.write_text(json.dumps({
                "version": 1,
                "url": server.base_url + "/resolved/encrypted/master.m3u8?" + server.resolved_token,
                "headers": {"X-Terebi-Fixture": "combined", "Authorization": "Bearer " + server.resolved_secret},
                "identity": identity, "title": f"Independent resolved episode {index + 1}",
                "subtitles": [{"url": server.base_url + f"/startup-{index}.vtt?" + server.resolved_token,
                               "format": "vtt", "language": "en", "label": f"Track {index}",
                               "default": True, "headers": {"X-Subtitle": f"track-{index}"}}]
            }))
            source.chmod(0o600)
            previous = set(states)
            invoke(["--new-instance", "--source-file", str(source), "--progress-file", str(report)])
            states = wait_states(count + index + 1, lambda snapshots: any(
                pid not in previous and state["media"] == f"Independent resolved episode {index + 1}"
                and state["status"] == 3 and state["hasVideo"] and state["hasAudio"]
                and state["canSeek"] and not state["busy"] and not state["seekPending"]
                and state["subtitleLabels"] == [f"Track {index}"] and state["selectedSubtitle"] == 0
                and state["subtitleText"] == f"Independent subtitle {index}" and not state["subtitleError"]
                for pid, state in snapshots.items()), "actual source-file CLI reaches registered GUI HLS and VTT playback")
            pid = next(pid for pid in states if pid not in previous)
            states = input_request(pid, count + index + 1, lambda state: state["status"] == 2,
                                   "resolved GUI pauses independently after decoded startup", action="pause")
            wait_report(report, [identity], lambda rows: any(
                row["event"] == "progress" and row["completion_eligible"] for row in rows),
                "resolved GUI owns v2 stable-identity reporting after CLI exit")
            resolved_pids.append(pid)
            resolved_reports.append(report)
            identities.append(identity)
        for pid, original in originals.items():
            if states[pid]["media"] != original["media"] or settings(states[pid]) != settings(original):
                raise RuntimeError("resolved launches disturbed an existing player")
        with server.lock:
            stats = server.resolved_stats.get("encrypted", {})
            if any(server.subtitle_requests[index] == 0 for index in range(2)) or any(
                    stats.get(kind, {}).get("requests", 0) == 0 or stats[kind]["denied"]
                    for kind in ("master", "variant", "key", "segment")):
                raise RuntimeError("actual CLI playback lost complete HLS URLs, credentials, or independent VTT headers")
        for index in reversed(range(2)):
            pid = resolved_pids[index]
            quit_instance(pid, count + index + 1)
            wait_report(resolved_reports[index], [identities[index]],
                        lambda rows: bool(rows) and rows[-1]["event"] == "closed",
                        "resolved GUI quit drains only its stable episode identity")
            report_bytes = resolved_reports[index].read_bytes()
            if not report_bytes.endswith(b"\n") or any(secret in report_bytes for secret in (
                    server.resolved_token.encode(), server.resolved_secret.encode(), server.base_url.encode())):
                raise RuntimeError("resolved report is incomplete or exposes transport secrets")
        wait_states(count, lambda _: True, "original players survive independent resolved GUI quits")
        print("PASS actual resolved CLI handoff: two isolated GUI episodes, authenticated encrypted HLS, independent default VTT, v2 progress and quit")

    resolved_fixtures(fixtures)
    with StartupStreamingServer(fixtures) as server:
        if extractor and runtime:
            choices = []
            for index, url in enumerate([canonical, *reported_urls[:3]]):
                metadata = root / f"youtube-{index}.json"
                metadata.write_text(json.dumps({
                    "id": url.split("v=")[1],
                    "title": "Independent YouTube fixture" if index == 0 else f"Reported YouTube fixture {index}",
                    "duration": 120, "is_live": False, "webpage_url": url, "extractor": "youtube",
                    "extractor_key": "Youtube", "formats": [{
                        "format_id": str(height), "height": height, "width": height * 16 // 9,
                        "url": server.base_url + "/media/cache.mp4", "protocol": "http", "ext": "mp4",
                        "vcodec": "mpeg4", "acodec": "aac", "dynamic_range": "SDR",
                        "http_headers": {"X-Terebi-Fixture": "combined"}} for height in (180, 360)]}))
                choices.append(f"    if (!strcmp(url, {json.dumps(url)})) info = {json.dumps(str(metadata))};")
            wrapper = app / "Contents/MacOS/yt-dlp"
            source = root / "offline-extractor.c"
            source.write_text(
                '#include <stdio.h>\n#include <string.h>\n#include <unistd.h>\n'
                'int main(int argc, char **argv) {\n'
                '    if (argc < 3) return 1;\n'
                '    const char *url = argv[argc - 1];\n    char *info = NULL;\n'
                + "\n".join(choices)
                + '\n    if (!info) { fputs("Fixture extraction deliberately failed\\n", stderr); return 1; }\n'
                + f'    char *args[argc + 2]; args[0] = {json.dumps(str(extractor))};\n'
                + '    args[1] = "--load-info-json"; args[2] = info;\n'
                + '    for (int i = 1; i < argc - 2; ++i) args[i + 2] = argv[i];\n'
                + '    args[argc] = NULL;\n    execv(args[0], args); perror("execv"); return 1;\n}\n')
            subprocess.run(["xcrun", "clang", "-O2", str(source), "-o", str(wrapper)], check=True)
            subprocess.run(["codesign", "--force", "--sign", "-", str(wrapper)], check=True)
        runtime_settings = {"directory": str(root), "path": str(root) + os.pathsep + os.environ.get("PATH", "")}
        (app / "Contents/Resources/cli-check.json").write_text(json.dumps(runtime_settings))
        subprocess.run(["codesign", "--force", "--sign", "-", str(app)], check=True)
        if os.environ.get("TEREBI_CHECK_RESOLVED_ONLY") == "1":
            check_resolved_instances(server, 0)
            return
        movie = fixtures / "cache.mp4"
        audio = root / "audio with spaces.wav"
        shutil.copy2(fixtures / "audio.wav", audio)

        invoke([str(movie)])
        first = wait_states(1, lambda states: all(state["media"] == movie.name and state["status"] == 3
                            and state["position"] > 0 and state["hasVideo"] for state in states.values()),
                            "cold default CLI opens one actual player window")
        first_pid = next(iter(first))
        if settings(first[first_pid]) != (False, False, True) or config.parent.exists():
            raise RuntimeError(f"missing config must use defaults without creating a file: {first}")
        invoke([])
        wait_states(1, lambda states: states[first_pid]["media"] == movie.name,
                    "warm default CLI reuses the existing process without replacing media")
        input_request(first_pid, 1, lambda state: state["status"] == 2, "default Space pauses once",
                      events=[{"key": " ", "code": 49}])
        input_request(first_pid, 1, lambda state: state["status"] == 2, "held Space cannot oscillate",
                      events=[{"key": " ", "code": 49, "repeat": True}])
        input_request(first_pid, 1, lambda state: state["status"] == 3, "same Space resumes",
                      events=[{"key": " ", "code": 49}])
        input_request(first_pid, 1, lambda state: state["status"] == 3, "rapid pause/play preserves latest intent",
                      events=[{"key": " ", "code": 49}, {"key": " ", "code": 49}])
        input_request(first_pid, 1, lambda state: state["compact"], "P enters the same compact window",
                      events=[{"key": "p", "code": 35}])
        input_request(first_pid, 1, lambda state: state["compact"], "held P cannot exit compact mode",
                      events=[{"key": "p", "code": 35, "repeat": True}])
        input_request(first_pid, 1, lambda state: not state["compact"], "same P restores normal mode",
                      events=[{"key": "p", "code": 35}])

        config.parent.mkdir(parents=True)
        config_text = ('[playback]\nautomatic_quality = true\n[ui]\n"buffer_info" = true # optional display\n'
                       'compact_snap_to_edges = false\n[keys]\nplay = ["cmd+8", "cmd+x", "x"]\n'
                       'pause = ["cmd+8", "cmd+x", "x"]\nshow_buffer_info = "b"\nhide_buffer_info = "b"\n')
        config.write_text(config_text)
        invoke(["--new-instance"])
        second = wait_states(2, lambda states: any(state["media"] == "" and state["status"] == 0
                             and settings(state) == (True, True, False)
                             for pid, state in states.items() if pid != first_pid),
                             "empty independent CLI creates exactly one extra player")
        empty_pid = next(pid for pid in second if pid != first_pid)
        if settings(second[first_pid]) != (False, False, True) or config.read_text() != config_text:
            raise RuntimeError(f"config loading changed an existing session or wrote the file: {second}")
        input_request(empty_pid, 2, lambda state: state["status"] == 3, "configured player loads real video",
                      media=movie)
        input_request(empty_pid, 2, lambda state: state["status"] == 2, "configured X pauses actual playback",
                      events=[{"key": "x"}])
        input_request(empty_pid, 2, lambda state: state["status"] == 2, "old K binding is removed",
                      events=[{"key": "k", "code": 40}])
        input_request(empty_pid, 2, lambda state: state["status"] == 2 and state["editorText"] == "x",
                      "typing X reaches the editor without controlling playback",
                      events=[{"key": "x", "editing": True}])
        input_request(empty_pid, 2, lambda state: state["status"] == 2, "configured Cmd+8 leaves text editing alone",
                      events=[{"key": "8", "code": 28, "flags": 1 << 20, "editing": True}])
        input_request(empty_pid, 2, lambda state: state["status"] == 3, "configured Cmd+X resumes actual playback",
                      events=[{"key": "x", "flags": 1 << 20}])
        input_request(empty_pid, 2, lambda state: state["status"] == 3, "held modified playback chord does not toggle",
                      events=[{"key": "x", "flags": 1 << 20, "repeat": True}])
        input_request(empty_pid, 2, lambda state: state["status"] == 2, "same Cmd+X pauses actual playback",
                      events=[{"key": "x", "flags": 1 << 20}])
        input_request(empty_pid, 2, lambda state: state["status"] == 3, "configured native menu Cmd+8 resumes",
                      events=[{"key": "8", "code": 28, "flags": 1 << 20, "menu": True}])
        input_request(empty_pid, 2, lambda state: state["status"] == 2, "native Pause item applies current state",
                      events=[{"select": "Pause"}])
        input_request(empty_pid, 2, lambda state: state["status"] == 3, "native Play item resumes the same player",
                      events=[{"select": "Play"}])
        input_request(empty_pid, 2, lambda state: not state["showBufferInfo"], "paired B hides buffer info",
                      events=[{"key": "b", "code": 11}])
        input_request(empty_pid, 2, lambda state: state["showBufferInfo"], "same B shows buffer info",
                      events=[{"key": "b", "code": 11}])
        input_request(empty_pid, 2, lambda state: state["status"] == 0, "configured close-file menu restores empty player",
                      events=[{"key": "w", "code": 13, "flags": (1 << 20) | (1 << 17), "menu": True}])
        invoke(["--new-instance", str(audio)])
        third = wait_states(3, lambda states: any(state["media"] == audio.name and state["status"] == 3
                            and state["hasAudio"] and not state["hasVideo"]
                            for pid, state in states.items() if pid not in second),
                            "independent local media reaches only the freshly launched process")
        if third[first_pid]["media"] != movie.name or third[first_pid]["status"] != 3 or third[empty_pid]["media"]:
            raise RuntimeError(f"independent launch disturbed an existing player: {third}")
        config.write_text("[ui]\nbuffer_info = [\n")
        failure = invoke(["--new-instance"], expected_code=78)
        if str(config) not in failure.stderr:
            raise RuntimeError(f"configuration failure omitted its path: {failure.stderr}")
        invoke(["--no-config"])
        unchanged = wait_states(3, lambda _: True, "bypass reuses an existing process without resetting its settings")
        if any(settings(unchanged[pid]) != settings(third[pid]) for pid in third):
            raise RuntimeError(f"warm bypass changed an existing session: {third}; {unchanged}")
        invoke(["--no-config", "--new-instance", str(movie)])
        bypass = wait_states(4, lambda states: any(state["media"] == movie.name and state["status"] == 3
                            and settings(state) == (False, False, True)
                            for pid, state in states.items() if pid not in third),
                            "fresh bypass skips invalid config and plays with built-in settings")
        invoke(["--no-config", "--new-instance", "--help"])
        wait_states(4, lambda _: True, "help skips invalid config without starting another process")
        config.write_text(config_text)
        count = 4
        if extractor and runtime:
            invoke(["--new-instance", "--url", canonical])
            fourth = wait_states(5, lambda states: any(
                state["media"] == "Independent YouTube fixture" and state["status"] == 3
                and state["position"] > 0 and state["hasAudio"] and state["hasVideo"]
                and settings(state) == (True, True, False)
                for pid, state in states.items() if pid not in bypass),
                "independent YouTube playback reaches only one new registered process")
            if fourth[first_pid]["media"] != movie.name or fourth[first_pid]["status"] != 3 or fourth[empty_pid]["media"]:
                raise RuntimeError(f"independent YouTube launch disturbed existing playback: {fourth}")
            count = 5
        else:
            print("SKIP independent YouTube playback: yt-dlp and a supported JS runtime are required")

        check_resolved_instances(server, count)
        existing_report = root / "existing progress.jsonl"
        sentinel = b"existing destination must remain byte-for-byte unchanged\n"
        existing_report.write_bytes(sentinel)
        rejection = invoke(["--new-instance", "--url", reported_urls[0],
                            "--progress-file", str(existing_report)], expected_code=73)
        if str(existing_report) not in rejection.stderr or existing_report.read_bytes() != sentinel:
            raise RuntimeError(f"existing report rejection lost its path or altered data: {rejection.stderr}")
        missing_target = root / "missing progress target.jsonl"
        dangling_report = root / "dangling progress.jsonl"
        dangling_report.symlink_to(missing_target)
        rejection = invoke(["--new-instance", "--url", reported_urls[0],
                            "--progress-file", str(dangling_report)], expected_code=73)
        if str(dangling_report) not in rejection.stderr or not dangling_report.is_symlink() or missing_target.exists():
            raise RuntimeError(f"dangling report rejection altered its destination: {rejection.stderr}")
        originals = wait_states(count, lambda _: True, "report preflight failures never launch a GUI")

        if extractor and runtime:
            report_paths = [root / f"progress session {index} {uuid.uuid4().hex}.jsonl" for index in (1, 2)]
            reporting_pids = []
            allowed_urls = ({reported_urls[0], reported_urls[2]}, {reported_urls[1]})
            states = originals
            for index, path in enumerate(report_paths):
                previous_pids = set(states)
                # Deliberately put the space-containing absolute report path after --url.
                invoke(["--new-instance", "--url", reported_urls[index], "--progress-file", str(path)])
                states = wait_states(count + index + 1, lambda snapshots: any(
                    pid not in previous_pids and state["media"] == f"Reported YouTube fixture {index + 1}"
                    and state["status"] == 3 and state["position"] > 0
                    and state["hasVideo"] and state["canSeek"] and not state["seekPending"]
                    for pid, state in snapshots.items()),
                    "reporting CLI returned before actual registered GUI playback observations")
                pid = next(pid for pid in states if pid not in previous_pids)
                reporting_pids.append(pid)
                wait_report(path, allowed_urls[index], lambda records: any(
                    record["event"] == "progress" and record["media_url"] == reported_urls[index]
                    and record["completion_eligible"] and record["duration_seconds"] > 0
                    for record in records), "registered GUI emits an initial actual decoded report")
            total = count + 2
            first_reporter, second_reporter = reporting_pids
            first_path, second_path = report_paths
            for pid, original in originals.items():
                if states[pid]["media"] != original["media"] or settings(states[pid]) != settings(original):
                    raise RuntimeError(f"reporting launches changed an original nonreporting player: {original}; {states[pid]}")

            before_pause = len(read_reports(first_path, allowed_urls[0]))
            paused = input_request(first_reporter, total, lambda state: state["status"] == 2,
                                   "reporting pause is acknowledged by the actual backend", action="pause")
            paused_position = paused[first_reporter]["position"]
            wait_report(first_path, allowed_urls[0], lambda records: any(
                progress_at(record, reported_urls[0], paused_position) for record in records[before_pause:]),
                "acknowledged playing-to-paused transition reports native position")
            input_request(second_reporter, total, lambda state: state["status"] == 2,
                          "second reporting stream pauses independently", action="pause")

            bookmark = paused[first_reporter]["duration"] * 0.4
            input_request(first_reporter, total, lambda state: at_position(state, bookmark),
                          "paused seek reaches an acknowledged decoded position", action="seek", position=bookmark)
            wait_report(first_path, allowed_urls[0], lambda records: any(
                progress_at(record, reported_urls[0], bookmark) for record in records),
                "seek report contains acknowledged native PTS, not just a requested target")
            before_quality = len(read_reports(first_path, allowed_urls[0]))
            wait_states(total, lambda snapshots: snapshots[first_reporter]["canChangeQuality"],
                        "manual quality control becomes available after the acknowledged seek")
            input_request(first_reporter, total,
                          lambda state: state["youtubeHeight"] == 180 and at_position(state, bookmark),
                          "actual quality reopen preserves the paused acknowledged timeline",
                          action="quality", height=180)
            quality_records = wait_report(first_path, allowed_urls[0], lambda records: any(
                progress_at(record, reported_urls[0], bookmark) for record in records[before_quality:]),
                "same-identity quality reopen emits its acknowledged initial record")
            if any(record["event"] == "closed" for record in quality_records):
                raise RuntimeError(f"quality reopen incorrectly closed its reporting identity: {quality_records}")

            input_request(first_reporter, total, lambda state: state["media"] == "Reported YouTube fixture 3"
                          and state["status"] == 3 and not state["busy"] and not state["seekPending"],
                          "new YouTube identity replaces only its reporting player",
                          action="youtube", url=reported_urls[2])
            wait_report(first_path, allowed_urls[0], lambda records: (
                any(record["event"] == "closed" and record["media_url"] == reported_urls[0]
                    and abs(record["position_seconds"] - bookmark) < 0.5 for record in records)
                and any(record["event"] == "progress" and record["media_url"] == reported_urls[2]
                        for record in records)), "replacement closes cached old identity and begins only the new one")
            input_request(first_reporter, total, lambda state: state["status"] == 0,
                          "semantic close invokes actual native media close", action="close")
            closed = wait_report(first_path, allowed_urls[0], lambda records: any(
                record["event"] == "closed" and record["media_url"] == reported_urls[2] for record in records),
                "explicit close reports the replacement identity")
            before_resume = len(closed)
            input_request(first_reporter, total, lambda state: state["media"] == "Reported YouTube fixture 1"
                          and state["status"] == 3 and not state["busy"] and not state["seekPending"]
                          and abs(state["position"] - bookmark) < 1,
                          "native history resumes a previously acknowledged position",
                          action="youtube", url=reported_urls[0])
            resumed = wait_report(first_path, allowed_urls[0], lambda records: any(
                record["event"] == "progress" and record["media_url"] == reported_urls[0]
                for record in records[before_resume:]),
                "first ready report follows the backend's initial resume acknowledgement")
            initial_resume = next(record for record in resumed[before_resume:]
                                  if record["event"] == "progress" and record["media_url"] == reported_urls[0])
            if abs(initial_resume["position_seconds"] - bookmark) >= 1:
                raise RuntimeError(f"initial resumed report exposed an unacknowledged zero timeline: {initial_resume}")
            input_request(first_reporter, total, lambda state: state["status"] == 2,
                          "resumed reporting stream can pause", action="pause")
            crossing = paused[first_reporter]["duration"] * 0.96
            input_request(first_reporter, total, lambda state: at_position(state, crossing),
                          "seek acknowledges a watch-fact crossing", action="seek", position=crossing)
            wait_report(first_path, allowed_urls[0], lambda records: any(
                record["media_url"] == reported_urls[0] and watch_fact(record) for record in records),
                "watch fact requires eligible actual decoded ratio, not launcher success")
            input_request(first_reporter, total, lambda state: state["status"] == 3,
                          "native play continues from the acknowledged crossing", action="play")
            ended = wait_states(total, lambda snapshots: snapshots[first_reporter]["status"] == 4,
                                "actual playback reaches its decoded end")
            if ended[first_reporter]["position"] / ended[first_reporter]["duration"] < 0.95:
                raise RuntimeError(f"native ended timeline unexpectedly reset before reporting: {ended[first_reporter]}")
            wait_report(first_path, allowed_urls[0], lambda records: any(
                record["event"] == "ended" and record["media_url"] == reported_urls[0] and watch_fact(record)
                for record in records), "ended report retains actual final position rather than bookmark-reset zero")

            input_request(first_reporter, total, lambda state: state["media"] == audio.name
                          and state["status"] == 3 and state["hasAudio"] and not state["hasVideo"],
                          "local audio replacement remains playable without reporting local media", media=audio)
            local_boundary = wait_report(first_path, allowed_urls[0], lambda records: (
                records[-1]["event"] == "closed" and records[-1]["media_url"] == reported_urls[0]
                and watch_fact(records[-1])) if records else False,
                "local replacement closes the final YouTube snapshot")
            input_request(first_reporter, total, lambda state: state["status"] == 2,
                          "unreported local pause uses the same real model", action="pause")
            input_request(first_reporter, total, lambda state: bool(state["error"]) and not state["busy"],
                          "fixture extraction failure reaches the actual GUI error path",
                          action="youtube", url=reported_urls[3], allow_error=True)
            failed_boundary = read_reports(first_path, allowed_urls[0])
            if any(record["event"] != "closed" or record["media_url"] != reported_urls[0]
                   for record in failed_boundary[len(local_boundary):]):
                raise RuntimeError(f"local media or failed resolution emitted a progress observation: {failed_boundary}")
            input_request(first_reporter, total, lambda state: state["media"] == "Reported YouTube fixture 3"
                          and state["status"] == 3 and not state["busy"] and not state["seekPending"]
                          and not state["error"],
                          "successful resolution recovers after an unreported failed open",
                          action="youtube", url=reported_urls[2], allow_error=True)
            wait_report(first_path, allowed_urls[0], lambda records: any(
                record["event"] == "progress" and record["media_url"] == reported_urls[2]
                for record in records[len(failed_boundary):]), "recovery reports only successfully opened identity")

            second_position = states[second_reporter]["duration"] * 0.2
            input_request(second_reporter, total, lambda state: at_position(state, second_position),
                          "second file receives only its own acknowledged seek", action="seek", position=second_position)
            wait_report(second_path, allowed_urls[1], lambda records: any(
                progress_at(record, reported_urls[1], second_position) for record in records),
                "second progress stream never receives first player's media identity")
            input_request(second_reporter, total, lambda state: state["editorText"] == "q",
                          "plain q remains text input rather than Quit",
                          events=[{"key": "q", "code": 12, "editing": True}])
            quit_instance(second_reporter, total, events=[
                {"key": "q", "code": 12, "flags": 1 << 20, "editing": True, "queued": True},
            ])
            final_second = wait_report(second_path, allowed_urls[1], lambda records: (
                records[-1]["event"] == "closed"
                and abs(records[-1]["position_seconds"] - second_position) < 0.5) if records else False,
                "normal GUI quit drains the second player's final acknowledged snapshot")
            quit_instance(first_reporter, total - 1)
            final_first = wait_report(first_path, allowed_urls[0], lambda records: (
                records[-1]["event"] == "closed" and records[-1]["media_url"] == reported_urls[2])
                if records else False, "normal GUI quit drains the first player's current identity")
            if not final_first or not final_second or any(not path.read_bytes().endswith(b"\n") for path in report_paths):
                raise RuntimeError("normal quit left an incomplete JSONL record")
            remaining = wait_states(count, lambda _: True, "original nonreporting GUIs survive both reporting quits")
            for pid, original in originals.items():
                if remaining[pid]["media"] != original["media"] or settings(remaining[pid]) != settings(original):
                    raise RuntimeError(f"reporting actions changed an unrelated session: {original}; {remaining[pid]}")
            print("PASS real progress launch: absolute spaced paths after --url, two isolated GUI streams, decoded initial/pause/seek/resume/quality/replacement/end/quit, eligible watch fact, local and failed-open exclusions")
        else:
            print("SKIP reporting GUI playback: yt-dlp and a supported JS runtime are required")
        before = wait_states(count, lambda _: True, "all independent processes remain registered")
        invoke(["--new-instance", "--help"])
        wait_states(count, lambda _: True, "help never creates an extra application")
        replacement = root / "replacement.mp4"
        shutil.copy2(fixtures / "av.mp4", replacement)
        invoke([str(replacement)])
        after = wait_states(count, lambda states: sum(state["media"] == replacement.name
                            for state in states.values()) == 1,
                            "default media delivery still reuses exactly one existing process")
        changed = [pid for pid in before if before[pid]["media"] != after[pid]["media"]]
        if len(changed) != 1:
            raise RuntimeError(f"default media open changed multiple players: {before}; {after}")
        if config.read_text() != config_text:
            raise RuntimeError("native app startup modified config.toml")
        print("PASS actual CLI instances: native startup config/defaults/invalid-file error/bypass/read-only settings, cold/default reuse, independent empty/local/YouTube, one window per PID, targeted media, original playback retained, help, warm replacement")



app_source, library, fixtures, dependencies_root = map(lambda path: Path(path).resolve(), sys.argv[1:])
with tempfile.TemporaryDirectory(prefix="terebi-startup-") as directory:
    root = Path(directory)
    app = root / "terebi.app"
    shutil.copytree(app_source, app)
    identifier = "local.personal.terebi.startup-check." + uuid.uuid4().hex
    info_path = app / "Contents/Info.plist"
    info = plistlib.loads(info_path.read_bytes())
    info["CFBundleIdentifier"] = identifier
    info_path.write_bytes(plistlib.dumps(info))
    shutil.copy2(library, app / "Contents/Frameworks/libterebi-player.dylib")
    subprocess.run([sys.executable, str(Path(__file__).resolve().parent.parent / "Scripts/bundle.py"),
                    "--app", str(app), "--dependencies", str(dependencies_root)], check=True)
    pipe = root / "result.fifo"
    os.mkfifo(pipe)
    descriptor = os.open(pipe, os.O_RDONLY | os.O_NONBLOCK)
    launch = None
    try:
        with selectors.DefaultSelector() as selector:
            selector.register(descriptor, selectors.EVENT_READ)
            launch = subprocess.Popen(
                ["open", "-g", "-n", "-W", str(app), "--args", str(pipe), str(fixtures)],
                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
            )
            if not selector.select(timeout=30):
                raise RuntimeError("packaged app did not report its startup state")
            result = json.loads(os.read(descriptor, 4096))
        stdout, stderr = launch.communicate(timeout=10)
        if launch.returncode != 0 or not result["passed"] or result["bundleIdentifier"] != identifier:
            raise RuntimeError(f"packaged startup failed: {result}; {stdout}{stderr}")
        print("PASS packaged bootstrap: registered app PID/identity, in-process player load, actual playback and Now Playing")
        check_cli_instances(root, app, fixtures)
    finally:
        os.close(descriptor)
        subprocess.run(
            ["xcrun", "swift", "-e",
             'import AppKit; for app in NSWorkspace.shared.runningApplications { '
             'if app.bundleIdentifier == CommandLine.arguments[1] { app.forceTerminate() } }',
             identifier],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
        )
        if launch is not None and launch.poll() is None:
            launch.terminate()
            launch.communicate(timeout=5)
        subprocess.run(
            ["/System/Library/Frameworks/CoreServices.framework/Frameworks/LaunchServices.framework/Support/lsregister", "-u", str(app)],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
        )
        for domain in [identifier, identifier + ".history"]:
            subprocess.run(["defaults", "delete", domain], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)