feat(plan): establish reusable workload and postinstall contracts (#8383)

* feat(plan): add verified atomic workload product restoration

* perf(plan): probe product availability without downloading payloads

* refactor(ci): formalize workspace initialization plans

* feat(plan): add versioned workload execution contracts

* refactor(plan): colocate postinstall defaults and docs

* fix(convergence): enforce cache result integrity

.github/scripts/lib/github.py

@@ -77,6 +77,26 @@ def run_artifacts(repository: str, run_id: int) -> list[dict[str, Any]]:
         page += 1
 
 
+def run_jobs(repository: str, run_id: int, attempt: int) -> list[dict[str, Any]]:
+    """Read exactly one attempt; never borrow a previous attempt's success."""
+    if type(run_id) is not int or run_id <= 0 or type(attempt) is not int or attempt <= 0:
+        raise GitHubError("job lookup requires a positive run id and attempt")
+    quoted_repository = "/".join(urllib.parse.quote(part, safe="") for part in repository.split("/"))
+    jobs: list[dict[str, Any]] = []
+    page = 1
+    while True:
+        value = api_json(f"/repos/{quoted_repository}/actions/runs/{run_id}/attempts/{attempt}/jobs?per_page=100&page={page}")
+        if not isinstance(value, dict) or not isinstance(value.get("jobs"), list):
+            raise GitHubError("GitHub jobs response has an invalid shape")
+        batch = value["jobs"]
+        if any(not isinstance(job, dict) for job in batch):
+            raise GitHubError("GitHub job entry has an invalid shape")
+        jobs.extend(batch)
+        if len(batch) < 100:
+            return jobs
+        page += 1
+
+
 def unique_run_artifact(repository: str, run_id: int, name: str) -> dict[str, Any] | None:
     matches = [
         artifact

.github/scripts/lib/postinstall_plan.py

@@ -0,0 +1,123 @@
+"""Resolve the canonical workspace state requested by a postinstall intent.
+
+The Plan describes the delivered state. Invocation IDs, concurrency and cache
+strategy stay outside it. Bump PLAN_SCHEMA_VERSION when the meaning of a
+serialized Plan changes without changing its serialized target state.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Callable
+from typing import Any
+
+
+PLAN_SCHEMA_VERSION = 2
+INSTALL_PROFILES = {
+    "workspace",
+    "source-web",
+    "release-executor",
+    "release-tools",
+    "release-validation",
+    "mac-runtime",
+}
+
+JsonLoader = Callable[[str], dict[str, Any]]
+
+
+def canonical_json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
+
+
+def _targets(config: dict[str, Any]) -> list[str]:
+    if config.get("schemaVersion") != PLAN_SCHEMA_VERSION:
+        raise ValueError("postinstall workflow config has an unsupported schemaVersion")
+    targets = config.get("targets")
+    if not isinstance(targets, list) or not targets or any(not isinstance(item, str) or not item for item in targets):
+        raise ValueError("postinstall workflow config requires non-empty string targets")
+    if len(set(targets)) != len(targets):
+        raise ValueError("postinstall workflow targets must be unique")
+    return targets
+
+
+def _dependency_map(load_json: JsonLoader, targets: list[str]) -> dict[str, list[str]]:
+    names: dict[str, str] = {}
+    manifests: dict[str, dict[str, Any]] = {}
+    for target in targets:
+        manifest = load_json(f"{target}/package.json")
+        manifests[target] = manifest
+        name = manifest.get("name")
+        if isinstance(name, str) and name:
+            names[name] = target
+
+    result: dict[str, list[str]] = {}
+    for target in targets:
+        dependencies: list[str] = []
+        for field in ("dependencies", "devDependencies", "optionalDependencies", "peerDependencies"):
+            values = manifests[target].get(field, {})
+            if not isinstance(values, dict):
+                continue
+            for name, specifier in values.items():
+                dependency = names.get(name)
+                if dependency and isinstance(specifier, str) and specifier.startswith("workspace:"):
+                    dependencies.append(dependency)
+        result[target] = list(dict.fromkeys(dependencies))
+    return result
+
+
+def _closure(load_json: JsonLoader, requested: list[str], targets: list[str]) -> list[str]:
+    unknown = [target for target in requested if target not in targets]
+    if unknown:
+        raise ValueError(f"postinstall intent references unknown targets: {unknown}")
+    dependencies = _dependency_map(load_json, targets)
+    selected: set[str] = set()
+
+    def include(target: str) -> None:
+        if target in selected:
+            return
+        selected.add(target)
+        for dependency in dependencies[target]:
+            include(dependency)
+
+    for target in requested:
+        include(target)
+    return [target for target in targets if target in selected]
+
+
+def resolve_plan(intent: str, load_json: JsonLoader) -> dict[str, Any]:
+    """Return the cache and invocation independent target state for an intent."""
+    workflow_config = load_json(".github/config/postinstall.json")
+    if workflow_config.get("schemaVersion") != PLAN_SCHEMA_VERSION:
+        raise ValueError("postinstall workflow config has an unsupported schemaVersion")
+    intents = workflow_config.get("intents")
+    if not isinstance(intents, dict) or intent not in intents:
+        raise ValueError(f"unknown postinstall intent: {intent}")
+    recipe = intents[intent]
+    if not isinstance(recipe, dict) or set(recipe) != {"installProfile", "requestedTargets"}:
+        raise ValueError(f"postinstall intent {intent} has an invalid recipe")
+    install_profile = recipe["installProfile"]
+    if install_profile not in INSTALL_PROFILES:
+        raise ValueError(f"postinstall intent {intent} has an invalid install profile")
+
+    targets = _targets(workflow_config)
+    requested_value = recipe["requestedTargets"]
+    requested = list(targets) if requested_value == "all" else requested_value
+    if not isinstance(requested, list) or any(not isinstance(item, str) or not item for item in requested):
+        raise ValueError(f"postinstall intent {intent} has invalid requestedTargets")
+    requested = list(dict.fromkeys(requested))
+    partial = install_profile != "workspace"
+    return {
+        "schemaVersion": PLAN_SCHEMA_VERSION,
+        "installProfile": install_profile,
+        "requestedTargets": requested,
+        "resolvedTargets": _closure(load_json, requested, targets),
+        "requirements": {
+            "materializeDomToPptx": not partial,
+            "probeNativeDependencies": not partial,
+        },
+    }
+
+
+def plan_digest(plan: dict[str, Any]) -> str:
+    return hashlib.sha256(canonical_json(plan).encode("utf-8")).hexdigest()

.github/scripts/lib/r2.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import datetime as dt
 import hashlib
 import hmac
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -51,6 +52,8 @@ class R2Client:
         bucket: str,
         credentials: R2Credentials,
         region: str = "auto",
+        retry_delay: float = 1.0,
+        sleep=time.sleep,
         timeout: float = 15.0,
     ) -> None:
         parsed = urllib.parse.urlparse(endpoint.rstrip("/"))
@@ -64,6 +67,8 @@ class R2Client:
         self.bucket = bucket
         self.credentials = credentials
         self.region = region
+        self.retry_delay = retry_delay
+        self.sleep = sleep
         self.timeout = timeout
 
     def put_file(
@@ -142,17 +147,26 @@ class R2Client:
             f"SignedHeaders={signed_headers}, Signature={signature}"
         )
         request = urllib.request.Request(url, data=body, headers=headers, method="PUT")
-        try:
-            with urllib.request.urlopen(request, timeout=self.timeout) as response:
-                if response.status not in {200, 201}:
-                    raise R2Error(f"R2 PUT returned HTTP {response.status}")
-        except urllib.error.HTTPError as error:
-            if error.code == 412:
-                raise R2PreconditionFailed(f"R2 object already exists: {key}") from error
-            detail = error.read(2048).decode("utf-8", "replace")
-            raise R2Error(f"R2 PUT failed with HTTP {error.code}: {detail}") from error
-        except urllib.error.URLError as error:
-            raise R2Error(f"R2 PUT failed: {error.reason}") from error
+        for attempt in range(2):
+            try:
+                with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                    if response.status not in {200, 201}:
+                        raise R2Error(f"R2 PUT returned HTTP {response.status}")
+                return
+            except urllib.error.HTTPError as error:
+                if error.code == 412:
+                    raise R2PreconditionFailed(f"R2 object already exists: {key}") from error
+                if attempt == 0 and error.code in {408, 429, 500, 502, 503, 504}:
+                    self.sleep(self.retry_delay)
+                    continue
+                detail = error.read(2048).decode("utf-8", "replace")
+                raise R2Error(f"R2 PUT failed with HTTP {error.code}: {detail}") from error
+            except (urllib.error.URLError, TimeoutError) as error:
+                if attempt == 0:
+                    self.sleep(self.retry_delay)
+                    continue
+                reason = error.reason if isinstance(error, urllib.error.URLError) else str(error)
+                raise R2Error(f"R2 PUT failed: {reason}") from error
 
 
 def self_check() -> None:
@@ -188,3 +202,23 @@ def self_check() -> None:
     authorization = request.get_header("Authorization") or ""
     if not authorization.startswith("AWS4-HMAC-SHA256 Credential=access/"):
         raise R2Error("R2 self-check produced an invalid authorization header")
+
+    attempts = 0
+    def timeout_once(request: urllib.request.Request, **_kwargs):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise urllib.error.URLError(TimeoutError("timed out"))
+        return Response()
+
+    retrying = R2Client(
+        endpoint="https://account.r2.cloudflarestorage.com",
+        bucket="results",
+        credentials=R2Credentials("access", "secret"),
+        retry_delay=0,
+        sleep=lambda _seconds: None,
+    )
+    with patch("urllib.request.urlopen", timeout_once):
+        retrying.put_bytes(key="path/retry.json", body=b"{}")
+    if attempts != 2:
+        raise R2Error("R2 self-check did not retry one transient write failure")

.github/scripts/lib/workload_products.py

@@ -0,0 +1,62 @@
+"""Restore opaque workload products as one verified, immutable local set.
+
+See docs/ci/workload-products.md for the producer and consumer contract.
+"""
+from __future__ import annotations
+
+import hashlib
+import os
+import tempfile
+import urllib.request
+from pathlib import Path
+from typing import Any, Callable
+
+from lib.config import ConfigError
+
+
+def materialize_products(
+    products: dict[str, Any],
+    destination: Path,
+    request: Callable[[str], urllib.request.Request],
+    *,
+    timeout: float,
+) -> dict[str, int]:
+    """Caller validates manifest identities; bytes stay opaque, never extracted.
+
+    A fresh destination is mandatory. A failed download or hash check leaves no
+    published set; callers can execute the workload instead. Never merge into
+    existing outputs, where stale files could masquerade as a cache hit.
+    """
+    if not products:
+        raise ConfigError("cannot materialize an empty product set")
+    if destination.exists() or destination.is_symlink():
+        raise ConfigError("product destination must not already exist")
+    for name, product in products.items():
+        if Path(name).name != name or name in {".", ".."} or "\\" in name:
+            raise ConfigError("unsafe product name")
+        digest = product.get("data", {}).get("sha256")
+        if not isinstance(digest, str) or len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest):
+            raise ConfigError("materialized products require SHA-256")
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    sizes: dict[str, int] = {}
+    with tempfile.TemporaryDirectory(prefix=".workload-products-", dir=destination.parent) as temporary:
+        stage = Path(temporary) / "set"
+        stage.mkdir()
+        for name, product in products.items():
+            digest = hashlib.sha256()
+            size = 0
+            with urllib.request.urlopen(request(product["source"]), timeout=timeout) as source:
+                if source.status != 200:
+                    raise OSError(f"unexpected product HTTP status {source.status}")
+                with (stage / name).open("xb") as target:
+                    while chunk := source.read(1024 * 1024):
+                        target.write(chunk)
+                        digest.update(chunk)
+                        size += len(chunk)
+            if digest.hexdigest() != product["data"]["sha256"]:
+                raise ConfigError(f"product digest mismatch: {name}")
+            sizes[name] = size
+        # Single rename after all products validate. The caller owns this unique
+        # destination; concurrent writers to the same output are unsupported.
+        os.rename(stage, destination)
+    return sizes