open-design/.github/scripts/postinstall.py

#!/usr/bin/env python3

"""Produce and consume frozen workspace-initialization plans for workflows."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import subprocess
from typing import Any

from lib.postinstall_plan import (
    PLAN_SCHEMA_VERSION,
    canonical_json,
    plan_digest,
    resolve_plan,
)

SCHEMA_VERSION = PLAN_SCHEMA_VERSION


def load_object(path: Path, label: str) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError(f"{label} must be a JSON object")
    return value


def repository_root() -> Path:
    return Path(__file__).resolve().parents[2]


def append_outputs(values: dict[str, str]) -> None:
    output_path = os.environ.get("GITHUB_OUTPUT")
    if not output_path:
        return
    with open(output_path, "a", encoding="utf-8") as output:
        for key, value in values.items():
            output.write(f"{key}={value}\n")


def create_plan(args: argparse.Namespace) -> dict[str, Any]:
    root = repository_root()
    def load_repository_json(path: str) -> dict[str, Any]:
        configured = args.config if path == ".github/config/postinstall.json" else root / path
        return load_object(configured, path)

    canonical = resolve_plan(args.intent, load_repository_json)
    install_profile = canonical["installProfile"]
    requested = canonical["requestedTargets"]
    resolved = canonical["resolvedTargets"]
    requirements = canonical["requirements"]
    cache_tools = args.cache_tools == "true"
    plan: dict[str, Any] = {
        "schemaVersion": SCHEMA_VERSION,
        "id": f"{os.environ.get('GITHUB_WORKFLOW', 'workflow')}/{os.environ.get('GITHUB_JOB', 'job')}/{args.intent}",
        "intent": args.intent,
        "installProfile": install_profile,
        "cacheTools": cache_tools,
        "requestedTargets": requested,
        "resolvedTargets": resolved,
        "requirements": requirements,
        "entries": {
            "dependencies": {
                "materializeDomToPptx": requirements["materializeDomToPptx"],
                "probeNativeDependencies": requirements["probeNativeDependencies"],
                "resolvedTargets": [],
                "concurrency": args.concurrency,
            },
            "build": {
                "materializeDomToPptx": False,
                "probeNativeDependencies": False,
                "resolvedTargets": resolved,
                "concurrency": args.concurrency,
            },
            "all": {
                "materializeDomToPptx": requirements["materializeDomToPptx"],
                "probeNativeDependencies": requirements["probeNativeDependencies"],
                "resolvedTargets": resolved,
                "concurrency": args.concurrency,
            },
        },
    }
    plan["digest"] = plan_digest(canonical)
    return plan


def plan_command(args: argparse.Namespace) -> int:
    plan = create_plan(args)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    append_outputs({
        "path": str(args.output),
        "json": canonical_json(plan),
        "digest": plan["digest"],
        "install-profile": plan["installProfile"],
    })
    print(canonical_json(plan))
    return 0


def load_receipts(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        return []
    receipts: list[dict[str, Any]] = []
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line:
            continue
        value = json.loads(line)
        if not isinstance(value, dict):
            raise ValueError("postinstall receipt line must be an object")
        receipts.append(value)
    return receipts


def consume_command(args: argparse.Namespace) -> int:
    plan = load_object(args.plan, "postinstall plan")
    canonical = {key: plan.get(key) for key in (
        "schemaVersion", "installProfile", "requestedTargets", "resolvedTargets", "requirements",
    )}
    if plan.get("schemaVersion") != SCHEMA_VERSION or plan.get("digest") != plan_digest(canonical):
        raise ValueError("postinstall plan digest is invalid")
    receipts = load_receipts(args.receipts)
    for receipt in receipts:
        if receipt.get("schemaVersion") != SCHEMA_VERSION:
            raise ValueError("postinstall receipt has an unsupported schemaVersion")
        if receipt.get("planId") != plan.get("id") or receipt.get("planDigest") != plan.get("digest"):
            raise ValueError("postinstall receipt does not belong to the frozen plan")
        if receipt.get("status") != "success":
            raise ValueError("postinstall receipt did not succeed")

    profile = plan["installProfile"]
    cache_tools = bool(plan["cacheTools"])
    expected_entries = ["build"] if profile != "workspace" else (["dependencies"] if cache_tools else ["all"])
    receipt_entries = {receipt.get("entry") for receipt in receipts}
    missing = [entry for entry in expected_entries if entry not in receipt_entries]
    if missing:
        raise ValueError(f"postinstall receipts are missing required entries: {missing}")

    build_receipt = next((receipt for receipt in receipts if receipt.get("entry") in {"build", "all"}), None)
    restored_targets: list[str] = []
    executed_targets: list[str] = []
    if cache_tools and args.tools_cache_hit == "true":
        restored_targets = list(plan["resolvedTargets"])
    else:
        if build_receipt is None:
            raise ValueError("postinstall build receipt is required when the tool closure was not restored")
        executed_targets = build_receipt.get("executedTargets", [])
        if executed_targets != plan["resolvedTargets"]:
            raise ValueError("postinstall executed targets differ from the frozen plan")

    result = {
        "schemaVersion": SCHEMA_VERSION,
        "planId": plan["id"],
        "planDigest": plan["digest"],
        "intent": plan["intent"],
        "installProfile": profile,
        "requiredTargets": plan["resolvedTargets"],
        "executedTargets": executed_targets,
        "restoredTargets": restored_targets,
        "status": "success",
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    append_outputs({"json": canonical_json(result), "path": str(args.output)})
    print(canonical_json(result))
    return 0


def validate_command(args: argparse.Namespace) -> int:
    root = repository_root()
    config = load_object(args.config, "postinstall workflow config")
    intents = config.get("intents", {})
    if not isinstance(intents, dict) or not intents:
        raise ValueError("postinstall workflow config requires intents")
    for intent in intents:
        namespace = argparse.Namespace(
            cache_tools="false", concurrency=1, config=args.config, intent=intent,
        )
        create_plan(namespace)
    targets = config.get("targets", [])
    describe_environment = {
        key: value for key, value in os.environ.items()
        if not key.startswith("OPEN_DESIGN_POSTINSTALL_")
    }
    described = json.loads(subprocess.check_output(
        ["node", "scripts/postinstall.mjs", "describe"], cwd=root, env=describe_environment, text=True,
    ))
    if described != targets:
        raise ValueError("postinstall workflow targets differ from consumer capabilities")
    print(canonical_json({"schemaVersion": SCHEMA_VERSION, "intents": sorted(intents), "targets": targets}))
    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Produce and consume workflow postinstall plans.")
    parser.add_argument("--config", type=Path, default=repository_root() / ".github/config/postinstall.json")
    commands = parser.add_subparsers(dest="command", required=True)
    plan = commands.add_parser("plan")
    plan.add_argument("--intent", required=True)
    plan.add_argument("--cache-tools", choices=["true", "false"], default="false")
    plan.add_argument("--concurrency", type=int, default=1)
    plan.add_argument("--output", type=Path, required=True)
    consume = commands.add_parser("consume")
    consume.add_argument("--plan", type=Path, required=True)
    consume.add_argument("--receipts", type=Path, required=True)
    consume.add_argument("--tools-cache-hit", choices=["true", "false"], default="false")
    consume.add_argument("--output", type=Path, required=True)
    commands.add_parser("validate")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    if args.command == "plan":
        if args.concurrency < 1:
            raise ValueError("postinstall concurrency must be positive")
        return plan_command(args)
    if args.command == "consume":
        return consume_command(args)
    return validate_command(args)


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, json.JSONDecodeError, subprocess.SubprocessError) as error:
        print(f"postinstall plan error: {error}", file=os.sys.stderr)
        raise SystemExit(2)