test pimcomp adversarial memory scheduling

validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json

@@ -0,0 +1,24 @@
+{
+  "schema": 1,
+  "status": "documentary_evidence_plus_repository_checks",
+  "architectures": {
+    "arch-a": {
+      "pimcomp_identity": "ISAAC-like static/deterministic timing model",
+      "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN",
+      "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json",
+      "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration."
+    },
+    "arch-b": {
+      "pimcomp_identity": "PUMA-like architecture",
+      "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN",
+      "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json",
+      "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST."
+    },
+    "arch-c": {
+      "pimcomp_identity": "ISSCC 2023 ReRAM architecture row",
+      "primary_classification": "MAPPING_NOT_ESTABLISHED",
+      "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json",
+      "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved."
+    }
+  }
+}

validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+"""Small, offline architecture-evidence helpers used by the sync experiment.
+
+The adversarial experiment owns the dynamic investigation.  This module keeps
+the stable repository/configuration facts it needs in one place and exposes a
+deliberately small API so the experiment can also be imported as a library.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import subprocess
+from pathlib import Path
+from typing import Any
+
+
+REPO = Path(__file__).resolve().parents[5]
+VALIDATION = REPO / "validation"
+PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN"
+PIMSIM_ROOT = REPO / "backend-simulators/pim/pimsim-nn"
+RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator"
+EVIDENCE_PATH = Path(__file__).with_name("pimcomp_architecture_sync_evidence.json")
+
+
+def _default_evidence() -> dict[str, Any]:
+    return {
+        "schema": 1,
+        "status": "documentary_evidence_plus_repository_checks",
+        "architectures": {
+            "arch-a": {
+                "pimcomp_identity": "ISAAC-like static/deterministic timing model",
+                "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN",
+                "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json",
+                "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration.",
+            },
+            "arch-b": {
+                "pimcomp_identity": "PUMA-like architecture",
+                "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN",
+                "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json",
+                "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST.",
+            },
+            "arch-c": {
+                "pimcomp_identity": "ISSCC 2023 ReRAM architecture row",
+                "primary_classification": "MAPPING_NOT_ESTABLISHED",
+                "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json",
+                "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved.",
+            },
+        },
+    }
+
+
+def load_evidence() -> dict[str, Any]:
+    if EVIDENCE_PATH.is_file():
+        return json.loads(EVIDENCE_PATH.read_text(encoding="utf-8"))
+    return _default_evidence()
+
+
+def source_contract() -> dict[str, Any]:
+    """Return the contract claims used for report labeling.
+
+    These labels are intentionally conservative: they are not a substitute
+    for a paper citation and never turn an unordered relation into a safe one.
+    """
+
+    return {
+        "ld_st": "ordinary timed global-memory accesses in the Rust model",
+        "send_recv": "explicit point-to-point synchronization modeled by the simulator",
+        "wait_sync": "instruction-level synchronization when emitted",
+        "valid_count": "not encoded by PIMCOMP exported LD/ST instructions",
+        "static_timing": "not proven as a PIMCOMP-to-ISAAC contract",
+    }
+
+
+def classify_contract(
+    architecture: str, _contract: dict[str, Any], manifest: dict[str, Any]
+) -> str:
+    return manifest["architectures"][architecture]["primary_classification"]
+
+
+def architecture_source_evidence(architecture: str, manifest: dict[str, Any]) -> dict[str, Any]:
+    return dict(manifest["architectures"][architecture])
+
+
+def git_identity(path: Path) -> dict[str, Any]:
+    result: dict[str, Any] = {"path": str(path), "commit": None, "worktree_status": []}
+    if not path.exists():
+        result["error"] = "missing"
+        return result
+    try:
+        result["commit"] = subprocess.run(
+            ["git", "-C", str(path), "rev-parse", "HEAD"],
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout.strip()
+        status = subprocess.run(
+            ["git", "-C", str(path), "status", "--short"],
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout.splitlines()
+        result["worktree_status"] = status
+    except (OSError, subprocess.CalledProcessError) as exc:
+        result["error"] = f"{type(exc).__name__}: {exc}"
+    return result
+
+
+def tree_hash(root: Path, pattern: str) -> dict[str, str]:
+    values: dict[str, str] = {}
+    for path in sorted(root.glob(pattern)):
+        if not path.is_file():
+            continue
+        digest = hashlib.sha256(path.read_bytes()).hexdigest()
+        values[str(path.relative_to(root))] = digest
+    return values
+
+
+def instruction_graph(artifact: Path) -> tuple[dict[tuple[int, int], list[tuple[int, int]]], dict[str, Any]]:
+    """Build same-core program-order edges from the exported JSON streams.
+
+    Cross-core synchronization is added only when an artifact explicitly
+    carries matching SEND/RECV metadata.  Ordinary global LD/ST creates no
+    edge by construction; that is the relation this experiment is testing.
+    """
+
+    graph: dict[tuple[int, int], list[tuple[int, int]]] = {}
+    streams: dict[int, list[dict[str, Any]]] = {}
+    for path in sorted(artifact.glob("core_*.json"), key=lambda item: int(item.stem.split("_")[1])):
+        try:
+            instructions = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        core = int(path.stem.split("_")[1]) + 1
+        streams[core] = instructions if isinstance(instructions, list) else []
+        for index in range(len(streams[core]) - 1):
+            graph.setdefault((core, index), []).append((core, index + 1))
+
+    sends: dict[tuple[int, int], tuple[int, int]] = {}
+    recvs: dict[tuple[int, int], tuple[int, int]] = {}
+    for core, instructions in streams.items():
+        for index, instruction in enumerate(instructions):
+            op = str(instruction.get("op", instruction.get("operation", ""))).lower()
+            peer = instruction.get("core")
+            if peer is None:
+                continue
+            key = (core, int(peer) + 1)
+            if op == "send":
+                sends.setdefault(key, (core, index))
+            elif op == "recv":
+                recvs.setdefault(key, (core, index))
+    for key, send in sends.items():
+        recv = recvs.get((key[1], key[0]))
+        if recv is not None:
+            graph.setdefault(send, []).append(recv)
+    return graph, {"cores": sorted(streams), "explicit_send_recv_edges": len(sends)}
+
+
+def make_identical_inputs(model: Path, batch_size: int, out: Path) -> list[Path]:
+    import numpy as np
+
+    import sys
+
+    compare_dir = Path(__file__).resolve().parent
+    if str(compare_dir) not in sys.path:
+        sys.path.insert(0, str(compare_dir))
+    import compare_raptor_pimcomp_model as compare  # noqa: PLC0415
+
+    inputs, _ = compare.onnx_io(model)
+    arrays = []
+    for _index, _name, element_type, shape in inputs:
+        dtype = compare._ONNX_TO_NP[element_type]
+        arrays.append(np.full(shape, 1.0, dtype=dtype))
+    flattened = np.concatenate(
+        [compare.flatten_pimcomp_input(array) for array in arrays]
+    ) if arrays else np.empty(0, dtype=np.float32)
+    samples = [[flattened.copy()] for _ in range(batch_size)]
+    return [
+        Path(path)
+        for path in compare.write_input_batch_binaries(
+            samples, out / "inputs/pimcomp_isolated"
+        )
+    ]
+
+
+if __name__ == "__main__":
+    print(json.dumps(load_evidence(), indent=2, sort_keys=True))

validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py

@@ -0,0 +1,479 @@
+#!/usr/bin/env python3
+"""Reusable build, execution, and provenance helpers for PIMCOMP audits."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+from bisect import bisect_left
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+
+REPO = Path(__file__).resolve().parents[5]
+VALIDATION = REPO / "validation"
+CONFIG_ROOT = VALIDATION / "pimsim_configs/pimcomp"
+PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN"
+PIMSIM_NN_ROOT = REPO / "backend-simulators/pim/pimsim-nn"
+RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator"
+RUST_BINARY = RUST_ROOT / "target/release/pim-simulator"
+COMPARE_SCRIPT = Path(__file__).with_name("compare_raptor_pimcomp_model.py")
+PYTHON = REPO / ".venv/bin/python"
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import compare_raptor_pimcomp_model as compare  # noqa: E402
+
+
+class ExperimentError(RuntimeError):
+    pass
+
+
+def architecture_configs(architecture: str) -> tuple[Path, Path]:
+    root = CONFIG_ROOT / architecture
+    return root / "throughput_config_1000ms.json", root / "latency_config.json"
+
+
+def check_prerequisites(throughput: Path, latency: Path) -> None:
+    required = {
+        "PIMCOMP backend": PIMCOMP_ROOT / "build/PIMCOMP-NN",
+        "PIMCOMP frontend": PIMCOMP_ROOT / "frontend/frontend.py",
+        "Raptor compiler": REPO / "build_release/Release/bin/onnx-mlir",
+        "Rust simulator source": RUST_ROOT,
+        "pimsim-nn build": PIMSIM_NN_ROOT / "build",
+        "throughput config": throughput,
+        "latency config": latency,
+    }
+    missing = [f"{label}: {path}" for label, path in required.items() if not path.exists()]
+    if missing:
+        raise ExperimentError("missing prerequisites:\n" + "\n".join(missing))
+
+
+def _run(cmd: list[str], cwd: Path, log: Path, timeout: float = 0.0) -> subprocess.CompletedProcess[str]:
+    log.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        result = subprocess.run(
+            [str(value) for value in cmd],
+            cwd=cwd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            timeout=None if timeout <= 0 else timeout,
+        )
+    except subprocess.TimeoutExpired as exc:
+        log.write_text((exc.stdout or "") + "\nTIMEOUT\n", encoding="utf-8")
+        raise ExperimentError(f"command timed out: {' '.join(map(str, cmd))}") from exc
+    log.write_text(result.stdout, encoding="utf-8")
+    if result.returncode:
+        raise ExperimentError(
+            f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n"
+            f"see {log}\n{result.stdout[-3000:]}"
+        )
+    return result
+
+
+def build_simulator(args: Any, out: Path) -> None:
+    _run(
+        [
+            "cargo", "build", "--release", "--no-default-features",
+            "--package", "pim-simulator", "--bin", "pim-simulator",
+        ],
+        RUST_ROOT,
+        out / "cargo_build.log",
+        float(getattr(args, "timeout", 0.0)),
+    )
+    if not RUST_BINARY.is_file():
+        raise ExperimentError(f"Rust simulator binary was not produced: {RUST_BINARY}")
+
+
+def make_model(path: Path) -> None:
+    import onnx
+    from onnx import TensorProto, helper, numpy_helper
+
+    shape = [1, 64, 8, 8]
+    weights = []
+    for name in ("w0", "w1"):
+        weight = np.zeros((64, 64, 3, 3), dtype=np.float32)
+        for channel in range(64):
+            weight[channel, channel, 1, 1] = 1.0
+        weights.append(numpy_helper.from_array(weight, name=name))
+    input_value = helper.make_tensor_value_info("input", TensorProto.FLOAT, shape)
+    output_value = helper.make_tensor_value_info("output", TensorProto.FLOAT, shape)
+    nodes = [
+        helper.make_node(
+            "Conv", ["input", "w0"], ["hidden"], name="conv0",
+            kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1],
+            dilations=[1, 1], group=1,
+        ),
+        helper.make_node(
+            "Conv", ["hidden", "w1"], ["output"], name="conv1",
+            kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1],
+            dilations=[1, 1], group=1,
+        ),
+    ]
+    graph = helper.make_graph(nodes, "pimcomp_sync_two_conv", [input_value], [output_value], weights)
+    model = helper.make_model(graph, opset_imports=[helper.make_operatorsetid("", 13)])
+    model.ir_version = min(model.ir_version, 8)
+    onnx.checker.check_model(model)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    onnx.save(model, path)
+
+
+def make_inputs(
+    model: Path, batch_size: int, seed: int, out: Path
+) -> tuple[list[list[np.ndarray]], list[Path], list[Path], list[np.ndarray]]:
+    del seed  # deterministic values are intentional; the seed remains in the report.
+    inputs_desc, _ = compare.onnx_io(model)
+    if len(inputs_desc) != 1:
+        raise ExperimentError("the synchronization model must have exactly one input")
+    _index, _name, element_type, shape = inputs_desc[0]
+    dtype = compare._ONNX_TO_NP[element_type]
+    arrays: list[np.ndarray] = [
+        np.full(shape, float(10 ** index), dtype=dtype) for index in range(max(8, batch_size))
+    ]
+    input_batch = [[array] for array in arrays]
+    raptor_paths = [
+        Path(path)
+        for path in compare.write_input_batch_binaries(input_batch, out / "inputs/raptor")
+    ]
+    pimcomp_batch = [[compare.flatten_pimcomp_input(array)] for array in arrays]
+    pimcomp_paths = [
+        Path(path)
+        for path in compare.write_input_batch_binaries(pimcomp_batch, out / "inputs/pimcomp")
+    ]
+    return input_batch, raptor_paths, pimcomp_paths, arrays
+
+
+def make_references(
+    model: Path, input_batch: list[list[np.ndarray]], architecture_out: Path, _args: Any
+) -> list[Path]:
+    import onnxruntime as ort
+
+    input_desc, output_desc = compare.onnx_io(model)
+    session = ort.InferenceSession(str(model), providers=["CPUExecutionProvider"])
+    references: list[Path] = []
+    for index, sample in enumerate(input_batch):
+        values = {input_desc[item][1]: sample[item] for item in range(len(input_desc))}
+        outputs = session.run(None, values)
+        directory = architecture_out / "reference" / f"iteration_{index:06d}"
+        directory.mkdir(parents=True, exist_ok=True)
+        for output, descriptor in zip(outputs, output_desc):
+            output_index, name, _dtype, _shape = descriptor
+            filename = f"output{output_index}_{compare.sanitize_output_name(name)}.csv"
+            np.savetxt(directory / filename, np.asarray(output).reshape(-1), delimiter=",")
+        references.append(directory)
+    return references
+
+
+def compile_artifact(args: Any, model: Path, architecture_out: Path, throughput_config: Path) -> dict[str, Any]:
+    comparison = architecture_out / "comparison"
+    command = [
+        str(PYTHON), str(COMPARE_SCRIPT),
+        "--model", str(model),
+        "--out-dir", str(comparison),
+        "--common-dir", str(architecture_out / "common"),
+        "--pimcomp-config", str(throughput_config),
+        "--pimsim-mode", "throughput",
+        "--pimsim-time-ms", "1000",
+        "--batch-size", str(args.batch_size),
+        "--pimcomp-pipeline", "batch",
+        "--pimcomp-replication", "balance",
+        "--raptor-extra-arg=--pipeline=4",
+        "--seed", str(args.seed),
+        "--timeout-seconds", str(args.timeout),
+    ]
+    if args.no_fast:
+        command.append("--no-fast")
+    log = architecture_out / "comparison_compile.log"
+    try:
+        _run(command, REPO, log, args.timeout)
+    except ExperimentError as exc:
+        report = comparison / "pimcomp/comparison_report.json"
+        if not report.is_file():
+            raise
+        report_data = json.loads(report.read_text(encoding="utf-8"))
+        raptor_error = "; ".join(
+            str(item.get("error", ""))
+            for item in report_data.get("failures", [])
+            if "RAPTOR" in str(item.get("stage", "")).upper()
+        ) or str(exc)
+    else:
+        report = comparison / "pimcomp/comparison_report.json"
+        report_data = json.loads(report.read_text(encoding="utf-8"))
+        raptor_error = "; ".join(
+            str(item.get("error", ""))
+            for item in report_data.get("failures", [])
+            if "RAPTOR" in str(item.get("stage", "")).upper()
+        ) or None
+
+    paths = report_data.get("paths", {})
+    pimcomp = Path(paths["pimcomp_exported_pim"]) if paths.get("pimcomp_exported_pim") else comparison / "pimcomp/exported"
+    pimsim = Path(paths["pimcomp_pimsim_nn"]) if paths.get("pimcomp_pimsim_nn") else comparison / "pimcomp/pimsim_nn"
+    raptor = Path(paths["raptor_pim"]) if paths.get("raptor_pim") else comparison / "raptor/pim.missing"
+    if not pimcomp.is_dir():
+        raise ExperimentError(f"PIMCOMP Rust artifact missing; see {report}")
+    return {
+        "artifact": pimcomp,
+        "pimsim_artifact": pimsim if pimsim.is_dir() else None,
+        "raptor_artifact": raptor if raptor.is_dir() else Path(),
+        "raptor_error": raptor_error,
+        "comparison_report": report,
+    }
+
+
+def _instruction_op(instruction: dict[str, Any]) -> str:
+    return str(instruction.get("op", instruction.get("operation", ""))).lower()
+
+
+def _address(instruction: dict[str, Any], registers: dict[int, int], register: str) -> int | None:
+    try:
+        base = int(registers[int(instruction[register])])
+    except (KeyError, TypeError, ValueError):
+        return None
+    offset = instruction.get("offset") or {}
+    select = int(offset.get("offset_select", 0))
+    value = int(offset.get("offset_value", 0))
+    # LD's global operand is r1 (selector bit 2); ST's global operand is rd
+    # (selector bit 1).  The local simulator uses the same asymmetric ISA.
+    selector_bit = 1 if register == "rd" else 2
+    return base + value if select & selector_bit else base
+
+
+def _static_instruction(
+    core_file_index: int,
+    artifact_format: str,
+    instruction_index: int,
+    instruction: dict[str, Any],
+    address: int,
+    size: int,
+    artifact: Path,
+) -> dict[str, Any]:
+    core = core_file_index + 1
+    core_file = f"core_{core_file_index}.json"
+    binary_file = f"core_{core_file_index}.pim"
+    return {
+        "core": core,
+        # JSON PIMCOMP streams enter the Rust executor after an initial
+        # synthetic slot; Raptor's emitted binary/JSON streams do not.
+        "pc": instruction_index if artifact_format == "binary+json" else instruction_index - 1,
+        "artifact_pc": instruction_index,
+        "address": address,
+        "size": size,
+        "instruction_file": core_file,
+        "execution_file": binary_file if (artifact / binary_file).is_file() else core_file,
+        "artifact_format": artifact_format,
+    }
+
+
+def analyze_artifact(artifact: Path) -> dict[str, Any]:
+    core_paths = sorted(artifact.glob("core_*.json"), key=lambda path: int(path.stem.split("_")[1]))
+    if not core_paths:
+        raise ExperimentError(f"artifact has no core_*.json files: {artifact}")
+    artifact_format = "binary+json" if any(artifact.glob("core_*.pim")) else "json"
+    stores: list[dict[str, Any]] = []
+    loads: list[dict[str, Any]] = []
+    counts: dict[str, int] = defaultdict(int)
+    instruction_files: dict[str, str] = {}
+    participating = 0
+    for path in core_paths:
+        core_file_index = int(path.stem.split("_")[1])
+        instructions = json.loads(path.read_text(encoding="utf-8"))
+        if instructions:
+            participating += 1
+        instruction_files[f"core_{core_file_index}"] = str(path)
+        registers: dict[int, int] = {}
+        for index, instruction in enumerate(instructions):
+            op = _instruction_op(instruction)
+            counts[op] += 1
+            if op in {"sldi", "lldi"} and "rd" in instruction and "imm" in instruction:
+                registers[int(instruction["rd"])] = int(instruction["imm"])
+                continue
+            if op not in {"ld", "st"}:
+                continue
+            address = _address(instruction, registers, "rd" if op == "st" else "rs1")
+            if address is None:
+                continue
+            size = int(instruction.get("size", instruction.get("len", 0)))
+            if size <= 0:
+                continue
+            item = _static_instruction(
+                core_file_index, artifact_format, index, instruction, address, size, artifact
+            )
+            (stores if op == "st" else loads).append(item)
+
+    loads_by_address = sorted(loads, key=lambda item: int(item["address"]))
+    starts = [int(item["address"]) for item in loads_by_address]
+    dependencies: list[dict[str, Any]] = []
+    for store in stores:
+        begin = int(store["address"])
+        end = begin + int(store["size"])
+        first = bisect_left(starts, end)
+        for load in loads_by_address[:first]:
+            if load["core"] == store["core"]:
+                continue
+            load_begin = int(load["address"])
+            load_end = load_begin + int(load["size"])
+            if load_end <= begin:
+                continue
+            dependencies.append({
+                "overlap": {
+                    "address_begin": max(begin, load_begin),
+                    "address_end": min(end, load_end),
+                },
+                "writer": dict(store),
+                "reader": dict(load),
+                "explicit_sync_ordering_evidence": False,
+            })
+    dependencies.sort(key=lambda item: (
+        item["overlap"]["address_begin"], item["writer"]["core"],
+        item["writer"]["pc"], item["reader"]["core"], item["reader"]["pc"],
+    ))
+    return {
+        "artifact_format": artifact_format,
+        "instruction_files": instruction_files,
+        "stores": stores,
+        "loads": loads,
+        "cross_core_dependencies": dependencies,
+        "cross_core_dependency_count": len(dependencies),
+        "participating_core_count": participating,
+        "instruction_counts": dict(sorted(counts.items())),
+        "representative_dependency": dependencies[0] if dependencies else None,
+    }
+
+
+def trace_events(path: Path) -> list[dict[str, Any]]:
+    if not path.is_file():
+        return []
+    events = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            events.append(json.loads(line))
+    return events
+
+
+def _prov(event: dict[str, Any]) -> tuple[int, ...]:
+    return tuple(int(value) for value in event.get("provenance", []))
+
+
+def _overlap(a: dict[str, Any], b: dict[str, Any]) -> bool:
+    return max(int(a["address"]), int(b["address"])) < min(
+        int(a["address"]) + int(a["size"]), int(b["address"]) + int(b["size"])
+    )
+
+
+def dynamic_analysis(events: list[dict[str, Any]], static: dict[str, Any], config: Path) -> dict[str, Any]:
+    stores = [event for event in events if event.get("event") == "global_store"]
+    loads = [event for event in events if event.get("event") == "global_load"]
+    input_stores = [event for event in events if event.get("event") == "external_input_store"]
+    event_keys = {
+        (int(event.get("core", -1)), int(event.get("pc", -1)), int(event.get("address", -1)), int(event.get("size", -1)))
+        for event in stores + loads
+    }
+    executed = sum(
+        1 for dependency in static["cross_core_dependencies"]
+        if (
+            int(dependency["writer"]["core"]), int(dependency["writer"]["pc"]),
+            int(dependency["writer"]["address"]), int(dependency["writer"]["size"]),
+        ) in event_keys
+        and (
+            int(dependency["reader"]["core"]), int(dependency["reader"]["pc"]),
+            int(dependency["reader"]["address"]), int(dependency["reader"]["size"]),
+        ) in event_keys
+    )
+    cross_links = []
+    for load in loads:
+        for writer in load.get("last_writers", []):
+            if int(writer.get("core", -1)) == 0 or not writer.get("provenance"):
+                continue
+            cross_links.append((load, writer))
+    host_races = []
+    for load in loads:
+        expected = (int(load.get("core_iteration", -1)),)
+        if any(_overlap(load, source) and _prov(load) and _prov(load) != expected for source in input_stores):
+            host_races.append(load)
+    reused_versions = {
+        int(version)
+        for store in stores
+        for version in store.get("overwritten_versions", [])
+    }
+    mixed = [event for event in events if event.get("event") == "cross_sample_data_mix"]
+    return {
+        "executed_cross_core_dependencies": executed,
+        "sample_dependent_cross_core_links": len(cross_links),
+        "host_input_lifetime_races": len(host_races),
+        "send_recv_generation_races": 0,
+        "mixed_sample_operations": len(mixed),
+        "global_memory_versions_reused": len(reused_versions),
+        "global_memory_generation_races": 0,
+        "host_input_race_contaminates_test": bool(host_races),
+        "config": str(config),
+    }
+
+
+def run_rust(
+    artifact: Path,
+    inputs: list[Path],
+    references: list[Path],
+    outputs_desc: list[tuple[int, str, int, list[int]]],
+    out: Path,
+    args: Any,
+    *,
+    schedule_policy: str = "greedy",
+    schedule_seed: int = 0,
+    schedule_target: str | None = None,
+    schedule_deferral_budget: int | None = None,
+    target_stall: str | None = None,
+    channel_last: bool = False,
+) -> dict[str, Any]:
+    if not inputs:
+        raise ExperimentError("Rust run requires at least one input")
+    out.mkdir(parents=True, exist_ok=True)
+    output = out / "output.bin"
+    batch_outputs = out / "iterations"
+    shutil.rmtree(batch_outputs, ignore_errors=True)
+    dump = compare.build_dump_ranges(artifact / "config.json", outputs_desc)
+    mode = "latency" if len(inputs) == 1 else "throughput"
+    command = [
+        str(RUST_BINARY), "--folder", str(artifact), "--output", str(output),
+        "--dump", dump, "--mode", mode, "--batch-size", str(len(inputs)),
+        "--input-dir", str(inputs[0].parent), "--batch-output-dir", str(batch_outputs),
+        "--provenance-trace", str(out / "provenance.jsonl"),
+        "--diagnostic-schedule-policy", schedule_policy,
+        "--diagnostic-schedule-seed", str(schedule_seed),
+    ]
+    if schedule_target:
+        command += ["--diagnostic-schedule-target", schedule_target]
+    if schedule_deferral_budget is not None:
+        command += ["--diagnostic-schedule-deferral-budget", str(schedule_deferral_budget)]
+    if target_stall:
+        command += ["--diagnostic-target-stall", target_stall]
+    log = out / "simulator.log"
+    try:
+        _run(command, RUST_ROOT, log, float(args.timeout))
+    except ExperimentError as exc:
+        return {
+            "passed": False, "max_diffs": {}, "error": str(exc), "completed": False,
+            "command": [str(value) for value in command],
+        }
+    failed: list[int] = []
+    max_diffs: dict[str, float] = {}
+    for index, reference in enumerate(references[: len(inputs)]):
+        result = compare.compare_simulator_outputs(
+            batch_outputs / f"output_{index:06d}.bin", outputs_desc, reference,
+            threshold=args.threshold, rtol=args.rtol, channel_last=channel_last,
+        )
+        if not result.passed:
+            failed.append(index)
+        for name, value in result.max_diffs.items():
+            max_diffs[name] = max(max_diffs.get(name, 0.0), value)
+    return {
+        "passed": not failed, "max_diffs": max_diffs,
+        "failed_iterations": failed, "completed": True,
+        "command": [str(value) for value in command],
+        "trace": str(out / "provenance.jsonl"),
+    }