Blacksite/scripts/editor/build_progress.py

#!/usr/bin/env python3
"""Build the Blacksite editor while publishing exact crate progress."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import subprocess
import sys
from typing import Iterable


ROOT = Path(__file__).resolve().parents[2]
DEFAULT_CARGO_LANE = ROOT / "scripts/codex/cargo_lane.py"


def write_status(
    path: Path,
    phase: str,
    detail: str,
    progress: int,
    state: str = "running",
    crate_count: str = "",
) -> None:
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(
        f"{phase}\n{detail}\n{progress}\n{state}\n{crate_count}\n",
        encoding="utf-8",
    )
    os.replace(temporary, path)


def completed_crate(
    message: object, completed_package_ids: set[str]
) -> tuple[bool, str]:
    if not isinstance(message, dict) or message.get("reason") != "compiler-artifact":
        return False, ""
    package_id = message.get("package_id")
    if not isinstance(package_id, str) or package_id in completed_package_ids:
        return False, ""
    completed_package_ids.add(package_id)
    target = message.get("target")
    if isinstance(target, dict) and isinstance(target.get("name"), str):
        return True, target["name"]
    return True, package_id.rsplit("#", 1)[-1]


def managed_cargo(cargo_lane: Path, lane: str, *arguments: str) -> list[str]:
    return [
        sys.executable,
        str(cargo_lane),
        "exec",
        lane,
        "--",
        "cargo",
        *arguments,
    ]


def host_target() -> str:
    completed = subprocess.run(
        ["rustc", "-vV"],
        cwd=ROOT,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        check=False,
    )
    for line in completed.stdout.splitlines():
        if line.startswith("host:"):
            return line.partition(":")[2].strip()
    return ""


def cargo_metadata(cargo_lane: Path, lane: str, target: str) -> list[str]:
    arguments = ["metadata", "--format-version", "1"]
    if target:
        arguments.extend(["--filter-platform", target])
    return managed_cargo(
        cargo_lane,
        lane,
        *arguments,
    )


def cargo_build(cargo_lane: Path, lane: str) -> list[str]:
    return managed_cargo(
        cargo_lane,
        lane,
        "build",
        "-p",
        "editor",
        "--bins",
        "--message-format=json-render-diagnostics",
        "--color",
        "never",
    )


def editor_package_closure(metadata_output: str) -> set[str]:
    document = json.loads(metadata_output)
    packages = document.get("packages")
    resolve = document.get("resolve")
    if not isinstance(packages, list) or not isinstance(resolve, dict):
        raise ValueError("Cargo metadata did not include packages and a resolve graph")

    editor_ids = {
        package.get("id")
        for package in packages
        if isinstance(package, dict)
        and package.get("name") == "editor"
        and isinstance(package.get("id"), str)
    }
    if len(editor_ids) != 1:
        raise ValueError("Cargo metadata did not identify exactly one editor package")
    root_id = next(iter(editor_ids))

    nodes = resolve.get("nodes")
    if not isinstance(nodes, list):
        raise ValueError("Cargo metadata resolve graph did not include nodes")
    nodes_by_id = {
        node["id"]: node
        for node in nodes
        if isinstance(node, dict) and isinstance(node.get("id"), str)
    }

    closure: set[str] = set()
    pending = [root_id]
    while pending:
        package_id = pending.pop()
        if package_id in closure:
            continue
        closure.add(package_id)
        node = nodes_by_id.get(package_id)
        if not isinstance(node, dict):
            continue
        dependencies = node.get("deps")
        if not isinstance(dependencies, list):
            continue
        for dependency in dependencies:
            if not isinstance(dependency, dict):
                continue
            dependency_id = dependency.get("pkg")
            kinds = dependency.get("dep_kinds")
            if not isinstance(dependency_id, str) or not isinstance(kinds, list):
                continue
            if any(
                isinstance(kind, dict) and kind.get("kind") in {None, "build"}
                for kind in kinds
            ):
                pending.append(dependency_id)
    return closure


def rendered_lines(message: object, raw_line: str) -> Iterable[str]:
    if not isinstance(message, dict):
        yield raw_line
        return
    if message.get("reason") == "compiler-message":
        diagnostic = message.get("message")
        if isinstance(diagnostic, dict):
            rendered = diagnostic.get("rendered")
            if isinstance(rendered, str) and rendered:
                yield rendered.rstrip("\n")


def run(status_path: Path, log_path: Path, cargo_lane: Path, lane: str) -> int:
    write_status(
        status_path,
        "Preparing build",
        "Resolving editor crate graph",
        8,
    )
    metadata = subprocess.run(
        cargo_metadata(cargo_lane, lane, host_target()),
        cwd=ROOT,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    try:
        crates = (
            editor_package_closure(metadata.stdout)
            if metadata.returncode == 0
            else set()
        )
    except (json.JSONDecodeError, ValueError):
        crates = set()
    if not crates:
        with log_path.open("a", encoding="utf-8") as log:
            log.write(metadata.stdout)
            log.write(metadata.stderr)
        write_status(
            status_path,
            "Build graph failed",
            "Open the launch log for Cargo details",
            8,
            "error",
        )
        return metadata.returncode or 1

    total = len(crates)
    write_status(
        status_path,
        "Compiling project",
        "Preparing editor crates",
        12,
        crate_count=f"0/{total}",
    )

    process = subprocess.Popen(
        cargo_build(cargo_lane, lane),
        cwd=ROOT,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,
    )
    assert process.stdout is not None
    completed_package_ids: set[str] = set()
    with log_path.open("a", encoding="utf-8") as log:
        for raw_line in process.stdout:
            stripped = raw_line.rstrip("\n")
            try:
                message: object = json.loads(stripped)
            except json.JSONDecodeError:
                message = stripped

            for rendered in rendered_lines(message, stripped):
                log.write(rendered + "\n")
                log.flush()

            changed, crate_name = completed_crate(message, completed_package_ids)
            if not changed:
                continue
            completed = len(completed_package_ids.intersection(crates))
            progress = min(78, 12 + (66 * completed // total))
            write_status(
                status_path,
                "Compiling project",
                f"{crate_name} ready",
                progress,
                crate_count=f"{completed}/{total}",
            )

    return_code = process.wait()
    completed = len(completed_package_ids.intersection(crates))
    if return_code != 0:
        write_status(
            status_path,
            "Build failed",
            "Open the launch log for compiler details",
            max(12, min(78, 12 + (66 * completed // total))),
            "error",
            f"{completed}/{total}",
        )
        return return_code

    write_status(
        status_path,
        "Build complete",
        "Preparing engine process",
        78,
        crate_count=f"{total}/{total}",
    )
    return 0


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("status_path", type=Path)
    parser.add_argument("log_path", type=Path)
    parser.add_argument("--cargo-lane", type=Path, default=DEFAULT_CARGO_LANE)
    parser.add_argument("--lane", default="dev")
    arguments = parser.parse_args()
    return run(
        arguments.status_path,
        arguments.log_path,
        arguments.cargo_lane,
        arguments.lane,
    )


if __name__ == "__main__":
    raise SystemExit(main())