#!/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())