H3 serving smoke
'
- 'One hardware configuration; four measured requests at each concurrency. Warmups are separate. '
+ f'
One hardware configuration; {matrix["requests_per_configuration"]} measured requests at each concurrency. Warmups are separate. '
'Latency is submit → downloaded media for technically valid clips. Throughput is valid clips / delivery wall seconds. '
- 'Four samples do not establish P90/P95 or sustainable capacity. Failed and unstarted requests remain counted.
'
+ 'Percentiles from small samples are preliminary and do not establish sustainable capacity. Failed and unstarted requests remain counted.'
'Download summary and raw-evidence links
'
'| Concurrency | Status | Scheduled | '
'Attempted | Valid | Failed | Not started | Delivery median (s) | Valid clips/s | '
@@ -60,18 +69,21 @@ def _report(root: Path, matrix: dict) -> None:
+ '', encoding="utf-8")
-def run_matrix(spec: dict, root: Path) -> dict:
+def run_matrix(spec: dict, root: Path, *, concurrencies=CONCURRENCIES) -> dict:
from .mvp_power import analyze_power
spec = validate_spec(spec)
+ concurrencies = validate_concurrencies(concurrencies)
+ count = len(spec["plan"]["cases"]) * spec["plan"]["repetitions"]
deadline = time.monotonic() + spec["limits"]["job_seconds"]
matrix = {"schema_version": "1.0.0", "bundle_type": "h3_serving_smoke_matrix", "status": "running",
- "started_at": gpu._now(), "plan": spec["plan"], "runtime": spec["baseline"], "gpu_uuids": spec["gpu_uuids"],
- "scheduled": 12, "warmup_per_configuration": spec["plan"]["warmup_runs"],
+ "started_at": gpu._now(), "plan": spec["plan"], "runtime": spec["baseline"], "server": spec["server"], "gpu_uuids": spec["gpu_uuids"],
+ "scheduled": count * len(concurrencies), "requests_per_configuration": count,
+ "warmup_per_configuration": spec["plan"]["warmup_runs"],
"ci_accepted": False, "release_qualified": False,
"cells": [{"concurrency": concurrency, "status": "not_started", "verified": False,
- "completion": {"scheduled": 4, "attempted": 0, "completed": 0, "valid": 0, "failed": 4, "not_started": 4, "unfinished": 0}}
- for concurrency in CONCURRENCIES]}
+ "completion": {"scheduled": count, "attempted": 0, "completed": 0, "valid": 0, "failed": count, "not_started": count, "unfinished": 0}}
+ for concurrency in concurrencies]}
path = root / "serving-smoke.json"
gpu._write(path, matrix)
try:
@@ -93,19 +105,19 @@ def run_matrix(spec: dict, root: Path) -> dict:
if run_path.is_file():
raw = gpu._read(run_path)
cell["run"] = {"path": run_path.relative_to(root).as_posix(), "sha256": gpu._hash(run_path)}
- summary = _summary(raw["records"], 4, raw["measurement"]["wall_seconds"])
+ summary = _summary(raw["records"], count, raw["measurement"]["wall_seconds"])
cell["completion"] = {key: summary[key] for key in ("scheduled", "completed", "valid", "failed")}
finished = {r["slot_id"] for r in raw["records"] if r["phase"] == "measurement" and r["attempted"]}
journal = directory / "baseline/events.jsonl"
events = [json.loads(line) for line in journal.read_text().splitlines()] if journal.exists() else []
started = finished | {event["slot_id"] for event in events if event["event"] == "attempt_started" and event["slot_id"].startswith("measurement-")}
- cell["completion"].update(attempted=len(started), not_started=4-len(started), unfinished=len(started-finished))
+ cell["completion"].update(attempted=len(started), not_started=count-len(started), unfinished=len(started-finished))
gpu._write(path, matrix)
verified = verify_measurement_job(directory, deadline=deadline, require_success=True, serving_smoke=True)
run, role = verified["runs"]["baseline"], receipt["roles"]["baseline"]
cell.update(status="complete", verified=True, metrics={
"client_ready_p50_seconds": run["serving"]["client_ready_latency_seconds"]["p50"],
- "valid_clips_per_second": _summary(run["records"], 4, run["measurement"]["wall_seconds"])["valid_clips_per_second"],
+ "valid_clips_per_second": _summary(run["records"], count, run["measurement"]["wall_seconds"])["valid_clips_per_second"],
"serving": run["serving"], "measurement": run["measurement"],
})
samples = [json.loads(line) for line in (directory / role["telemetry_path"]).read_text().splitlines()]
diff --git a/experimental/video-generation/export_ci.py b/experimental/video-generation/export_ci.py
index 9874e82cd4..45f970dccb 100644
--- a/experimental/video-generation/export_ci.py
+++ b/experimental/video-generation/export_ci.py
@@ -57,9 +57,9 @@ def verified_execution(run_id: str, *, inventory: bool = False) -> tuple[dict, d
and ((run["status"] == "completed" and run["conclusion"] == "success") or current_export),
"Source must be a successful manual InferenceX execution or this run's completed H3 job")
jobs = api(f"actions/runs/{run_id}/attempts/{run['run_attempt']}/jobs")
- job_name = "H3 H200 hardware inventory" if inventory else "H3 video H200 smoke"
+ job_pattern = re.escape("H3 H200 hardware inventory") if inventory else r"H3 video (?:H100|H200|B200|MI355X) smoke"
selected = [job for job in jobs["jobs"] if re.fullmatch(
- r"(?:h3-video / )?p[0-9]+(?:\.[0-9]+)? \| " + re.escape(job_name), job["name"])]
+ r"(?:h3-video / )?p[0-9]+(?:\.[0-9]+)? \| " + job_pattern, job["name"])]
ci.need(len(selected) == 1 and selected[0]["status"] == "completed"
and selected[0]["conclusion"] == "success", "Source lacks a successful H3 Slurm job")
name = f"h3-{'hardware' if inventory else 'video'}-{run_id}-{run['run_attempt']}"
diff --git a/experimental/video-generation/inspect_amd_node.py b/experimental/video-generation/inspect_amd_node.py
new file mode 100644
index 0000000000..6ad873494b
--- /dev/null
+++ b/experimental/video-generation/inspect_amd_node.py
@@ -0,0 +1,185 @@
+"""Bounded AMD node inventory using the existing Slurm ownership receipts."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import pwd
+import re
+import shutil
+import subprocess
+
+import ci
+
+
+def step_gpu_indices(value: str) -> set[int]:
+ indices = set()
+ for part in value.split(","):
+ ci.need(re.fullmatch(r"\d{1,2}(?:-\d{1,2})?", part) is not None, "Missing AMD step GPU assignment")
+ limits = [int(item) for item in part.split("-")]
+ first, last = limits[0], limits[-1]
+ ci.need(0 <= first <= last < 8, "Unexpected AMD step GPU assignment")
+ indices.update(range(first, last + 1))
+ return indices
+
+
+def observation(argv: list[str]) -> dict:
+ try:
+ result = subprocess.run(argv, capture_output=True, text=True, timeout=30)
+ return {"argv": argv, "exit_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr}
+ except (OSError, subprocess.TimeoutExpired) as error:
+ return {"argv": argv, "error": str(error)}
+
+
+def inspect_node(run_dir: Path) -> None:
+ context = ci.read(run_dir / "context.json")
+ job = context["allocation"]["identity"]["JobId"]
+ ci.need(os.environ.get("SLURM_JOB_ID") == job
+ and os.environ.get("SLURMD_NODENAME") == context["node"], "Wrong AMD inventory allocation")
+ ci.need(step_gpu_indices(os.environ.get("SLURM_STEP_GPUS", "")) == set(range(8)),
+ "AMD inventory requires the full eight-GPU step binding")
+ binding = {"job_id": job, "step_id": os.environ.get("SLURM_STEP_ID"),
+ "node": context["node"], "cpu_affinity": sorted(os.sched_getaffinity(0)),
+ "slurm": {name: os.environ.get(name) for name in
+ ("SLURM_JOB_GPUS", "SLURM_STEP_GPUS", "ROCR_VISIBLE_DEVICES", "HIP_VISIBLE_DEVICES", "CUDA_VISIBLE_DEVICES")}}
+ ci.write(run_dir / "binding.json", binding)
+ from evaluator.mvp_amd_gpu import observe_system_monitor
+ ci.write(run_dir / "amd-system-monitor.json", observe_system_monitor(10))
+ commands = [["uname", "-a"], ["enroot", "list", "-f"], ["srun", "--help"],
+ ["bash", "-c", "command -v amd-smi rocm-smi rocminfo python3; ls -ld /opt/rocm* /var/lib/enroot /run/enroot /etc/enroot 2>/dev/null"]]
+ smi = shutil.which("amd-smi")
+ if not smi and Path("/opt/rocm/bin/amd-smi").is_file():
+ smi = "/opt/rocm/bin/amd-smi"
+ if smi:
+ commands += [[smi, option, "--json"] for option in ("list", "static", "metric", "process")]
+ commands += [[smi, "version", "--json"]]
+ processes = observation([smi, "process", "--json"])
+ if processes.get("exit_code") == 0:
+ pids = {item["process_info"]["pid"] for gpu in json.loads(processes["stdout"])
+ for item in gpu.get("process_list", []) if isinstance(item.get("process_info", {}).get("pid"), int)}
+ identities = []
+ for pid in sorted(pids):
+ proc = Path("/proc") / str(pid)
+ identity = {"pid": pid}
+ for field in ("exe", "comm", "status", "cgroup"):
+ try:
+ identity[field] = str((proc / field).resolve(strict=True)) if field == "exe" else (proc / field).read_text().strip()
+ except OSError as error:
+ identity[field + "_error"] = str(error)
+ identities.append(identity)
+ ci.write(run_dir / "observed-processes.json", identities)
+ cache = Path("/var/lib/squash")
+ images = [{"path": str(p), "size_bytes": p.stat().st_size} for p in sorted(cache.glob("*sglang*rocm*.sqsh"))]
+ ci.write(run_dir / "node-inventory.json", {
+ "schema_version": 1, "bundle_type": "h3_amd_node_inventory", "binding": binding,
+ "cached_images": images, "observations": [observation(argv) for argv in commands],
+ "generation_executed": False, "runtime_compatibility": "not_tested", "observed_at": ci.now(),
+ })
+ if context.get("prepare_runtime"):
+ from prepare_amd_runtime import prepare_on_node
+ prepare_on_node(Path(context["workspace"]), run_dir)
+
+
+def inspect(workspace: Path, output: Path, *, prepare_runtime: bool = False, serving_continuation: bool = False) -> int:
+ ci.need(not serving_continuation or prepare_runtime, "Serving continuation requires runtime inspection")
+ ci.need(workspace.is_absolute(), "Persistent workspace must be absolute")
+ run_id, attempt = os.environ["H3_RUN_ID"], os.environ["H3_RUN_ATTEMPT"]
+ ci.need(run_id.isdigit() and attempt.isdigit(), "Invalid CI identity")
+ root = workspace / "results/h3-cross-hardware"
+ root.mkdir(parents=True, exist_ok=True)
+ run_dir = root / (f"github-{run_id}-{attempt}" + ("-runtime" if serving_continuation else ""))
+ run_dir.mkdir(exist_ok=False)
+ account = pwd.getpwuid(os.getuid()).pw_name
+ ci.need(account == "cameronamd@semianalysis.com", "Unexpected AMD scheduler identity")
+ config = {"task_id": "h3-cross-hardware", "mode": "serving-smoke", "allocation_receipts": [],
+ "site": {"cluster": "mi355x-amds", "partition": "compute", "account": account, "gpu_model": "MI355X"},
+ "resources": {"gpus": 8, "allocated_gpus": 8, "cpus": 8, "memory_gb": 64, "minutes": 15}}
+ if prepare_runtime:
+ config["resources"].update(cpus=32, memory_gb=256, minutes=60)
+ if serving_continuation:
+ config["resources"].update(memory_gb=1024, minutes=120)
+ state = {"status": "starting", "started_at": ci.now(), "config": config,
+ "purpose": "Observe actual AMD device identities, telemetry formats and cached runtimes before adding a GPU adapter",
+ "source_sha": os.environ.get("H3_SOURCE_SHA"), "generation_executed": False}
+ receipt, reused, code = None, False, 2
+ control = workspace / "campaigns/h3-cross-hardware"
+ control.mkdir(parents=True, exist_ok=True)
+ with ci.task_lock(control / ".node-inventory.lock"):
+ try:
+ if prepare_runtime:
+ from prepare_amd_runtime import prepare_source, interrupted_rootfs, CONTAINER, IMAGE
+ prepare_source(workspace)
+ rootfs = workspace.parent / "enroot-data" / CONTAINER
+ if rootfs.is_dir():
+ origin = ci.read(rootfs.with_suffix(".image.json"))
+ if origin == {"image": str(IMAGE), "status": "creating"}:
+ interrupted_rootfs(workspace)
+ else:
+ ci.need(origin in ({"image": str(IMAGE), "status": "created"},
+ {"image": str(IMAGE), "status": "recovered"}), "AMD rootfs identity differs")
+ if not serving_continuation:
+ config["resources"]["minutes"] = 15
+ ci.need(not serving_continuation or rootfs.is_dir(), "Serving continuation requires the retained rootfs before allocation")
+ recovery = ci.recover(config, root)
+ ci.write(run_dir / "recovery.json", recovery)
+ ci.need(recovery["action"] != "wait", "Task-owned AMD allocation is waiting; do not submit another")
+ if recovery["action"] == "reuse":
+ ci.need(not recovery["active_steps"].strip(), "Task-owned AMD allocation has active steps")
+ receipt, reused = recovery["receipt"], True
+ ci.write(run_dir / "allocation.json", receipt)
+ else:
+ receipt = ci.allocate(config, run_dir)
+ record = ci.job_record(receipt["identity"]["JobId"])
+ state.update(allocation=receipt, allocation_reused=reused, slurm_job=record)
+ ci.write(run_dir / "slurm-job.json", record)
+ ci.verify_identity(receipt, record, config["task_id"])
+ ci.need(record["JobState"] == "RUNNING", "Owned AMD allocation is " + record["JobState"])
+ # AMD AllocTRES omits GPU accounting even for a granted --gres=gpu:8
+ # request. This read-only inventory checks CPU/memory/time here and
+ # verifies the full eight-GPU Slurm step binding before device queries.
+ reason = ci.capacity(record, {**config["resources"], "gpus": 0})
+ ci.need(reason is None, "Owned AMD allocation: " + str(reason))
+ ci.write(run_dir / "context.json", {"allocation": receipt, "node": record["NodeList"],
+ "workspace": str(workspace), "prepare_runtime": prepare_runtime})
+ # The source checkout and result directory are on the shared filesystem.
+ step_minutes = 10 if serving_continuation else config["resources"]["minutes"] - 5
+ argv = ["srun", "--jobid=" + record["JobId"], "--nodelist=" + record["NodeList"],
+ "--nodes=1", "--ntasks=1", "--gres=gpu:8", "--cpus-per-task=" + str(config["resources"]["cpus"]), "--cpu-bind=cores",
+ "--time=" + str(step_minutes), "--export=NONE", "/usr/bin/python3", str(Path(__file__).resolve()),
+ "--inside", str(run_dir)]
+ ci.write(run_dir / "step-command.json", argv)
+ code = ci.run_step(argv, run_dir / "srun.log", (step_minutes + 1) * 60)
+ ci.need(code == 0 and (run_dir / "node-inventory.json").is_file(), "AMD inventory step failed; inspect retained logs")
+ state["status"] = "complete"
+ except Exception as error:
+ state.update(status="failed", error=str(error))
+ code = 2
+ finally:
+ if receipt is None and (run_dir / "allocation.json").exists():
+ receipt = ci.read(run_dir / "allocation.json")
+ try:
+ if receipt:
+ state["step_cleanup"] = ci.drain_step(receipt, config["task_id"], run_dir)
+ retain = reused or (serving_continuation and state["status"] == "complete")
+ state["allocation_cleanup"] = {"status": "retained"} if retain else ci.stop_allocation(receipt, config["task_id"])
+ except Exception as error:
+ state.update(status="failed", cleanup_error=str(error))
+ code = 2
+ state.update(finished_at=ci.now(), exit_code=code)
+ ci.write(run_dir / "inventory-status.json", state)
+ ci.collect(run_dir, output)
+ return code
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--workspace", type=Path)
+ parser.add_argument("--output", type=Path)
+ parser.add_argument("--inside", type=Path)
+ parser.add_argument("--prepare-runtime", action="store_true")
+ args = parser.parse_args()
+ if args.inside:
+ inspect_node(args.inside)
+ else:
+ raise SystemExit(inspect(args.workspace, args.output, prepare_runtime=args.prepare_runtime))
diff --git a/experimental/video-generation/prepare_amd_runtime.py b/experimental/video-generation/prepare_amd_runtime.py
new file mode 100644
index 0000000000..85c4dcfa3a
--- /dev/null
+++ b/experimental/video-generation/prepare_amd_runtime.py
@@ -0,0 +1,228 @@
+"""Prepare one task-owned cached ROCm runtime; generation is a separate gate."""
+from __future__ import annotations
+
+import argparse
+import os
+from pathlib import Path
+import subprocess
+
+import ci
+
+REVISION = "71de97b264b04dcd514cf904003028aefe9775c8"
+IMAGE = Path("/var/lib/squash/lmsysorg_sglang-rocm_v0.5.18-rocm720-mi35x-20260828.sqsh")
+CONTAINER = "wenyao-minimax-h3-rocm"
+
+
+CPU_PROBE = r'''
+import importlib, importlib.metadata as metadata, json, sys
+from pathlib import Path
+result = {"python": sys.executable, "packages": {}, "imports": {}, "gpu_execution": False}
+for name in ("torch", "torchvision", "av", "numpy", "diffusers", "transformers", "sglang", "aiter", "triton", "amdsmi"):
+ try:
+ module = importlib.import_module(name)
+ result["imports"][name] = {"path": getattr(module, "__file__", None)}
+ try: result["packages"][name] = metadata.version(name)
+ except metadata.PackageNotFoundError: pass
+ except Exception as error:
+ result["imports"][name] = {"error": str(error)}
+Path(sys.argv[1]).write_text(json.dumps(result, indent=2) + "\n")
+'''
+
+
+def interrupted_rootfs(workspace: Path) -> None:
+ rootfs = workspace.parent / "enroot-data" / CONTAINER
+ previous = workspace / "results/h3-cross-hardware/github-34344130223-1"
+ ci.need(rootfs.is_dir() and rootfs.stat().st_uid == os.getuid(), "Task-owned partial rootfs missing")
+ ci.need(ci.read(rootfs.with_suffix(".image.json")) == {"image": str(IMAGE), "status": "creating"},
+ "Unexpected rootfs creation receipt")
+ failed = ci.read(previous / "inventory-status.json")
+ ci.need(failed["allocation_cleanup"]["status"] == "released", "Prior allocation cleanup is not recorded")
+ log = (previous / "srun.log").read_text()
+ ci.need("Ignoring xattrs in filesystem" in log and "created 464452 files" in log
+ and "created 11757 symlinks" in log, "Prior extraction did not reach the recorded completion footer")
+ ci.need((rootfs / "etc/rc").is_file(), "Extracted Enroot entrypoint missing")
+
+
+def recover_rootfs(workspace: Path, output: Path) -> None:
+ root = workspace.parent
+ rootfs = root / "enroot-data" / CONTAINER
+ origin = rootfs.with_suffix(".image.json")
+ previous = workspace / "results/h3-cross-hardware/github-34344130223-1"
+ record = {"rootfs": str(rootfs), "image": str(IMAGE), "source_revision": REVISION,
+ "started_at": ci.now(), "gpu_allocation": False, "gpu_execution": False,
+ "predecessor_run": str(previous), "new_extraction": False}
+ output.mkdir(parents=True, exist_ok=True)
+ control = workspace / "campaigns/h3-cross-hardware"
+ with ci.task_lock(control / ".node-inventory.lock"), ci.task_lock(root / ".session.lock"):
+ try:
+ interrupted_rootfs(workspace)
+ record["entrypoint"] = (rootfs / "etc/rc").read_text()
+ prepare_source(workspace)
+ env = {**os.environ, "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+ "ENROOT_DATA_PATH": str(root / "enroot-data"), "ENROOT_CACHE_PATH": str(root / "cache"),
+ "ENROOT_RUNTIME_PATH": str(root / "runtime-cpu-recovery"), "ENROOT_TEMP_PATH": str(root / "tmp-cpu-recovery"),
+ "ROCR_VISIBLE_DEVICES": "", "HIP_VISIBLE_DEVICES": "", "CUDA_VISIBLE_DEVICES": ""}
+ record["permission_normalization"] = "Prior full-tree normalization exceeded 300s; test entry before repairing any specific permission"
+ probe = output / "runtime-cpu-probe.py"
+ probe.write_text(CPU_PROBE)
+ persistent = control / f"rootfs-recovery-{os.environ['H3_RUN_ID']}-{os.environ['H3_RUN_ATTEMPT']}"
+ persistent.mkdir(exist_ok=False)
+ (persistent / probe.name).write_text(CPU_PROBE)
+ inside = Path("/work") / persistent.relative_to(workspace)
+ argv = ["/usr/local/bin/enroot", "start", "--rw", "--mount", str(workspace) + ":/work",
+ "--env", "PYTHONDONTWRITEBYTECODE=1", "--env", "SGLANG_USE_AITER=1",
+ "--env", "PYTHONPATH=/work/runtime-sglang-" + REVISION + "/python",
+ "--env", "ROCR_VISIBLE_DEVICES=", "--env", "HIP_VISIBLE_DEVICES=", "--env", "CUDA_VISIBLE_DEVICES=",
+ CONTAINER, "python3", str(inside / probe.name), str(inside / "runtime-cpu-probe.json")]
+ ci.write(output / "runtime-cpu-command.json", argv)
+ subprocess.run(argv, env=env, check=True, timeout=300)
+ result = ci.read(persistent / "runtime-cpu-probe.json")
+ ci.write(output / "runtime-cpu-probe.json", result)
+ record.update(status="recovered", probe=result,
+ compatibility="CPU entry and imports only; HIP and H3 generation unverified")
+ ci.write(origin, {"image": str(IMAGE), "status": "recovered"})
+ ci.write(control / "rootfs-recovered.json", record)
+ except Exception as error:
+ record.update(status="failed", error=str(error))
+ raise
+ finally:
+ record["finished_at"] = ci.now()
+ ci.write(output / "rootfs-recovery.json", record)
+
+
+def prepare_source(workspace: Path) -> Path:
+ source = workspace / ("runtime-sglang-" + REVISION)
+ if not source.exists():
+ subprocess.run(["git", "init", "-b", "feat/h3-runtime-preparation", str(source)], check=True)
+ subprocess.run(["git", "-C", str(source), "remote", "add", "origin", "https://github.com/sgl-project/sglang.git"], check=True)
+ subprocess.run(["git", "-C", str(source), "fetch", "--depth", "1", "origin", REVISION], check=True, timeout=300)
+ subprocess.run(["git", "-C", str(source), "checkout", "--detach", REVISION], check=True)
+ ci.need(ci.command(["git", "-C", str(source), "rev-parse", "HEAD"]).strip() == REVISION, "Prepared AMD source has a different revision")
+ ci.need(not ci.command(["git", "-C", str(source), "status", "--porcelain", "--untracked-files=all"]).strip(), "Prepared AMD source is not clean")
+ return source
+
+
+PROBE = r'''
+import ctypes, importlib, importlib.metadata as metadata, json, os, shutil, subprocess, sys
+from pathlib import Path
+out = Path(sys.argv[1])
+result = {"python": sys.executable, "packages": {}, "imports": {}, "generation_executed": False}
+for name in ("torch", "torchvision", "av", "numpy", "diffusers", "transformers", "sglang", "aiter", "triton", "amdsmi"):
+ try:
+ module = importlib.import_module(name)
+ result["imports"][name] = {"path": getattr(module, "__file__", None)}
+ try: result["packages"][name] = metadata.version(name)
+ except metadata.PackageNotFoundError: pass
+ except Exception as error:
+ result["imports"][name] = {"error": str(error)}
+try:
+ import torch
+ result["torch_hip"] = torch.version.hip
+ result["torch_devices"] = [{"ordinal": i, "name": torch.cuda.get_device_name(i),
+ "total_memory_bytes": torch.cuda.get_device_properties(i).total_memory} for i in range(torch.cuda.device_count())]
+ hip = ctypes.CDLL("libamdhip64.so")
+ hip.hipGetDeviceCount.argtypes = [ctypes.POINTER(ctypes.c_int)]
+ hip.hipDeviceGetPCIBusId.argtypes = [ctypes.c_char_p, ctypes.c_int, ctypes.c_int]
+ count = ctypes.c_int()
+ assert hip.hipGetDeviceCount(ctypes.byref(count)) == 0
+ devices = []
+ for i in range(count.value):
+ bdf = ctypes.create_string_buffer(32)
+ assert hip.hipDeviceGetPCIBusId(bdf, len(bdf), i) == 0
+ devices.append({"hip_ordinal": i, "pci_bdf": bdf.value.decode().lower()})
+ result["hip_devices"] = devices
+except Exception as error:
+ result["device_error"] = str(error)
+result["smi_observations"] = []
+smi = shutil.which("amd-smi")
+if smi:
+ for option in ("version", "list", "static", "metric", "process"):
+ try:
+ call = subprocess.run([smi, option, "--json"], capture_output=True, text=True, timeout=20)
+ result["smi_observations"].append({"option": option, "path": smi, "exit_code": call.returncode,
+ "stdout": call.stdout, "stderr": call.stderr})
+ except Exception as error:
+ result["smi_observations"].append({"option": option, "error": str(error)})
+try:
+ import amdsmi
+ amdsmi.amdsmi_init()
+ result["smi_memory_bytes"] = [{"pci_bdf": amdsmi.amdsmi_get_gpu_device_bdf(handle),
+ "uuid": amdsmi.amdsmi_get_gpu_device_uuid(handle),
+ "total_bytes": amdsmi.amdsmi_get_gpu_memory_total(handle, amdsmi.AmdSmiMemoryType.VRAM),
+ "used_bytes": amdsmi.amdsmi_get_gpu_memory_usage(handle, amdsmi.AmdSmiMemoryType.VRAM)}
+ for handle in amdsmi.amdsmi_get_processor_handles()]
+ amdsmi.amdsmi_shut_down()
+except Exception as error:
+ result["smi_api_error"] = str(error)
+result["environment"] = {k: os.environ.get(k) for k in ("ROCR_VISIBLE_DEVICES", "HIP_VISIBLE_DEVICES", "CUDA_VISIBLE_DEVICES", "PYTHONPATH")}
+out.write_text(json.dumps(result, indent=2) + "\n")
+print(json.dumps(result))
+'''
+
+
+def prepare_on_node(workspace: Path, run_dir: Path) -> None:
+ # Called only after inspect_amd_node has verified this job, node and all 8 GPUs.
+ root = workspace.parent
+ enroot = Path("/usr/local/bin/enroot")
+ ci.need(enroot.is_file(), "Expected approved /usr/local/bin/enroot is unavailable")
+ rootfs = root / "enroot-data" / CONTAINER
+ record = {"rootfs": str(rootfs), "existed": rootfs.is_dir(), "image": str(IMAGE),
+ "source_revision": REVISION, "generation_executed": False, "started_at": ci.now()}
+ ci.write(run_dir / "runtime-preparation.json", record)
+ env = os.environ.copy()
+ env.update(PATH="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+ ENROOT_DATA_PATH=str(root / "enroot-data"), ENROOT_CACHE_PATH=str(root / "cache"),
+ ENROOT_RUNTIME_PATH=str(root / "runtime-preparation"), ENROOT_TEMP_PATH=str(root / "tmp-preparation"),
+ ENROOT_MAX_PROCESSORS="8", PYTHONDONTWRITEBYTECODE="1")
+ for key in ("ENROOT_DATA_PATH", "ENROOT_CACHE_PATH", "ENROOT_RUNTIME_PATH", "ENROOT_TEMP_PATH"):
+ Path(env[key]).mkdir(parents=True, exist_ok=True)
+ origin = rootfs.with_suffix(".image.json")
+ if rootfs.is_dir():
+ if origin.is_file() and ci.read(origin) == {"image": str(IMAGE), "status": "creating"}:
+ interrupted_rootfs(workspace)
+ record["recovery"] = "Reuse completed extraction after CPU runner denied user namespaces; no extraction or full-tree permission rescan"
+ else:
+ ci.need(origin.is_file() and ci.read(origin) in (
+ {"image": str(IMAGE), "status": "created"}, {"image": str(IMAGE), "status": "recovered"}),
+ "Existing AMD rootfs has no completed task-owned image receipt; inspect before reuse")
+ if not rootfs.is_dir():
+ ci.need(IMAGE.is_file(), "Validated cached ROCm image is missing on this node")
+ record.update(create_reason="Task-owned named rootfs is missing", image_size_bytes=IMAGE.stat().st_size)
+ ci.write(run_dir / "runtime-preparation.json", record)
+ ci.write(origin, {"image": str(IMAGE), "status": "creating"})
+ subprocess.run([str(enroot), "create", "--name", CONTAINER, str(IMAGE)], env=env, check=True, timeout=2400)
+ ci.write(origin, {"image": str(IMAGE), "status": "created"})
+ rc = rootfs / "etc/rc"
+ record["entrypoint"] = rc.read_text() if rc.is_file() else None
+ ci.write(run_dir / "runtime-preparation.json", record)
+ probe = run_dir / "runtime-probe.py"
+ probe.write_text(PROBE)
+ container_dir = Path("/work") / run_dir.relative_to(workspace)
+ source = Path("/work") / ("runtime-sglang-" + REVISION)
+ # Discover the image's Python and retained version module before importing the
+ # pinned checkout. No package upgrades or model inference occur in this step.
+ script = 'set -eu; command -v python3; python3 "$1" "$2"'
+ entry = ["bash", "-c", script, "probe", str(container_dir / probe.name), str(container_dir / "runtime-probe.json")]
+ if record["entrypoint"] and 'exec bash "$@"' in record["entrypoint"]:
+ entry = entry[1:]
+ argv = [str(enroot), "start", "--rw", "--mount", str(workspace) + ":/work",
+ "--mount", "/dev/kfd:/dev/kfd", "--mount", "/dev/dri:/dev/dri",
+ "--env", "PYTHONDONTWRITEBYTECODE=1", "--env", "SGLANG_USE_AITER=1",
+ "--env", "PYTHONPATH=" + str(source / "python"),
+ "--env", "ROCR_VISIBLE_DEVICES=" + os.environ["ROCR_VISIBLE_DEVICES"], CONTAINER, *entry]
+ ci.write(run_dir / "runtime-command.json", argv)
+ subprocess.run(argv, env=env, check=True, timeout=300)
+ result = ci.read(run_dir / "runtime-probe.json")
+ ci.write(origin, {"image": str(IMAGE), "status": "recovered"})
+ record.update(finished_at=ci.now(), probe=result,
+ status="inspected", compatibility="Imports and device enumeration only; H3 generation untested")
+ ci.write(run_dir / "runtime-preparation.json", record)
+ ci.write(workspace / "campaigns/h3-cross-hardware/runtime-inspected.json", record)
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--workspace", required=True, type=Path)
+ parser.add_argument("--output", required=True, type=Path)
+ args = parser.parse_args()
+ recover_rootfs(args.workspace, args.output)
diff --git a/experimental/video-generation/run_amd_serving_ci.py b/experimental/video-generation/run_amd_serving_ci.py
new file mode 100644
index 0000000000..e56d84b31a
--- /dev/null
+++ b/experimental/video-generation/run_amd_serving_ci.py
@@ -0,0 +1,147 @@
+"""Keep AMD runtime inspection, warmup and twenty C1 requests in one lease."""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+import os
+from pathlib import Path
+import shutil
+import signal
+
+import ci
+import inspect_amd_node
+import stage_amd_site
+from prepare_amd_runtime import prepare_source
+from stage_model_ci import source_spec
+
+
+def allocation_budget(job: dict) -> tuple[int, datetime]:
+ end = datetime.fromisoformat(job["EndTime"]).replace(tzinfo=timezone.utc)
+ minutes = min(110, int((end - datetime.now(timezone.utc)).total_seconds() // 60) - 1)
+ ci.need(minutes >= 75, "Fewer than 75 serving minutes remain; do not start the twenty-request campaign")
+ ready_by = datetime.fromisoformat(job["StartTime"]).replace(tzinfo=timezone.utc) + timedelta(minutes=15)
+ ci.need(ready_by > datetime.now(timezone.utc), "Allocation-wide runtime readiness deadline expired")
+ return minutes, ready_by
+
+
+def ready_before(run_dir: Path, deadline: datetime) -> dict:
+ path = run_dir / "gpu/c1/gpu-job.json"
+ ci.need(path.is_file(), "AMD runtime was not ready within 15 minutes of allocation start")
+ role = ci.read(path).get("roles", {}).get("baseline", {})
+ seconds = role.get("startup_seconds")
+ start = role.get("startup_timing_window", {}).get("start_utc")
+ ci.need(type(seconds) in (int, float) and seconds >= 0 and isinstance(start, str),
+ "AMD runtime readiness was not recorded before the deadline")
+ ready_at = datetime.fromisoformat(start) + timedelta(seconds=seconds)
+ ci.need(ready_at <= deadline, "AMD runtime exceeded the 15-minute readiness deadline")
+ return {"ready_at": ready_at.isoformat(), "deadline": deadline.isoformat(),
+ "evidence": "gpu/c1/gpu-job.json: roles.baseline.startup_timing_window and startup_seconds"}
+
+
+def run(source_run_id: str, output: Path) -> int:
+ workspace = stage_amd_site.WORKSPACE
+ run_id = f"github-{os.environ['H3_RUN_ID']}-{os.environ['H3_RUN_ATTEMPT']}"
+ control = workspace / "campaigns/h3-cross-hardware" / run_id
+ control.mkdir(parents=True, exist_ok=False)
+ output.mkdir(parents=True, exist_ok=True)
+ preparation = control / "preparation"
+ run_dir = workspace / "results/h3-cross-hardware" / run_id
+ prep_dir = run_dir.with_name(run_id + "-runtime")
+ record = {"status": "preparing", "started_at": ci.now(), "allocation_minutes_cap": 120,
+ "allocated_gpus": 8, "participating_gpus": 4, "planned_measured_requests": 20,
+ "warmups": 1, "runtime_probe_seconds_cap": 660, "allocation_to_readiness_seconds_cap": 900,
+ "cleanup_reserve_minutes": 10, "replacement_allocation_allowed": False}
+ receipt, owned, code, ready_by = None, False, 2, None
+ def interrupted(signum, frame):
+ raise InterruptedError("AMD serving orchestration interrupted")
+ def readiness_expired(signum, frame):
+ record["readiness"] = ready_before(run_dir, ready_by)
+ previous = {sig: signal.signal(sig, handler) for sig, handler in
+ ((signal.SIGTERM, interrupted), (signal.SIGALRM, readiness_expired))}
+ try:
+ source_output = control / "source"
+ source_output.mkdir()
+ spec, record["source_provenance"] = source_spec(source_run_id, source_output)
+ ci.need(spec["plan"] == ci.read(stage_amd_site.INPUTS / "formal-8s-plan.json"),
+ "Source workload differs from the frozen twenty-request plan; no allocation requested")
+ stage_amd_site.timing_source(prepare_source(workspace))
+ code = inspect_amd_node.inspect(workspace, preparation, prepare_runtime=True, serving_continuation=True)
+ inspected = ci.read(preparation / "inventory-status.json")
+ receipt = inspected.get("allocation")
+ owned = receipt is not None and not inspected.get("allocation_reused", False)
+ record["runtime_inspection"] = inspected
+ ci.need(code == 0, "AMD runtime inspection failed; measured requests were not started")
+ stage_amd_site.runtime_probe(ci.read(workspace / "campaigns/h3-cross-hardware/runtime-inspected.json"))
+ job = ci.job_record(receipt["identity"]["JobId"])
+ ci.verify_identity(receipt, job, "h3-cross-hardware")
+ minutes, ready_by = allocation_budget(job)
+ record["readiness_deadline"] = ready_by.isoformat()
+ signal.setitimer(signal.ITIMER_REAL, (ready_by - datetime.now(timezone.utc)).total_seconds())
+ staged = control / "prepared-site"
+ staged.mkdir()
+ with ci.task_lock(workspace / "campaigns/h3-cross-hardware/.site-preparation.lock"):
+ site = stage_amd_site.stage(spec, staged, server_timing=True, allocation_minutes=minutes,
+ destination=control / "formal-c1")
+ config = ci.read(site["site_config"])
+ record.update(status="running", prepared_site=site, allocation=receipt,
+ serving_minutes_cap=minutes, supervisor_seconds_cap=minutes * 60 - 600)
+ ci.write(control / "amd-serving.json", record)
+ code = ci.launch(config, output, required_allocation=receipt["identity"]["JobId"])
+ if code == 0:
+ record["readiness"] = ready_before(run_dir, ready_by)
+ record["status"] = "complete" if code == 0 else "failed"
+ except (Exception, KeyboardInterrupt) as error:
+ record.update(status="failed", error=str(error))
+ code = 2
+ finally:
+ signal.setitimer(signal.ITIMER_REAL, 0)
+ try:
+ # Recover the persistent receipt even when artifact collection itself failed.
+ inspected = ci.read(prep_dir / "inventory-status.json") if (prep_dir / "inventory-status.json").is_file() else {}
+ if receipt is None:
+ receipt = inspected.get("allocation")
+ if receipt is None and (prep_dir / "allocation.json").is_file():
+ receipt = ci.read(prep_dir / "allocation.json")
+ recovery = ci.read(prep_dir / "recovery.json") if (prep_dir / "recovery.json").is_file() else {}
+ owned = receipt is not None and recovery.get("action") != "reuse"
+ if receipt and owned:
+ record["allocation_cleanup"] = (inspected["allocation_cleanup"]
+ if inspected.get("allocation_cleanup", {}).get("status") == "released"
+ else ci.stop_allocation(receipt, "h3-cross-hardware"))
+ elif receipt:
+ record["allocation_cleanup"] = {"status": "retained", "reason": "Borrowed allocation remains with its owner"}
+ except Exception as error:
+ record.update(status="failed", cleanup_error=str(error))
+ code = 2
+ record.update(finished_at=ci.now(), exit_code=code)
+ try:
+ ci.write(control / "amd-serving.json", record)
+ run_dir.mkdir(parents=True, exist_ok=True)
+ ci.write(run_dir / "amd-serving.json", record)
+ if prep_dir.is_dir():
+ ci.collect(prep_dir, run_dir / "preparation")
+ if (control / "prepared-site").is_dir():
+ shutil.copytree(control / "prepared-site", run_dir / "prepared-site", dirs_exist_ok=True)
+ if (run_dir / "ci.json").is_file():
+ state = ci.read(run_dir / "ci.json")
+ state.update(allocation_cleanup=record.get("allocation_cleanup"), exit_code=code)
+ if code != 0:
+ state.update(phase="failed", ci_accepted=False)
+ ci.write(run_dir / "ci.json", state)
+ manifest = ci.read(run_dir / "manifest.json")
+ manifest.update(exit_code=code, allocation_cleanup=record.get("allocation_cleanup"))
+ manifest["evidence"].update({name: ci.digest(run_dir / name) for name in ("ci.json", "amd-serving.json")})
+ ci.write(run_dir / "manifest.json", manifest)
+ ci.collect(run_dir, output)
+ finally:
+ for sig, handler in previous.items():
+ signal.signal(sig, handler)
+ return code
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--source-run-id", required=True)
+ parser.add_argument("--output", required=True, type=Path)
+ args = parser.parse_args()
+ raise SystemExit(run(args.source_run_id, args.output))
diff --git a/experimental/video-generation/runtime-entry.example.sh b/experimental/video-generation/runtime-entry.example.sh
index e8421a13a7..2c52d35f0a 100644
--- a/experimental/video-generation/runtime-entry.example.sh
+++ b/experimental/video-generation/runtime-entry.example.sh
@@ -34,7 +34,7 @@ for part in value.split(','):
elif 0 <= bounds[0] <= bounds[1] < 8:
ids.extend(range(bounds[0], bounds[1] + 1))
else:
- raise SystemExit('GPU range outside the single H200 node')
+ raise SystemExit('GPU range outside the single eight-GPU node')
if len(ids) != len(set(ids)) or not ids or any(i >= 8 for i in ids):
raise SystemExit('Invalid global GPU assignment')
devices = {}
@@ -52,10 +52,13 @@ PY
)
h3_gpu_rows=$(nvidia-smi --id="$h3_gpu_uuids" --query-gpu=uuid,name --format=csv,noheader)
H3_ASSIGNED_GPU_UUIDS=$(python3 - "$h3_gpu_rows" "$h3_gpu_uuids" <<'PY'
-import csv, re, sys
+import csv, os, re, sys
rows = list(csv.reader(sys.argv[1].splitlines()))
-if not rows or any(len(row) != 2 or 'H200' not in row[1] or not re.fullmatch(r'GPU-[0-9a-fA-F-]{36}', row[0].strip()) for row in rows):
- raise SystemExit('Assigned hardware is not a physical H200 GPU set')
+expected = os.environ.get('H3_EXPECTED_GPU_MODEL', 'H200')
+if expected not in {'H100', 'H200', 'B200'}:
+ raise SystemExit('Unsupported expected NVIDIA GPU model')
+if not rows or any(len(row) != 2 or not re.search(r'\b' + expected + r'\b', row[1]) or not re.fullmatch(r'GPU-[0-9a-fA-F-]{36}', row[0].strip()) for row in rows):
+ raise SystemExit('Assigned hardware does not match expected physical ' + expected + ' GPUs')
observed = [row[0].strip() for row in rows]
if sorted(observed) != sorted(sys.argv[2].split(',')):
raise SystemExit('NVIDIA query differs from assigned physical UUIDs')
diff --git a/experimental/video-generation/runtime-patches/sglang-71de97b-h3-server-timing.patch b/experimental/video-generation/runtime-patches/sglang-71de97b-h3-server-timing.patch
new file mode 100644
index 0000000000..f9e8b9667e
--- /dev/null
+++ b/experimental/video-generation/runtime-patches/sglang-71de97b-h3-server-timing.patch
@@ -0,0 +1,55 @@
+--- a/python/sglang/multimodal_gen/runtime/entrypoints/openai/video_api.py
++++ b/python/sglang/multimodal_gen/runtime/entrypoints/openai/video_api.py
+@@ -6,6 +6,7 @@
+ import shutil
+ import tempfile
+ import time
++from evaluator import mvp_runtime_timing as vgbench_timing
+ from collections.abc import Coroutine
+ from contextlib import suppress
+ from typing import Any, Dict, Optional
+@@ -527,6 +528,7 @@
+ update_fields, request_id=job_id, result=result
+ )
+ update_fields.update(final_media_fields)
++ vgbench_timing.emit(job_id, "media_ready")
+ await VIDEO_STORE.update_fields(job_id, update_fields)
+ except Exception as e:
+ logger.exception("Video job %s failed", job_id)
+@@ -599,6 +601,7 @@
+ ):
+ content_type = request.headers.get("content-type", "").lower()
+ request_id = generate_request_id()
++ vgbench_timing.emit(request_id, "http_received")
+
+ server_args = get_global_server_args()
+ task_type = server_args.pipeline_config.task_type
+@@ -861,6 +864,7 @@
+ )
+ job.update(sampling_params.project_video_queued_job_fields(batch))
+ await VIDEO_STORE.upsert(request_id, job)
++ vgbench_timing.emit(request_id, "http_accepted")
+ except Exception as e:
+ if batch is not None:
+ try:
+--- a/python/sglang/multimodal_gen/runtime/managers/scheduler.py
++++ b/python/sglang/multimodal_gen/runtime/managers/scheduler.py
+@@ -4,6 +4,7 @@
+ import dataclasses
+ import pickle
+ import time
++from evaluator import mvp_runtime_timing as vgbench_timing
+ from collections import deque
+ from contextlib import contextmanager
+ from copy import deepcopy
+@@ -303,7 +304,9 @@
+ )
+
+ if len(reqs) == 1 or not allow_dynamic_batching:
+- return self.worker.execute_forward(reqs)
++ with vgbench_timing.forward(reqs, replica_id=self.dp_replica,
++ leader=self.receiver is not None):
++ return self.worker.execute_forward(reqs)
+
+ if self.server_args.pipeline_config.supports_native_grouped_requests():
+ return self._execute_generation_grouped(reqs)
diff --git a/experimental/video-generation/site_preflight.py b/experimental/video-generation/site_preflight.py
new file mode 100644
index 0000000000..12d66f38e5
--- /dev/null
+++ b/experimental/video-generation/site_preflight.py
@@ -0,0 +1,134 @@
+"""Retain runner identity so SSH trust can be checked against exact CI provenance."""
+
+import argparse
+import json
+import os
+from pathlib import Path
+import platform
+import pwd
+import shutil
+import subprocess
+
+
+def amd_runtime_observations() -> dict:
+ """A failed extraction can leave reusable files; inspect before rebuilding."""
+ root = Path("/it-share/data/wenyao-minimax-h3")
+ rootfs = root / "enroot-data/wenyao-minimax-h3-rocm"
+ files = {}
+ for path in (rootfs.with_suffix(".image.json"), rootfs / "etc/rc",
+ rootfs / "etc/os-release", root / "work/campaigns/h3-cross-hardware/model-ready.json"):
+ files[str(path)] = path.read_text()[:16000] if path.is_file() else None
+ entries = {}
+ for name in ("usr/bin/python3", "usr/local/bin/python3", "opt/venv/bin/python3", "bin/bash", "etc/rc", "dev", "proc", "sys"):
+ path = rootfs / name
+ entries[name] = {"exists": path.exists(), "symlink": str(path.readlink()) if path.is_symlink() else None}
+ installed_sources = {}
+ for directory in (Path("/usr/local/lib/enroot"), Path("/usr/lib/enroot")):
+ for path in sorted(directory.glob("*.sh")):
+ lines = path.read_text().splitlines()
+ selected = set()
+ for i, line in enumerate(lines):
+ if any(word in line for word in ("unsquashfs", "runtime::create", "mksquashfs", "xattr")):
+ selected.update(range(max(0, i - 4), min(len(lines), i + 60)))
+ if selected:
+ installed_sources[str(path)] = "\n".join(f"{i + 1}: {lines[i]}" for i in sorted(selected))[:30000]
+ return {"rootfs": str(rootfs), "exists": rootfs.is_dir(), "files": files,
+ "entries": entries, "installed_enroot_sources": installed_sources,
+ "gpu_allocation": False, "rootfs_changed": False}
+
+
+def allocation_observations(root: Path) -> list[dict]:
+ """Read this task's saved scheduler receipts without requesting resources."""
+ records = []
+ for path in sorted(root.glob("*/allocation.json"))[-8:]:
+ receipt = json.loads(path.read_text())
+ identity = receipt.get("identity", {})
+ job = identity.get("JobId", "")
+ if receipt.get("task_id") != "h3-cross-hardware" or not str(job).isdigit():
+ continue
+ result = subprocess.run(["scontrol", "show", "job", "-o", str(job)],
+ capture_output=True, text=True, timeout=10,
+ env={**os.environ, "TZ": "UTC", "LC_ALL": "C"})
+ records.append({"receipt": str(path), "identity": identity, "exit_code": result.returncode,
+ "stdout": result.stdout, "stderr": result.stderr})
+ if result.returncode:
+ accounting = subprocess.run(
+ ["sacct", "-X", "-j", str(job), "--noheader", "--parsable2",
+ "--format=JobID,State,Start,End,AllocCPUS,AllocTRES,ReqTRES,NodeList,Elapsed,ExitCode"],
+ capture_output=True, text=True, timeout=10,
+ env={**os.environ, "TZ": "UTC", "LC_ALL": "C"})
+ records[-1]["accounting"] = {"exit_code": accounting.returncode, "stdout": accounting.stdout,
+ "stderr": accounting.stderr}
+ return records
+
+
+def inspect_site() -> dict:
+ user = pwd.getpwuid(os.getuid())
+ public_keys = {}
+ for algorithm in ("ed25519", "ecdsa", "rsa"):
+ path = Path(f"/etc/ssh/ssh_host_{algorithm}_key.pub")
+ if path.is_file():
+ public_keys[algorithm] = path.read_text().strip()
+ known_jumpbox = subprocess.run(
+ ["ssh-keygen", "-F", "64.139.223.123"], capture_output=True, text=True, timeout=5,
+ ) if shutil.which("ssh-keygen") else None
+ associations = subprocess.run(
+ ["sacctmgr", "-nP", "show", "assoc", "where", f"user={user.pw_name}",
+ "format=Account,Partition,QOS,DefaultQOS"], capture_output=True, text=True, timeout=10,
+ ) if shutil.which("sacctmgr") else None
+ defaults = subprocess.run(
+ ["sacctmgr", "-nP", "show", "user", "where", f"name={user.pw_name}",
+ "format=User,DefaultAccount"], capture_output=True, text=True, timeout=10,
+ ) if shutil.which("sacctmgr") else None
+ active_accounts = subprocess.run(
+ ["squeue", "--noheader", "--user=" + user.pw_name, "--format=%a"],
+ capture_output=True, text=True, timeout=10,
+ ) if shutil.which("squeue") else None
+ # Shared runtime/model cache candidates are metadata, not proof of compatibility.
+ candidates = {}
+ for name in ("/var/lib/squash", "/it-share/data", "/it-share/wenyao-minimax-h3",
+ "/data/home/sa-shared/wenyao-minimax-h3"):
+ path = Path(name)
+ candidates[name] = sorted(item.name for item in path.iterdir()
+ if any(word in item.name.lower() for word in ("sglang", "rocm", "h3", "minimax"))) if path.is_dir() else None
+ enroot_config = Path("/etc/enroot/enroot.conf")
+ enroot_paths = [line.strip() for line in enroot_config.read_text().splitlines()
+ if line.strip().startswith(("ENROOT_DATA_PATH", "ENROOT_CACHE_PATH", "ENROOT_RUNTIME_PATH"))] if enroot_config.is_file() else None
+ storage = {}
+ for name in ("/it-share", "/it-share/data", "/it-share/hf-hub-cache", "/it-share/gharunners2", user.pw_dir):
+ path = Path(name)
+ if path.is_dir():
+ info = path.stat()
+ mount = subprocess.run(["findmnt", "--target", name, "--noheadings", "--output", "TARGET,SOURCE,FSTYPE"],
+ capture_output=True, text=True, timeout=5) if shutil.which("findmnt") else None
+ storage[name] = {"uid": info.st_uid, "gid": info.st_gid, "mode": oct(info.st_mode & 0o777),
+ "writable": os.access(path, os.W_OK), "free_bytes": shutil.disk_usage(path).free,
+ "mount": mount.stdout.strip() if mount and mount.returncode == 0 else None}
+ else:
+ storage[name] = None
+ return {
+ "schema_version": 1, "bundle_type": "h3_site_preflight_no_gpu",
+ "hostname": platform.node(), "uid": os.getuid(), "username": user.pw_name,
+ "user_home": user.pw_dir,
+ "cluster": os.environ.get("H3_CLUSTER"), "runner": os.environ.get("RUNNER_NAME"),
+ "ci": {key: os.environ.get(key) for key in ("GITHUB_REPOSITORY", "GITHUB_SHA", "GITHUB_RUN_ID", "GITHUB_RUN_ATTEMPT")},
+ "commands": {name: shutil.which(name) for name in ("salloc", "srun", "enroot", "amd-smi", "nvidia-smi")},
+ "ssh_host_public_keys": public_keys,
+ "previously_known_amd_jumpbox": known_jumpbox.stdout if known_jumpbox and known_jumpbox.returncode == 0 else None,
+ "scheduler_associations": associations.stdout if associations and associations.returncode == 0 else None,
+ "scheduler_default_account": defaults.stdout if defaults and defaults.returncode == 0 else None,
+ "scheduler_active_accounts": sorted(set(active_accounts.stdout.split())) if active_accounts and active_accounts.returncode == 0 else None,
+ "runtime_candidates": candidates, "enroot_paths": enroot_paths, "persistent_storage": storage,
+ "saved_allocations": allocation_observations(Path("/it-share/data/wenyao-minimax-h3/work/results/h3-cross-hardware"))
+ if os.environ.get("H3_CLUSTER") == "mi355x-amds" and shutil.which("scontrol") else [],
+ "amd_runtime": amd_runtime_observations() if os.environ.get("H3_CLUSTER") == "mi355x-amds" else None,
+ "gpu_execution": False, "runtime_compatibility": "not_tested",
+ }
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--output", required=True, type=Path)
+ args = parser.parse_args()
+ args.output.mkdir(parents=True, exist_ok=True)
+ (args.output / "site-preflight.json").write_text(json.dumps(inspect_site(), indent=2) + "\n")
diff --git a/experimental/video-generation/stage_amd_site.py b/experimental/video-generation/stage_amd_site.py
new file mode 100644
index 0000000000..f93f8bc215
--- /dev/null
+++ b/experimental/video-generation/stage_amd_site.py
@@ -0,0 +1,148 @@
+"""Seal AMD C1 inputs after inspection, without allocating or installing."""
+from __future__ import annotations
+
+import argparse
+import copy
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+
+import ci
+from evaluator.mvp_gpu_job import source_file_manifest, validate_gpu_job
+from prepare_amd_runtime import CONTAINER, REVISION
+from stage_model_ci import source_spec
+
+WORKSPACE = Path("/it-share/data/wenyao-minimax-h3/work")
+INPUTS = Path(__file__).parent / "campaigns/h3-cross-hardware"
+
+
+def runtime_probe(record: dict) -> dict:
+ ci.need(record.get("status") in {"inspected", "recovered"} and record.get("source_revision") == REVISION,
+ "AMD runtime inspection is missing or has a different source")
+ probe = record["probe"]
+ required = ("torch", "torchvision", "av", "numpy", "diffusers", "transformers", "sglang", "aiter", "triton", "amdsmi")
+ missing = [name for name in required if not probe.get("imports", {}).get(name, {}).get("path")]
+ ci.need(not missing, "AMD runtime imports require preparation: " + ", ".join(missing))
+ if record["status"] == "inspected":
+ ci.need(probe.get("torch_hip") and not probe.get("device_error")
+ and len(probe.get("hip_devices", [])) == 8, "AMD HIP device enumeration is not verified")
+ devices = probe.get("torch_devices", [])
+ ci.need(len(devices) == 8 and all("MI355X" in item["name"] for item in devices),
+ "Runtime is not the inspected eight-MI355X node")
+ ci.need(Path(probe["python"]).is_absolute(), "Inspected Python path must be absolute")
+ return probe
+
+
+def timing_source(source: Path) -> tuple[Path, dict]:
+ from evaluator import mvp_runtime_timing as timing
+ destination = source.with_name(source.name + "-timing")
+ if not destination.exists():
+ subprocess.run(["git", "-C", str(source), "worktree", "add", "-b", "feat/h3-amd-serving-timing",
+ str(destination), REVISION], check=True)
+ subprocess.run(["git", "-C", str(destination), "apply", "--check", str(timing.PATCH)], check=True)
+ subprocess.run(["git", "-C", str(destination), "apply", str(timing.PATCH)], check=True)
+ subprocess.run(["git", "-C", str(destination), "add", *timing.PATCHED_FILES], check=True)
+ subprocess.run(["git", "-C", str(destination), "-c", "user.name=H3 Benchmark", "-c", "user.email=h3-benchmark@localhost",
+ "commit", "-m", "feat: record request-correlated H3 serving stages",
+ "-m", "记录 H3 请求的服务端阶段时间，保留原始运行时作为基线。"], check=True)
+ timing.validate_source(destination)
+ return destination, timing.identity()
+
+
+def stage(spec: dict, output: Path, *, server_timing: bool = False, allocation_minutes: int = 110, destination: Path | None = None) -> dict:
+ workspace = WORKSPACE
+ control = workspace / "campaigns/h3-cross-hardware"
+ readiness = control / "runtime-inspected.json"
+ if not readiness.is_file():
+ readiness = control / "rootfs-recovered.json"
+ runtime = ci.read(readiness)
+ probe = runtime_probe(runtime)
+ model = ci.read(control / "model-ready.json")
+ entries = spec["model"]["files"]
+ manifest = hashlib.sha256(json.dumps(entries, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+ ci.need(model.get("status") == "complete" and model.get("manifest_sha256") == manifest
+ and model.get("model_revision") == spec["model"]["revision"],
+ "Prepared AMD weights differ from the frozen source")
+ rootfs = workspace.parent / "enroot-data" / CONTAINER
+ ci.need(runtime["rootfs"] == str(rootfs) and rootfs.is_dir(), "Prepared AMD rootfs is missing or changed")
+ source = workspace / ("runtime-sglang-" + REVISION)
+ identity = source_file_manifest(source)
+ ci.need(identity["revision"] == REVISION, "AMD source revision changed after inspection")
+ instrumentation = None
+ if server_timing:
+ source, instrumentation = timing_source(source)
+ identity = source_file_manifest(source)
+ destination = destination or control / ("formal-c1-timing-v1" if server_timing else "formal-c1-v1")
+ ci.need(not destination.exists(), "AMD formal inputs already exist; inspect and reuse the sealed configuration")
+ source_server = copy.deepcopy(spec["server"])
+ spec = copy.deepcopy(spec)
+ if server_timing:
+ spec["server_timing"] = True
+ plan = ci.read(INPUTS / "formal-8s-plan.json")
+ spec.update(gpu_vendor="amd", job_id=plan["plan_id"], plan=plan,
+ gpu_uuids=[f"00000000-0000-0000-0000-{i:012d}" for i in range(4)],
+ lock_directory="/work/campaigns/h3-cross-hardware/control/gpu-locks", port=30317,
+ serving={"mode": "closed_loop", "concurrency": 1, "delivery_deadline_seconds": None})
+ spec["server"] = {"tp_size": 1, "ulysses_degree": 4, "encoder_parallel": "auto",
+ "performance_mode": "speed", "dit_cpu_offload": False, "attention_backend": "aiter"}
+ ci.need(30 <= allocation_minutes <= 110, "AMD serving requires 30–110 remaining allocation minutes")
+ spec["limits"].update(job_seconds=(allocation_minutes - 10) * 60, startup_seconds=900, request_seconds=900,
+ cleanup_seconds=60, command_seconds=30, telemetry_interval_seconds=1)
+ spec["authorization"]["approval_reference"] = (
+ "User authorized MI355X recovery, one warmup and exactly twenty measured attempts on 2026-09-09. "
+ "Retain source model-license approval. AMD initial C1 uses 20 measured requests and one separate warmup, "
+ "8 allocated GPUs and 4 participating GPUs, one allocation capped at 120 minutes, "
+ f"with {allocation_minutes} minutes remaining for serving and ten minutes reserved for cleanup. "
+ "Generation compatibility is unverified; retain failures and release the owned allocation.")
+ spec["model"]["path"] = str(Path("/work") / Path(model["model_path"]).relative_to(workspace))
+ for role in ("baseline", "candidate"):
+ spec[role] = {"source": "/work/" + source.name, "source_sha256": identity["source_sha256"],
+ "revision": identity["revision"], "python": probe["python"]}
+ spec = validate_gpu_job(spec)
+ destination.mkdir()
+ entry = destination / "entry-only.sh"
+ command = '-c \'exec "$@"\' h3-entry "$@"' if 'exec bash "$@"' in (runtime.get("entrypoint") or "") else '"$@"'
+ entry.write_text((INPUTS / "entry-only-amd.sh").read_text().replace("@ENTRY@", command)
+ .replace("runtime-inspected.json", readiness.name))
+ ci.write(destination / "gpu-spec.json", spec)
+ config = {"schema_version": 1, "task_id": "h3-cross-hardware", "site": ci.AMD_SITE,
+ "workspace": {"host": str(workspace), "container": "/work"},
+ "runtime": {"entry": str(entry), "entry_sha256": ci.digest(entry), "rootfs": str(rootfs),
+ "ready_marker": str(readiness), "python": probe["python"]},
+ "spec": {"path": str(destination / "gpu-spec.json"), "sha256": ci.digest(destination / "gpu-spec.json")},
+ "resources": {"gpus": 4, "allocated_gpus": 8, "cpus": 32, "memory_gb": 1024, "minutes": allocation_minutes},
+ "allocation_receipts": [], "mode": "serving-smoke", "concurrencies": [1]}
+ config = ci.validate_config(config)
+ ci.prepared_spec(config)
+ ci.write(destination / "site.json", config)
+ for name in ("entry-only.sh", "gpu-spec.json", "site.json"):
+ (output / name).write_bytes((destination / name).read_bytes())
+ return {"site_config": str(destination / "site.json"), "source": identity,
+ "runtime_inspection": runtime, "instrumentation": instrumentation,
+ "server_configuration_deviation": {"source_server": source_server, "executed_server": spec["server"],
+ "reason": "AMD uses TP1/Ulysses4 with AITER; NVIDIA source settings are retained for explicit comparison."},
+ "model_receipt": model, "generation_executed": False,
+ "status": "prepared", "compatibility": "Imports checked; allocated HIP identity and full video/audio warmup remain mandatory before measurement"}
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--source-run-id", required=True)
+ parser.add_argument("--output", required=True, type=Path)
+ parser.add_argument("--server-timing", action="store_true")
+ args = parser.parse_args()
+ args.output.mkdir(parents=True, exist_ok=True)
+ record = {"status": "preparing", "generation_executed": False,
+ "ci": {key: os.environ.get(key) for key in ("H3_RUN_ID", "H3_SOURCE_SHA")}}
+ try:
+ spec, provenance = source_spec(args.source_run_id, args.output)
+ record.update(provenance)
+ with ci.task_lock(WORKSPACE / "campaigns/h3-cross-hardware/.site-preparation.lock"):
+ record.update(stage(spec, args.output, server_timing=args.server_timing))
+ except Exception as error:
+ record.update(status="failed", error=str(error))
+ raise
+ finally:
+ ci.write(args.output / "site-preparation.json", record)
diff --git a/experimental/video-generation/stage_model_ci.py b/experimental/video-generation/stage_model_ci.py
new file mode 100644
index 0000000000..154720e553
--- /dev/null
+++ b/experimental/video-generation/stage_model_ci.py
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Stage the frozen model from an accepted H3 run, without allocating GPUs."""
+from __future__ import annotations
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+import hashlib
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+from urllib.parse import quote
+from urllib.request import urlopen
+
+import ci
+from evaluator.mvp_gpu_job import validate_gpu_job
+from export_ci import REPOSITORY, source_ids, verified_execution
+
+
+def matches(path: Path, entry: dict) -> bool:
+ return (path.is_file() and path.stat().st_size == entry["size_bytes"]
+ and ci.digest(path) == entry["sha256"])
+
+
+def fetch_weight(root: Path, revision: str, entry: dict) -> None:
+ target = root / entry["path"]
+ if matches(target, entry):
+ return
+ ci.need(not target.exists(), "Existing model file differs from the frozen manifest: " + entry["path"])
+ target.parent.mkdir(parents=True, exist_ok=True)
+ partial = target.with_name(target.name + ".partial")
+ url = "https://huggingface.co/MiniMaxAI/MiniMax-H3/resolve/" + revision + "/" + quote(entry["path"], safe="/")
+ digest, size = hashlib.sha256(), 0
+ # No GitHub or implicit Hugging Face credentials are sent to the model host.
+ with urlopen(url, timeout=60) as response, partial.open("wb") as stream:
+ while chunk := response.read(8 * 1024 * 1024):
+ size += len(chunk)
+ ci.need(size <= entry["size_bytes"], "Model download exceeds frozen size")
+ digest.update(chunk)
+ stream.write(chunk)
+ ci.need(size == entry["size_bytes"] and digest.hexdigest() == entry["sha256"],
+ "Downloaded model differs from frozen size or SHA256: " + entry["path"])
+ partial.replace(target)
+
+
+def stage_model(spec: dict, workspace: Path, candidates: list[Path]) -> dict:
+ spec = validate_gpu_job(spec)
+ ci.need(spec["plan"]["model_id"] == "MiniMaxAI/MiniMax-H3", "Only the approved H3 model is supported")
+ approval = spec["authorization"]
+ ci.need(approval["compute_approved"] and approval["model_license_reviewed"]
+ and approval["approval_reference"].strip(), "Source lacks recorded model approval")
+ model = spec["model"]
+ entries = model["files"]
+ observations = []
+ root = workspace / "models" / "MiniMax-H3" / model["revision"]
+ for candidate in [*candidates, root]:
+ present = candidate.is_dir()
+ valid = present and all(matches(candidate / entry["path"], entry) for entry in entries)
+ observations.append({"path": str(candidate), "present": present, "verified": valid})
+ if valid:
+ root = candidate
+ break
+ else:
+ root.mkdir(parents=True, exist_ok=True)
+ missing_bytes = sum(entry["size_bytes"] for entry in entries if not (root / entry["path"]).exists())
+ ci.need(shutil.disk_usage(root).free >= missing_bytes, "Insufficient persistent space for frozen model")
+ with ThreadPoolExecutor(max_workers=4) as pool:
+ list(pool.map(lambda entry: fetch_weight(root, model["revision"], entry), entries))
+ return {"model_path": str(root), "model_revision": model["revision"],
+ "manifest_sha256": hashlib.sha256(json.dumps(entries, sort_keys=True, separators=(",", ":")).encode()).hexdigest(),
+ "verified_files": len(entries), "total_bytes": sum(entry["size_bytes"] for entry in entries),
+ "reuse_candidates": observations, "verification": "complete frozen file sizes and SHA256"}
+
+
+def source_spec(run_id: str, output: Path) -> tuple[dict, dict]:
+ """Read the accepted artifact contract without re-downloading model weights."""
+ ci.need(source_ids(run_id) == [run_id], "One accepted source run is required")
+ source, artifact = verified_execution(run_id)
+ ci.write(output / "source-provenance.json", {"source_ci": source, "source_artifact": artifact})
+ original = output / "source"
+ subprocess.run(["gh", "run", "download", run_id, "--repo", REPOSITORY,
+ "--name", artifact["name"], "--dir", str(original)], check=True, timeout=180)
+ sums = dict(line.split(" ", 1)[::-1] for line in (original / "SHA256SUMS").read_text().splitlines())
+ path = "gpu/c1/spec.json"
+ ci.need(sums.get(path) == ci.digest(original / path), "Frozen source specification checksum differs")
+ return validate_gpu_job(ci.read(original / path)), {"source_ci": source, "source_artifact": artifact}
+
+
+def prepare(run_id: str, workspace: Path, output: Path) -> None:
+ ci.need(source_ids(run_id) == [run_id], "One accepted source run is required")
+ ci.need(workspace.is_absolute(), "Persistent workspace must be absolute")
+ output.mkdir(parents=True, exist_ok=True)
+ record = {"schema_version": 1, "bundle_type": "h3_model_preparation_no_gpu",
+ "gpu_allocation": False, "gpu_execution": False, "status": "preparing",
+ "workspace": {"host": str(workspace), "container": "/work"},
+ "ci": {key: os.environ.get(key) for key in ("GITHUB_RUN_ID", "GITHUB_RUN_ATTEMPT", "GITHUB_SHA")}}
+ ci.write(output / "model-preparation.json", record)
+ try:
+ spec, provenance = source_spec(run_id, output)
+ record.update(provenance)
+ workspace.mkdir(parents=True, exist_ok=True)
+ with ci.task_lock(workspace / ".model-preparation.lock"):
+ revision = spec["model"]["revision"]
+ candidates = [base / "models--MiniMaxAI--MiniMax-H3" / "snapshots" / revision
+ for base in (Path("/it-share/hf-hub-cache"), Path.home() / ".cache/huggingface/hub")]
+ record.update(stage_model(spec, workspace, candidates), status="complete")
+ receipt = workspace / "campaigns/h3-cross-hardware/model-ready.json"
+ receipt.parent.mkdir(parents=True, exist_ok=True)
+ ci.write(receipt, record)
+ except Exception as error:
+ record.update(status="failed", error=str(error))
+ raise
+ finally:
+ ci.write(output / "model-preparation.json", record)
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--source-run-id", required=True)
+ parser.add_argument("--workspace", required=True, type=Path)
+ parser.add_argument("--output", required=True, type=Path)
+ args = parser.parse_args()
+ prepare(args.source_run_id, args.workspace, args.output)
diff --git a/experimental/video-generation/tests/test_ci.py b/experimental/video-generation/tests/test_ci.py
index c8420d13b3..38453b451f 100644
--- a/experimental/video-generation/tests/test_ci.py
+++ b/experimental/video-generation/tests/test_ci.py
@@ -38,6 +38,73 @@ def save_receipt(root, receipt):
return path
+def test_h100_site_keeps_full_allocation_separate_from_participating_gpus(tmp_path, monkeypatch):
+ monkeypatch.setenv("RUNNER_NAME", "h3-test-runner")
+ cfg = config(tmp_path)
+ cfg.update(mode="serving-smoke", site={"cluster": "h100-dgxc", "partition": "hpc-gpu-1", "account": "customer", "gpu_model": "H100"})
+ cfg["resources"].update(gpus=4, allocated_gpus=8)
+ ci.validate_config(cfg)
+ commands = []
+ def run(argv, **kwargs):
+ commands.append(argv)
+ return SimpleNamespace(stdout="salloc: Granted job allocation 123", stderr="", returncode=0)
+ monkeypatch.setattr(ci.subprocess, "run", run)
+ monkeypatch.setattr(ci, "command", lambda argv: "tester")
+ receipt = ci.allocate(cfg, tmp_path)
+ assert "--partition=hpc-gpu-1" in commands[0] and "--account=customer" in commands[0]
+ assert "--exclusive" in commands[0] and "--gres=gpu:8" in commands[0]
+ assert receipt["site"] == cfg["site"]
+ _, record = allocation(tmp_path)
+ record.update(receipt["identity"])
+ ci.verify_identity(receipt, record, cfg["task_id"])
+ step = ci.step_argv(cfg, receipt, record, tmp_path, tmp_path)
+ assert "--gpus-per-task=4" in step
+ record["Account"] = "other"
+ with pytest.raises(ValueError, match="identity differs"):
+ ci.verify_identity(receipt, record, cfg["task_id"])
+
+
+def test_amd_granted_full_node_requires_explicit_gpu_evidence(tmp_path):
+ cfg = config(tmp_path)
+ cfg.update(mode="serving-smoke", site=dict(ci.AMD_SITE))
+ cfg["resources"].update(gpus=4, allocated_gpus=8)
+ ci.validate_config(cfg)
+ receipt, record = allocation(tmp_path)
+ record.update(Account=ci.AMD_SITE["account"], Partition="compute", AllocTRES="cpu=128,mem=512G,node=1,billing=128", TresPerNode="gres/gpu:8")
+ assert ci.allocated_gpu_count(record) == 8
+ assert ci.capacity(record, cfg["resources"]) is None
+ assert "--gres=gpu:8" in ci.step_argv(cfg, receipt, record, tmp_path, tmp_path)
+ record["OverSubscribe"] = "OK"
+ assert ci.allocated_gpu_count(record) is None
+ assert ci.capacity(record, cfg["resources"]) == "insufficient allocated GPU/CPU/memory capacity"
+ record.update(OverSubscribe="NO", AllocTRES="cpu=128,mem=512G,node=1,gres/gpu=4")
+ assert ci.allocated_gpu_count(record) == 4
+
+
+@pytest.mark.parametrize("change", [
+ {"site": {"cluster": "h100-dgxc", "partition": "hpc-gpu-1", "account": "customer", "gpu_model": "H200"}},
+ {"site": {"cluster": "unknown", "partition": "main", "account": "customer", "gpu_model": "H200"}},
+ {"resources": {"gpus": 4, "allocated_gpus": 2, "cpus": 32, "memory_gb": 512, "minutes": 90}},
+])
+def test_invalid_hardware_or_allocation_budget_is_rejected(tmp_path, change):
+ cfg = config(tmp_path)
+ cfg.update(change)
+ with pytest.raises(ValueError):
+ ci.validate_config(cfg)
+
+
+def test_explicit_concurrency_selection_requires_serving_mode(tmp_path):
+ cfg = config(tmp_path)
+ cfg["concurrencies"] = [4]
+ with pytest.raises(ValueError, match="requires serving-smoke"):
+ ci.validate_config(cfg)
+ cfg["mode"] = "serving-smoke"
+ ci.validate_config(cfg)
+ cfg["concurrencies"] = [1, 1]
+ with pytest.raises(ValueError, match="unique concurrency"):
+ ci.validate_config(cfg)
+
+
@pytest.mark.parametrize("mode", ["smoke", "serving-smoke"])
def test_allocation_submits_from_receipted_work_directory(tmp_path, monkeypatch, mode):
run_dir = tmp_path / "results"
@@ -209,10 +276,13 @@ def test_staging_reuses_identical_source_and_refuses_drift(tmp_path):
(source / "evaluator").mkdir(parents=True)
(source / "ci.py").write_text("entry")
(source / "evaluator" / "__init__.py").write_text("")
+ (source / "runtime-patches").mkdir()
+ (source / "runtime-patches" / "timing.patch").write_text("CPU patch fixture")
dest = tmp_path / "package"
original = ci.stage_package(source, dest)
assert ci.stage_package(source, dest) == original
- (dest / "ci.py").write_text("tampered")
+ assert (dest / "runtime-patches" / "timing.patch").read_text() == "CPU patch fixture"
+ (dest / "runtime-patches" / "timing.patch").write_text("tampered")
with pytest.raises(ValueError, match="source differs"):
ci.stage_package(source, dest)
@@ -338,3 +408,25 @@ def test_entry_resolves_device_minors_instead_of_nvml_indices(tmp_path, assignme
else:
assert result.returncode != 0
assert 'lack NVIDIA UUIDs' in result.stderr
+
+
+@pytest.mark.parametrize("decision", [{"action": "allocate"}, {"action": "reuse", "receipt": {"identity": {"JobId": "999"}}}])
+def test_required_lease_never_replaces_or_borrows_another_allocation(tmp_path, monkeypatch, decision):
+ cfg = config(tmp_path)
+ entry = Path(cfg["runtime"]["entry"])
+ entry.write_text("entry")
+ Path(cfg["runtime"]["ready_marker"]).write_text("synthetic readiness")
+ cfg["runtime"]["entry_sha256"] = ci.digest(entry)
+ monkeypatch.setenv("GITHUB_RUN_ID", "456")
+ monkeypatch.setenv("GITHUB_RUN_ATTEMPT", "1")
+ monkeypatch.setenv("H3_SOURCE_SHA", "a" * 40)
+ monkeypatch.setattr(ci, "prepared_spec", lambda cfg: {})
+ monkeypatch.setattr(ci, "command", lambda argv, **kw: "a" * 40 if "rev-parse" in argv else "")
+ monkeypatch.setattr(ci, "stage_package", lambda *args: {})
+ monkeypatch.setattr(ci, "recover", lambda *args: decision)
+ monkeypatch.setattr(ci, "allocate", lambda *args: pytest.fail("must not replace the required allocation"))
+ monkeypatch.setattr(ci, "run_step", lambda *args: pytest.fail("must not enter a different allocation"))
+ monkeypatch.setattr(ci, "stop_allocation", lambda *args: pytest.fail("outer owner releases its allocation"))
+ output = tmp_path / "output"
+ assert ci.launch(cfg, output, required_allocation="123") == 2
+ assert "must reuse its original allocation" in ci.read(output / "ci.json")["error"]
diff --git a/experimental/video-generation/tests/test_compare_serving_ci.py b/experimental/video-generation/tests/test_compare_serving_ci.py
new file mode 100644
index 0000000000..2905713652
--- /dev/null
+++ b/experimental/video-generation/tests/test_compare_serving_ci.py
@@ -0,0 +1,96 @@
+import json
+from pathlib import Path
+import shutil
+
+import pytest
+
+import ci
+import compare_serving_ci as fidelity
+
+
+SHA = "a" * 40
+
+
+def source(root, run_id="123"):
+ run = root / "gpu/c1/baseline/run.json"
+ run.parent.mkdir(parents=True)
+ ci.write(run, {"configuration": {"serving": {"concurrency": 1}}})
+ ci.write(root / "ci.json", {"run_id": run_id, "run_attempt": "1", "source_sha": SHA})
+ ci.write(root / "manifest.json", {"run_id": run_id, "run_attempt": "1", "git_commit": SHA,
+ "mode": "serving-smoke"})
+ ci.write(root / "serving-smoke.json", {"bundle_type": "h3_serving_smoke_matrix", "schema_version": "1.0.0",
+ "cells": [{"concurrency": 1, "run": {"path": "gpu/c1/baseline/run.json", "sha256": ci.digest(run)}}]})
+ seal(root)
+ return {"databaseId": int(run_id), "runAttempt": 1, "headSha": SHA}
+
+
+def seal(root):
+ (root / "SHA256SUMS").write_text("".join(f"{sha} {path}\n" for path, sha in ci.inventory(root).items()))
+
+
+def test_selects_sealed_c1_and_rejects_tampered_bytes_or_ci_identity(tmp_path):
+ metadata = source(tmp_path)
+ selected = fidelity.selected_run(tmp_path, metadata)
+ assert selected == tmp_path / "gpu/c1/baseline"
+ with pytest.raises(ValueError, match="CI identity"):
+ fidelity.selected_run(tmp_path, {**metadata, "headSha": "b" * 40})
+ (selected / "run.json").write_text("{}")
+ with pytest.raises(ValueError, match="checksum"):
+ fidelity.selected_run(tmp_path, metadata)
+
+
+@pytest.mark.parametrize("mutation", ["non_c1", "duplicate_c1", "escape"])
+def test_rejects_wrong_or_ambiguous_cell_even_when_outer_seal_is_updated(tmp_path, mutation):
+ metadata = source(tmp_path)
+ path = tmp_path / "serving-smoke.json"
+ matrix = ci.read(path)
+ if mutation == "non_c1":
+ run = tmp_path / "gpu/c1/baseline/run.json"
+ ci.write(run, {"configuration": {"serving": {"concurrency": 2}}})
+ matrix["cells"][0]["run"]["sha256"] = ci.digest(run)
+ elif mutation == "duplicate_c1":
+ matrix["cells"].append(matrix["cells"][0])
+ else:
+ matrix["cells"][0]["run"]["path"] = "../outside/run.json"
+ ci.write(path, matrix)
+ seal(tmp_path)
+ with pytest.raises(ValueError):
+ fidelity.selected_run(tmp_path, metadata)
+
+
+def test_publication_retains_uncalibrated_failure_and_original_seals(tmp_path, monkeypatch):
+ originals = tmp_path / "originals"
+ metadata = {run: source(originals / run, run) for run in ("123", "456")}
+ monkeypatch.setenv("GITHUB_SHA", SHA)
+ monkeypatch.setenv("GITHUB_RUN_ID", "789")
+ monkeypatch.setenv("GITHUB_RUN_ATTEMPT", "1")
+ monkeypatch.setenv("GITHUB_REPOSITORY", "SemiAnalysisAI/InferenceX")
+ monkeypatch.setattr(ci, "command", lambda *args: SHA)
+ monkeypatch.setattr(fidelity.export_ci, "verified_execution", lambda run: (metadata[run], {"name": "original"}))
+ def download(argv, **kwargs):
+ assert argv[:3] == ["gh", "run", "download"]
+ shutil.copytree(originals / argv[3], Path(argv[argv.index("--dir") + 1]))
+ monkeypatch.setattr(fidelity.subprocess, "run", download)
+ def compare(left, right, *, policy):
+ assert left.parts[-4:] == ("source-123", "gpu", "c1", "baseline")
+ assert right.parts[-4:] == ("source-456", "gpu", "c1", "baseline")
+ assert policy["calibration_status"] == "uncalibrated"
+ return {"overall_status": "fail", "release_qualified": False, "policy": policy,
+ "summary": {"matched_valid_pairs": 20}}
+ monkeypatch.setattr(fidelity, "compare_runs", compare)
+ def report(result, path):
+ assert path.suffix == ".html" and result["overall_status"] == "fail"
+ path.parent.mkdir()
+ path.write_text("fixture report")
+ monkeypatch.setattr(fidelity, "write_report", report)
+ output = tmp_path / "published"
+ fidelity.publish(["123", "456"], output)
+ assert (output / "sources/123/original-SHA256SUMS").read_bytes() == (originals / "123/SHA256SUMS").read_bytes()
+ assert ci.read(output / "sources/123/manifest.json")["git_commit"] == SHA
+ assert not (output / "source-123").exists()
+ assert not (output / "sources/123/gpu").exists()
+ result = json.loads((output / "comparison.json").read_text())
+ assert result["source_artifacts"][1]["ci"]["databaseId"] == 456
+ receipt = ci.read(output / "reprocessing.json")
+ assert receipt["status"] == "complete" and receipt["threshold_outcome"] == "fail"
+ assert receipt["release_qualified"] is False and receipt["generation_executed"] is False
diff --git a/experimental/video-generation/tests/test_export_ci.py b/experimental/video-generation/tests/test_export_ci.py
index 54922b6808..27ef23700c 100644
--- a/experimental/video-generation/tests/test_export_ci.py
+++ b/experimental/video-generation/tests/test_export_ci.py
@@ -46,6 +46,13 @@ def test_source_ids_preserve_order_without_shell_interpretation():
assert export_ci.source_ids("34293342829,34291306687") == ["34293342829", "34291306687"]
+@pytest.mark.parametrize("hardware", ["H100", "B200", "MI355X"])
+def test_serving_sources_accept_verified_cross_hardware_generation(github, hardware):
+ github[1]["jobs"][0]["name"] = f"h3-video / p1.5 | H3 video {hardware} smoke"
+ source, artifact = export_ci.verified_execution("123")
+ assert source["headSha"] == SHA and artifact["id"] == 789
+
+
def test_inventory_reuse_requires_the_hardware_job_and_artifact(github):
with pytest.raises(ValueError):
export_ci.verified_execution("123", inventory=True)
diff --git a/experimental/video-generation/tests/test_inspect_amd_node.py b/experimental/video-generation/tests/test_inspect_amd_node.py
new file mode 100644
index 0000000000..c7a4821435
--- /dev/null
+++ b/experimental/video-generation/tests/test_inspect_amd_node.py
@@ -0,0 +1,96 @@
+"""Fake scheduler checks for ownership and cleanup, not AMD hardware evidence."""
+from types import SimpleNamespace
+
+import pytest
+
+import ci
+import inspect_amd_node as amd
+from test_ci import allocation
+
+
+def test_wrong_node_stops_before_device_queries(tmp_path, monkeypatch):
+ ci.write(tmp_path / "context.json", {"allocation": {"identity": {"JobId": "123"}}, "node": "amd-node"})
+ monkeypatch.setenv("SLURM_JOB_ID", "123")
+ monkeypatch.setenv("SLURMD_NODENAME", "another-node")
+ monkeypatch.setattr(amd, "observation", lambda argv: pytest.fail("foreign node must not be queried"))
+ with pytest.raises(ValueError, match="Wrong AMD inventory allocation"):
+ amd.inspect_node(tmp_path)
+ assert not (tmp_path / "binding.json").exists()
+
+
+@pytest.mark.parametrize("value", ["0-7", "0,1,2,3,4,5,6,7"])
+def test_full_amd_step_assignment(value):
+ assert amd.step_gpu_indices(value) == set(range(8))
+
+
+@pytest.mark.parametrize("value", ["", "0-99", "7-0", "0;id"])
+def test_invalid_amd_step_assignment(value):
+ with pytest.raises(ValueError, match="AMD step GPU assignment"):
+ amd.step_gpu_indices(value)
+
+
+@pytest.mark.parametrize("reused", [False, True])
+@pytest.mark.parametrize("prepare_runtime", [False, True])
+def test_failed_inventory_drains_only_owned_step_and_preserves_borrowed_allocation(tmp_path, monkeypatch, reused, prepare_runtime):
+ monkeypatch.setenv("H3_RUN_ID", "456")
+ monkeypatch.setenv("H3_RUN_ATTEMPT", "1")
+ account = "cameronamd@semianalysis.com"
+ monkeypatch.setattr(amd.pwd, "getpwuid", lambda uid: SimpleNamespace(pw_name=account))
+ receipt, record = allocation(tmp_path)
+ site = {"cluster": "mi355x-amds", "partition": "compute", "account": account, "gpu_model": "MI355X"}
+ receipt.update(task_id="h3-cross-hardware", site=site)
+ receipt["identity"].update(Account=account, Partition="compute")
+ record.update(receipt["identity"])
+ monkeypatch.setattr(ci, "recover", lambda *args: {"action": "reuse" if reused else "allocate", "receipt": receipt, "active_steps": ""})
+ def allocate(config, root):
+ assert not reused
+ ci.write(root / "allocation.json", receipt)
+ return receipt
+ monkeypatch.setattr(ci, "allocate", allocate)
+ monkeypatch.setattr(ci, "job_record", lambda job: record)
+ import prepare_amd_runtime
+ monkeypatch.setattr(prepare_amd_runtime, "prepare_source", lambda workspace: workspace)
+ def fail_step(argv, log, timeout):
+ assert "--gres=gpu:8" in argv
+ assert "--time=" + ("55" if prepare_runtime else "10") in argv
+ assert timeout == (3360 if prepare_runtime else 660)
+ return 1
+ monkeypatch.setattr(ci, "run_step", fail_step)
+ cleanup = []
+ monkeypatch.setattr(ci, "drain_step", lambda owned, task, root: cleanup.append(("step", owned["identity"]["JobId"])) or {"status": "ended"})
+ monkeypatch.setattr(ci, "stop_allocation", lambda owned, task: cleanup.append(("allocation", owned["identity"]["JobId"])) or {"status": "released"})
+ output = tmp_path / "output"
+ assert amd.inspect(tmp_path / "work", output, prepare_runtime=prepare_runtime) == 2
+ assert cleanup == [("step", "123")] + ([] if reused else [("allocation", "123")])
+ assert ci.read(output / "inventory-status.json")["generation_executed"] is False
+ assert (output / "SHA256SUMS").is_file()
+
+
+def test_failed_cpu_entry_does_not_approve_partial_rootfs(tmp_path, monkeypatch):
+ import subprocess
+ import prepare_amd_runtime as runtime
+ workspace = tmp_path / "work"
+ (workspace / "campaigns/h3-cross-hardware").mkdir(parents=True)
+ rootfs = tmp_path / "enroot-data" / runtime.CONTAINER
+ (rootfs / "etc").mkdir(parents=True)
+ (rootfs / "etc/rc").write_text('exec "$@"\n')
+ origin = rootfs.with_suffix(".image.json")
+ ci.write(origin, {"image": str(runtime.IMAGE), "status": "creating"})
+ previous = workspace / "results/h3-cross-hardware/github-34344130223-1"
+ previous.mkdir(parents=True)
+ ci.write(previous / "inventory-status.json", {"allocation_cleanup": {"status": "released"}})
+ (previous / "srun.log").write_text("Ignoring xattrs in filesystem\ncreated 464452 files\ncreated 11757 symlinks\n")
+ monkeypatch.setenv("H3_RUN_ID", "987")
+ monkeypatch.setenv("H3_RUN_ATTEMPT", "1")
+ monkeypatch.setattr(runtime, "prepare_source", lambda path: path)
+ def fail_entry(argv, **kwargs):
+ if argv[0].endswith("enroot"):
+ assert not any("/dev/kfd" in arg or "/dev/dri" in arg for arg in argv)
+ raise subprocess.CalledProcessError(1, argv)
+ monkeypatch.setattr(runtime.subprocess, "run", fail_entry)
+ output = tmp_path / "output"
+ with pytest.raises(subprocess.CalledProcessError):
+ runtime.recover_rootfs(workspace, output)
+ assert ci.read(origin)["status"] == "creating"
+ assert ci.read(output / "rootfs-recovery.json")["status"] == "failed"
+ assert not (workspace / "campaigns/h3-cross-hardware/rootfs-recovered.json").exists()
diff --git a/experimental/video-generation/tests/test_mvp_amd_gpu.py b/experimental/video-generation/tests/test_mvp_amd_gpu.py
new file mode 100644
index 0000000000..32adbffc6e
--- /dev/null
+++ b/experimental/video-generation/tests/test_mvp_amd_gpu.py
@@ -0,0 +1,117 @@
+"""CPU checks for observed AMD SMI shapes; these are not GPU measurements."""
+import copy
+from types import SimpleNamespace
+
+import pytest
+
+from evaluator import mvp_amd_gpu as amd
+
+
+GPU = "75ff75a3-0000-1000-80e3-fd74aab3f72c"
+OTHER = "68ff75a3-0000-1000-8089-743843afe909"
+
+
+def observations():
+ # Reduced shape from CI 34341458378 on MI355X / AMD SMI 26.2.0.
+ return {
+ "list": [{"gpu": 0, "bdf": "0000:05:00.0", "uuid": GPU, "partition_id": 0}],
+ "static": [{"gpu": 0, "bus": {"bdf": "0000:05:00.0"},
+ "asic": {"market_name": "AMD Instinct MI355X"}, "driver": {"version": "6.16.6"},
+ "limit": {"socket_power": {"value": 1400, "unit": "W"}, "max_power": {"value": 1400, "unit": "W"}}}],
+ "metric": {"gpu_data": [{"gpu": 0,
+ "mem_usage": {"total_vram": {"value": 294896, "unit": "MB"}, "used_vram": {"value": 283, "unit": "MB"}},
+ "usage": {"gfx_activity": {"value": 0, "unit": "%"}},
+ "power": {"socket_power": {"value": 239, "unit": "W"}},
+ "temperature": {"hotspot": {"value": 36, "unit": "C"}}}]},
+ "process": [{"gpu": 0, "process_list": [{"process_info": {"pid": 15910, "memory_usage": {"vram_mem": {"value": 0, "unit": "B"}}}}]}],
+ }
+
+
+def test_snapshot_preserves_zero_memory_process_and_measured_power(monkeypatch):
+ data = observations()
+ monkeypatch.setattr(amd, "smi", lambda option, timeout: copy.deepcopy(data[option]))
+ probe = amd.AmdGpuProbe([GPU], 2)
+ result = probe.snapshot()
+ assert result["compute_apps"] == [{"gpu_uuid": GPU, "pid": 15910, "memory_used_mib": 0.0}]
+ assert result["gpus"][0]["power_watts"] == 239
+ assert result["gpus"][0]["memory_used_mib"] == 283
+ assert probe.power_configuration()["gpus"][0]["configured_limit_w"] == 1400
+ data["metric"]["gpu_data"][0]["power"]["socket_power"] = "N/A"
+ assert probe.snapshot()["gpus"][0]["power_watts"] is None
+
+
+def test_hip_order_joins_physical_identity_instead_of_smi_ordinal(monkeypatch):
+ rows = observations()["list"] + [{"gpu": 1, "bdf": "0000:15:00.0", "uuid": OTHER, "partition_id": 0}]
+ monkeypatch.setattr(amd, "smi", lambda *args: rows)
+ def count(pointer):
+ pointer._obj.value = 2
+ return 0
+ def bdf(buffer, size, ordinal):
+ buffer.value = [b"0000:15:00.0", b"0000:05:00.0"][ordinal]
+ return 0
+ monkeypatch.setattr(amd.ctypes, "CDLL", lambda _: SimpleNamespace(hipGetDeviceCount=count, hipDeviceGetPCIBusId=bdf))
+ assert amd.hip_devices() == [OTHER, GPU]
+
+
+@pytest.mark.parametrize("mutation", [
+ lambda rows: rows.append(dict(rows[0])),
+ lambda rows: rows[0].update(partition_id=1),
+ lambda rows: rows[0].update(bdf="../../other"),
+])
+def test_ambiguous_or_partitioned_device_inventory_is_rejected(mutation):
+ rows = observations()["list"]
+ mutation(rows)
+ with pytest.raises(RuntimeError):
+ amd.inventory(rows)
+
+
+@pytest.mark.parametrize("value", [{"value": 239, "unit": "mW"}, {"value": True, "unit": "W"}, {"value": float("nan"), "unit": "W"}])
+def test_invalid_power_unit_or_value_is_not_a_measurement(value):
+ assert amd.number(value, "W", optional=True) is None
+ with pytest.raises(RuntimeError):
+ amd.number(value, "W")
+
+
+@pytest.mark.parametrize("change", ["pid", "start_ticks", "executable", "uid", "memory", "unverified"])
+def test_monitor_exception_rejects_changed_or_active_context(monkeypatch, change):
+ identity = {"pid": 15910, "start_ticks": 50, "executable": "/opt/gpuagent/gpuagent", "uid": 0}
+ receipt = {"status": "verified", "process": identity}
+ app = {"pid": 15910, "memory_used_mib": 0}
+ observed = dict(identity)
+ if change == "memory":
+ app["memory_used_mib"] = 1
+ elif change == "unverified":
+ receipt["status"] = "unverified"
+ elif change == "executable":
+ observed[change] = "/tmp/other"
+ else:
+ observed[change] += 1
+ monkeypatch.setattr(amd, "monitor_process", lambda pid: observed)
+ assert not amd.is_system_monitor(app, receipt)
+
+
+def test_verified_monitor_is_retained_separately_from_workload_contexts(monkeypatch):
+ data = observations()
+ monkeypatch.setattr(amd, "smi", lambda option, timeout: copy.deepcopy(data[option]))
+ probe = amd.AmdGpuProbe([GPU], 2)
+ identity = {"pid": 15910, "start_ticks": 50, "executable": "/opt/gpuagent/gpuagent", "uid": 0}
+ probe.monitor = {"status": "verified", "service": "gpuagent.service", "process": identity}
+ monkeypatch.setattr(amd, "monitor_process", lambda pid: dict(identity))
+ result = probe.snapshot()
+ assert result["compute_apps"] == []
+ excluded = result["excluded_system_monitor_contexts"]
+ assert len(excluded) == 1 and excluded[0]["pid"] == 15910 and excluded[0]["gpu_uuid"] == GPU
+ assert excluded[0]["identity"]["service"] == "gpuagent.service"
+
+
+def test_service_pid_mismatch_never_approves_monitor(monkeypatch):
+ monkeypatch.setattr(amd, "_command", lambda *a, **k: "MainPID=42\nExecMainPID=43\nExecStart={ path=/opt/gpuagent/gpuagent ; }\nActiveState=active\nSubState=running\nControlGroup=/system.slice/gpuagent.service\nType=simple\n")
+ monkeypatch.setattr(amd, "monitor_process", lambda pid: pytest.fail("mismatched service PID must be rejected"))
+ assert amd.observe_system_monitor(1)["status"] == "unverified"
+
+
+def test_unreadable_monitor_identity_remains_a_foreign_process(monkeypatch):
+ def unreadable(pid):
+ raise PermissionError("process identity unavailable")
+ monkeypatch.setattr(amd, "monitor_process", unreadable)
+ assert not amd.is_system_monitor({"pid": 42, "memory_used_mib": 0}, {"status": "verified", "pid": 42})
diff --git a/experimental/video-generation/tests/test_mvp_gpu_job.py b/experimental/video-generation/tests/test_mvp_gpu_job.py
index a891115902..aa6779f393 100644
--- a/experimental/video-generation/tests/test_mvp_gpu_job.py
+++ b/experimental/video-generation/tests/test_mvp_gpu_job.py
@@ -57,6 +57,25 @@ def snapshot(*, used=50, pid=None):
"compute_apps": [{"gpu_uuid": GPU, "pid": pid, "memory_used_mib": used}] if pid else []}
+def test_amd_spec_binds_vendor_and_documented_attention_layout(spec, tmp_path, monkeypatch):
+ spec["gpu_vendor"] = "amd"
+ spec["gpu_uuids"] = [GPU.removeprefix("GPU-")]
+ spec["server"]["attention_backend"] = "aiter"
+ frozen = gpu.validate_gpu_job(spec)
+ assert gpu._server_argv(frozen, "baseline")[-2:] == ["--attention-backend", "aiter"]
+ monkeypatch.setenv("ROCR_VISIBLE_DEVICES", "3")
+ env = gpu._runtime_env("/pinned/source", frozen["gpu_uuids"], "nonce", tmp_path, "amd")
+ assert env["ROCR_VISIBLE_DEVICES"] == "3"
+ assert env["HIP_VISIBLE_DEVICES"] == env["CUDA_VISIBLE_DEVICES"] == "0"
+ assert env["SGLANG_USE_AITER"] == "1"
+ monkeypatch.delenv("ROCR_VISIBLE_DEVICES")
+ with pytest.raises(ValueError, match="bound ROCR"):
+ gpu._runtime_env("/pinned/source", frozen["gpu_uuids"], "nonce", tmp_path, "amd")
+ del spec["gpu_vendor"]
+ with pytest.raises(ValueError, match="declared vendor"):
+ gpu.validate_gpu_job(spec)
+
+
def controlled_receipt(spec):
roles = {}
for label in ("baseline", "candidate"):
diff --git a/experimental/video-generation/tests/test_mvp_power.py b/experimental/video-generation/tests/test_mvp_power.py
index 2642807570..ff2a44328f 100644
--- a/experimental/video-generation/tests/test_mvp_power.py
+++ b/experimental/video-generation/tests/test_mvp_power.py
@@ -55,6 +55,22 @@ def test_ramp_clipping_separates_phases_and_excludes_client_decode():
assert result["sample_series"][8]["aggregate_watts"] == 135
+def test_amd_energy_retains_sensor_source_and_rejects_mixed_sources():
+ role, run, samples = data()
+ for sample in samples:
+ sample["power_query"] = {"field": "amd-smi power.socket_power", "start_utc": sample["at"],
+ "start_monotonic_seconds": sample["monotonic_seconds"],
+ "end_monotonic_seconds": sample["monotonic_seconds"]}
+ result = analyze(role, run, samples)
+ assert result["valid"] is True
+ assert result["semantics"]["sensor"].startswith("amd-smi power.socket_power;")
+ assert result["phases"]["measurement"]["aggregate"]["energy_j"] == 450
+ samples[9]["power_query"]["field"] = "power.draw"
+ result = analyze(role, run, samples)
+ assert result["phases"]["measurement"]["aggregate"] is None
+ assert "power_sensor_source_inconsistent" in result["invalid_reasons"]
+
+
@pytest.mark.parametrize("defect", ["missing_power", "nan", "negative", "missing_gpu", "duplicate_gpu", "foreign", "identity", "partition", "nonmonotonic", "clock_jump"])
def test_invalid_telemetry_withholds_measurement_metrics(defect):
role, run, samples = data()
diff --git a/experimental/video-generation/tests/test_mvp_runtime_timing.py b/experimental/video-generation/tests/test_mvp_runtime_timing.py
new file mode 100644
index 0000000000..e2a6e42692
--- /dev/null
+++ b/experimental/video-generation/tests/test_mvp_runtime_timing.py
@@ -0,0 +1,148 @@
+"""CPU event/transport checks; these do not establish H3 timing overhead."""
+
+from copy import deepcopy
+import hashlib
+import importlib.util
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from evaluator import mvp_runtime_timing as timing
+from evaluator.mvp_serving import summarize
+from test_mvp_serving import sample_run
+from test_mvp_runner import execute, fixture_server, mocked_media, plan # noqa: F401
+
+
+@pytest.fixture
+def ledger(tmp_path, monkeypatch):
+ path = tmp_path / "server-timings.jsonl"
+ path.touch()
+ monkeypatch.setenv("VGBENCH_SERVER_TIMING_PATH", str(path))
+ monkeypatch.setenv("VGBENCH_SERVER_TIMING_INSTANCE", "cpu-fixture-instance")
+ monkeypatch.setattr(timing, "_clock_id", lambda: "linux:cpu-fixture:time-namespace:CLOCK_MONOTONIC")
+ return path
+
+
+def complete_request(request_id):
+ timing.emit(request_id, "http_received")
+ timing.emit(request_id, "http_accepted")
+ with timing.forward([SimpleNamespace(request_id=request_id, num_outputs_per_prompt=1)], replica_id=0, leader=True):
+ pass
+ timing.emit(request_id, "media_ready")
+
+
+def test_real_event_writer_correlates_request_and_derives_stage_windows(ledger, monkeypatch):
+ stamps = iter([1, 2, 7, 17, 19])
+ monkeypatch.setattr(timing.time, "monotonic_ns", lambda: next(stamps) * 1_000_000_000)
+ complete_request("video-1")
+ result = timing.collect("video-1")
+ assert result["status"] == "complete"
+ assert [result[name] for name in timing.DURATIONS] == [1, 5, 10, 2, 18]
+ assert result["observed_batch_size"] == 1 and result["replica_id"] == 0
+ assert timing.collect("different-video") is None
+ assert len(timing.read_events(ledger)) == 5
+
+
+def test_interleaved_requests_and_missing_stage_do_not_create_queue_time(ledger):
+ timing.emit("a", "http_received")
+ complete_request("b")
+ timing.emit("a", "http_accepted")
+ partial = timing.collect("a")
+ assert partial["status"] == "partial"
+ assert partial["queue_delay_seconds"] is None
+ assert partial["server_ready_latency_seconds"] is None
+ assert partial["observed_batch_size"] is None
+ assert timing.collect("b")["status"] == "complete"
+
+
+@pytest.mark.parametrize("defect", ["duplicate", "clock", "reverse", "instance", "batch"])
+def test_corrupt_or_incompatible_events_fail_closed(ledger, defect):
+ complete_request("a")
+ rows = timing.read_events(ledger)
+ if defect == "duplicate":
+ rows.append(rows[0])
+ elif defect == "clock":
+ rows[2]["clock_id"] = "another-host"
+ elif defect == "reverse":
+ rows[2]["monotonic_ns"] = rows[0]["monotonic_ns"] - 1
+ elif defect == "instance":
+ rows[2]["instance_id"] = "another-server"
+ else:
+ rows[2]["observed_batch_size"] = 2
+ with pytest.raises(ValueError):
+ timing.derive(rows, "a", "cpu-fixture-instance")
+
+
+def test_worker_followers_do_not_duplicate_events_and_failure_has_no_ready_time(ledger):
+ req = SimpleNamespace(request_id="a", num_outputs_per_prompt=1)
+ with timing.forward([req], replica_id=0, leader=False):
+ pass
+ assert timing.read_events(ledger) == []
+ with pytest.raises(RuntimeError, match="generation failed"):
+ with timing.forward([req], replica_id=0, leader=True):
+ raise RuntimeError("generation failed")
+ partial = timing.collect("a")
+ assert partial["execution_seconds"] >= 0
+ assert partial["server_ready_latency_seconds"] is None
+
+
+def test_http_client_retains_timings_in_original_record_and_journal(plan, mocked_media, fixture_server, ledger, tmp_path):
+ ordinal = 0
+ def record_fixture_timing():
+ nonlocal ordinal
+ ordinal += 1
+ complete_request(f"fixture-{ordinal}")
+ endpoint, _ = fixture_server(before_submit=record_fixture_timing)
+ output = tmp_path / "client"
+ result = execute(plan, output, endpoint)
+ assert all(record["server_timings"]["request_id"] == record["job_id"] for record in result["records"])
+ records = [row["record"] for row in map(json.loads, (output / "events.jsonl").read_text().splitlines())
+ if row["event"] == "attempt_finished"]
+ assert [row["server_timings"] for row in records] == [row["server_timings"] for row in result["records"]]
+
+
+def test_serving_summary_requires_full_valid_population_for_percentiles():
+ run = sample_run()
+ historical = summarize(run)
+ assert historical["queue_delay_seconds"] is None
+ assert "server_execution_seconds" not in historical
+ for record in run["records"][:10]:
+ record["server_timings"] = {"status": "complete", "queue_delay_seconds": 2,
+ "execution_seconds": 5, "server_ready_latency_seconds": 8,
+ "prequeue_seconds": .5, "postprocess_seconds": .5, "observed_batch_size": 1, "replica_id": 0}
+ observed = summarize(run)
+ assert observed["queue_delay_seconds"]["p90"] == 2
+ assert observed["server_execution_seconds"]["sample_count"] == 10
+ assert observed["observed_batch_sizes"] == [1] * 10
+ run["records"][0]["server_timings"] = None
+ incomplete = summarize(run)
+ assert incomplete["queue_delay_seconds"]["missing_count"] == 1
+ assert incomplete["queue_delay_seconds"]["p90"] is None
+ assert incomplete["observed_batch_sizes"] is None
+
+
+def test_offline_verification_binds_ledger_hash_instance_and_request(ledger, tmp_path):
+ complete_request("a")
+ run = {"records": [{"job_id": "a", "status": "succeeded", "server_timings": timing.collect("a")}]}
+ role = {"process_identity": {"launch_nonce": "cpu-fixture-instance"}, "server_timing_evidence": {
+ **timing.identity(), "path": ledger.name, "instance_id": "cpu-fixture-instance",
+ "sha256": hashlib.sha256(ledger.read_bytes()).hexdigest()}}
+ timing.verify_evidence(tmp_path, role, run, required=True)
+ changed = deepcopy(run)
+ changed["records"][0]["server_timings"]["queue_delay_seconds"] = 99
+ with pytest.raises(ValueError, match="raw events"):
+ timing.verify_evidence(tmp_path, role, changed, required=True)
+ ledger.write_text(ledger.read_text() + "{}\n")
+ with pytest.raises(ValueError, match="hash mismatch"):
+ timing.verify_evidence(tmp_path, role, run, required=True)
+
+
+def test_staged_runtime_helper_can_resolve_its_pinned_patch(tmp_path):
+ import ci
+ destination = tmp_path / "staged"
+ ci.stage_package(timing.PATCH.parents[1], destination)
+ spec = importlib.util.spec_from_file_location("staged_timing", destination / "evaluator/mvp_runtime_timing.py")
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ assert module.identity() == timing.identity()
diff --git a/experimental/video-generation/tests/test_mvp_serving_smoke.py b/experimental/video-generation/tests/test_mvp_serving_smoke.py
index 2cefef538f..24929ba138 100644
--- a/experimental/video-generation/tests/test_mvp_serving_smoke.py
+++ b/experimental/video-generation/tests/test_mvp_serving_smoke.py
@@ -38,9 +38,10 @@ def test_single_runtime_smoke_cannot_be_accepted_as_paired_evidence(spec, tmp_pa
@pytest.mark.parametrize("fail_second", [False, True])
-def test_matrix_preserves_twelve_requests_without_doubling_roles(spec, tmp_path, monkeypatch, fail_second):
+@pytest.mark.parametrize("requests", [4, 20])
+def test_matrix_preserves_scheduled_requests_without_doubling_roles(spec, tmp_path, monkeypatch, fail_second, requests):
spec["plan"]["cases"] = spec["plan"]["cases"][:1]
- spec["plan"]["repetitions"] = 4
+ spec["plan"]["repetitions"] = requests
spec["serving"] = {"concurrency": 1}
submitted = []
def execute(current, directory, *, serving_smoke):
@@ -51,15 +52,16 @@ def execute(current, directory, *, serving_smoke):
return saved_single(current, directory)
monkeypatch.setattr(gpu, "run_gpu_job", execute)
result = smoke.run_matrix(spec, tmp_path)
- assert result["completion"]["scheduled"] == 12
+ assert result["completion"]["scheduled"] == requests * 3
+ assert result["requests_per_configuration"] == requests
assert [s["serving"]["concurrency"] for s in submitted] == ([1, 2] if fail_second else [1, 2, 4])
assert all(s["plan"] == spec["plan"] for s in submitted)
- assert result["completion"]["valid"] == (4 if fail_second else 12)
- assert result["completion"]["not_started"] == (8 if fail_second else 0)
+ assert result["completion"]["valid"] == requests * (1 if fail_second else 3)
+ assert result["completion"]["not_started"] == (requests * 2 if fail_second else 0)
assert result["status"] == ("failed" if fail_second else "complete")
assert not result["ci_accepted"]
report = (tmp_path / "report/index.html").read_text()
- assert report.count("
|---|