feat(ci): enforce reusable workload plans (#8393)

* feat(ci): enforce reusable workload plans

* fix(ci): skip incompatible convergence upgrades

.github/scripts/convergence.py

@@ -32,7 +32,7 @@ from lib.github import (
     download_artifact,
     event_payload,
     run_jobs,
-    unique_run_artifact,
+    latest_run_artifact,
 )
 from lib.r2 import R2Client, R2Credentials, R2Error, R2PreconditionFailed, self_check as r2_self_check
 from lib.workload_products import materialize_products
@@ -1641,6 +1641,8 @@ def validate_admitted_plan(
 
 
 def admit_command(args: argparse.Namespace, contract: ConvergenceContract) -> int:
+    if getattr(args, "isolated", False):
+        return admit_isolated_command(args, contract)
     payload = event_payload()
     context = workflow_run_context(payload)
     if context["head_repository"] != context["repository"]:
@@ -1662,8 +1664,6 @@ def admit_command(args: argparse.Namespace, contract: ConvergenceContract) -> in
         if entry[field] != expected:
             raise ConfigError(f"convergence handoff {field} differs from workflow_run")
     workflow = contract.workflow(entry["workflow"])
-    if entry["policy"] != workflow.policy:
-        raise ConfigError("convergence handoff policy differs from trusted policy")
     base_sha = entry["base_sha"]
     head_sha = entry["head_sha"]
     subprocess.run(
@@ -1680,6 +1680,8 @@ def admit_command(args: argparse.Namespace, contract: ConvergenceContract) -> in
     elif git_differs("HEAD", base_sha, control_paths):
         reason = "producer-control-plane-superseded"
         publish = False
+    elif entry["policy"] != workflow.policy:
+        raise ConfigError("convergence handoff policy differs from trusted policy")
     if publish and contract.schema_version != 1:
         tree = authenticated_source_tree(entry, payload)
         root = args.root.resolve() if args.root else repository_root(__file__)
@@ -1695,6 +1697,45 @@ def admit_command(args: argparse.Namespace, contract: ConvergenceContract) -> in
     return 0
 
 
+def require_isolated_candidate(candidate: dict[str, Any]) -> None:
+    """Authorize same-run CI publication from the default branch only."""
+    payload = event_payload()
+    default_branch = payload.get("repository", {}).get("default_branch", "")
+    if (os.environ.get("GITHUB_EVENT_NAME") != "workflow_dispatch"
+            or os.environ.get("GITHUB_REPOSITORY") != "nexu-io/open-design"
+            or os.environ.get("GITHUB_REF") != f"refs/heads/{default_branch}"):
+        raise ConfigError("isolated publication requires the authorized CI default branch")
+    context = producer_context(payload)
+    if (candidate.get("workflow") != "ci"
+            or candidate.get("policy") != "ci-v2"
+            or candidate.get("repositoryId") != context["repositoryId"]
+            or candidate.get("repository") != context["repository"]):
+        raise ConfigError("isolated candidate repository/workflow/policy differs")
+    provenance = candidate.get("provenance", {})
+    for field in ("event", "runId", "runAttempt", "headSha", "baseSha", "treeSha"):
+        if provenance.get(field) != context["provenance"][field]:
+            raise ConfigError(f"isolated candidate {field} differs from current run")
+    with tempfile.TemporaryDirectory() as directory:
+        path = Path(directory) / "candidate.json"
+        write_json_atomic(path, candidate)
+        prepare_publication(path, Path(directory) / "receipts", require_urls=False)
+
+
+def admit_isolated_command(args: argparse.Namespace, contract: ConvergenceContract) -> int:
+    entries = handoff_contract.candidate_entry_dirs(args.handoff_root, "convergence")
+    if len(entries) != 1:
+        raise ConfigError("isolated publication requires exactly one successful gate handoff")
+    entry = handoff_contract.validate_convergence(entries[0])
+    candidate = load_json(Path(entry["candidate_path"]))
+    require_isolated_candidate(candidate)
+    if contract.workflow(candidate["workflow"]).policy != candidate["policy"]:
+        raise ConfigError("isolated policy differs from declaration")
+    root = args.root.resolve() if args.root else repository_root(__file__)
+    validate_admitted_plan(candidate, contract, root, candidate["provenance"]["treeSha"])
+    append_outputs({"candidate": entry["candidate_path"], "publish": "true"})
+    return 0
+
+
 def storage_config(*, required: bool) -> dict[str, str]:
     values = {key: os.environ.get(name, "") for key, name in STORAGE_ENV.items()}
     missing = [STORAGE_ENV[key] for key, value in values.items() if not value]
@@ -1721,7 +1762,7 @@ def stage_products_command(args: argparse.Namespace) -> int:
     args.output_dir.mkdir(parents=True, exist_ok=True)
     staged = []
     for source in candidate_product_sources(args.candidate):
-        artifact = unique_run_artifact(repository, run_id, source)
+        artifact = latest_run_artifact(repository, run_id, source)
         if artifact is None:
             raise ConfigError(f"current-run product artifact is missing: {source}")
         destination = args.output_dir / f"{source}.zip"
@@ -1984,6 +2025,8 @@ def contribute_command(args: argparse.Namespace, contract: ConvergenceContract)
 
 
 def publish_command(args: argparse.Namespace) -> int:
+    if getattr(args, "isolated", False):
+        require_isolated_candidate(load_json(args.candidate))
     storage = storage_config(required=True)
     origin = public_origin(storage["public_origin"])
     client = R2Client(
@@ -2147,6 +2190,7 @@ def parse_args() -> argparse.Namespace:
     batches.add_argument("--output-dir", type=Path, required=True)
     admit = sub.add_parser("admit")
     admit.add_argument("--handoff-root", type=Path, required=True)
+    admit.add_argument("--isolated", action="store_true")
     publication = sub.add_parser("prepare-publication")
     publication.add_argument("--candidate", type=Path, required=True)
     publication.add_argument("--output-dir", type=Path, required=True)
@@ -2160,6 +2204,7 @@ def parse_args() -> argparse.Namespace:
     publish.add_argument("--output-dir", type=Path, required=True)
     publish.add_argument("--products-root", type=Path, required=True)
     publish.add_argument("--timeout", type=float, default=15.0)
+    publish.add_argument("--isolated", action="store_true")
     publish.add_argument("--pending", type=Path, help="project consumer requests after trusted publication")
     return parser.parse_args()
 

.github/scripts/handoff.py

@@ -10,7 +10,7 @@ import tempfile
 from pathlib import Path
 from typing import Any
 
-from lib.github import GitHubError, append_outputs, event_payload, unique_run_artifact
+from lib.github import GitHubError, append_outputs, event_payload, latest_run_artifact
 
 SCHEMA_VERSION = 1
 KINDS = {"comment", "autofix", "report", "convergence"}
@@ -305,8 +305,12 @@ def write_convergence(root: Path, handoff_id: str, candidate: dict[str, Any]) ->
 
 def resolve_run_artifact(kind: str, handoff_id: str, run_id: int, repository: str) -> None:
     name = artifact_name(kind, handoff_id)
-    artifact = unique_run_artifact(repository, run_id, name)
-    append_outputs({"found": str(artifact is not None).lower(), "name": name if artifact else ""})
+    artifact = latest_run_artifact(repository, run_id, name)
+    append_outputs({
+        "found": str(artifact is not None).lower(),
+        "id": str(artifact["id"]) if artifact else "",
+        "name": name if artifact else "",
+    })
 
 
 

.github/scripts/lib/github.py

@@ -97,7 +97,13 @@ def run_jobs(repository: str, run_id: int, attempt: int) -> list[dict[str, Any]]
         page += 1
 
 
-def unique_run_artifact(repository: str, run_id: int, name: str) -> dict[str, Any] | None:
+def latest_run_artifact(repository: str, run_id: int, name: str) -> dict[str, Any] | None:
+    """Select the newest surviving artifact across workflow run attempts.
+
+    Failed-job reruns retain artifacts from earlier attempts and may upload the
+    same name again. Consumers still validate the downloaded handoff's attempt,
+    so an older artifact can never become accepted evidence for a newer attempt.
+    """
     matches = [
         artifact
         for artifact in run_artifacts(repository, run_id)
@@ -105,12 +111,13 @@ def unique_run_artifact(repository: str, run_id: int, name: str) -> dict[str, An
     ]
     if not matches:
         return None
-    if len(matches) != 1:
-        raise GitHubError(f"expected exactly one current-run artifact named {name!r}, found {len(matches)}")
-    artifact_id = matches[0].get("id")
-    if not isinstance(artifact_id, int) or artifact_id <= 0:
-        raise GitHubError(f"artifact {name!r} has an invalid id")
-    return matches[0]
+    for artifact in matches:
+        artifact_id = artifact.get("id")
+        if not isinstance(artifact_id, int) or artifact_id <= 0:
+            raise GitHubError(f"artifact {name!r} has an invalid id")
+        if not isinstance(artifact.get("created_at"), str) or not artifact["created_at"]:
+            raise GitHubError(f"artifact {name!r} has an invalid creation time")
+    return max(matches, key=lambda artifact: (artifact["created_at"], artifact["id"]))
 
 
 def download_artifact(repository: str, artifact_id: int, destination: Path) -> None:

.github/scripts/scopes.py

@@ -14,18 +14,35 @@ from lib.github import append_outputs, append_summary
 
 
 CONFIDENCE = {"medium": 0, "certain": 1}
+DAEMON_UNIT_WORKLOADS = {f"daemon_unit_{shard}" for shard in range(1, 5)}
+WEB_WORKSPACE_WORKLOADS = {f"web_workspace_{shard}" for shard in range(1, 3)}
+UI_P0_WORKLOADS = {
+    "ui_p0_entry_settings",
+    "ui_p0_project_workspace",
+    "ui_p0_project_workspace_editor",
+    "ui_p0_project_collab",
+    "ui_p0_project_runtime",
+    "ui_p0_workspace_restoration",
+}
 WORKLOADS = {
+    "platform_build",
+    "platform_tests",
     "static_gate",
     "preflight",
     "workspace_unit_tests",
-    "daemon_unit_tests",
+    *DAEMON_UNIT_WORKLOADS,
     "windows_tools_pack_payload_tests",
-    "web_workspace_tests",
+    *WEB_WORKSPACE_WORKLOADS,
     "e2e_vitest",
     "playwright_critical",
-    "ui_p0",
+    *UI_P0_WORKLOADS,
     "playwright_visual",
 }
+SELECTION_ALIASES = {
+    "daemon_unit_tests": DAEMON_UNIT_WORKLOADS,
+    "web_workspace_tests": WEB_WORKSPACE_WORKLOADS,
+    "ui_p0": UI_P0_WORKLOADS,
+}
 
 
 class ScopeContract:
@@ -185,15 +202,17 @@ def enabled_workloads(outputs, ci_mode, full_lanes):
     broad = full_lanes or ci_mode == "hot" or any_scope
     ui_p0 = full_lanes or outputs["ui_p0_validation_required"]
     enabled = {
+        "platform_build": broad,
+        "platform_tests": broad,
         "static_gate": True,
         "preflight": True,
         "workspace_unit_tests": broad,
-        "daemon_unit_tests": outputs["daemon_tests_required"],
+        **{name: outputs["daemon_tests_required"] for name in DAEMON_UNIT_WORKLOADS},
         "windows_tools_pack_payload_tests": full_lanes or outputs["windows_tools_pack_payload_tests_required"],
-        "web_workspace_tests": full_lanes or outputs["web_tests_required"],
+        **{name: full_lanes or outputs["web_tests_required"] for name in WEB_WORKSPACE_WORKLOADS},
         "e2e_vitest": full_lanes or outputs["web_tests_required"] or outputs["ui_p0_validation_required"],
         "playwright_critical": outputs["ui_critical_validation_required"] and not ui_p0,
-        "ui_p0": ui_p0,
+        **{name: ui_p0 for name in UI_P0_WORKLOADS},
         "playwright_visual": full_lanes or outputs["visual_validation_required"],
     }
     if set(enabled) != WORKLOADS:
@@ -302,6 +321,7 @@ def emit_plan(plan, output_path=None):
         output_path.parent.mkdir(parents=True, exist_ok=True)
         output_path.write_text(json.dumps(plan, indent=2, sort_keys=True) + "\n", encoding="utf-8")
     append_outputs({
+        "enabled": compact_json(plan["enabled"]),
         "scopes": compact_json(plan["scopes"]),
         "ui_p0_matrix": compact_json(plan["matrices"]["ui_p0"]),
         "visual_matrix": compact_json(plan["matrices"]["visual"]),
@@ -348,7 +368,19 @@ def main():
         return 0
     if args.command == "github-output":
         source, files, threshold, mode, full_lanes, derive, resolved = changed_files_for_environment()
-        emit_plan(build_plan(contract, files, source, threshold, mode, full_lanes, derive, resolved), args.output)
+        plan = build_plan(contract, files, source, threshold, mode, full_lanes, derive, resolved)
+        selection = os.environ.get("CI_WORKLOADS", "").strip()
+        if selection:
+            if os.environ.get("GITHUB_EVENT_NAME") != "workflow_dispatch":
+                raise ConfigError("explicit workload selection requires workflow_dispatch")
+            requested = set(selection.split(","))
+            if not requested.issubset(WORKLOADS | set(SELECTION_ALIASES)):
+                raise ConfigError("explicit selection contains unknown workload identities")
+            selected = set().union(*(SELECTION_ALIASES.get(name, {name}) for name in requested))
+            plan["enabled"] = {name: name in selected for name in WORKLOADS}
+            plan["source"] = "workflow_dispatch:selected"
+            plan["selection"] = sorted(requested)
+        emit_plan(plan, args.output)
         return 0
     files = list(args.files)
     if args.files_from:

.github/scripts/template.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""Render strict text templates owned by GitHub automation."""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+
+TEMPLATES_ROOT = Path(__file__).resolve().parents[1] / "templates"
+ALLOWED_SUFFIXES = {".md", ".txt"}
+PLACEHOLDER = re.compile(r"{{\s*([a-z][a-z0-9_]*)\s*}}")
+
+
+class TemplateError(ValueError):
+    pass
+
+
+def resolve_template(path_value: str) -> Path:
+    relative = Path(path_value)
+    if relative.is_absolute():
+        raise TemplateError("template path must be relative to .github/templates")
+    path = (TEMPLATES_ROOT / relative).resolve()
+    try:
+        path.relative_to(TEMPLATES_ROOT.resolve())
+    except ValueError as error:
+        raise TemplateError("template path must stay within .github/templates") from error
+    if path.suffix not in ALLOWED_SUFFIXES:
+        raise TemplateError("template must use a .md or .txt extension")
+    if not path.is_file():
+        raise TemplateError(f"template does not exist: {path_value}")
+    return path
+
+
+def template_parameters(source: str) -> set[str]:
+    parameters = set(PLACEHOLDER.findall(source))
+    remainder = PLACEHOLDER.sub("", source)
+    if "{{" in remainder or "}}" in remainder:
+        raise TemplateError("template contains an invalid placeholder")
+    return parameters
+
+
+def render_template(path_value: str, params: dict[str, str]) -> str:
+    source = resolve_template(path_value).read_text(encoding="utf-8")
+    expected = template_parameters(source)
+    missing = sorted(expected - params.keys())
+    unknown = sorted(params.keys() - expected)
+    if missing:
+        raise TemplateError(f"missing template parameters: {', '.join(missing)}")
+    if unknown:
+        raise TemplateError(f"unknown template parameters: {', '.join(unknown)}")
+    return PLACEHOLDER.sub(lambda match: params[match.group(1)], source)
+
+
+def parse_args(argv: list[str]) -> tuple[str, dict[str, str], Path | None]:
+    if not argv or argv[0] in {"-h", "--help"}:
+        print(
+            "usage: template.py <path> [--params-<name> <value>]... [--output <path>]",
+            file=sys.stdout,
+        )
+        raise SystemExit(0)
+
+    path_value = argv[0]
+    params: dict[str, str] = {}
+    output: Path | None = None
+    index = 1
+    while index < len(argv):
+        argument = argv[index]
+        if "=" in argument:
+            option, value = argument.split("=", 1)
+            consumed = 1
+        else:
+            option = argument
+            if index + 1 >= len(argv):
+                raise TemplateError(f"missing value for {option}")
+            value = argv[index + 1]
+            consumed = 2
+
+        if option == "--output":
+            if output is not None:
+                raise TemplateError("--output may only be provided once")
+            output = Path(value)
+        elif option.startswith("--params-"):
+            name = option.removeprefix("--params-").replace("-", "_")
+            if not re.fullmatch(r"[a-z][a-z0-9_]*", name):
+                raise TemplateError(f"invalid template parameter option: {option}")
+            if name in params:
+                raise TemplateError(f"duplicate template parameter: {name}")
+            params[name] = value
+        else:
+            raise TemplateError(f"unknown option: {option}")
+        index += consumed
+    return path_value, params, output
+
+
+def main(argv: list[str] | None = None) -> int:
+    try:
+        path_value, params, output = parse_args(sys.argv[1:] if argv is None else argv)
+        rendered = render_template(path_value, params)
+        if output is None:
+            sys.stdout.write(rendered)
+        else:
+            output.write_text(rendered, encoding="utf-8")
+        return 0
+    except TemplateError as error:
+        print(f"template error: {error}", file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())