searchi/tools/package_linux.py

"""Package an already-built Linux x86-64 headless development candidate.

No artifact is executed. readelf inspects dependency closure; native execution,
ABI compatibility, GPU hardware and production installers remain release gates.
"""
import argparse
import gzip
import hashlib
import io
import json
import lzma
import os
from pathlib import Path, PurePosixPath
import re
import struct
import subprocess
import tarfile
import tempfile
from embeddinggemma2.runtime_provenance import verify as verify_provenance
from stage_extraction_worker import verify as verify_extraction
from stage_speech_worker import verify_directory as verify_speech

ROOT = Path(__file__).resolve().parents[1]
MAX_FILE = 256 * 1024 * 1024
MAX_TOTAL = 1024 * 1024 * 1024
CPU_VARIANTS = {"x64", "sse42", "sandybridge", "ivybridge", "piledriver", "haswell",
                "skylakex", "cannonlake", "cascadelake", "icelake", "cooperlake", "zen4",
                "alderlake", "sapphirerapids"}
SYSTEM = {"libc.so.6", "libm.so.6", "libdl.so.2", "libpthread.so.0", "librt.so.1",
          "libmvec.so.1", "libstdc++.so.6", "libgcc_s.so.1", "ld-linux-x86-64.so.2"}


def safe_name(name):
    if (not isinstance(name, str) or not name or "\\" in name or ":" in name
            or str(PurePosixPath(name)) != name or PurePosixPath(name).is_absolute()
            or any(part in {".", ".."} for part in name.split("/"))):
        raise ValueError("Unsafe package path")
    return name


def regular(path, parent):
    path, parent = Path(path).absolute(), Path(parent).resolve()
    if not path.is_relative_to(parent) or not path.resolve().is_relative_to(parent):
        raise ValueError("Package source escapes its directory")
    candidate = path
    while candidate != parent:
        if candidate.is_symlink():
            raise ValueError("Package source contains a symlink")
        candidate = candidate.parent
    if not path.is_file() or not 0 < path.stat().st_size <= MAX_FILE:
        raise ValueError(f"Missing or oversized package source: {path.name}")


def digest(path):
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def elf_x64(path):
    with path.open("rb") as stream:
        header = stream.read(64)
    if (len(header) != 64 or header[:7] != b"\x7fELF\x02\x01\x01"
            or struct.unpack_from("<H", header, 18)[0] != 62
            or struct.unpack_from("<H", header, 16)[0] not in {2, 3}):
        raise ValueError(f"Expected Linux x86-64 ELF artifact: {path.name}")


def dynamic(path):
    output = subprocess.check_output(["readelf", "--dynamic", str(path)], text=True, timeout=30)
    values = {}
    for key, value in re.findall(r"\((NEEDED|SONAME|RUNPATH|RPATH)\).*?\[([^\]]+)\]", output):
        values.setdefault(key, []).append(value)
    return values


def runtime_files(runtime, root=ROOT):
    inventory = runtime / "runtime-libraries.json"
    regular(inventory, runtime)
    if inventory.stat().st_size > 65536:
        raise ValueError("Oversized runtime inventory")
    inventory_bytes = inventory.read_bytes()
    manifest = json.loads(inventory_bytes)
    records = manifest.get("libraries") if isinstance(manifest, dict) else None
    if not isinstance(manifest, dict) or manifest.get("platform") != "linux" or not isinstance(records, list) or not 1 <= len(records) <= 64:
        raise ValueError("Invalid Linux runtime inventory")
    verify_provenance(manifest, root)
    expected = {"searchi-gemma-worker": {"sha256": manifest.get("worker_sha256")}}
    core = {}
    for record in records:
        if not isinstance(record, dict):
            raise ValueError("Invalid library record")
        name = record.get("path")
        safe_name(name)
        match = re.fullmatch(r"lib(ggml|ggml-base|llama|mtmd)\.so(?:\.\d+)*", name)
        if match:
            if match[1] in core:
                raise ValueError("Duplicate runtime core library")
            core[match[1]] = name
        elif name != "libggml-vulkan.so" and name not in {f"libggml-cpu-{v}.so" for v in CPU_VARIANTS}:
            raise ValueError("Unapproved runtime library")
        if name in expected or type(record.get("bytes")) is not int or record["bytes"] <= 0:
            raise ValueError("Duplicate or invalid runtime library")
        expected[name] = record
    if set(core) != {"ggml", "ggml-base", "llama", "mtmd"} or not {"libggml-cpu-x64.so", "libggml-vulkan.so"}.issubset(expected):
        raise ValueError("Runtime lacks baseline CPU, Vulkan or core library")
    for name, record in expected.items():
        path = runtime / name
        regular(path, runtime)
        checksum = record.get("sha256")
        if not isinstance(checksum, str) or not re.fullmatch(r"[0-9a-f]{64}", checksum):
            raise ValueError("Invalid runtime checksum")
        if digest(path) != checksum or ("bytes" in record and path.stat().st_size != record["bytes"]):
            raise ValueError(f"Runtime integrity mismatch: {name}")
        record["bytes"] = path.stat().st_size
        elf_x64(path)
        metadata = dynamic(path)
        if metadata.get("RPATH") or metadata.get("RUNPATH") != ["$ORIGIN"]:
            raise ValueError("Runtime requires origin-only RUNPATH")
        allowed = set(expected) | SYSTEM
        if name == "libggml-vulkan.so":
            allowed.add("libvulkan.so.1")
        if set(metadata.get("NEEDED", [])) - allowed:
            raise ValueError(f"Unpackaged runtime dependency: {name}")
        if name in core.values() and metadata.get("SONAME") != [name]:
            raise ValueError("Core SONAME differs from packaged filename")
    expected[inventory.name] = {"sha256": hashlib.sha256(inventory_bytes).hexdigest(), "bytes": len(inventory_bytes)}
    return expected


def media_files(root, media):
    manifest_path = root / "tools/media-runtime/manifest.json"
    regular(manifest_path, root)
    if manifest_path.stat().st_size > 65536:
        raise ValueError("Oversized media manifest")
    manifest = json.loads(manifest_path.read_text())
    artifact = manifest["platforms"]["linux-x86_64"]
    archive_name = safe_name(artifact["archive"])
    if "/" in archive_name or not archive_name.endswith(".tar.xz"):
        raise ValueError("Invalid media archive name")
    archive = media / archive_name
    regular(archive, media)
    if archive.stat().st_size != artifact["bytes"] or digest(archive) != artifact["sha256"]:
        raise ValueError("Media archive provenance mismatch")
    selected = {}
    with tarfile.open(archive, "r:xz") as package:
        members = 0
        declared_total = 0
        for member in package:
            members += 1
            declared_total += member.size
            if (members > 10000 or member.size < 0 or member.size > MAX_FILE
                    or declared_total > MAX_TOTAL):
                raise ValueError("Media archive exceeds member or size ceiling")
            basename = member.name.replace("\\", "/").split("/")[-1]
            if basename in {"ffmpeg", "ffprobe"}:
                target = f"bin/{basename}"
            elif basename.lower().startswith(("license", "copying")) or basename.lower() in {"readme.txt", "readme.md"}:
                target = "licenses/" + safe_name(member.name)
            else:
                continue
            if not member.isfile() or not 0 < member.size <= MAX_FILE or target in selected:
                raise ValueError("Invalid or duplicate media member")
            path = media / target
            regular(path, media)
            with package.extractfile(member) as stream:
                checksum = hashlib.file_digest(stream, "sha256").hexdigest()
            if path.stat().st_size != member.size or digest(path) != checksum:
                raise ValueError(f"Media helper/notice differs from pinned archive: {target}")
            selected[target] = {"sha256": checksum, "bytes": member.size}
    if not {"bin/ffmpeg", "bin/ffprobe"}.issubset(selected) or not any(n.startswith("licenses/") for n in selected):
        raise ValueError("Media archive lacks helpers or notices")
    selected = {"media-runtime/" + name: (media / name, record) for name, record in selected.items()}
    selected["media-runtime/manifest.json"] = (manifest_path, {"sha256": digest(manifest_path), "bytes": manifest_path.stat().st_size})
    return selected


def package(root, executable, runtime, media, output, llama_license=None, compression="gzip", speech_runtime=None):
    if compression not in {"gzip", "xz"}:
        raise ValueError("Unsupported package compression")
    root, executable, runtime, media, output = map(Path, (root, executable, runtime, media, output))
    regular(executable, executable.parent)
    elf_x64(executable)
    expected = runtime_files(runtime, root)
    extraction = verify_extraction(runtime, root, "linux", "x86_64")
    helper = runtime / "searchi-extract-worker"
    regular(helper, runtime)
    regular(runtime / "extraction-worker.json", runtime)
    elf_x64(helper)
    helper_metadata = dynamic(helper)
    if (helper_metadata.get("RPATH") or helper_metadata.get("RUNPATH")
            or set(helper_metadata.get("NEEDED", [])) - SYSTEM):
        raise ValueError("Extraction helper has unexpected native dependency or search path")
    expected.update(extraction)
    cli_metadata = dynamic(executable)
    if (cli_metadata.get("RPATH") or cli_metadata.get("RUNPATH")
            or set(cli_metadata.get("NEEDED", [])) - SYSTEM):
        raise ValueError("CLI has unexpected native dependency or search path")
    selected = {name: (runtime / name, record) for name, record in expected.items()}
    if speech_runtime is not None:
        speech_runtime = Path(speech_runtime)
        speech_records = verify_speech(speech_runtime, root, 'linux', 'x86_64')
        selected.update({name: (speech_runtime / PurePosixPath(name).name, record)
                         for name, record in speech_records.items()})
    selected.update(media_files(root, media))
    selected["searchi"] = (executable, {"sha256": digest(executable), "bytes": executable.stat().st_size})
    license_path = Path(llama_license) if llama_license else root / ".searchi/references/llama.cpp/LICENSE"
    for path, name, parent in [(root / "LICENSE", "LICENSE.txt", root), (license_path, "LLAMA-LICENSE.txt", license_path.parent)]:
        regular(path, parent)
        selected[name] = (path, {"sha256": digest(path), "bytes": path.stat().st_size})
    for name in ("media-runtime/bin/ffmpeg", "media-runtime/bin/ffprobe"):
        elf_x64(selected[name][0])
        metadata = dynamic(selected[name][0])
        if metadata.get("RPATH") or metadata.get("RUNPATH") or set(metadata.get("NEEDED", [])) - SYSTEM:
            raise ValueError("Media helper has unpackaged dependencies")
    if sum(record["bytes"] for _, record in selected.values()) > MAX_TOTAL:
        raise ValueError("Package exceeds total size ceiling")
    if output.resolve() in {path.resolve() for path, _ in selected.values()}:
        raise ValueError("Package output would overwrite an input")
    output.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(dir=output.parent, suffix=".partial", delete=False) as temporary:
        temporary_path = Path(temporary.name)
    try:
        records = []
        with temporary_path.open("wb") as raw, (lzma.LZMAFile(raw, "w", preset=6) if compression == "xz" else gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0)) as compressed, tarfile.open(fileobj=compressed, mode="w|") as archive:
            for name, (path, expected_record) in sorted(selected.items()):
                safe_name(name)
                regular(path, path.parent)
                # Bounded immutable snapshot: archive and manifest describe the same bytes.
                with path.open("rb") as stream:
                    data = stream.read(MAX_FILE + 1)
                checksum = hashlib.sha256(data).hexdigest()
                if len(data) != expected_record["bytes"] or checksum != expected_record["sha256"]:
                    raise ValueError(f"Source changed during packaging: {name}")
                mode = 0o755 if name in {"searchi", "searchi-gemma-worker", "searchi-extract-worker", "speech-runtime/searchi-speech-worker", "media-runtime/bin/ffmpeg", "media-runtime/bin/ffprobe"} or ".so" in name else 0o644
                add(archive, name, data, mode)
                records.append({"path": name, "bytes": len(data), "sha256": checksum, "mode": oct(mode)})
            speech_instructions = ("\nExperimental local speech search is included. Download verified recognition assets (about 149 MB):\n"
                "./searchi setup --speech --model-dir ./speech-models\n"
                "Then index recognized passages alongside Google media embeddings:\n"
                "./searchi embed --db ./searchi.sqlite --model-dir ./models --embedding-worker ./searchi-gemma-worker --speech-worker ./speech-runtime/searchi-speech-worker --speech-model-dir ./speech-models\n"
                "Search uses the same command above; spoken passages carry estimated source timestamps.\n"
                "Recognition and search remain local. Music/no-answer relevance and precise timestamps remain experimental.\n") if speech_runtime is not None else ""
            instructions = ("searchi — Find what matters. Locally.\n\n"
                "Extract the entire archive into its own folder. This is a headless development candidate, not a production installer.\n"
                "Use ./searchi --help. Setup downloads about 865 MB of verified Google model assets:\n"
                "./searchi setup --model-dir ./models\n"
                "./searchi index --db ./searchi.sqlite --root /path/to/files\n"
                "./searchi embed --db ./searchi.sqlite --model-dir ./models --embedding-worker ./searchi-gemma-worker\n"
                "./searchi search --db ./searchi.sqlite 'what you remember' --model-dir ./models --embedding-worker ./searchi-gemma-worker\n"
                "Media helpers are discovered automatically beside this CLI. Ordinary search stays local.\n"
                "The normal runtime includes Vulkan and CPU fallback. Graphics drivers and a compatible system Vulkan loader are optional.\n"
                "Requires a compatible Linux libc/C++ ABI; clean-machine and distribution coverage remain open.\n" + speech_instructions).encode()
            add(archive, "START-HERE.txt", instructions, 0o644)
            records.append({"path": "START-HERE.txt", "bytes": len(instructions),
                "sha256": hashlib.sha256(instructions).hexdigest(), "mode": oct(0o644)})
            metadata = {"platform": "linux", "architecture": "x86_64",
                "status": "unsigned headless development candidate; distribution ABI and GPU hardware validation pending",
                "system_libraries": sorted(SYSTEM), "optional_gpu_loader": "libvulkan.so.1",
                "files": records}
            if speech_runtime is not None:
                metadata['features'] = {'local_speech': True, 'speech_status': 'experimental',
                                        'speech_runtime': 'speech-runtime/searchi-speech-worker'}
            add(archive, "PACKAGE.json", json.dumps(metadata, indent=2).encode(), 0o644)
        os.replace(temporary_path, output)
    finally:
        temporary_path.unlink(missing_ok=True)
    return "x86_64"


def add(archive, name, data, mode):
    info = tarfile.TarInfo(safe_name(name))
    info.size, info.mode, info.mtime = len(data), mode, 0
    info.uid = info.gid = 0
    info.uname = info.gname = ""
    archive.addfile(info, io.BytesIO(data))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--executable", type=Path, default=ROOT / "target/release/searchi")
    parser.add_argument("--runtime", type=Path, default=ROOT / ".searchi/runtime/auto")
    parser.add_argument("--media", type=Path, default=ROOT / ".searchi/media-runtime")
    parser.add_argument("--speech-runtime", type=Path, help="Verified normal auto speech runtime directory")
    parser.add_argument("--llama-license", type=Path)
    parser.add_argument("--compression", choices=["gzip", "xz"], default="gzip")
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    if args.output is None:
        args.output = ROOT / (".searchi/packages/searchi-linux-x86_64.tar.xz" if args.compression == "xz" else ".searchi/packages/searchi-linux-x86_64.tar.gz")
    package(ROOT, args.executable, args.runtime, args.media, args.output, args.llama_license, args.compression, args.speech_runtime)
    print(f"Packaged Linux x86_64 development candidate: {args.output}")


if __name__ == "__main__":
    main()