diff --git a/.gitignore b/.gitignore index cf068d0a0..aad9ac2ee 100755 --- a/.gitignore +++ b/.gitignore @@ -71,3 +71,5 @@ tasks/flydsl2flydsl/*/.rocprofv3/ tasks/flydsl2flydsl/*/performance_report.json # SIKL generation drafts, source snapshots and validation evidence. sikl_task_builder_runs/ +# Tasks generated from a SIKL v5 checkout; they track SIKL commits and are rebuilt, not committed. +/tasks/SIKL-v5/ diff --git a/README.md b/README.md index c659603bb..79b6bbb89 100755 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ The platform provides: - **Resumable experiments**: Resume a run without repeating tasks whose framework completion and source evidence remain valid. - **Held-out evaluation**: Test optimized kernels on unseen shapes and measure the generalization gap. - **Task validation and visualization**: Validate task quality with a dedicated agent and compare local run reports in a dashboard. -- **SIKL task authoring**: Generate, repair, and validate standard Arena tasks from a dataset folder with [`sikl_task_builder`](docs/how-to/sikl-task-builder.md). +- **SIKL task authoring**: Generate and validate standard Arena tasks from a SIKL checkout or a legacy dataset folder with [`sikl_task_builder`](docs/how-to/sikl-task-builder.md). AgentKernelArena supplies an environment and objective reward signals; it does not currently include an RL trainer, replay buffer, or policy-update loop. Its per-task workspaces provide reproducibility and concurrent-run separation, not a security sandbox: agent processes run permissively inside a privileged container and can access mounted repository and authentication state. diff --git a/agents/sikl_task_builder/__init__.py b/agents/sikl_task_builder/__init__.py index 2e0f6844a..642983f8f 100644 --- a/agents/sikl_task_builder/__init__.py +++ b/agents/sikl_task_builder/__init__.py @@ -1,3 +1,3 @@ """SIKL task generation campaign (runs before Arena task discovery).""" -BUILDER_VERSION = 2 +BUILDER_VERSION = 3 diff --git a/agents/sikl_task_builder/__main__.py b/agents/sikl_task_builder/__main__.py index 0bcc4b483..9279ff8f5 100644 --- a/agents/sikl_task_builder/__main__.py +++ b/agents/sikl_task_builder/__main__.py @@ -5,11 +5,13 @@ import argparse import json import logging +import sys from pathlib import Path -from .bundle import inspect_bundle -from .config import load_config -from .orchestrator import REPO, resolve_paths, run +from .bundle import ImportProblem, inspect_bundle +from .checkout import check_mountable, inspect_checkout +from .config import CheckoutConfig, load_config +from .orchestrator import REPO, resolve_paths, run, run_checkout def main(argv=None): @@ -19,9 +21,19 @@ def main(argv=None): parser.add_argument("--input-dir") parser.add_argument("--run-id") args = parser.parse_args(argv) - config = load_config(args.config) - if args.input_dir: - config.input_dir = args.input_dir + if args.action == "resume" and not args.run_id: + parser.error("resume requires --run-id") + if args.action == "run" and args.run_id: + parser.error("use resume with --run-id") + # The Docker runner captures mount-info's stdout, so its errors go where the user sees them. + errors = sys.stderr if args.action == "mount-info" else sys.stdout + try: + config = load_config(args.config, args.input_dir) + if isinstance(config, CheckoutConfig): + return checkout_action(args, config) + except ImportProblem as exc: + print(json.dumps({"ok": False, "code": exc.code, "message": str(exc)}), file=errors) + return 1 if args.action == "inspect": tasks = inspect_bundle(Path(config.input_dir), config.selections) result = {"ok": True, "tasks": [t.summary() for t in tasks], @@ -30,24 +42,47 @@ def main(argv=None): source, output, artifacts = resolve_paths(config) if not source.is_dir(): raise ValueError(f"Missing input directory: {source}") - values = [str(source), str(args.config.resolve()), str(artifacts.relative_to(REPO)), - str(output.relative_to(REPO)), " ".join(sorted({"codex", config.validator.get("backend", "codex")}))] - if any("\n" in v or ":" in v for v in values): - raise ValueError("Mount paths cannot contain newline or colon characters") - artifacts.mkdir(parents=True, exist_ok=True) - output.mkdir(parents=True, exist_ok=True) - print("\n".join(values)) + print_mounts(args, config, source, output, artifacts, + " ".join(sorted({"codex", config.validator.get("backend", "codex")}))) return 0 else: - if args.action == "resume" and not args.run_id: - parser.error("resume requires --run-id") - if args.action == "run" and args.run_id: - parser.error("use resume with --run-id") logging.basicConfig(level=logging.INFO) result = run(config, args.run_id) print(json.dumps(result, indent=2)) return 0 if result["ok"] else 1 +def checkout_action(args, config: CheckoutConfig) -> int: + from src.preprocessing import _resolve_gfx_arch + source, output, artifacts = resolve_paths(config) + if args.action == "inspect": + commit, tasks = inspect_checkout(source, config.tasks) + arch = _resolve_gfx_arch(config.target_gpu_model) + result = {"ok": True, "input": "sikl-v5", "commit": commit, + "tasks": [t.summary(arch, config.target_gpu_model) for t in tasks], + "task_count": len(tasks), "case_count": sum(len(t.cases) for t in tasks)} + elif args.action == "mount-info": + inspect_checkout(source, config.tasks) + check_mountable(source) + # A placeholder, not an empty line: `$(...)` would drop a trailing empty line. + print_mounts(args, config, source, output, artifacts, "none") + return 0 + else: + logging.basicConfig(level=logging.INFO) + result = run_checkout(config, args.run_id) + print(json.dumps(result, indent=2)) + return 0 if result["ok"] else 1 + + +def print_mounts(args, config, source: Path, output: Path, artifacts: Path, agents: str) -> None: + values = [str(source), str(args.config.resolve()), str(artifacts.relative_to(REPO)), + str(output.relative_to(REPO)), agents] + if any("\n" in v or ":" in v for v in values): + raise ValueError("Mount paths cannot contain newline or colon characters") + artifacts.mkdir(parents=True, exist_ok=True) + output.mkdir(parents=True, exist_ok=True) + print("\n".join(values)) + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/agents/sikl_task_builder/checkout.py b/agents/sikl_task_builder/checkout.py new file mode 100644 index 000000000..3c6733263 --- /dev/null +++ b/agents/sikl_task_builder/checkout.py @@ -0,0 +1,162 @@ +"""Read committed SIKL v5 schemas from a git checkout without importing or executing them.""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path + +from .bundle import ImportProblem, parse_json + +SCHEMA_PATH = re.compile(r"^sikl/ops/(?P[a-z0-9_]+)/impls/(?Pflydsl|triton)/" + r"(?P[a-z0-9_]+)/(?P[0-9a-f]{16})\.json$") +SOURCE_PATH = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*\.py$") +IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +# SIKL's code-lint check-added-large-files limit (default maxkb=500). +MAX_SCHEMA_BYTES = 500 * 1024 + + +def is_checkout(path: Path) -> bool: + return (Path(path) / "sikl" / "ops").is_dir() + + +def git(root: Path, *args: str) -> bytes: + # Neither inherited GIT_* variables nor global/system configuration or hooks may steer these reads. + env = {key: value for key, value in os.environ.items() if not key.startswith("GIT_")} + env.update(GIT_CONFIG_NOSYSTEM="1", GIT_CONFIG_SYSTEM=os.devnull, GIT_CONFIG_GLOBAL=os.devnull, + GIT_TERMINAL_PROMPT="0") + command = ["git", "-c", "core.hooksPath=/dev/null", "-c", f"safe.directory={root}", "-C", str(root), *args] + result = subprocess.run(command, env=env, capture_output=True) + if result.returncode: + raise ImportProblem("not_a_sikl_checkout", + f"git {' '.join(args[:2])} in {root}: {result.stderr.decode(errors='replace').strip()}") + return result.stdout + + +def _validate_schema(path: str, match: re.Match, data) -> None: + def invalid(reason): + return ImportProblem("invalid_schema", f"{path}: {reason}") + if not isinstance(data, dict): + raise invalid("not a JSON object") + if type(data.get("schema_version")) is not int or data["schema_version"] != 5: + raise ImportProblem("unsupported_schema_version", f"{path}: schema_version {data.get('schema_version')!r}") + for key, part in (("name", "name"), ("op_type", "op")): + if data.get(key) != match[part]: + raise invalid(f"{key} {data.get(key)!r} != {match[part]!r}") + kernel = data.get("kernel") + if not isinstance(kernel, dict) or kernel.get("language") != match["language"]: + raise invalid(f"kernel.language is not {match['language']!r}") + sources = kernel.get("sources") + if not isinstance(sources, list) or not sources: + raise invalid("kernel.sources is empty") + paths = [] + for source in sources: + if not isinstance(source, dict) or set(source) != {"path", "content"}: + raise invalid("a kernel source is not {path, content}") + name, content = source["path"], source["content"] + if not isinstance(name, str) or not SOURCE_PATH.match(name) or name == "__init__.py": + raise invalid(f"source path {name!r} is not one .py segment") + if not isinstance(content, str) or not content: + raise invalid(f"source {name} is empty") + paths.append(name) + if paths != sorted(set(paths)): + raise invalid("kernel.sources paths are not sorted and unique") + entry = kernel.get("entry_point") + file, _, symbol = entry.partition("::") if isinstance(entry, str) else ("", "", "") + if file not in paths or not IDENTIFIER.match(symbol): + raise invalid(f"entry_point {entry!r} is not ::") + targets = kernel.get("targets") + if not isinstance(targets, list) or not targets or any( + not isinstance(t, dict) or not t or set(t) - {"arch", "hardware"} + or any(not isinstance(v, str) or not v for v in t.values()) for t in targets): + raise invalid("kernel.targets must be a non-empty list of {arch, hardware} subsets") + workloads = data.get("workloads") + if not isinstance(workloads, list) or not workloads: + raise invalid("workloads is empty") + seen = set() + for workload in workloads: + axes = workload.get("axes") if isinstance(workload, dict) else None + if not isinstance(axes, dict) or not axes or any( + not isinstance(k, str) or type(v) is not int for k, v in axes.items()): + raise invalid("a workload's axes is not a non-empty {str: int}") + key = json.dumps(axes, sort_keys=True) + if key in seen: + raise invalid(f"duplicate workload axes {key}") + seen.add(key) + + +@dataclass(frozen=True) +class SchemaTask: + path: str + op: str + language: str + name: str + uuid: str + oid: str + text: str + data: dict + + @property + def task_id(self) -> str: + return f"{self.name}/{self.uuid}" + + @property + def cases(self) -> list[str]: + return [",".join(f"{a}={v}" for a, v in w["axes"].items()) for w in self.data["workloads"]] + + def supports(self, arch: str | None, gpu_model: str) -> bool: + # SIKL's DeviceTarget.normalize_hardware: drop the "AMD Instinct " prefix, compare upper case. + targets = self.data["kernel"]["targets"] + hardware = [t.get("hardware", "").strip().upper().removeprefix("AMD INSTINCT ") for t in targets] + return any(t.get("arch") == arch for t in targets) or gpu_model.upper() in hardware + + def summary(self, arch: str | None, gpu_model: str) -> dict: + return {"task_id": self.task_id, "path": self.path, "op_type": self.op, "language": self.language, + "cases": len(self.cases), "targets": self.data["kernel"]["targets"], + "platform_supported": self.supports(arch, gpu_model)} + + +def inspect_checkout(root: Path, selectors: list[str]) -> tuple[str, list[SchemaTask]]: + """Return HEAD and its schemas; uncommitted files never become tasks.""" + root = Path(root).resolve() + if not root.is_dir() or Path(git(root, "rev-parse", "--show-toplevel").decode().strip()).resolve() != root: + raise ImportProblem("not_a_sikl_checkout", f"{root} is not the top level of a git checkout") + commit = git(root, "rev-parse", "--verify", "HEAD^{commit}").decode().strip() + tasks = [] + for entry in git(root, "ls-tree", "-r", "-z", "-l", "--full-tree", "HEAD", "--", "sikl/ops").split(b"\0"): + meta, _, path = entry.decode(errors="replace").partition("\t") + match = SCHEMA_PATH.match(path) + if not match: + continue + mode, kind, oid, size = meta.split() + if mode != "100644" or kind != "blob": + raise ImportProblem("bad_mode", f"{path}: mode {mode} {kind}") + if int(size) > MAX_SCHEMA_BYTES: + raise ImportProblem("schema_too_large", f"{path}: {size} bytes > {MAX_SCHEMA_BYTES}") + try: + text = git(root, "cat-file", "blob", oid).decode("utf-8") + data = parse_json(text, path) + except (UnicodeDecodeError, ImportProblem) as exc: + raise ImportProblem("invalid_schema", f"{path}: {exc}") from exc + _validate_schema(path, match, data) + tasks.append(SchemaTask(path, match["op"], match["language"], match["name"], match["uuid"], oid, text, data)) + if not tasks: + raise ImportProblem("no_schemas", f"no sikl/ops//impls///.json at {commit}") + if selectors: + missing = sorted(set(selectors) - {t.name for t in tasks} - {t.task_id for t in tasks}) + if missing: + raise ValueError(f"Unknown task selectors: {missing}") + tasks = [t for t in tasks if t.name in selectors or t.task_id in selectors] + return commit, tasks + + +def check_mountable(root: Path) -> None: + # The Docker runner mounts only this directory; its object store must be self-contained. + if not (root / ".git").is_dir(): + raise ImportProblem("checkout_not_mountable", f"{root}/.git is not a directory (git worktree?)") + if (root / ".git" / "objects" / "info" / "alternates").exists(): + raise ImportProblem("checkout_not_mountable", + f"{root} borrows objects through alternates; run `git repack -a -d` and remove them") diff --git a/agents/sikl_task_builder/config.py b/agents/sikl_task_builder/config.py index 1054d2240..bb892f641 100644 --- a/agents/sikl_task_builder/config.py +++ b/agents/sikl_task_builder/config.py @@ -79,8 +79,45 @@ def arena_task_id(self, definition: str) -> str: return (output.relative_to(repo / "tasks") / definition).as_posix() -def load_config(path: Path) -> Config: +@dataclass +class CheckoutConfig: + """A SIKL v5 checkout input; legacy bundle keys have no meaning here.""" + + input_dir: str + output_dir: str = "tasks/SIKL-v5" + artifact_root: str = "sikl_task_builder_runs" + target_gpu_model: str = "MI355X" + tasks: list[str] = field(default_factory=list) + command_timeout: int = 3600 + max_task_seconds: int = 14400 + + def __post_init__(self): + for name in ("input_dir", "output_dir", "artifact_root", "target_gpu_model"): + if not isinstance(getattr(self, name), str) or not getattr(self, name).strip(): + raise ValueError(f"{name} must be a nonempty string") + if not isinstance(self.tasks, list) or any(not isinstance(t, str) for t in self.tasks): + raise ValueError("tasks must be a list of or / selectors") + for name in ("command_timeout", "max_task_seconds"): + if type(getattr(self, name)) is not int or getattr(self, name) <= 0: + raise ValueError(f"{name} must be a positive integer") + + mapping = Config.mapping + arena_task_id = Config.arena_task_id + + +def load_config(path: Path, input_dir: str | None = None) -> Config | CheckoutConfig: + from .bundle import ImportProblem + from .checkout import is_checkout raw = yaml.safe_load(path.read_text()) if not isinstance(raw, dict): raise ValueError("Configuration must be a mapping") + if input_dir: + raw["input_dir"] = input_dir + repo = Path(__file__).resolve().parents[2] + if isinstance(raw.get("input_dir"), str) and is_checkout(repo / Path(raw["input_dir"]).expanduser()): + unused = sorted(set(raw) - set(CheckoutConfig.__dataclass_fields__)) + if unused: + raise ImportProblem("config_not_v5", f"SIKL v5 input does not use: {unused}; " + "use CONFIG=agents/sikl_task_builder/sikl_v5_config.yaml") + return CheckoutConfig(**raw) return Config(**raw) diff --git a/agents/sikl_task_builder/materialize.py b/agents/sikl_task_builder/materialize.py index 16493a2db..be9501c3e 100644 --- a/agents/sikl_task_builder/materialize.py +++ b/agents/sikl_task_builder/materialize.py @@ -10,10 +10,31 @@ from . import BUILDER_VERSION from .bundle import ImportProblem, TaskSpec, case_manifest, fingerprint, source_files, solution_targets +from .checkout import SchemaTask from .config import Config TEMPLATES = Path(__file__).with_name("templates") EDITABLE = "scripts/task_inputs.py" +RUNTIME_ROOT = "/opt/sikl" +SCHEMA_DESCRIPTION = f"""\ +Make the SIKL v5 kernel in source/ faster on every workload of start.json. start.json is the committed +schema: its kernel ABI, workloads, reference and error bounds are fixed. Edit only the files under source/; +do not add or remove files, and keep the entry point a top-level def. The kernel must be correct for every +scalar value inside its declared range. + +Each evaluation rebuilds the schema from source/ and runs `python -m sikl validate --stages audit,bench` +at the pinned SIKL commit, as a SIKL pull request that adds this kernel would be judged. + +A committed kernel is exempt from SIKL's source rules SA0-SA9 only while its bytes are unchanged. Once +edited it is audited as a new kernel, and committed kernels rarely pass that audit, so the first change +must make the whole file comply: no `global`/`nonlocal` (SA1), no module-level mutable state or caches +(SA2), no functools caching except the single `@functools.cache` compile function (SA3), imports limited to +a few standard modules, torch, flydsl and triton (SA4), no forbidden names (SA5), no mutation of non-local +objects or inputs (SA6, SA7), no allocation or other side effect in launch (SA8). The rules are in +{RUNTIME_ROOT}/sikl/ops/README.md; the authority is +{RUNTIME_ROOT}/sikl/core/validation/stages/_audit/_rules.py (AKA_SIKL_ROOT replaces {RUNTIME_ROOT}). +A correctness FAIL names every rejected line or failing workload. +""" def authoring_editable(task: TaskSpec) -> list[str]: @@ -99,14 +120,51 @@ def task_files(task: TaskSpec, config: Config) -> dict[str, str]: def materialize_task(task: TaskSpec, config: Config, destination: Path) -> dict: files = task_files(task, config) + write_files(destination, files) + return {"task_id": task.task_id, "files": sorted(files), "editable": authoring_editable(task)} + + +def write_files(destination: Path, files: dict[str, str]) -> None: if destination.exists(): raise ImportProblem("destination_exists", f"Refusing to replace existing draft: {destination}") destination.mkdir(parents=True) for relative, content in files.items(): path = destination / relative path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(content) - return {"task_id": task.task_id, "files": sorted(files), "editable": authoring_editable(task)} + path.write_text(content, encoding="utf-8") + + +def schema_task_files(task: SchemaTask, config, commit: str) -> dict[str, str]: + """A v5 task judges source/ with `sikl validate`, so it carries the committed schema, not semantics.""" + from src.preprocessing import _resolve_gfx_arch + arch = _resolve_gfx_arch(config.target_gpu_model) + kernel = task.data["kernel"] + entry_file, _, symbol = kernel["entry_point"].partition("::") + cfg = { + "schema_version": 2, + "description": SCHEMA_DESCRIPTION, + "kernel_identity": {"logical_operator": task.name}, + "candidate": {"language": task.language, "initial_state": "implemented", + "editable": [{"path": f"source/{s['path']}", "scope": "file"} for s in kernel["sources"]], + "entrypoints": [{"file": f"source/{entry_file}", "kind": "function", "symbol": symbol}]}, + "baseline": {"kind": "provided", "language": task.language, "source_files": ["start.json"]}, + "evaluation": {"runner": ["python3", "-I", "-B", "scripts/task_runner.py"], "timeout_s": config.command_timeout}, + "platform_support": {"required_arch": arch, "status": "active"}, + "exports": [{"format": "sikl-schema", "output": "export.json", + "command": ["python3", "-I", "-B", "scripts/export_schema.py"], "timeout_s": 60}], + } + from src.task_spec import TaskSpec as ArenaTaskSpec + ArenaTaskSpec.from_mapping(cfg, task_id=config.arena_task_id(task.task_id)) + files = {"config.yaml": yaml.safe_dump(cfg, sort_keys=False), "start.json": task.text} + for source in kernel["sources"]: + files[f"source/{source['path']}"] = source["content"] + for name in ("task_runner.py", "export_schema.py"): + files[f"scripts/{name}"] = (TEMPLATES / "sikl_v5" / name).read_text() + files["scripts/provenance.json"] = json.dumps({ + "builder_version": BUILDER_VERSION, "sikl_commit": commit, "schema_path": task.path, + "oid": task.oid, "name": task.name, "uuid": task.uuid, "arch": arch, + }, indent=2, sort_keys=True) + "\n" + return files def task_tree(root: Path) -> dict[str, str]: diff --git a/agents/sikl_task_builder/orchestrator.py b/agents/sikl_task_builder/orchestrator.py index d0b8e41e0..18d66fd77 100644 --- a/agents/sikl_task_builder/orchestrator.py +++ b/agents/sikl_task_builder/orchestrator.py @@ -15,9 +15,10 @@ from . import BUILDER_VERSION from .backend import generate from .bundle import ImportProblem, fingerprint, inspect_bundle -from .config import Config -from .materialize import check_contract, materialize_task, task_digest, task_tree -from .validation import runtime_identity, validate_task +from .checkout import inspect_checkout +from .config import CheckoutConfig, Config +from .materialize import check_contract, materialize_task, schema_task_files, task_digest, task_tree, write_files +from .validation import runtime_identity, validate_schema_task, validate_task REPO = Path(__file__).resolve().parents[2] LOG = logging.getLogger(__name__) @@ -48,6 +49,16 @@ def implementation_digest(): return task_digest(Path(__file__).parent) +def enclosing_task(output: Path, destination: Path) -> Path | None: + """A task config between output and destination, which would make the new task a nested one.""" + for parent in destination.parents: + if parent == output or not parent.is_relative_to(output): + return None + if (parent / "config.yaml").exists(): + return parent / "config.yaml" + return None + + def install_task(draft: Path, output: Path, task_id: str, evidence: dict) -> Path: if not evidence.get("ok") or evidence.get("task_digest") != task_digest(draft): raise ImportProblem("stale_validation", "Task changed or has no accepted validation") @@ -57,6 +68,7 @@ def install_task(draft: Path, output: Path, task_id: str, evidence: dict) -> Pat if task_digest(destination) == evidence["task_digest"]: return destination raise ImportProblem("output_conflict", f"Existing task differs; refusing overwrite: {destination}") + destination.parent.mkdir(parents=True, exist_ok=True) temporary = Path(tempfile.mkdtemp(prefix=".sikl-install-", dir=output)) try: # Copy exactly the accepted files, excluding command-generated caches. @@ -70,7 +82,7 @@ def install_task(draft: Path, output: Path, task_id: str, evidence: dict) -> Pat # an existing task, including an empty directory created by another run. with (output / ".sikl-install.lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX) - if destination.exists(): + if destination.exists() or enclosing_task(output, destination): raise ImportProblem("output_conflict", str(destination)) temporary.rename(destination) return destination @@ -79,6 +91,20 @@ def install_task(draft: Path, output: Path, task_id: str, evidence: dict) -> Pat shutil.rmtree(temporary) +def run_root(artifacts: Path, run_id: str | None) -> tuple[Path, bool]: + resuming = run_id is not None + run_id = run_id or (time.strftime("%Y%m%d_%H%M%S") + "_" + uuid.uuid4().hex[:8]) + if not run_id or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-" for c in run_id): + raise ValueError("Invalid run ID") + root = artifacts / run_id + if resuming: + if not root.is_dir(): + raise ValueError(f"Run does not exist: {run_id}") + else: + root.mkdir(parents=True, exist_ok=False) + return root, resuming + + def run(config: Config, run_id: str | None = None, *, repo: Path = REPO, generator=generate, validator=validate_task, runtime=runtime_identity) -> dict: source, output, artifacts = resolve_paths(config, repo) @@ -94,16 +120,7 @@ def run(config: Config, run_id: str | None = None, *, repo: Path = REPO, "implementation": implementation_digest(), "runtime": environment, "tasks": {t.task_id: t.digest for t in tasks}, "bundle_files": bundle_files}) - resuming = run_id is not None - run_id = run_id or (time.strftime("%Y%m%d_%H%M%S") + "_" + uuid.uuid4().hex[:8]) - if not run_id or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-" for c in run_id): - raise ValueError("Invalid run ID") - root = artifacts / run_id - if resuming: - if not root.is_dir(): - raise ValueError(f"Run does not exist: {run_id}") - else: - root.mkdir(parents=True, exist_ok=False) + root, resuming = run_root(artifacts, run_id) with (root / ".lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) if resuming: @@ -112,7 +129,7 @@ def run(config: Config, run_id: str | None = None, *, repo: Path = REPO, raise ValueError("Resume refused: input, configuration, builder or runtime changed") else: shutil.copytree(source, root / "bundle") - state = {"run_id": run_id, "identity": identity, "environment": environment, "tasks": {}} + state = {"run_id": root.name, "identity": identity, "environment": environment, "tasks": {}} atomic_json(root / "config.json", config.mapping()) atomic_json(root / "state.json", state) frozen = task_digest(root / "bundle") @@ -194,3 +211,81 @@ def verify_sources(): atomic_json(root / "state.json", state) atomic_json(root / "summary.json", state) return state + + +def run_checkout(config: CheckoutConfig, run_id: str | None = None, *, repo: Path = REPO) -> dict: + """Build one task per committed SIKL v5 schema: deterministic files, the initial-action gate, install.""" + from src.preprocessing import _resolve_gfx_arch + source, output, artifacts = resolve_paths(config, repo) + environment = runtime_identity(config) + commit, tasks = inspect_checkout(source, config.tasks) + arch = _resolve_gfx_arch(config.target_gpu_model) + identity = fingerprint({"config": config.mapping(), "builder": BUILDER_VERSION, + "implementation": implementation_digest(), "runtime": environment, + "commit": commit, "tasks": {t.task_id: t.oid for t in tasks}}) + root, resuming = run_root(artifacts, run_id) + with (root / ".lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + if resuming: + state = json.loads((root / "state.json").read_text()) + if state["identity"] != identity: + raise ValueError("Resume refused: input, configuration, builder or runtime changed") + else: + state = {"run_id": root.name, "identity": identity, "environment": environment, + "commit": commit, "tasks": {}} + atomic_json(root / "config.json", config.mapping()) + atomic_json(root / "state.json", state) + for task in tasks: + record = state["tasks"].setdefault(task.task_id, {"state": "discovered", "attempts": 0}) + destination = output / task.task_id + if record["state"] == "installed": + if not destination.is_dir() or task_digest(destination) != record["task_digest"]: + raise ImportProblem("installed_changed", f"Installed task changed: {task.task_id}") + continue + # Materialization is deterministic, so only a failed gate is worth retrying. + if record["state"] in {"platform_deferred", "output_conflict", "baseline_rejected"}: + continue + if not task.supports(arch, config.target_gpu_model): + record.update(state="platform_deferred") + atomic_json(root / "state.json", state) + continue + draft = root / "drafts" / task.task_id + shutil.rmtree(draft, ignore_errors=True) + write_files(draft, schema_task_files(task, config, commit)) + digest = task_digest(draft) + if destination.exists() or enclosing_task(output, destination): + if destination.is_dir() and task_digest(destination) == digest: + record.update(state="installed", task_digest=digest, destination=str(destination)) + else: + record.update(state="output_conflict", + error={"code": "output_conflict", "message": f"{destination} differs or is nested"}) + atomic_json(root / "state.json", state) + continue + record["attempts"] += 1 + record["state"] = "validating" + atomic_json(root / "state.json", state) + LOG.info("Building %s (%d cases), attempt %d", task.task_id, len(task.cases), record["attempts"]) + try: + evidence = validate_schema_task(draft, root / "evidence" / task.task_id / str(record["attempts"]), + config, task.task_id) + outcome = evidence["outcome"] + if evidence["ok"]: + install_task(draft, output, task.task_id, evidence) + record.update(state="installed", task_digest=evidence["task_digest"], + destination=str(destination), validation_id=evidence["validation_id"]) + record.pop("error", None) + else: + record.update(state=outcome["state"], error={"code": outcome["code"], + "message": f"see evidence attempt {record['attempts']}"}) + except Exception as exc: + code = getattr(exc, "code", "execution_error") + # Another run installed a different task here after the pre-gate check. + record.update(state="output_conflict" if code == "output_conflict" else "failed", + error={"code": code, "message": str(exc)}) + LOG.exception("Task %s failed", task.task_id) + atomic_json(root / "state.json", state) + state["ok"] = not any(t["state"] in ("failed", "baseline_rejected", "output_conflict") + for t in state["tasks"].values()) + atomic_json(root / "state.json", state) + atomic_json(root / "summary.json", state) + return state diff --git a/agents/sikl_task_builder/sikl_runtime.Dockerfile b/agents/sikl_task_builder/sikl_runtime.Dockerfile new file mode 100644 index 000000000..0c654060c --- /dev/null +++ b/agents/sikl_task_builder/sikl_runtime.Dockerfile @@ -0,0 +1,19 @@ +# Runtime of SIKL v5 tasks: SIKL CI's image and pins, with the SIKL checkout at /opt/sikl owned by root. +# Build context: a clean SIKL clone (full .git, no alternates, submodules initialized). BASE is ci.yaml's +# SIKL_CI_IMAGE with its digest; SIKL_COMMIT is that clone's HEAD. +ARG BASE +FROM ${BASE} +ARG SIKL_COMMIT +LABEL sikl.commit=${SIKL_COMMIT} +COPY --chown=root:root . /opt/sikl +# ci.yaml setup_sikl: install_triton.sh, then its pinned Triton; setup.py's FlyDSL pin. No editable install: +# task runners import aiter and sikl from a private checkout of the pinned commit. +RUN chmod -R a+rX,go-w /opt/sikl \ + && bash /opt/sikl/.github/scripts/install_triton.sh \ + && python3 -m pip install --no-deps --force-reinstall \ + "$(grep -oE -- "--force-reinstall 'triton==[0-9.]+'" /opt/sikl/.github/workflows/ci.yaml | grep -oE 'triton==[0-9.]+')" \ + && python3 -m pip install "$(sed -nE 's/^FLYDSL_VERSION = "(flydsl==[^"]+)"$/\1/p' /opt/sikl/setup.py)" \ + && rm -rf /tmp/aiter_configs && mkdir -m 1777 /tmp/aiter_configs +# aiter merges its tuned configs into the fixed /tmp/aiter_configs; tasks run as a non-root user, and the +# base image's root-owned merges come from its own aiter, not from the pinned checkout's. +ENV AKA_SIKL_ROOT=/opt/sikl PREBUILD_KERNELS=0 AITER_USE_SYSTEM_TRITON=1 diff --git a/agents/sikl_task_builder/sikl_v5_config.yaml b/agents/sikl_task_builder/sikl_v5_config.yaml new file mode 100644 index 000000000..6e98a9776 --- /dev/null +++ b/agents/sikl_task_builder/sikl_v5_config.yaml @@ -0,0 +1,11 @@ +# Configuration for a SIKL v5 checkout input (sikl/ops//impls///.json). +# Legacy bundle keys (policy, generator, validator, selections, ...) are rejected for this input. +input_dir: SIKL +output_dir: tasks/SIKL-v5 +artifact_root: sikl_task_builder_runs +# Read by the Docker runner with sed, so keep it in this file. +target_gpu_model: MI355X +# Empty selects every committed schema; entries are or /. +tasks: [] +command_timeout: 3600 +max_task_seconds: 14400 diff --git a/agents/sikl_task_builder/templates/sikl_v5/export_schema.py b/agents/sikl_task_builder/templates/sikl_v5/export_schema.py new file mode 100644 index 000000000..d85618693 --- /dev/null +++ b/agents/sikl_task_builder/templates/sikl_v5/export_schema.py @@ -0,0 +1,51 @@ +"""Arena sikl-schema export: the validated, faster candidate's repo path and text, or nothing.""" +import json +import os +import sys +from pathlib import Path + +# `python3 -I` leaves the script's directory off sys.path. +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import task_runner as runner # noqa: E402 + + +def validated_candidate(workspace: Path) -> tuple[str, str]: + reports = runner.report_dir(workspace, "candidate") + if not (reports / "exit_code").is_file(): + raise runner.Failed(f"no report for source/ {reports.name}") + # The runner copies a schema only for a candidate new at the base; an unchanged kernel stops here. + files = sorted(reports.glob("sikl/ops/*/impls/*/*/*.json")) + if len(files) != 1: + raise runner.Failed(f"{len(files)} candidate files in {reports}") + exit_code = int((reports / "exit_code").read_text()) + if exit_code: + raise runner.Failed(f"validate exit code {exit_code}") + bench = runner.stage(json.loads((reports / "report.json").read_text(encoding="utf-8")), "bench") + status = bench["status"] if bench else "missing" + if status != "passed": + raise runner.Failed(f"bench stage is {status}") + path = files[0].relative_to(reports).as_posix() + commit = runner.read_json(workspace, "scripts/provenance.json")["sikl_commit"] + if runner.git(runner.SIKL_ROOT, "ls-tree", commit, "--", path).strip(): + raise runner.Failed(f"{path} exists at base") + # SIKL holds a new file to "retain" (>= 0.99) against the kernel it would displace; only gains are exported. + slow = [runner.axes_text(r["axes"]) for r in bench["workload_records"] + if any(t["threshold"] == "retain" and t["ratio_lower_bound"] <= 1.0 for t in r["timings"])] + if slow: + raise runner.Failed("not faster than the served kernel at " + ", ".join(slow)) + return path, files[0].read_text(encoding="utf-8") + + +def main() -> int: + try: + path, text = validated_candidate(Path.cwd()) + except Exception as exc: # Arena records the export as FAIL with this message, never a traceback + print(f"no export: {exc or type(exc).__name__}", file=sys.stderr) + return 1 + Path(os.environ["ARENA_EXPORT_PATH"]).write_text( + json.dumps({"path": path, "text": text}, ensure_ascii=False), encoding="utf-8") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/agents/sikl_task_builder/templates/sikl_v5/task_runner.py b/agents/sikl_task_builder/templates/sikl_v5/task_runner.py new file mode 100644 index 000000000..e8af1d800 --- /dev/null +++ b/agents/sikl_task_builder/templates/sikl_v5/task_runner.py @@ -0,0 +1,398 @@ +"""arena-eval-v1 actions of a SIKL v5 schema task, judged by `python -m sikl validate` at the pinned SIKL commit. + +Standard library only; run as `python3 -I -B`. Generated by agents/sikl_task_builder; it imports nothing from +Arena or the builder. +""" +import atexit +import contextlib +import functools +import hashlib +import json +import os +import re +import shutil +import statistics +import subprocess +import sys +import tempfile +from importlib import metadata +from pathlib import Path + +SIKL_ROOT = Path(os.environ.get("AKA_SIKL_ROOT", "/opt/sikl")).resolve() +CK = "3rdparty/composable_kernel" +# Shared across validations as in SIKL CI; every other kernel cache is private to one validation. +AITER_JIT_DIR = "/tmp/aka-sikl-aiter-jit" +# SIKL's code-lint check-added-large-files limit (default maxkb=500). +MAX_SCHEMA_BYTES = 500 * 1024 +REPORT_VERSION = 2 +# Findings of a correct kernel that is merely slow: they decide performance, never correctness. +SLOW_BUT_CORRECT_CODES = ("not_faster", "regressed") +COMPILE_CODES = ("build_failed", "schema_invalid", "kernel_missing") +LOG_TAIL_CHARS = 4000 +# SIKL retries a tracer fault in a fresh process, but a bench runner killed by a signal mid-measurement is a +# candidate_crash with no retry; validate reruns once when every failure is one of these. +SIGNAL_CRASH = re.compile(r"exited with code -\d+$") +# Untracked non-schema file in the private checkout: with it, a path already at the base is still benched, +# and the base exempts that shipped kernel from the source rules, as a SIKL PR that also changes code would. +BASE_MARKER = ".sikl-task-base" +TRITON_PIN = re.compile(r"--force-reinstall 'triton==([0-9.]+)'") +FLYDSL_PIN = re.compile(r'^FLYDSL_VERSION = "flydsl==([^"]+)"$', re.M) +ROLES = ("baseline", "candidate") +ACTIONS = ("compile", "correctness", "performance") +# Run with cwd = the private checkout: `-c` puts cwd first on sys.path, a script file would not. +GENERATE = """ +import json, sys +from pathlib import Path +from sikl.core.contract.expr import json_file_text +workspace, out = Path(sys.argv[1]), Path(sys.argv[2]) +schema = json.loads((workspace / "start.json").read_text(encoding="utf-8")) +for source in schema["kernel"]["sources"]: + source["content"] = (workspace / "source" / source["path"]).read_text(encoding="utf-8") +for workload in schema["workloads"]: + workload.pop("speedup", None) + workload.pop("accuracy", None) +out.write_text(json_file_text(schema), encoding="utf-8") +""" + + +class Failed(Exception): + """The action fails for this reason.""" + + +def tail(text: str) -> str: + return text[-LOG_TAIL_CHARS:] + + +@functools.cache +def git_config() -> str: + # The runner's own global config, never the agent's: a local clone runs upload-pack on the source's git + # dir with GIT_CONFIG_PARAMETERS cleared, so `-c safe.directory` would not reach it. + fd, path = tempfile.mkstemp(prefix="sikl-gitconfig-") + with os.fdopen(fd, "w") as f: + f.write("[core]\n\thooksPath = /dev/null\n[safe]\n") + f.writelines(f"\tdirectory = {p}\n" for p in (SIKL_ROOT, SIKL_ROOT / ".git", SIKL_ROOT / CK)) + atexit.register(os.unlink, path) + return path + + +def git(cwd, *args: str) -> str: + env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")} + env.update(GIT_CONFIG_NOSYSTEM="1", GIT_CONFIG_SYSTEM=os.devnull, GIT_CONFIG_GLOBAL=git_config(), + GIT_TERMINAL_PROMPT="0") + return subprocess.run(["git", "-c", "core.hooksPath=/dev/null", "-c", f"safe.directory={Path(cwd).resolve()}", + "-C", str(cwd), *args], env=env, capture_output=True, text=True, check=True).stdout + + +def axes_text(axes: dict) -> str: + return ",".join(f"{a}={v}" for a, v in axes.items()) + + +def read_json(workspace: Path, relative: str) -> dict: + return json.loads((workspace / relative).read_text(encoding="utf-8")) + + +def case_ids(start: dict) -> list[str]: + return [axes_text(w["axes"]) for w in start["workloads"]] + + +def schema_dir(start: dict) -> str: + return f"sikl/ops/{start['op_type']}/impls/{start['kernel']['language']}/{start['name']}" + + +def content_hash(workspace: Path, role: str) -> str: + h = hashlib.sha256() + for source in read_json(workspace, "start.json")["kernel"]["sources"]: + if role == "baseline": + data = source["content"].encode() + else: + data = (workspace / "source" / source["path"]).read_bytes() + h.update(source["path"].encode() + b"\0" + data + b"\0") + return h.hexdigest()[:16] + + +def report_dir(workspace: Path, role: str) -> Path: + return workspace / "reports" / f"{role}-{content_hash(workspace, role)}" + + +def stage(report: dict, name: str) -> dict | None: + return next((s for s in report["stages"] if s["stage_name"] == name), None) + + +@contextlib.contextmanager +def private_checkout(commit: str): + # Resolved: SIKL prints resolved paths, and the candidate's export check line is matched on it. + tmp = Path(tempfile.mkdtemp(prefix="sikl-")).resolve() + try: + checkout = tmp / "SIKL" + git(tmp, "clone", "-q", "--shared", "--no-checkout", str(SIKL_ROOT), str(checkout)) + git(checkout, "checkout", "-q", "--detach", commit) + for name in ("tmp", "home"): + (tmp / name).mkdir() + yield checkout + finally: + shutil.rmtree(tmp, ignore_errors=True) + + +def sikl_env(checkout: Path, arch: str) -> dict: + tmp = checkout.parent / "tmp" + env = {k: v for k, v in os.environ.items() if k != "AITER_ASM_DIR" and not k.startswith("GIT_")} + env.update( + PYTHONPATH=str(checkout), + PYTHONNOUSERSITE="1", + PYTHONDONTWRITEBYTECODE="1", + HOME=str(checkout.parent / "home"), + GPU_ARCHS=arch, + CK_DIR=str(SIKL_ROOT / CK), + AITER_JIT_DIR=AITER_JIT_DIR, + TMPDIR=str(tmp), + FLYDSL_RUNTIME_CACHE_DIR=str(tmp / "flydsl-runtime"), + FLYDSL_AUTOTUNE_CACHE_DIR=str(tmp / "flydsl-autotune"), + TRITON_CACHE_DIR=str(tmp / "triton"), + XDG_CACHE_HOME=str(tmp / "xdg"), + SIKL_KERNEL_CACHE_DIR=str(tmp / "sikl-kernels"), + ) + return env + + +def sikl(checkout: Path, env: dict, *args: str) -> subprocess.CompletedProcess: + return subprocess.run([sys.executable, "-s", "-m", "sikl", *args], cwd=checkout, env=env, + capture_output=True, text=True) + + +def build_candidate(checkout: Path, env: dict, workspace: Path, start: dict, reports: Path) -> tuple[str, bool]: + """Write the candidate into the checkout; return its repo path and whether it is new at the base.""" + out = checkout / schema_dir(start) / "candidate.json" + p = subprocess.run([sys.executable, "-s", "-c", GENERATE, str(workspace), str(out)], cwd=checkout, + env=env, capture_output=True, text=True) + if p.returncode: + raise Failed("cannot build the candidate schema: " + tail(p.stderr)) + size = out.stat().st_size + if size > MAX_SCHEMA_BYTES: + raise Failed(f"candidate schema is {size} bytes, over the {MAX_SCHEMA_BYTES}-byte limit of SIKL code-lint") + p = sikl(checkout, env, "export", "check") + moved = f"{out}: not at /" + mine = [line for line in p.stderr.splitlines() if line.startswith(f"{out}: ")] + if len(mine) != 1 or not mine[0].startswith(moved): + raise Failed("export check: " + ("\n".join(mine) or tail(p.stderr))) + path = "sikl/ops/" + mine[0][len(moved):] + if git(checkout, "ls-tree", "HEAD", "--", path).strip(): + out.unlink() + return path, False + os.rename(out, checkout / path) + p = sikl(checkout, env, "export", "check") + if p.returncode: + raise Failed("export check after rename: " + tail(p.stderr)) + (reports / path).parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(checkout / path, reports / path) + return path, True + + +def ensure_report(workspace: Path, role: str, start: dict, fresh: bool) -> Path: + reports = report_dir(workspace, role) + if reports.is_symlink(): + reports.unlink() + if fresh: + shutil.rmtree(reports, ignore_errors=True) + elif (reports / "exit_code").exists(): + return reports + provenance = read_json(workspace, "scripts/provenance.json") + commit = provenance["sikl_commit"] + with private_checkout(commit) as checkout: + env = sikl_env(checkout, provenance["arch"]) + reports.mkdir(parents=True, exist_ok=True) + if role == "baseline": + # validate-task checked that start.json is this committed blob. + path, is_new = provenance["schema_path"], False + else: + path, is_new = build_candidate(checkout, env, workspace, start, reports) + if not is_new: + (checkout / BASE_MARKER).write_text("") + argv = ["validate", path, "--base", commit, "--stages", "audit,bench", "--devices", "0", + "--out", str(reports)] + log = "" + for attempt in (1, 2): + (reports / "report.json").unlink(missing_ok=True) + p = sikl(checkout, env, *argv) + log += p.stdout + p.stderr + (reports / "validate.log").write_text(log, encoding="utf-8") + if not (reports / "report.json").is_file(): + raise Failed(f"validate wrote no report (rc {p.returncode}): " + tail(p.stdout + p.stderr)) + if attempt == 2 or not platform_flake(json.loads((reports / "report.json").read_text())): + break + # Written last: its presence marks a complete report, the cache hit test above. + (reports / "exit_code").write_text(f"{p.returncode}\n") + return reports + + +def platform_flake(report: dict) -> bool: + bad = [f for s in report["stages"] for f in s["findings"] if f["code"] not in SLOW_BUT_CORRECT_CODES] + return bool(bad) and all(f["code"] == "platform_failure" + or (f["code"] == "candidate_crash" and SIGNAL_CRASH.search(f["detail"])) for f in bad) + + +def finding_line(f: dict) -> str: + return f"{f['code']} {axes_text(f['axes']) if f['axes'] else '-'}: {f['detail']}" + + +def aborted_reason(report: dict) -> str: + deciding = next((s for status in ("failed", "aborted") for s in report["stages"] if s["status"] == status), None) + if deciding is None: # validate killed after writing a passed report.json + return "validation aborted" + findings = dict.fromkeys(f"{f['code']}: {f['detail']}" for s in report["stages"] for f in s["findings"]) + return "\n".join([f"validation {deciding['status']} ({deciding['stage_name']}): {deciding['detail']}", *findings]) + + +def correctness(ids: list[str], findings: list, report: dict, exit_code: int) -> tuple[str, list, str | None]: + bad = [f for f in findings if f["code"] not in SLOW_BUT_CORRECT_CODES] + if exit_code == 0 or (exit_code == 1 and findings and not bad): + return "PASS", [{"test_case_id": i, "status": "PASS"} for i in ids], None + aborted = exit_code not in (0, 1) + everyone = aborted or not bad or any(f["axes"] is None for f in bad) + located = {axes_text(f["axes"]) for f in bad if f["axes"] is not None} + cases = [{"test_case_id": i, "status": "FAIL" if everyone or i in located else "PASS"} for i in ids] + if aborted: + reason = aborted_reason(report) + else: + reason = "\n".join(finding_line(f) for f in bad) or f"validate exit code {exit_code} without findings" + return "FAIL", cases, reason + + +def judge(action: str, start: dict, report: dict, exit_code: int) -> tuple[str, list, str | None]: + if report.get("version") != REPORT_VERSION: + raise Failed(f"unsupported SIKL report version {report.get('version')!r}") + ids = case_ids(start) + findings = [f for s in report["stages"] for f in s["findings"]] + if action == "compile": + if exit_code not in (0, 1): + return "FAIL", [], aborted_reason(report) + build = [f for f in findings if f["code"] in COMPILE_CODES + or (f["code"] == "source_rejected" and f["detail"].startswith("SA0 "))] + if build: + return "FAIL", [], "\n".join(finding_line(f) for f in build) + return "PASS", [], None + status, cases, reason = correctness(ids, findings, report, exit_code) + if action == "correctness" or status == "FAIL": + return status, cases, reason + bench = stage(report, "bench") + records = bench.get("workload_records", []) if bench else [] + cases, untimed = [], [] + for workload in start["workloads"]: + case_id = axes_text(workload["axes"]) + record = next((r for r in records if r["axes"] == workload["axes"]), None) + timing = next((t for t in (record["timings"] if record else []) if t["versus"] == "baseline"), None) + if timing and timing["candidate_milliseconds"]: + cases.append({"test_case_id": case_id, "status": "PASS", "benchmark_method": "cuda_graph", + "execution_time_ms": statistics.median(timing["candidate_milliseconds"])}) + else: + cases.append({"test_case_id": case_id, "status": "FAIL"}) + untimed.append(case_id) + if untimed: + return "FAIL", cases, "no bench timing versus baseline at " + ", ".join(untimed) + return "PASS", cases, None + + +def evaluate(workspace: Path, role: str, action: str) -> tuple[str, list, str | None, dict | None]: + start = read_json(workspace, "start.json") + # performance always measures afresh: a cached report could be one the agent wrote. + reports = ensure_report(workspace, role, start, fresh=action == "performance") + report = json.loads((reports / "report.json").read_text(encoding="utf-8")) + exit_code = int((reports / "exit_code").read_text()) + return (*judge(action, start, report, exit_code), None) + + +def pinned(text: str, pattern: re.Pattern, what: str) -> str: + match = pattern.search(text) + if not match: + raise Failed(f"no {what} pin at the SIKL commit") + return match.group(1) + + +def installed_version(distribution: str) -> str: + try: + return metadata.version(distribution).split("+")[0] + except metadata.PackageNotFoundError: + return "not installed" + + +def validate_task(workspace: Path) -> tuple[str, list, str | None, dict | None]: + start = read_json(workspace, "start.json") + provenance = read_json(workspace, "scripts/provenance.json") + commit = provenance["sikl_commit"] + missing = [f"source/{s['path']}" for s in start["kernel"]["sources"] + if not (workspace / "source" / s["path"]).is_file() + or not (workspace / "source" / s["path"]).stat().st_size] + if missing: + raise Failed("candidate source missing or empty: " + ", ".join(missing)) + try: + git(SIKL_ROOT, "cat-file", "-e", f"{commit}^{{commit}}") + except subprocess.CalledProcessError: + raise Failed(f"sikl_commit {commit} not in {SIKL_ROOT}") from None + listed = git(SIKL_ROOT, "ls-tree", commit, "--", provenance["schema_path"]).split() + expected = listed[2] if len(listed) > 2 else "none" + actual = git(SIKL_ROOT, "hash-object", "--no-filters", str(workspace / "start.json")).strip() + if actual != expected: + raise Failed(f"start.json oid {actual} != {expected} at {provenance['schema_path']}") + for distribution, file, pattern in (("triton", ".github/workflows/ci.yaml", TRITON_PIN), + ("flydsl", "setup.py", FLYDSL_PIN)): + pin = pinned(git(SIKL_ROOT, "show", f"{commit}:{file}"), pattern, distribution) + version = installed_version(distribution) + if version != pin: + raise Failed(f"{distribution} {version} != pin {pin}") + gitlink = git(SIKL_ROOT, "ls-tree", commit, "--", CK).split() + expected = gitlink[2] if len(gitlink) > 2 else "none" + try: + # An uninitialized submodule directory would resolve to the superproject's HEAD. + if not (SIKL_ROOT / CK / ".git").exists(): + raise OSError + actual = git(SIKL_ROOT / CK, "rev-parse", "HEAD").strip() + except (subprocess.CalledProcessError, OSError): + raise Failed(f"CK submodule missing: {SIKL_ROOT / CK}") from None + if actual != expected: + raise Failed(f"CK submodule {actual} != {expected}") + for path in (SIKL_ROOT, SIKL_ROOT / ".git", SIKL_ROOT / ".git" / "objects"): + if os.access(path, os.W_OK): + raise Failed(f"AKA_SIKL_ROOT {SIKL_ROOT} is writable by this uid") + cases = [{"test_case_id": i, "status": "PASS", "checks": ["correctness", "performance"]} for i in case_ids(start)] + return "PASS", cases, None, {"candidate_state": "implemented"} + + +def failed_cases(workspace: Path, action: str) -> list: + if action in ("validate-task", "compile"): + return [] + try: + return [{"test_case_id": i, "status": "FAIL"} for i in case_ids(read_json(workspace, "start.json"))] + except Exception: + return [] + + +def main(argv: list[str] | None = None) -> int: + argv = sys.argv[1:] if argv is None else argv + if argv == ["validate-task"]: + role, action = "task", "validate-task" + elif len(argv) == 2 and argv[0] in ROLES and argv[1] in ACTIONS: + role, action = argv + else: + print(f"usage: task_runner.py validate-task | {{{','.join(ROLES)}}} {{{','.join(ACTIONS)}}}", file=sys.stderr) + return 2 + workspace = Path.cwd() + try: + if role == "task": + status, cases, reason, metadata_ = validate_task(workspace) + else: + status, cases, reason, metadata_ = evaluate(workspace, role, action) + except Exception as exc: # AKA treats a missing result line as a protocol error, not a FAIL + stderr = getattr(exc, "stderr", None) + reason = (str(exc) or type(exc).__name__) + (f"\n{tail(stderr)}" if stderr else "") + status, metadata_ = "FAIL", None + cases = failed_cases(workspace, action) + report = {"protocol": "arena-eval-v1", "role": role, "action": action, "status": status, "cases": cases} + if reason is not None: + report["reason"] = reason + if metadata_ is not None: + report["metadata"] = metadata_ + print("ARENA_EVAL_RESULT=" + json.dumps(report, allow_nan=False, separators=(",", ":"))) + return 0 if status == "PASS" else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/agents/sikl_task_builder/validation.py b/agents/sikl_task_builder/validation.py index 9464d87ed..2961c451e 100644 --- a/agents/sikl_task_builder/validation.py +++ b/agents/sikl_task_builder/validation.py @@ -5,8 +5,10 @@ import json import logging import os +import re import shutil import sys +import time import uuid from pathlib import Path @@ -132,12 +134,103 @@ def validate_task(draft: Path, artifacts: Path, config: Config, timeout=None) -> return result +GATE_ACTIONS = ("task/validate-task", "baseline/compile", "baseline/correctness", "baseline/performance", + "candidate/compile", "candidate/correctness", "candidate/performance") +DEADLINE_ERROR = re.compile(r"exceeded its .*s action deadline") +# Exit code of the initial-mode subprocess when the task deadline passed before an action started. +DEADLINE_EXIT = 3 +# Baseline findings that say nothing about the committed kernel itself. +PLATFORM_CODES = {"platform_failure", "not_faster", "regressed"} + + +def gate_outcome(process: dict, commands: dict, initial: dict | None, baseline_codes: list[str]) -> dict: + """Map one gate attempt to a run state; only a complete, passing gate is installable.""" + errors = [c["execution_error"] or "" for c in commands.values() if c["execution_error"]] + if process["timed_out"] or process["exit_code"] == DEADLINE_EXIT or any(DEADLINE_ERROR.search(e) for e in errors): + return {"state": "failed", "code": "gate_timeout"} + failed = [key for key, c in commands.items() if c["result"] and not c["ok"]] + if errors or initial is None: + return {"state": "failed", "code": "execution_error"} + if any(key.startswith("baseline/") for key in failed): + if set(baseline_codes) - PLATFORM_CODES: + return {"state": "baseline_rejected", "code": "baseline_rejected"} + return {"state": "failed", "code": "execution_error"} + if "task/validate-task" in failed: + return {"state": "failed", "code": "validate_task_failed"} + if failed: + return {"state": "failed", "code": "candidate_rejected"} + if process["exit_code"] != 0 or not initial.get("accepted") or set(commands) != set(GATE_ACTIONS): + return {"state": "failed", "code": "execution_error"} + return {"state": "installable", "code": None} + + +def validate_schema_task(draft: Path, artifacts: Path, config, task_id: str) -> dict: + """The seven initial Arena actions of a SIKL v5 task in a fresh workspace, bounded by one subprocess.""" + environment = runtime_identity(config) + artifacts.mkdir(parents=True, exist_ok=False) + workspace = artifacts / "workspace" + source_digest = task_digest(draft) + copy_task_files(draft, workspace) + request = artifacts / "gate_request.json" + request.write_text(json.dumps({"mode": "initial", "workspace": str(workspace), + "task_id": config.arena_task_id(task_id)})) + env = build_subprocess_env() + # Inner deadline first, so the action supervisor reaps its own process group cleanly. + env["ARENA_TASK_DEADLINE"] = str(time.time() + max(1, config.max_task_seconds - 60)) + process = run_process([sys.executable, "-m", "agents.sikl_task_builder.validation", str(request)], + Path(__file__).resolve().parents[2], artifacts / "gate.log", config.max_task_seconds, env) + session = artifacts / "session" + initial_path = session / "initial_validation.json" + initial = json.loads(initial_path.read_text()) if initial_path.is_file() else None + commands = {} + for path in sorted(session.glob("action-*.json")): + record = json.loads(path.read_text()) + result = record.get("result") + key = f"{record.get('role') or (result or {}).get('role')}/{record.get('action') or (result or {}).get('action')}" + commands[key] = {"ok": bool(result) and result.get("status") == "PASS", "result": result, + "execution_error": record.get("execution_error")} + baseline_codes, unreadable = set(), [] + for report in sorted((session / "baseline" / "reports").glob("baseline-*/report.json")): + try: + baseline_codes |= {f["code"] for s in json.loads(report.read_text())["stages"] for f in s["findings"]} + except (OSError, ValueError, KeyError, TypeError): + unreadable.append(str(report)) + baseline_codes = sorted(baseline_codes) + outcome = gate_outcome(process, commands, initial, baseline_codes) + if outcome["state"] == "installable" and task_digest(draft) != source_digest: + outcome = {"state": "failed", "code": "stale_validation"} + evidence = {"validation_id": uuid.uuid4().hex, "task_digest": source_digest, "environment": environment, + "initial_validation": initial, "commands": commands, "baseline_codes": baseline_codes, + "unreadable_reports": unreadable, + "gate_process": process, "outcome": outcome, "ok": outcome["state"] == "installable"} + (artifacts / "evidence.json").write_text(json.dumps(evidence, indent=2) + "\n") + return evidence + + +def _validate_initial(request: dict) -> int: + from src.task_runtime import bind_session_runtime + from src.task_session import TaskSession + from src.task_spec import load_task_spec + workspace = Path(request["workspace"]) + spec = load_task_spec(workspace / "config.yaml", task_id=request["task_id"]) + session = TaskSession.create(spec, workspace, workspace.parent / "session") + bind_session_runtime(session) + try: + initial = session.validate_initial() + except TimeoutError: + return DEADLINE_EXIT + session.verify_baseline_sources() + return 0 if initial.accepted else 1 + + def _main(): - from agents.task_validator.launch_agent import launch_agent request = json.loads(Path(sys.argv[1]).read_text()) + logging.basicConfig(level=logging.INFO) + if request.get("mode") == "initial": + return _validate_initial(request) + from agents.task_validator.launch_agent import launch_agent config = Config(**request["config"]) workspace = Path(request["workspace"]) - logging.basicConfig(level=logging.INFO) from src.task_spec import load_task_spec from src.task_session import TaskSession from src.task_run import validate_task_session diff --git a/docs/how-to/sikl-task-builder.md b/docs/how-to/sikl-task-builder.md index c056befff..6a95640e3 100644 --- a/docs/how-to/sikl-task-builder.md +++ b/docs/how-to/sikl-task-builder.md @@ -1,19 +1,148 @@ -# Build Arena tasks from a SIKL bundle +# Build Arena tasks from SIKL -`sikl_task_builder` turns a local SIKL dataset folder into standard, self-contained -Arena tasks under `tasks/SIKL-task/`. It combines deterministic import tools, a -Codex agent that reviews generation (and repairs legacy input adapters), and an independent `task_validator`. -Only accepted tasks reach the output directory. Future bundles using the supported -schema can use the same workflow without a new handwritten task adapter per JSONL. +`sikl_task_builder` turns SIKL kernels into standard Arena tasks. It is a workflow +**before task discovery**, invoked with its own Make command, not an `agent.template` +for `main.py`. The resulting tasks use the shared [`schema_version: 2` task contract](add-task.md) +with an explicit candidate, provided baseline and `arena-eval-v1` runner, and can be selected by +ordinary optimization runs. -This is an agent workflow **before task discovery**, invoked with its own Make -command. It is not an `agent.template` for `main.py`: its input is a dataset -folder and its output is multiple tasks. The resulting tasks use the shared -[`schema_version: 2` task contract](add-task.md), with an explicit candidate, -provided baseline, and `arena-eval-v1` runner. They can be selected by ordinary optimization runs. -It does not require KernelForge or a task-specific SIKL integration. +It reads two inputs, chosen by the shape of `--input-dir`: -## Input and task boundaries +- **A SIKL checkout** (a directory with `sikl/ops/`): SIKL `main` since AMD-AGI/SIKL#52, whose + kernels are committed v5 schema files. See [SIKL v5 checkouts](#sikl-v5-checkouts). +- **A legacy bundle** (`definitions/`, `workloads/`, `solutions/`), exported by SIKL before #52. + See [Legacy bundles](#legacy-bundles-sikl-before-52). + +## SIKL v5 checkouts + +SIKL `main` keeps one kernel per file, `sikl/ops//impls///.json` +([kernel schema](https://github.com/AMD-AGI/SIKL/blob/main/sikl/ops/README.md)). The reference, +input distributions, error bounds and source rules live in the SIKL package, not in the file, so a +task cannot carry its own semantics: every action is judged by SIKL's own `python -m sikl validate`. + +### Input, selection and output + +```bash +python3 -m agents.sikl_task_builder inspect \ + --config agents/sikl_task_builder/sikl_v5_config.yaml --input-dir /path/to/SIKL +``` + +`inspect` runs on the host, reads only committed git objects at `HEAD` (no torch, no SIKL import, no +code execution) and lists one task per schema file with its workloads and whether its +`kernel.targets` match `target_gpu_model`. Errors print one JSON line +`{"ok": false, "code": ..., "message": ...}` and exit 1; codes: `not_a_sikl_checkout`, `bad_mode`, +`schema_too_large`, `invalid_schema`, `unsupported_schema_version`, `no_schemas`, +`checkout_not_mountable`, `config_not_v5`, `installed_changed`. + +The v5 input takes only the keys of +[`sikl_v5_config.yaml`](../../agents/sikl_task_builder/sikl_v5_config.yaml): `input_dir`, +`output_dir` (default `tasks/SIKL-v5`), `artifact_root`, `target_gpu_model`, `tasks`, +`command_timeout`, `max_task_seconds`. Legacy keys (`policy`, `generator`, `validator`, +`selections`, `target_language`, ...) are rejected with `config_not_v5`. A `tasks` entry is a +schema name (all its files) or `/`. + +Each schema becomes `tasks/SIKL-v5///`: + +```text +├── config.yaml +├── start.json # the committed schema bytes, protected +├── source/ # the only editable files, starting from the committed kernel +└── scripts/ + ├── task_runner.py # arena-eval-v1 actions -> sikl validate + ├── export_schema.py # sikl-schema export + └── provenance.json # builder version, SIKL commit, schema path, blob oid, arch +``` + +Generated tasks are not committed (`/tasks/SIKL-v5/` is ignored): they follow SIKL commits and +are rebuilt deterministically. Their timing comes from SIKL's bench, which the perf-helper audit +does not recognize, so `make check-perf-helpers` fails in a checkout where they are installed. + +### Runtime image + +The tasks need SIKL CI's runtime and a read-only SIKL checkout. Build it from a clean SIKL clone +(full `.git`, no alternates, `git submodule update --init --recursive`): + +```bash +docker build --network host -f agents/sikl_task_builder/sikl_runtime.Dockerfile \ + --build-arg BASE="$(sed -nE 's/^ +SIKL_CI_IMAGE: +&ci_image +//p' /path/to/SIKL/.github/workflows/ci.yaml)" \ + --build-arg SIKL_COMMIT="$(git -C /path/to/SIKL rev-parse HEAD)" --iidfile sikl-runtime.id /path/to/SIKL +export AKA_DOCKER_IMAGE="$(cat sikl-runtime.id)" +``` + +The image copies the clone to `/opt/sikl` owned by root, installs `ci.yaml`'s Triton pin and +`setup.py`'s FlyDSL pin, and sets `AKA_SIKL_ROOT=/opt/sikl`. It adds a copy of SIKL (with its +`.git` and submodules) to the base image. Ordinary runs of these tasks must use it, through +`AKA_DOCKER_IMAGE` or the run config's `docker_image`. The `forge` agent additionally needs +Hyperloom installed on top of it. + +### Build + +```bash +AKA_DOCKER_IMAGE="$(cat sikl-runtime.id)" AKA_DOCKER_PRIVILEGED=0 AKA_SKIP_DEV_MEM=1 \ + make docker-sikl-task-builder INPUT=/path/to/SIKL CONFIG=agents/sikl_task_builder/sikl_v5_config.yaml +``` + +Materialization is deterministic: no generator agent and no repair loop, because nothing in the task +is authored. The gate copies each draft to a fresh workspace and runs the framework's seven initial +actions (`validate-task`, then baseline and candidate compile, correctness and performance) in one +subprocess bounded by `max_task_seconds`; the task is installed only if all of them pass. No agent CLI +or credential is mounted. Run states, in `sikl_task_builder_runs//summary.json`: + +| State | Meaning | On `resume` | +| --- | --- | --- | +| `installed` | gate passed, or an identical task is already installed | skipped | +| `platform_deferred` | `kernel.targets` does not include the runtime GPU | skipped | +| `output_conflict` | a different task already exists at the destination, or it would nest in another task | skipped; remove it and start a new run | +| `baseline_rejected` | the committed kernel itself fails in this runtime | skipped | +| `failed` | `gate_timeout`, `execution_error`, `validate_task_failed`, `candidate_rejected` or `stale_validation` | retried | + +Evidence for every attempt is under `evidence////` (`evidence.json`, the gate +log, the session and its SIKL reports). Resume refuses a changed configuration, input commit, +builder or runtime. Upgrading the builder package also refuses to resume legacy runs started +before the upgrade. + +### Judging + +- **Candidate**: `source/` is rebuilt into a schema (SIKL's `json_file_text`, `speedup` and + `accuracy` dropped), `sikl export check` names its `.json`, and + `sikl validate --base --stages audit,bench --devices 0` runs in a private + checkout of the pinned commit: a SIKL PR that adds this one file. +- **Baseline**: the committed file, validated with an untracked marker in the private checkout so it + is benched as a PR that also changes code would be. +- Differences from SIKL CI: one path instead of the change set, `--devices 0`, base = the pinned commit, + no `--allow-measured-edits`, no e2e stage. +- `compile` fails on `build_failed`, `schema_invalid`, `kernel_missing` or an SA0 (parse) rejection; + `correctness` fails on any finding but `not_faster`/`regressed`, located to a workload when the + finding has axes; `performance` re-measures every time, fails if correctness would, and reports the + median versus-baseline time of every workload. Baseline time is taken from the initial validation. +- A committed kernel is exempt from SIKL's source rules (SA0–SA9) only while its bytes are unchanged. + Committed kernels rarely pass them as new files, so an agent's first change must make the whole + file comply. +- `validate-task` checks, in order, that the pinned commit is in `AKA_SIKL_ROOT`, that `start.json` + is the committed blob, that Triton and FlyDSL match that commit's pins, that the CK submodule is at + its gitlink, and that `AKA_SIKL_ROOT` is not writable by the current user. + +The `sikl-schema` export writes `{"path", "text"}` only for a candidate that passed `validate`, is +new at the pinned commit and is faster (`retain` lower bound above 1.0) than the kernel it would +displace on every workload; the path is a second kernel file under the same name. + +### Trust model + +The SIKL checkout is trusted code: the gate and the runner execute it. The task protects what the +judge reads: the pinned commit, a root-owned `/opt/sikl`, protected `start.json` and `scripts/`, git +without global configuration or hooks, SIKL subprocesses without user site-packages or the agent's +`HOME`, and re-measured performance. It is not a sandbox: the agent runs as the same user, and the +AITER JIT cache is shared across validations as in SIKL CI. SIKL CI, which re-validates an exported +schema in a clean environment, remains the authority. + +## Legacy bundles (SIKL before #52) + +The legacy path turns a SIKL dataset folder into self-contained tasks under `tasks/SIKL-task/`. It +combines deterministic import tools, a Codex agent that reviews generation (and repairs legacy input +adapters), and an independent `task_validator`. Only accepted tasks reach the output directory. +It does not require KernelForge. + +### Input and task boundaries Pass an extracted directory, not a tar file: @@ -94,7 +223,7 @@ top-k routing. Other policies need an input adapter supported by the definition and original code. An unresolved baseline/reference disagreement cannot be repaired by rewriting the reference or relaxing tolerances. -## Run +### Run Inspect metadata on the host; this does not execute solution code, import torch, create tensors, call an agent or use a GPU: @@ -125,7 +254,7 @@ not get an automatic dependency-installation or source-rewriting escape hatch. `target_language` accepts `triton` (default) or `flydsl`. The seeded candidate is declared `implemented`, with the selected baseline's actual `initial_language`. -## Workflow and tools +### Workflow and tools 1. Inspect and join source records; snapshot the bundle, configuration and runtime identity, and compute fingerprints. @@ -183,7 +312,7 @@ inputs, reference self-comparison, and rejection of deliberately incorrect finite outputs for every case. Canonical performance helpers are materialized by the normal workspace machinery. -## Outputs, repair limits and resume +### Outputs, repair limits and resume ```text tasks/SIKL-task// diff --git a/src/scripts/docker_benchmark.sh b/src/scripts/docker_benchmark.sh index eb39b7d9b..53db775ca 100755 --- a/src/scripts/docker_benchmark.sh +++ b/src/scripts/docker_benchmark.sh @@ -1319,7 +1319,8 @@ build_docker_args() { add_mount "$SIKL_BUILDER_CONFIG_HOST" /sikl-config.yaml ro add_mount "$HOST_ROOT/$SIKL_BUILDER_ARTIFACT_REL" "$CONTAINER_WORKDIR/$SIKL_BUILDER_ARTIFACT_REL" add_mount "$HOST_ROOT/$SIKL_BUILDER_OUTPUT_REL" "$CONTAINER_WORKDIR/$SIKL_BUILDER_OUTPUT_REL" - docker_args+=(-e "AGENT_KERNEL_ARENA_IMAGE=$SELECTED_IMAGE") + # SIKL v5 task runners judge against this checkout during the builder's gate. + docker_args+=(-e "AGENT_KERNEL_ARENA_IMAGE=$SELECTED_IMAGE" -e "AKA_SIKL_ROOT=/sikl-input") elif [[ -n "$QUALITY_LOOP_ARTIFACT_REL" || -n "$QUALITY_LOOP_WORKTREE_REL" ]]; then [[ -n "$QUALITY_LOOP_ARTIFACT_REL" && -n "$QUALITY_LOOP_WORKTREE_REL" ]] \ || die "quality_loop requires both artifact and worktree mount paths" @@ -2091,6 +2092,8 @@ case "${1:-}" in SIKL_BUILDER_ARTIFACT_REL="${sikl_mounts[2]}" SIKL_BUILDER_OUTPUT_REL="${sikl_mounts[3]}" REQUIRED_AGENTS="${sikl_mounts[4]}" + # SIKL v5 inputs need no agent CLI; mount-info prints a placeholder for the empty list. + [[ "$REQUIRED_AGENTS" != none ]] || REQUIRED_AGENTS="" AGENTS_STRICT=1 AGENT_HOME_ISOLATION=1 AKA_CONTAINER_HOME="/tmp/aka-sikl-builder-${BASHPID}" diff --git a/tests/test_sikl_task_builder.py b/tests/test_sikl_task_builder.py index dc1446599..e8bb9f99f 100644 --- a/tests/test_sikl_task_builder.py +++ b/tests/test_sikl_task_builder.py @@ -1,4 +1,5 @@ """SIKL import fidelity and acceptance gates; no model or GPU needed.""" +import importlib.util import json import os import subprocess @@ -10,12 +11,16 @@ import yaml from agents.sikl_task_builder.bundle import ImportProblem, inspect_bundle -from agents.sikl_task_builder.config import Config +from agents.sikl_task_builder.checkout import inspect_checkout +from agents.sikl_task_builder.config import CheckoutConfig, Config, load_config from agents.sikl_task_builder.execution import run_process -from agents.sikl_task_builder.materialize import check_contract, materialize_task, task_digest +from agents.sikl_task_builder.materialize import (check_contract, materialize_task, schema_task_files, + task_digest, write_files) from agents.sikl_task_builder.orchestrator import install_task, run from agents.sikl_task_builder import validation +from agents.sikl_task_builder.validation import GATE_ACTIONS, gate_outcome from agents.task_validator.report_schema import finalize_report +from src.task_spec import load_task_spec from tests.test_task_validator_v2 import context as validation_context, draft as validation_draft from agents.task_validator.report_v2 import DRAFT_FILENAME @@ -736,3 +741,209 @@ def test_report_identity_follows_output_suite(): assert Config('bundle', output_dir='tasks/../tasks/imported').arena_task_id('gemm') == 'imported/gemm' root = Path(__file__).resolve().parents[1] assert Config('bundle', output_dir=str(root / 'tasks/imported')).arena_task_id('gemm') == 'imported/gemm' + + +# SIKL v5 checkout input: inspection, materialization, runner verdicts and gate states. + +V5_TEMPLATES = Path(__file__).resolve().parents[1] / "agents/sikl_task_builder/templates/sikl_v5" +SCHEMA = "sikl/ops/gemm/impls/flydsl/gemm_n32_k64/0123456789abcdef.json" +KERNEL = '"""Tiny kernel."""\n\n\ndef run(*, a, b, out):\n pass\n' + + +def v5_schema(**overrides): + value = { + "schema_version": 5, "name": "gemm_n32_k64", "contract_id": "gemm@1.0", "description": "d", + "op_type": "gemm", "axes": {"m": {"type": "var"}}, "attrs": {}, "inputs": {}, "outputs": {}, + "workloads": [{"axes": {"m": 1}, "accuracy": "passed", "speedup": 1.1}, {"axes": {"m": 8}}], + "layout_variants": {}, "compare": {}, + "kernel": {"language": "flydsl", "targets": [{"arch": "gfx950", "hardware": "MI355X"}], + "entry_point": "kernel.py::run", "sources": [{"path": "kernel.py", "content": KERNEL}]}, + } + value.update(overrides) + return value + + +def run_git(root, *args): + return subprocess.run(["git", "-C", str(root), *args], check=True, capture_output=True, text=True).stdout + + +def commit_files(root, files): + for relative, value in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(value if isinstance(value, str) else json.dumps(value, indent=2) + "\n") + run_git(root, "add", "-A") + run_git(root, "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "c") + return run_git(root, "rev-parse", "HEAD").strip() + + +@pytest.fixture +def checkout(tmp_path): + root = tmp_path / "SIKL" + root.mkdir() + run_git(root, "init", "-q") + commit_files(root, {SCHEMA: v5_schema(), "sikl/ops/gemm/contract.py": "x = 1\n"}) + return root + + +def test_inspect_reads_committed_schemas_only(checkout): + (checkout / SCHEMA).write_text("not json") + extra = checkout / "sikl/ops/gemm/impls/flydsl/gemm_n32_k64/fedcba9876543210.json" + extra.write_text(json.dumps(v5_schema())) + _, tasks = inspect_checkout(checkout, []) + assert [t.task_id for t in tasks] == ["gemm_n32_k64/0123456789abcdef"] + assert tasks[0].cases == ["m=1", "m=8"] + assert tasks[0].supports("gfx950", "MI355X") + assert not tasks[0].supports("gfx942", "MI300X") + + +@pytest.mark.parametrize("bad, code", [ + ({"schema_version": 4}, "unsupported_schema_version"), + ({"name": "other"}, "invalid_schema"), + ({"workloads": [{"axes": {"m": 1}}, {"axes": {"m": 1}}]}, "invalid_schema"), + ({"kernel": {**v5_schema()["kernel"], "entry_point": "missing.py::run"}}, "invalid_schema"), + ({"kernel": {**v5_schema()["kernel"], "sources": [{"path": "b.py", "content": "x"}, + {"path": "a.py", "content": "x"}]}}, "invalid_schema"), +]) +def test_inspect_rejects_malformed_schemas(checkout, bad, code): + commit_files(checkout, {SCHEMA: v5_schema(**bad)}) + with pytest.raises(ImportProblem) as error: + inspect_checkout(checkout, []) + assert error.value.code == code + + +def test_inspect_rejects_unknown_selectors(checkout): + assert inspect_checkout(checkout, ["gemm_n32_k64"])[1] + with pytest.raises(ValueError, match="Unknown task selectors"): + inspect_checkout(checkout, ["gemm_n64_k64"]) + + +def test_materialized_task_is_a_valid_arena_task(checkout, tmp_path): + commit, tasks = inspect_checkout(checkout, []) + task = tmp_path / "task" + write_files(task, schema_task_files(tasks[0], CheckoutConfig(input_dir=str(checkout), command_timeout=600), commit)) + spec = load_task_spec(task / "config.yaml", task_id="SIKL-v5/gemm_n32_k64/0123456789abcdef") + assert spec.baseline.source_files == ("start.json",) + assert [e.path for e in spec.candidate.editable] == ["source/kernel.py"] + assert yaml.safe_load((task / "config.yaml").read_text())["platform_support"]["required_arch"] == "gfx950" + assert (task / "start.json").read_bytes() == run_git(checkout, "show", f"HEAD:{SCHEMA}").encode() + assert (task / "source/kernel.py").read_text() == KERNEL + provenance = json.loads((task / "scripts/provenance.json").read_text()) + assert (provenance["sikl_commit"], provenance["schema_path"]) == (commit, SCHEMA) + + +def test_v5_config_rejects_legacy_keys(checkout, tmp_path): + config = tmp_path / "config.yaml" + config.write_text(yaml.safe_dump({"input_dir": str(checkout), "tasks": []})) + assert isinstance(load_config(config), CheckoutConfig) + config.write_text(yaml.safe_dump({"input_dir": str(checkout), "policy": {}})) + with pytest.raises(ImportProblem) as error: + load_config(config) + assert error.value.code == "config_not_v5" + + +@pytest.fixture(scope="module") +def runner(): + spec = importlib.util.spec_from_file_location("sikl_v5_task_runner", V5_TEMPLATES / "task_runner.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def report(findings=(), records=(), status="passed"): + return {"version": 2, "stages": [ + {"stage_name": "audit", "status": "passed", "detail": "", "findings": []}, + {"stage_name": "bench", "status": status, "detail": "bench failed", "findings": list(findings), + "workload_records": list(records)}]} + + +def finding(code, axes=None, detail="d"): + return {"code": code, "axes": axes, "detail": detail} + + +def timing(*ms): + return {"versus": "baseline", "threshold": "faster", "ratio_lower_bound": 1.0, "candidate_milliseconds": list(ms)} + + +START = v5_schema() + + +def test_slow_but_correct_candidate_passes_correctness(runner): + status, cases, _ = runner.judge("correctness", START, report([finding("not_faster", {"m": 8})], status="failed"), 1) + assert status == "PASS" and [c["status"] for c in cases] == ["PASS", "PASS"] + + +def test_wrong_workload_fails_only_that_case(runner): + status, cases, reason = runner.judge("correctness", START, + report([finding("accuracy_failed", {"m": 8}, "max error 3")], status="failed"), 1) + assert status == "FAIL" and [c["status"] for c in cases] == ["PASS", "FAIL"] + assert "m=8" in reason and "max error 3" in reason + + +def test_finding_without_axes_fails_every_case(runner): + status, cases, _ = runner.judge("correctness", START, report([finding("source_rejected", None)], status="failed"), 1) + assert status == "FAIL" and {c["status"] for c in cases} == {"FAIL"} + + +def test_aborted_validation_fails_correctness_and_names_its_findings(runner): + crash = [finding("platform_failure", {"m": m}, "RuntimeError: No CUDA GPUs are available") for m in (1, 8)] + status, cases, reason = runner.judge("correctness", START, report(crash, status="aborted"), 3) + assert status == "FAIL" and {c["status"] for c in cases} == {"FAIL"} and "aborted" in reason + assert reason.count("No CUDA GPUs are available") == 1 + + +@pytest.mark.parametrize("code, detail, status", [ + ("build_failed", "x", "FAIL"), ("source_rejected", "SA0 too deep", "FAIL"), + ("source_rejected", "SA7 mutates input", "PASS"), ("accuracy_failed", "x", "PASS"), +]) +def test_compile_fails_only_on_build_problems(runner, code, detail, status): + assert runner.judge("compile", START, report([finding(code, None, detail)], status="failed"), 1)[0] == status + + +def test_performance_reports_median_baseline_relative_time(runner): + records = [{"axes": {"m": 1}, "timings": [timing(3.0, 1.0, 2.0)]}, + {"axes": {"m": 8}, "timings": [timing(5.0, 4.0)]}] + status, cases, _ = runner.judge("performance", START, report(records=records), 0) + assert status == "PASS" + assert [(c["test_case_id"], c["execution_time_ms"]) for c in cases] == [("m=1", 2.0), ("m=8", 4.5)] + + +def test_performance_fails_when_a_workload_is_untimed(runner): + records = [{"axes": {"m": 1}, "timings": [timing(1.0, 1.0)]}] + status, cases, reason = runner.judge("performance", START, report(records=records), 0) + assert status == "FAIL" and [c["status"] for c in cases] == ["PASS", "FAIL"] and "m=8" in reason + + +def test_performance_fails_when_correctness_fails(runner): + records = [{"axes": {"m": w}, "timings": [timing(1.0, 1.0)]} for w in (1, 8)] + bad = report([finding("accuracy_failed", {"m": 1})], records, status="failed") + assert runner.judge("performance", START, bad, 1)[0] == "FAIL" + + +def test_unknown_report_version_is_refused(runner): + with pytest.raises(runner.Failed, match="report version"): + runner.judge("correctness", START, {**report(), "version": 3}, 0) + + +def gate_commands(failed=(), error=None): + return {key: {"ok": key not in failed, "result": {"status": "FAIL" if key in failed else "PASS"}, + "execution_error": error if key == GATE_ACTIONS[-1] else None} for key in GATE_ACTIONS} + + +PROCESS = {"timed_out": False, "exit_code": 0} + + +@pytest.mark.parametrize("process, cmds, initial, codes, state, code", [ + (PROCESS, gate_commands(), {"accepted": True}, [], "installable", None), + ({"timed_out": True, "exit_code": None}, gate_commands(), None, [], "failed", "gate_timeout"), + (PROCESS, gate_commands(error="exceeded its 60s action deadline"), {"accepted": True}, [], "failed", "gate_timeout"), + (PROCESS, gate_commands(["baseline/correctness"]), {"accepted": False}, ["accuracy_failed"], + "baseline_rejected", "baseline_rejected"), + (PROCESS, gate_commands(["baseline/performance"]), {"accepted": False}, ["platform_failure"], + "failed", "execution_error"), + (PROCESS, gate_commands(["task/validate-task"]), {"accepted": False}, [], "failed", "validate_task_failed"), + (PROCESS, gate_commands(["candidate/performance"]), {"accepted": False}, [], "failed", "candidate_rejected"), + ({"timed_out": False, "exit_code": 1}, gate_commands(), {"accepted": False}, [], "failed", "execution_error"), +]) +def test_gate_outcome_maps_results_to_run_states(process, cmds, initial, codes, state, code): + assert gate_outcome(process, cmds, initial, codes) == {"state": state, "code": code}