[BUG] Pip updates can create invalid dist-info / missing RECORD through scope changes and concurrent dependency writes #5488

"""Windows-only standalone CLI reproduction; uses no existing package environment.

Run: python minimal-repro.py [--pip-wheel pip-26.2.1-py3-none-any.whl]
Without --pip-wheel, download the official pip wheel and verify its published SHA.
The temporary runtime is automatically removed after all child processes exit.
No production package, pip source, or UniGetUI configuration is modified.
"""
import argparse
import base64
import csv
import hashlib
import io
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import urllib.request
import zipfile

CORE, CLIENT = "unigetui-probe-core", "unigetui-probe-client"
PIP_VERSION = "26.2.1"
PIP_SHA256 = "71138adf1f4ca900cdb7d289c21b7494329f2332b6d85f0e1c42108c0384ed3e"


def build_wheel(folder, name, version):
    module = name.replace("-", "_")
    info = f"{module}-{version}.dist-info"
    requires = f"Requires-Dist: {CORE} (>= {version}, < 1.0)\n" if name == CLIENT else ""
    payload = {
        module + "/__init__.py": f'__version__ = "{version}"\n'.encode(),
        info + "/METADATA": f"Metadata-Version: 2.1\nName: {name}\nVersion: {version}\n{requires}".encode(),
        info + "/WHEEL": b"Wheel-Version: 1.0\nRoot-Is-Purelib: true\nTag: py3-none-any\n",
    }
    record = io.StringIO()
    writer = csv.writer(record, lineterminator="\n")
    for filename, data in payload.items():
        value = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).rstrip(b"=").decode()
        writer.writerow([filename, "sha256=" + value, len(data)])
    writer.writerow([info + "/RECORD", "", ""])
    payload[info + "/RECORD"] = record.getvalue().encode()
    wheel = folder / f"{module}-{version}-py3-none-any.whl"
    with zipfile.ZipFile(wheel, "w") as archive:
        for filename, data in payload.items():
            archive.writestr(filename, data)


PROBE = r'''
import base64,hashlib,importlib.metadata as m,json,pathlib,site
import unigetui_probe_core as core,unigetui_probe_client as client
rows=[]; errors=[]; checked=0
for root in [*site.getsitepackages(),site.getusersitepackages()]:
 for path in pathlib.Path(root).glob('unigetui_probe_*.dist-info'):
  d=m.Distribution.at(path); record=d.read_text('RECORD')
  rows.append({'dir':path.name,'METADATA':(path/'METADATA').is_file(),
   'RECORD':record is not None,'REQUESTED':(path/'REQUESTED').is_file(),
   'tracked_REQUESTED':record is not None and '/REQUESTED,' in record})
  for f in d.files or []:
   p=pathlib.Path(d.locate_file(f))
   if not p.is_file(): errors.append(['missing',str(f)])
   elif f.hash:
    got=base64.urlsafe_b64encode(hashlib.new(f.hash.mode,p.read_bytes()).digest()).rstrip(b'=').decode()
    if got!=f.hash.value: errors.append(['hash',str(f)])
    checked+=1
print(json.dumps({'core':[core.__version__,m.version('unigetui-probe-core')],
 'client':[client.__version__,m.version('unigetui-probe-client')],
 'rows':rows,'hashes_checked':checked,'record_errors':errors}))
'''


def reproduce(root, supplied_wheel):
    runtime = root / "runtime"
    runtime.mkdir()
    original = Path(sys.base_prefix)
    for path in original.iterdir():
        if path.is_file() and path.suffix.lower() in (".exe", ".dll"):
            shutil.copyfile(path, runtime / path.name)
    shutil.copytree(original / "Lib", runtime / "Lib",
                    ignore=shutil.ignore_patterns("site-packages", "__pycache__"))
    shutil.copytree(original / "DLLs", runtime / "DLLs")
    python = runtime / "python.exe"
    env = {k: v for k, v in os.environ.items() if not k.upper().startswith(("PIP_", "PYTHON"))}
    env.update(PYTHONUSERBASE=str(root / "userbase"), PIP_CONFIG_FILE=os.devnull,
               PIP_DISABLE_PIP_VERSION_CHECK="1", PYTHONUTF8="1")

    def run(*args, expected=0):
        command = [str(python), *map(str, args)]
        result = subprocess.run(command, env=env, cwd=root, capture_output=True,
                                text=True, encoding="utf-8", errors="replace", timeout=180)
        output = result.stdout + result.stderr
        public = "$ " + subprocess.list2cmdline(command) + "\n" + output
        for source, replacement in ((str(root), "<SANDBOX>"), (str(Path.home()), "<USER_HOME>")):
            public = re.sub(re.escape(source), lambda _: replacement, public, flags=re.I)
        print(public, flush=True)
        if expected is not None:
            assert result.returncode == expected, (command, result.returncode)
        return result, output

    wheel = supplied_wheel
    if wheel is None:
        with urllib.request.urlopen(f"https://pypi.org/pypi/pip/{PIP_VERSION}/json") as response:
            manifest = json.load(response)
        artifact = next(item for item in manifest["urls"] if item["filename"].endswith(".whl"))
        assert artifact["digests"]["sha256"] == PIP_SHA256
        wheel = root / artifact["filename"]
        urllib.request.urlretrieve(artifact["url"], wheel)
    assert hashlib.sha256(wheel.read_bytes()).hexdigest() == PIP_SHA256
    run("-m", "ensurepip", "--default-pip")
    run("-m", "pip", "install", "--no-index", "--no-deps", "--no-cache-dir", "--force-reinstall", wheel)

    def pip_integrity():
        checked = 0
        with zipfile.ZipFile(wheel) as archive:
            for member in archive.namelist():
                if member.startswith("pip/") and not member.endswith("/"):
                    assert (runtime / "Lib/site-packages" / member).read_bytes() == archive.read(member)
                    checked += 1
        print("OFFICIAL_PIP_FILES_IDENTICAL", checked, flush=True)

    indexes = {}
    for version in ("0.1.0", "0.2.0", "0.3.0"):
        indexes[version] = root / ("index-" + version)
        indexes[version].mkdir()
        for name in (CORE, CLIENT):
            build_wheel(indexes[version], name, version)

    def install(version, *targets):
        return run("-m", "pip", "install", "--no-index", "--find-links", indexes[version],
                   "--no-input", "--no-color", "--no-cache-dir", *targets)

    def state():
        result, _ = run("-c", PROBE)
        return json.loads(result.stdout)

    def healthy(version, count, allow_untracked=False):
        snapshot = state()
        assert snapshot["core"] == [version, version]
        assert snapshot["client"] == [version, version]
        assert len(snapshot["rows"]) == count
        assert not snapshot["record_errors"]
        assert all(r["METADATA"] and r["RECORD"] for r in snapshot["rows"])
        if not allow_untracked:
            assert all(not r["REQUESTED"] or r["tracked_REQUESTED"] for r in snapshot["rows"])
        run("-m", "pip", "check")

    pip_integrity()
    install("0.1.0", CORE + "==0.1.0", CLIENT + "==0.1.0")
    for version in ("0.2.0", "0.3.0"):
        install(version, "--upgrade", CORE + "==" + version)
        install(version, "--upgrade", CLIENT + "==" + version)
        healthy(version, 2)
    print("SINGLE_SCHEME_SEQUENTIAL_CONTROL_PASSED", flush=True)

    install("0.1.0", "--force-reinstall", CORE + "==0.1.0", CLIENT + "==0.1.0")
    install("0.1.0", "--user", "--ignore-installed", CORE + "==0.1.0")
    healthy("0.1.0", 3)
    for version in ("0.2.0", "0.3.0"):
        install(version, "--user", "--upgrade", CORE + "==" + version)
        install(version, "--upgrade", CLIENT + "==" + version)
        healthy(version, 3)
    print("PRESERVED_USER_SCHEME_CONTROL_PASSED", flush=True)
    install("0.1.0", "--user", "--force-reinstall", CORE + "==0.1.0")
    install("0.1.0", "--no-deps", "--force-reinstall", CLIENT + "==0.1.0")
    healthy("0.1.0", 3)
    print("TWO_SCHEMES_HEALTHY_BASELINE_PASSED", flush=True)
    install("0.2.0", "--upgrade", CORE + "==0.2.0")
    changed = state()
    assert changed["core"] == ["0.2.0", "0.1.0"]
    install("0.2.0", "--upgrade", CLIENT + "==0.2.0")
    healthy("0.2.0", 2, allow_untracked=True)
    overlay = state()
    assert any(r["REQUESTED"] and not r["tracked_REQUESTED"] for r in overlay["rows"])
    install("0.3.0", "--upgrade", CORE + "==0.3.0")
    orphan = state()
    assert orphan["core"] == ["0.3.0", None]
    result, output = run("-m", "pip", "install", "--no-index", "--find-links", indexes["0.3.0"],
                         "--upgrade", CLIENT + "==0.3.0", expected=None)
    assert result.returncode != 0 and "uninstall-no-record-file" in output
    pip_integrity()
    print("CLEAN_CLI_REPRODUCED_MISSING_RECORD; pip sources stayed identical", flush=True)


if __name__ == "__main__":
    sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace")
    parser = argparse.ArgumentParser()
    parser.add_argument("--pip-wheel", type=Path)
    args = parser.parse_args()
    assert os.name == "nt", "This runtime-copy reproduction is for Windows."
    with tempfile.TemporaryDirectory(prefix="unigetui-pip-clean-") as temporary:
        reproduce(Path(temporary), args.pip_wheel.resolve() if args.pip_wheel else None)