diff --git a/README.md b/README.md index a75703fb3..aff59cb10 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,8 @@ yoyo apply src/migrations -d postgresql://user:password@localhost/clusterdev See [docs/database.md](docs/database.md) for migration patterns and creating new migrations. For production incident triage, see [docs/production-debugging.md](docs/production-debugging.md). +For nightly end-to-end kernel checks, see +[docs/application-validation.md](docs/application-validation.md). ### Environment Variables @@ -52,6 +54,7 @@ PROBLEM_DEV_DIR=examples DISABLE_SSL=1 # Set for local development GITHUB_TOKEN_BACKUP= # Fallback token for rate limiting ADMIN_TOKEN= # Token for admin API endpoints +APPLICATION_VALIDATION_ENABLED=true # Discord bot (only needed if testing Discord integration) # See docs/discord.md for setup instructions diff --git a/docs/application-validation.md b/docs/application-validation.md new file mode 100644 index 000000000..46a9b9f4e --- /dev/null +++ b/docs/application-validation.md @@ -0,0 +1,66 @@ +# Application validation + +KernelBot can run a small, problem-owned workload against the current top 10 +submissions. A problem opts in with two fields in `reference-kernels`: + +```yaml +validation: + version: cholesky-natural-gradient-v1 + script: validation.py +``` + +The version invalidates old results when the workload changes. KernelBot loads +the script beside the problem, adds it to the problem's normal source files, +and runs it on B200 through the existing Modal app. + +The script receives the version in `KERNELBOT_VALIDATION_CONFIG` and prints one +JSON result: + +```json +{ + "passed_shapes": 8, + "total_shapes": 8, + "fully_validated": true, + "geomean_sync_wall_speedup": 1.42, + "results": [] +} +``` + +## Schedule + +Every day at 22:00 `America/Los_Angeles`, one KernelBot replica claims each +`(leaderboard, GPU, contract version, local date)` sweep. It snapshots the +current best submission from each of the top 10 users and runs at most two +Modal jobs concurrently. The database claim prevents duplicate sweeps. + +Set `APPLICATION_VALIDATION_ENABLED=false` to disable the scheduler. A failed +job is recorded as `VALIDATION ERROR`; a completed job is stored as `X/Y +VALIDATED`. + +## Local debug + +Modal already reads `MODAL_ENVIRONMENT`, so KernelBot does not need separate +environment plumbing: + +```bash +modal environment create cholesky-validation-debug +modal deploy --env cholesky-validation-debug src/runners/modal_runner_archs.py + +MODAL_ENVIRONMENT=cholesky-validation-debug \ +APPLICATION_VALIDATION_ENABLED=false \ +python src/kernelbot/main.py --api-only --debug +``` + +Run one sweep synchronously: + +```bash +kernelbot-admin --api-url http://localhost:8000 \ + validate-top10 cholesky B200 --wait +``` + +Omit `--wait` to enqueue the sweep and return immediately. The command reads +`ADMIN_TOKEN` and, unless `--api-url` is supplied, +`DISCORD_CLUSTER_MANAGER_API_BASE_URL` from the environment. + +Roll out KernelBot's migration and runner first, then the reference-kernels +contract, then the Kernelboard badge. diff --git a/pyproject.toml b/pyproject.toml index ba778820d..27038f2f6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,9 @@ dev = [ "pytest-asyncio" ] +[project.scripts] +kernelbot-admin = "kernelbot.admin_cli:main" + [tool.setuptools.packages.find] where = ["src"] diff --git a/src/kernelbot/admin_cli.py b/src/kernelbot/admin_cli.py new file mode 100644 index 000000000..8a605d242 --- /dev/null +++ b/src/kernelbot/admin_cli.py @@ -0,0 +1,62 @@ +import argparse +import json +import os +from urllib.parse import quote + +import requests + + +def _validate_top10(args: argparse.Namespace) -> int: + api_url = args.api_url or os.getenv("DISCORD_CLUSTER_MANAGER_API_BASE_URL") + token = os.getenv("ADMIN_TOKEN") + if not api_url: + raise SystemExit( + "Set DISCORD_CLUSTER_MANAGER_API_BASE_URL or pass --api-url." + ) + if not token: + raise SystemExit("Set ADMIN_TOKEN.") + + path = "/admin/application-validations/{}/{}".format( + quote(args.leaderboard, safe=""), + quote(args.gpu, safe=""), + ) + response = requests.post( + f"{api_url.rstrip('/')}{path}", + headers={"Authorization": f"Bearer {token}"}, + params={"wait": str(args.wait).lower()}, + timeout=None if args.wait else 30, + ) + response.raise_for_status() + print(json.dumps(response.json(), indent=2, sort_keys=True)) + return 0 + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="kernelbot-admin") + parser.add_argument( + "--api-url", + help="Kernelbot API URL (defaults to DISCORD_CLUSTER_MANAGER_API_BASE_URL)", + ) + commands = parser.add_subparsers(required=True) + validate = commands.add_parser( + "validate-top10", + help="Run application validation for a leaderboard's current top 10", + ) + validate.add_argument("leaderboard") + validate.add_argument("gpu") + validate.add_argument( + "--wait", + action="store_true", + help="Wait for all validation jobs and print their results", + ) + validate.set_defaults(run=_validate_top10) + return parser + + +def main() -> int: + args = _parser().parse_args() + return args.run(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/kernelbot/api/main.py b/src/kernelbot/api/main.py index 7644faea3..e68d04510 100644 --- a/src/kernelbot/api/main.py +++ b/src/kernelbot/api/main.py @@ -12,6 +12,7 @@ from fastapi.responses import JSONResponse, StreamingResponse from kernelbot.env import env +from libkernelbot.application_validation import ApplicationValidationService from libkernelbot.backend import KernelBackend from libkernelbot.background_submission_manager import BackgroundSubmissionManager from libkernelbot.consts import SubmissionMode @@ -54,6 +55,7 @@ def json_serializer(obj): backend_instance: KernelBackend = None background_submission_manager: BackgroundSubmissionManager = None +application_validation_service: ApplicationValidationService = None _last_action = time.time() _submit_limiter = asyncio.Semaphore(3) @@ -89,6 +91,12 @@ def init_background_submission_manager(_manager: BackgroundSubmissionManager): return background_submission_manager +def init_application_validation_service(_service: ApplicationValidationService): + global application_validation_service + application_validation_service = _service + return application_validation_service + + @app.exception_handler(KernelBotError) async def kernel_bot_error_handler(req: Request, exc: KernelBotError): return JSONResponse(status_code=exc.http_code, content={"message": str(exc)}) @@ -477,6 +485,35 @@ async def admin_unban_user( return {"status": "ok", "user_id": user_id, "banned": False} +@app.post("/admin/application-validations/{leaderboard_name}/{gpu_type}") +async def admin_run_application_validation( + leaderboard_name: str, + gpu_type: str, + _: Annotated[None, Depends(require_admin)], + wait: bool = Query(False), +) -> dict: + if application_validation_service is None: + raise HTTPException( + status_code=503, + detail="Application validation service is not initialized", + ) + if wait: + return await application_validation_service.run_sweep( + leaderboard_name, + gpu_type, + scheduled_for=None, + ) + application_validation_service.enqueue_manual_sweep( + leaderboard_name, + gpu_type, + ) + return { + "status": "accepted", + "leaderboard": leaderboard_name, + "gpu_type": gpu_type, + } + + @app.post("/{leaderboard_name}/{gpu_type}/{submission_mode}") async def run_submission( # noqa: C901 leaderboard_name: str, diff --git a/src/kernelbot/env.py b/src/kernelbot/env.py index 0c3d11f14..1d9456609 100644 --- a/src/kernelbot/env.py +++ b/src/kernelbot/env.py @@ -42,6 +42,10 @@ # PostgreSQL-specific constants env.DATABASE_URL = os.getenv("DATABASE_URL") env.DISABLE_SSL = os.getenv("DISABLE_SSL") +env.APPLICATION_VALIDATION_ENABLED = os.getenv( + "APPLICATION_VALIDATION_ENABLED", + "true", +).lower() in {"1", "true", "yes"} def init_environment(skip_discord: bool = False): diff --git a/src/kernelbot/main.py b/src/kernelbot/main.py index 398d17166..64de6b262 100644 --- a/src/kernelbot/main.py +++ b/src/kernelbot/main.py @@ -4,7 +4,12 @@ import discord import uvicorn -from api.main import app, init_api, init_background_submission_manager +from api.main import ( + app, + init_api, + init_application_validation_service, + init_background_submission_manager, +) from cogs.admin_cog import AdminCog from cogs.leaderboard_cog import LeaderboardCog from cogs.misc_cog import BotManagerCog @@ -15,6 +20,7 @@ from env import env, init_environment from libkernelbot import consts +from libkernelbot.application_validation import ApplicationValidationService from libkernelbot.backend import KernelBackend from libkernelbot.background_submission_manager import BackgroundSubmissionManager from libkernelbot.launchers import GitHubLauncher, ModalLauncher @@ -50,11 +56,19 @@ async def run_api_server(backend: KernelBackend): init_api(backend) manager = init_background_submission_manager(BackgroundSubmissionManager(backend)) await manager.start() + validation_service = init_application_validation_service( + ApplicationValidationService( + backend, + enabled=env.APPLICATION_VALIDATION_ENABLED, + ) + ) + await validation_service.start() server = create_uvicorn_server() try: await server.serve() finally: + await validation_service.stop() await manager.stop() @@ -241,9 +255,9 @@ async def start_bot(self, token: str): raise e -async def start_api_only(): +async def start_api_only(debug_mode: bool = False): """Start only the FastAPI server without Discord bot.""" - backend = create_backend(debug_mode=False) + backend = create_backend(debug_mode=debug_mode) await run_api_server(backend) @@ -258,6 +272,13 @@ async def start_bot_and_api(debug_mode: bool): init_api(bot_instance.backend) manager = init_background_submission_manager(BackgroundSubmissionManager(bot_instance.backend)) await manager.start() + validation_service = init_application_validation_service( + ApplicationValidationService( + bot_instance.backend, + enabled=env.APPLICATION_VALIDATION_ENABLED, + ) + ) + await validation_service.start() server = create_uvicorn_server() try: @@ -266,6 +287,7 @@ async def start_bot_and_api(debug_mode: bool): server.serve(), ) finally: + await validation_service.stop() await manager.stop() def on_unhandled_exception(loop, context): @@ -283,9 +305,9 @@ def main(): if args.api_only: logger.info("Starting API server only (no Discord bot)...") - with asyncio.Runner() as runner: + with asyncio.Runner(debug=args.debug) as runner: runner.get_loop().set_exception_handler(on_unhandled_exception) - runner.run(start_api_only()) + runner.run(start_api_only(debug_mode=args.debug)) else: logger.info("Starting kernelbot and API server...") with asyncio.Runner(debug=args.debug) as runner: diff --git a/src/libkernelbot/application_validation.py b/src/libkernelbot/application_validation.py new file mode 100644 index 000000000..203a23dfe --- /dev/null +++ b/src/libkernelbot/application_validation.py @@ -0,0 +1,301 @@ +"""Nightly application-level validation for ranked kernel submissions.""" + +from __future__ import annotations + +import asyncio +import contextlib +import datetime +import math +from typing import Any +from zoneinfo import ZoneInfo + +from libkernelbot.consts import get_gpu_by_name +from libkernelbot.task import LeaderboardTask, build_validation_config +from libkernelbot.utils import KernelBotError, setup_logging + +logger = setup_logging(__name__) + +SCHEDULE = datetime.time(22, 0) +TIMEZONE = ZoneInfo("America/Los_Angeles") +TOP_K = 10 +MAX_CONCURRENCY = 2 + + +def _due_date(now: datetime.datetime) -> datetime.date | None: + if now.tzinfo is None: + now = now.replace(tzinfo=datetime.timezone.utc) + local_now = now.astimezone(TIMEZONE) + if local_now.time() < SCHEDULE: + return None + return local_now.date() + + +class ApplicationValidationService: + def __init__(self, backend, *, enabled: bool = False, poll_seconds: int = 60): + self.backend = backend + self.enabled = enabled + self.poll_seconds = poll_seconds + self._scheduler_task: asyncio.Task | None = None + self._manual_tasks: set[asyncio.Task] = set() + + async def start(self) -> None: + if not self.enabled or self._scheduler_task is not None: + return + logger.info("Starting application validation scheduler") + self._scheduler_task = asyncio.create_task( + self._scheduler_loop(), + name="application-validation-scheduler", + ) + + async def stop(self) -> None: + tasks = list(self._manual_tasks) + if self._scheduler_task is not None: + self._scheduler_task.cancel() + tasks.append(self._scheduler_task) + self._scheduler_task = None + for task in tasks: + task.cancel() + for task in tasks: + with contextlib.suppress(asyncio.CancelledError): + await task + self._manual_tasks.clear() + + async def _scheduler_loop(self) -> None: + while True: + try: + await self.run_due_once() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Application validation scheduler iteration failed") + await asyncio.sleep(self.poll_seconds) + + async def run_due_once( + self, + now: datetime.datetime | None = None, + ) -> list[dict[str, Any]]: + scheduled_for = _due_date( + now or datetime.datetime.now(datetime.timezone.utc) + ) + if scheduled_for is None: + return [] + + with self.backend.db as db: + leaderboards = db.get_leaderboards() + + summaries = [] + for leaderboard in leaderboards: + if leaderboard["task"].validation is None: + continue + for gpu_type in leaderboard["gpu_types"]: + summaries.append( + await self.run_sweep( + leaderboard["name"], + gpu_type, + scheduled_for=scheduled_for, + ) + ) + return summaries + + def enqueue_manual_sweep( + self, + leaderboard_name: str, + gpu_type: str, + ) -> None: + task = asyncio.create_task( + self.run_sweep(leaderboard_name, gpu_type, scheduled_for=None), + name=f"manual-application-validation-{leaderboard_name}-{gpu_type}", + ) + self._manual_tasks.add(task) + task.add_done_callback(self._manual_task_done) + + def _manual_task_done(self, task: asyncio.Task) -> None: + self._manual_tasks.discard(task) + if not task.cancelled() and task.exception() is not None: + logger.error( + "Manual application validation sweep failed", + exc_info=task.exception(), + ) + + async def _validate_submission( + self, + entry: dict, + *, + task: LeaderboardTask, + gpu_type: str, + semaphore: asyncio.Semaphore, + ) -> dict[str, Any]: + submission_id = entry["submission_id"] + contract_version = task.validation.version + async with semaphore: + try: + with self.backend.db as db: + source = db.get_submission_code_for_validation(submission_id) + config = build_validation_config(task, source) + + gpu = get_gpu_by_name(gpu_type) + if gpu is None: + raise KernelBotError(f"Unknown GPU type {gpu_type!r}") + launcher = self.backend.launcher_map.get(gpu.value) + if launcher is None: + raise KernelBotError( + f"No validation launcher is registered for {gpu_type!r}" + ) + response = await launcher.run_validation(config, gpu) + status = ( + "completed" + if response.get("status") == "completed" + else "failed" + ) + result = response.get("result", {}) + if not isinstance(result, dict): + raise ValueError("validator result must be an object") + + total_shapes = int(result.get("total_shapes", 0)) + passed_shapes = int(result.get("passed_shapes", 0)) + if status == "completed" and ( + total_shapes <= 0 + or not 0 <= passed_shapes <= total_shapes + ): + raise ValueError("validator returned invalid shape counts") + + speedup = result.get("geomean_sync_wall_speedup") + if speedup is not None: + speedup = float(speedup) + if not math.isfinite(speedup) or speedup <= 0: + raise ValueError("validator returned an invalid speedup") + fully_validated = bool( + status == "completed" + and passed_shapes == total_shapes + and result.get("fully_validated") + ) + + with self.backend.db as db: + db.upsert_submission_validation( + submission_id=submission_id, + gpu_type=gpu_type, + contract_version=contract_version, + status=status, + passed_shapes=passed_shapes, + total_shapes=total_shapes, + fully_validated=fully_validated, + geomean_sync_wall_speedup=speedup, + result=result, + error=response.get("error"), + ) + return { + "submission_id": submission_id, + "status": status, + "passed_shapes": passed_shapes, + "total_shapes": total_shapes, + "fully_validated": fully_validated, + "geomean_sync_wall_speedup": speedup, + } + except Exception as exc: + error = f"{type(exc).__name__}: {exc}"[:1000] + logger.warning( + "Application validation failed for submission=%s: %s", + submission_id, + error, + ) + with self.backend.db as db: + db.upsert_submission_validation( + submission_id=submission_id, + gpu_type=gpu_type, + contract_version=contract_version, + status="failed", + passed_shapes=0, + total_shapes=0, + fully_validated=False, + geomean_sync_wall_speedup=None, + result={}, + error=error, + ) + return { + "submission_id": submission_id, + "status": "failed", + "passed_shapes": 0, + "total_shapes": 0, + "fully_validated": False, + } + + async def run_sweep( + self, + leaderboard_name: str, + gpu_type: str, + *, + scheduled_for: datetime.date | None, + ) -> dict[str, Any]: + with self.backend.db as db: + leaderboard = db.get_leaderboard(leaderboard_name) + validation = leaderboard["task"].validation + if validation is None: + raise KernelBotError( + f"Leaderboard {leaderboard_name!r} has no application validation", + code=400, + ) + if gpu_type not in leaderboard["gpu_types"]: + raise KernelBotError( + f"GPU {gpu_type!r} is not configured for {leaderboard_name!r}", + code=400, + ) + sweep_id = db.claim_validation_sweep( + leaderboard_id=leaderboard["id"], + gpu_type=gpu_type, + contract_version=validation.version, + scheduled_for=scheduled_for, + ) + if sweep_id is None: + return { + "status": "already_claimed", + "leaderboard": leaderboard_name, + "gpu_type": gpu_type, + "scheduled_for": scheduled_for, + } + submissions = db.get_leaderboard_submissions( + leaderboard_name, + gpu_type, + limit=TOP_K, + ) + + logger.info( + "Running application validation sweep %s for leaderboard=%s gpu=%s submissions=%s", + sweep_id, + leaderboard_name, + gpu_type, + len(submissions), + ) + semaphore = asyncio.Semaphore(MAX_CONCURRENCY) + + try: + results = await asyncio.gather( + *( + self._validate_submission( + entry, + task=leaderboard["task"], + gpu_type=gpu_type, + semaphore=semaphore, + ) + for entry in submissions + ) + ) + with self.backend.db as db: + db.complete_validation_sweep(sweep_id, status="completed") + return { + "status": "completed", + "sweep_id": sweep_id, + "leaderboard": leaderboard_name, + "gpu_type": gpu_type, + "scheduled_for": scheduled_for, + "contract_version": validation.version, + "results": results, + } + except Exception as exc: + error = f"{type(exc).__name__}: {exc}"[:1000] + with self.backend.db as db: + db.complete_validation_sweep( + sweep_id, + status="failed", + error=error, + ) + raise diff --git a/src/libkernelbot/launchers/launcher.py b/src/libkernelbot/launchers/launcher.py index 33ef7757e..9643efa1f 100644 --- a/src/libkernelbot/launchers/launcher.py +++ b/src/libkernelbot/launchers/launcher.py @@ -28,6 +28,9 @@ def __init__(self, name: str, gpus: Type[Enum]): async def run_submission(self, config: dict, gpu_type: GPU, status: RunProgressReporter): raise NotImplementedError() + async def run_validation(self, config: dict, gpu_type: GPU) -> dict: + raise NotImplementedError(f"{self.name} does not support application validation") + async def get_queue_status( self, gpu_type: GPU, config: dict | None = None ) -> RunnerQueueStatus: diff --git a/src/libkernelbot/launchers/modal.py b/src/libkernelbot/launchers/modal.py index 1512dfbf8..ce7ae06ee 100644 --- a/src/libkernelbot/launchers/modal.py +++ b/src/libkernelbot/launchers/modal.py @@ -35,6 +35,16 @@ async def run_submission( return result + async def run_validation(self, config: dict, gpu_type: GPU) -> dict: + func_name = f"run_validation_script_{gpu_type.value.lower()}" + logger.info( + "Starting Modal application validation using %s for contract %s", + func_name, + config.get("version"), + ) + function = modal.Function.from_name("discord-bot-runner", func_name) + return await function.remote.aio(config=config) + def _function_name(self, config: dict, gpu_type: GPU) -> str: func_type = "pytorch" if config["lang"] == "py" else "cuda" return f"run_{func_type}_script_{gpu_type.value.lower()}" diff --git a/src/libkernelbot/leaderboard_db.py b/src/libkernelbot/leaderboard_db.py index 2ed15acb3..06357e7b3 100644 --- a/src/libkernelbot/leaderboard_db.py +++ b/src/libkernelbot/leaderboard_db.py @@ -880,6 +880,139 @@ def set_leaderboard_visibility(self, leaderboard_name: str, visibility: str): logger.exception("Error setting leaderboard visibility", exc_info=e) raise KernelBotError("Error setting leaderboard visibility") from e + def claim_validation_sweep( + self, + *, + leaderboard_id: int, + gpu_type: str, + contract_version: str, + scheduled_for: Optional[datetime.date], + ) -> Optional[int]: + """Atomically claim a nightly validation sweep. + + ``scheduled_for=None`` is reserved for explicit admin-triggered sweeps and + intentionally permits more than one run. + """ + try: + self.cursor.execute( + """ + INSERT INTO leaderboard.validation_sweep ( + leaderboard_id, gpu_type, contract_version, scheduled_for, + status + ) + VALUES (%s, %s, %s, %s, 'running') + ON CONFLICT (leaderboard_id, gpu_type, contract_version, scheduled_for) + DO NOTHING + RETURNING id + """, + (leaderboard_id, gpu_type, contract_version, scheduled_for), + ) + row = self.cursor.fetchone() + self.connection.commit() + return row[0] if row is not None else None + except psycopg2.Error as e: + self.connection.rollback() + logger.exception("Error claiming application validation sweep", exc_info=e) + raise KernelBotError("Could not claim application validation sweep") from e + + def complete_validation_sweep( + self, + sweep_id: int, + *, + status: str, + error: Optional[str] = None, + ) -> None: + if status not in {"completed", "failed"}: + raise ValueError("validation sweep status must be completed or failed") + try: + self.cursor.execute( + """ + UPDATE leaderboard.validation_sweep + SET status = %s, error = %s, completed_at = NOW() + WHERE id = %s + """, + (status, error, sweep_id), + ) + self.connection.commit() + except psycopg2.Error as e: + self.connection.rollback() + logger.exception("Error completing application validation sweep", exc_info=e) + raise KernelBotError("Could not complete application validation sweep") from e + + def get_submission_code_for_validation(self, submission_id: int) -> str: + """Fetch submitted source for an isolated validation runner. + + Callers must not log the returned value. + """ + self.cursor.execute( + """ + SELECT c.code + FROM leaderboard.submission s + JOIN leaderboard.code_files c ON c.id = s.code_id + WHERE s.id = %s + """, + (submission_id,), + ) + row = self.cursor.fetchone() + if row is None: + raise KernelBotError(f"Submission {submission_id} does not exist", code=404) + return bytes(row[0]).decode("utf-8") + + def upsert_submission_validation( + self, + *, + submission_id: int, + gpu_type: str, + contract_version: str, + status: str, + passed_shapes: int, + total_shapes: int, + fully_validated: bool, + geomean_sync_wall_speedup: Optional[float], + result: dict, + error: Optional[str] = None, + ) -> None: + if status not in {"completed", "failed"}: + raise ValueError("submission validation status must be completed or failed") + try: + self.cursor.execute( + """ + INSERT INTO leaderboard.submission_validation ( + submission_id, gpu_type, contract_version, + status, passed_shapes, total_shapes, fully_validated, + geomean_sync_wall_speedup, result, error + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s) + ON CONFLICT (submission_id, gpu_type, contract_version) + DO UPDATE SET + status = EXCLUDED.status, + passed_shapes = EXCLUDED.passed_shapes, + total_shapes = EXCLUDED.total_shapes, + fully_validated = EXCLUDED.fully_validated, + geomean_sync_wall_speedup = EXCLUDED.geomean_sync_wall_speedup, + result = EXCLUDED.result, + error = EXCLUDED.error, + checked_at = NOW() + """, + ( + submission_id, + gpu_type, + contract_version, + status, + passed_shapes, + total_shapes, + fully_validated, + geomean_sync_wall_speedup, + json.dumps(result), + error, + ), + ) + self.connection.commit() + except psycopg2.Error as e: + self.connection.rollback() + logger.exception("Error saving submission application validation", exc_info=e) + raise KernelBotError("Could not save submission application validation") from e + def get_leaderboard_submissions( self, leaderboard_name: str, @@ -1025,7 +1158,6 @@ def get_leaderboard_submissions( raise KernelBotError( f"Invalid GPU type '{gpu_name}' for leaderboard '{leaderboard_name}'" ) - return result def generate_stats(self, last_day: bool, leaderboard_name: Optional[str] = None): diff --git a/src/libkernelbot/task.py b/src/libkernelbot/task.py index e0fb0a410..ddd819451 100644 --- a/src/libkernelbot/task.py +++ b/src/libkernelbot/task.py @@ -27,6 +27,16 @@ class PythonTaskData: TestCaseType = Dict[str, Union[int, str]] +@dataclasses.dataclass +class ApplicationValidation: + version: str + source: str + + def __post_init__(self): + if not self.version or not self.source: + raise ValueError("validation version and source must be non-empty") + + @dataclasses.dataclass class LeaderboardTask: """ @@ -62,6 +72,7 @@ class LeaderboardTask: ranking_by: RankCriterion = RankCriterion.LAST seed: Optional[int] = None multi_gpu: bool = False + validation: Optional[ApplicationValidation] = None def __post_init__(self): if self.lang == Language.Python and not isinstance(self.config, PythonTaskData): @@ -77,6 +88,8 @@ def from_dict(cls, data: dict): data_["lang"] = lang data_["ranking_by"] = criterion data_["multi_gpu"] = data.get("multi_gpu", False) + if data.get("validation") is not None: + data_["validation"] = ApplicationValidation(**data["validation"]) if lang == Language.Python: data_["config"] = PythonTaskData(**data["config"]) else: @@ -143,6 +156,10 @@ def make_task_definition(yaml_file: str | Path) -> LeaderboardDefinition: # noq raw["files"] = file_dict + validation = raw.get("validation") + if validation is not None: + validation["source"] = (root / validation.pop("script")).read_text() + # load template files templates = {} for lang, source in raw.get("templates", {}).items(): @@ -219,3 +236,23 @@ def build_task_config( "include_dirs": task.config.include_dirs, **common, } + + +def build_validation_config( + task: LeaderboardTask, + submission_content: str, +) -> dict: + validation = task.validation + if validation is None: + raise KernelBotError("leaderboard does not define application validation") + sources = { + name: submission_content if content == "@SUBMISSION@" else content + for name, content in task.files.items() + } + sources["validation.py"] = validation.source + return { + "version": validation.version, + "main": "validation.py", + "sources": sources, + "timeout": 900, + } diff --git a/src/libkernelbot/validation_runtime.py b/src/libkernelbot/validation_runtime.py new file mode 100644 index 000000000..cff912a3b --- /dev/null +++ b/src/libkernelbot/validation_runtime.py @@ -0,0 +1,122 @@ +"""Isolated runtime for problem-owned application validation jobs.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +MAX_RESULT_BYTES = 1024 * 1024 + + +def _safe_destination(root: Path, name: str) -> Path: + destination = (root / name).resolve() + if destination == root or root not in destination.parents: + raise ValueError(f"validation source path escapes workspace: {name!r}") + return destination + + +def _read_tail(path: Path, limit: int = MAX_RESULT_BYTES) -> str: + with path.open("rb") as handle: + handle.seek(0, os.SEEK_END) + size = handle.tell() + handle.seek(max(0, size - limit)) + return handle.read().decode("utf-8", errors="replace") + + +def _parse_result(stdout_path: Path, returncode: int) -> dict: + output_lines = [ + line.strip() + for line in _read_tail(stdout_path).splitlines() + if line.strip() + ] + if not output_lines: + return { + "status": "failed", + "error": f"validation exited {returncode} without a result", + } + try: + result = json.loads(output_lines[-1]) + except json.JSONDecodeError: + return { + "status": "failed", + "error": f"validation exited {returncode} without valid JSON", + } + if not isinstance(result, dict): + return { + "status": "failed", + "error": "validation result must be a JSON object", + } + if returncode != 0 or result.get("status") == "failed": + return { + "status": "failed", + "error": result.get( + "failure_reason", + f"validation exited {returncode}", + ), + } + return {"status": "completed", "result": result} + + +def run_validation_config(config: dict) -> dict: + """Execute one validation entrypoint without exposing submission output.""" + sources = config.get("sources") + main = config.get("main") + timeout = int(config.get("timeout", 900)) + if not isinstance(sources, dict) or not sources: + return {"status": "failed", "error": "validation sources are missing"} + if not isinstance(main, str) or main not in sources: + return {"status": "failed", "error": "validation main source is missing"} + if timeout <= 0: + return {"status": "failed", "error": "validation timeout must be positive"} + + public_config = { + key: value + for key, value in config.items() + if key not in {"sources", "main", "timeout"} + } + try: + with tempfile.TemporaryDirectory(prefix="kernelbot-validation-") as directory: + root = Path(directory).resolve() + for name, content in sources.items(): + destination = _safe_destination(root, name) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(content) + + stdout_path = root / "stdout.txt" + stderr_path = root / "stderr.txt" + env = os.environ.copy() + env.update( + { + "KERNELBOT_VALIDATION_CONFIG": json.dumps(public_config), + "PYTHONUNBUFFERED": "1", + "TORCH_EXTENSIONS_DIR": str(root / "torch-extensions"), + "TRITON_CACHE_DIR": str(root / "triton-cache"), + } + ) + with stdout_path.open("wb") as stdout, stderr_path.open("wb") as stderr: + try: + completed = subprocess.run( + [sys.executable, str(_safe_destination(root, main))], + cwd=root, + env=env, + stdout=stdout, + stderr=stderr, + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired: + return { + "status": "failed", + "error": f"validation timed out after {timeout} seconds", + } + + return _parse_result(stdout_path, completed.returncode) + except Exception as exc: + return { + "status": "failed", + "error": f"{type(exc).__name__}: {exc}", + } diff --git a/src/migrations/20260726_01_application-validation.py b/src/migrations/20260726_01_application-validation.py new file mode 100644 index 000000000..6d9c45725 --- /dev/null +++ b/src/migrations/20260726_01_application-validation.py @@ -0,0 +1,54 @@ +""" +Persist problem-owned application validation results for leaderboard submissions. +""" + +from yoyo import step + +__depends__ = {"20260319_01_allow-zero-rate-limits"} + +steps = [ + step( + """ + CREATE TABLE leaderboard.validation_sweep ( + id BIGSERIAL PRIMARY KEY, + leaderboard_id INTEGER NOT NULL + REFERENCES leaderboard.leaderboard(id) ON DELETE CASCADE, + gpu_type TEXT NOT NULL, + contract_version TEXT NOT NULL, + scheduled_for DATE, + status TEXT NOT NULL + CHECK (status IN ('running', 'completed', 'failed')), + started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + error TEXT, + UNIQUE (leaderboard_id, gpu_type, contract_version, scheduled_for) + ); + + CREATE TABLE leaderboard.submission_validation ( + id BIGSERIAL PRIMARY KEY, + submission_id INTEGER NOT NULL + REFERENCES leaderboard.submission(id) ON DELETE CASCADE, + gpu_type TEXT NOT NULL, + contract_version TEXT NOT NULL, + status TEXT NOT NULL + CHECK (status IN ('completed', 'failed')), + passed_shapes INTEGER NOT NULL DEFAULT 0 CHECK (passed_shapes >= 0), + total_shapes INTEGER NOT NULL DEFAULT 0 CHECK (total_shapes >= 0), + fully_validated BOOLEAN NOT NULL DEFAULT FALSE, + geomean_sync_wall_speedup DOUBLE PRECISION, + result JSONB NOT NULL DEFAULT '{}'::jsonb, + error TEXT, + checked_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (submission_id, gpu_type, contract_version) + ); + + CREATE INDEX submission_validation_latest_idx + ON leaderboard.submission_validation + (submission_id, gpu_type, checked_at DESC); + """, + """ + DROP TABLE leaderboard.submission_validation; + DROP TABLE leaderboard.validation_sweep; + """, + ) +] diff --git a/src/runners/modal_runner_archs.py b/src/runners/modal_runner_archs.py index e768020ca..55f025a0a 100644 --- a/src/runners/modal_runner_archs.py +++ b/src/runners/modal_runner_archs.py @@ -2,6 +2,8 @@ # Modal apps on specific devices. We will fix this later. from modal_runner import MODAL_RUN_TIMEOUT_SECONDS, app, cuda_image, modal_run_config +from libkernelbot.validation_runtime import run_validation_config + gpus = ["T4", "L4", "L4:4", "A100-80GB", "H100!", "B200"] for gpu in gpus: gpu_slug = gpu.lower().split("-")[0].strip("!").replace(":", "x") @@ -19,3 +21,11 @@ serialized=True, timeout=MODAL_RUN_TIMEOUT_SECONDS, )(modal_run_config) + +app.function( + gpu="B200", + image=cuda_image, + name="run_validation_script_b200", + serialized=True, + timeout=MODAL_RUN_TIMEOUT_SECONDS, +)(run_validation_config) diff --git a/tests/conftest.py b/tests/conftest.py index 9e17d480c..db5666a68 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -48,7 +48,8 @@ def docker_compose(project_root: Path): def _nuke_contents(db): db.cursor.execute( - "TRUNCATE leaderboard.code_files, leaderboard.submission, leaderboard.runs, " + "TRUNCATE leaderboard.submission_validation, leaderboard.validation_sweep, " + "leaderboard.code_files, leaderboard.submission, leaderboard.runs, " "leaderboard.leaderboard, leaderboard.user_info, leaderboard.templates, " "leaderboard.gpu_type, leaderboard.leaderboard_invite_scope, " "leaderboard.leaderboard_invite, leaderboard.rate_limit RESTART IDENTITY CASCADE" diff --git a/tests/test_admin_api.py b/tests/test_admin_api.py index 485f1ce53..5b2feb47c 100644 --- a/tests/test_admin_api.py +++ b/tests/test_admin_api.py @@ -26,13 +26,32 @@ def mock_background_manager(): @pytest.fixture -def test_client(mock_backend, mock_background_manager): +def mock_validation_service(): + service = MagicMock() + service.run_sweep = AsyncMock( + return_value={ + "status": "completed", + "leaderboard": "cholesky", + "gpu_type": "B200", + } + ) + return service + + +@pytest.fixture +def test_client(mock_backend, mock_background_manager, mock_validation_service): """Create a test client with mocked backend.""" # Patch env before importing the app with patch.dict('os.environ', {'ADMIN_TOKEN': 'test_token'}): - from kernelbot.api.main import app, init_api, init_background_submission_manager + from kernelbot.api.main import ( + app, + init_api, + init_application_validation_service, + init_background_submission_manager, + ) init_api(mock_backend) init_background_submission_manager(mock_background_manager) + init_application_validation_service(mock_validation_service) yield TestClient(app) @@ -91,6 +110,46 @@ def test_admin_stop(self, test_client, mock_backend): assert mock_backend.accepts_jobs is False +class TestAdminApplicationValidation: + def test_manual_validation_can_wait_for_debug_result( + self, + test_client, + mock_validation_service, + ): + response = test_client.post( + "/admin/application-validations/cholesky/B200?wait=true", + headers={"Authorization": "Bearer test_token"}, + ) + + assert response.status_code == 200 + assert response.json()["status"] == "completed" + mock_validation_service.run_sweep.assert_awaited_once_with( + "cholesky", + "B200", + scheduled_for=None, + ) + + def test_manual_validation_is_async_by_default( + self, + test_client, + mock_validation_service, + ): + response = test_client.post( + "/admin/application-validations/cholesky/B200", + headers={"Authorization": "Bearer test_token"}, + ) + + assert response.status_code == 200 + assert response.json() == { + "status": "accepted", + "leaderboard": "cholesky", + "gpu_type": "B200", + } + mock_validation_service.enqueue_manual_sweep.assert_called_once_with( + "cholesky", + "B200", + ) + class TestRunnerQueue: def test_get_runner_queue(self, test_client, mock_backend): """GET /runner_queue/{gpu_type} returns runner backlog.""" diff --git a/tests/test_admin_cli.py b/tests/test_admin_cli.py new file mode 100644 index 000000000..ce784d6b1 --- /dev/null +++ b/tests/test_admin_cli.py @@ -0,0 +1,78 @@ +from unittest.mock import Mock, patch + +import pytest + +from kernelbot import admin_cli + + +def test_validate_top10_enqueues_current_top_ten(capsys): + response = Mock() + response.json.return_value = { + "status": "accepted", + "leaderboard": "cholesky", + "gpu_type": "B200", + } + + with ( + patch.dict( + "os.environ", + { + "DISCORD_CLUSTER_MANAGER_API_BASE_URL": "https://kernelbot.test/", + "ADMIN_TOKEN": "secret", + }, + ), + patch("kernelbot.admin_cli.requests.post", return_value=response) as post, + patch("sys.argv", ["kernelbot-admin", "validate-top10", "cholesky", "B200"]), + ): + assert admin_cli.main() == 0 + + post.assert_called_once_with( + "https://kernelbot.test/admin/application-validations/cholesky/B200", + headers={"Authorization": "Bearer secret"}, + params={"wait": "false"}, + timeout=30, + ) + response.raise_for_status.assert_called_once_with() + assert '"status": "accepted"' in capsys.readouterr().out + + +def test_validate_top10_can_wait_and_url_encodes_names(): + response = Mock() + response.json.return_value = {"status": "completed", "results": []} + + with ( + patch.dict("os.environ", {"ADMIN_TOKEN": "secret"}, clear=True), + patch("kernelbot.admin_cli.requests.post", return_value=response) as post, + patch( + "sys.argv", + [ + "kernelbot-admin", + "--api-url", + "https://kernelbot.test", + "validate-top10", + "batched cholesky", + "B200", + "--wait", + ], + ), + ): + assert admin_cli.main() == 0 + + assert post.call_args.kwargs["params"] == {"wait": "true"} + assert post.call_args.kwargs["timeout"] is None + assert post.call_args.args[0].endswith( + "/application-validations/batched%20cholesky/B200" + ) + + +def test_validate_top10_requires_admin_token(): + with ( + patch.dict( + "os.environ", + {"DISCORD_CLUSTER_MANAGER_API_BASE_URL": "https://kernelbot.test"}, + clear=True, + ), + patch("sys.argv", ["kernelbot-admin", "validate-top10", "cholesky", "B200"]), + pytest.raises(SystemExit, match="Set ADMIN_TOKEN"), + ): + admin_cli.main() diff --git a/tests/test_application_validation.py b/tests/test_application_validation.py new file mode 100644 index 000000000..1002be7b9 --- /dev/null +++ b/tests/test_application_validation.py @@ -0,0 +1,161 @@ +import asyncio +import datetime +from types import SimpleNamespace + +import pytest + +from libkernelbot.application_validation import ( + ApplicationValidationService, + _due_date, +) +from libkernelbot.task import LeaderboardTask + + +def _task() -> LeaderboardTask: + return LeaderboardTask.from_dict( + { + "lang": "py", + "files": {"submission.py": "@SUBMISSION@"}, + "config": {"main": "submission.py"}, + "validation": { + "version": "v1", + "source": "print('validator')", + }, + } + ) + + +class FakeDB: + def __init__(self): + self.task = _task() + self.leaderboard = { + "id": 7, + "name": "cholesky", + "task": self.task, + "gpu_types": ["B200"], + } + self.submissions = [ + { + "submission_id": submission_id, + "submission_name": f"submission-{submission_id}.py", + } + for submission_id in range(1, 13) + ] + self.claims = set() + self.saved = [] + self.sweeps = [] + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def get_leaderboards(self): + return [self.leaderboard] + + def get_leaderboard(self, name): + assert name == "cholesky" + return self.leaderboard + + def claim_validation_sweep( + self, + *, + leaderboard_id, + gpu_type, + contract_version, + scheduled_for, + ): + key = (leaderboard_id, gpu_type, contract_version, scheduled_for) + if scheduled_for is not None and key in self.claims: + return None + self.claims.add(key) + return len(self.claims) + + def get_leaderboard_submissions(self, _name, _gpu, limit): + return self.submissions[:limit] + + def get_submission_code_for_validation(self, submission_id): + return f"# submission {submission_id}" + + def upsert_submission_validation(self, **values): + self.saved.append(values) + + def complete_validation_sweep(self, sweep_id, *, status, error=None): + self.sweeps.append((sweep_id, status, error)) + + +class FakeLauncher: + def __init__(self): + self.active = 0 + self.max_active = 0 + + async def run_validation(self, config, _gpu): + assert "@SUBMISSION@" not in config["sources"]["submission.py"] + self.active += 1 + self.max_active = max(self.max_active, self.active) + await asyncio.sleep(0) + self.active -= 1 + return { + "status": "completed", + "result": { + "passed_shapes": 2, + "total_shapes": 2, + "fully_validated": True, + "geomean_sync_wall_speedup": 1.25, + "results": [ + { + "shape": {"n": 1}, + "passed": True, + "unexpected_private_field": "drop me", + } + ], + }, + } + + +def test_due_date_uses_problem_timezone(): + assert _due_date( + datetime.datetime(2026, 7, 27, 4, 59, tzinfo=datetime.timezone.utc), + ) is None + assert _due_date( + datetime.datetime(2026, 7, 27, 5, 0, tzinfo=datetime.timezone.utc), + ) == datetime.date(2026, 7, 26) + + +@pytest.mark.asyncio +async def test_sweep_validates_top_ten_with_bounded_concurrency(): + database = FakeDB() + launcher = FakeLauncher() + backend = SimpleNamespace(db=database, launcher_map={"B200": launcher}) + service = ApplicationValidationService(backend) + + summary = await service.run_sweep( + "cholesky", + "B200", + scheduled_for=datetime.date(2026, 7, 26), + ) + + assert summary["status"] == "completed" + assert len(summary["results"]) == 10 + assert len(database.saved) == 10 + assert launcher.max_active == 2 + assert database.sweeps == [(1, "completed", None)] + assert database.saved[0]["fully_validated"] is True + assert database.saved[0]["result"]["results"][0]["unexpected_private_field"] == "drop me" + + +@pytest.mark.asyncio +async def test_nightly_sweep_is_claimed_once(): + database = FakeDB() + launcher = FakeLauncher() + backend = SimpleNamespace(db=database, launcher_map={"B200": launcher}) + service = ApplicationValidationService(backend) + now = datetime.datetime(2026, 7, 27, 5, 1, tzinfo=datetime.timezone.utc) + + first = await service.run_due_once(now) + second = await service.run_due_once(now) + + assert first[0]["status"] == "completed" + assert second[0]["status"] == "already_claimed" + assert len(database.saved) == 10 diff --git a/tests/test_leaderboard_db.py b/tests/test_leaderboard_db.py index 6e5eb535b..063f66edc 100644 --- a/tests/test_leaderboard_db.py +++ b/tests/test_leaderboard_db.py @@ -1,6 +1,7 @@ import copy import dataclasses import datetime +import json import pytest from test_report import sample_compile_result, sample_run_result, sample_system_info @@ -90,6 +91,103 @@ def test_nested_enter(database): assert db_inner.get_leaderboards() == [] +def test_application_validation_persistence(database, submit_leaderboard): + submitted_at = datetime.datetime.now(tz=datetime.timezone.utc) + source = "def custom_kernel(matrix): return matrix" + + with database as db: + db.cursor.execute( + """ + UPDATE leaderboard.leaderboard + SET task = jsonb_set( + task, + '{validation}', + %s::jsonb + ) + WHERE id = %s + """, + ( + json.dumps( + { + "version": "v1", + "source": "print('ok')", + } + ), + submit_leaderboard, + ), + ) + db.connection.commit() + submission_id = db.create_submission( + "submit-leaderboard", + "submission.py", + 5, + source, + submitted_at, + user_name="validator", + ) + assert db.get_submission_code_for_validation(submission_id) == source + + scheduled_for = datetime.date(2026, 7, 26) + sweep_id = db.claim_validation_sweep( + leaderboard_id=submit_leaderboard, + gpu_type="B200", + contract_version="v1", + scheduled_for=scheduled_for, + ) + assert sweep_id is not None + assert db.claim_validation_sweep( + leaderboard_id=submit_leaderboard, + gpu_type="B200", + contract_version="v1", + scheduled_for=scheduled_for, + ) is None + + db.upsert_submission_validation( + submission_id=submission_id, + gpu_type="B200", + contract_version="v1", + status="completed", + passed_shapes=7, + total_shapes=8, + fully_validated=False, + geomean_sync_wall_speedup=1.2, + result={"passed_shapes": 7, "total_shapes": 8}, + ) + db.cursor.execute( + """ + SELECT passed_shapes, total_shapes, fully_validated, + geomean_sync_wall_speedup + FROM leaderboard.submission_validation + WHERE submission_id = %s AND gpu_type = 'B200' + """, + (submission_id,), + ) + assert db.cursor.fetchone() == (7, 8, False, 1.2) + + db.upsert_submission_validation( + submission_id=submission_id, + gpu_type="B200", + contract_version="v1", + status="completed", + passed_shapes=8, + total_shapes=8, + fully_validated=True, + geomean_sync_wall_speedup=1.3, + result={"passed_shapes": 8, "total_shapes": 8}, + ) + db.cursor.execute( + """ + SELECT fully_validated, geomean_sync_wall_speedup + FROM leaderboard.submission_validation + WHERE submission_id = %s AND gpu_type = 'B200' + """, + (submission_id,), + ) + assert db.cursor.fetchone() == (True, 1.3) + + db.complete_validation_sweep(sweep_id, status="completed") + + def test_leaderboard_basics(database, task_directory): """ This test creates an empty leaderboard and checks its properties. diff --git a/tests/test_modal.py b/tests/test_modal.py index fd3a9c327..f3ac827b2 100644 --- a/tests/test_modal.py +++ b/tests/test_modal.py @@ -49,6 +49,31 @@ async def test_modal_submission_uses_native_async_api(): assert reporter.updates == ["✅ Waiting for modal run to finish... Done"] +@pytest.mark.asyncio +async def test_modal_validation_uses_b200_validation_function(): + launcher = ModalLauncher(add_include_dirs=[]) + function = MagicMock() + expected = {"status": "completed", "result": {"fully_validated": True}} + function.remote.aio = AsyncMock(return_value=expected) + config = {"version": "v1"} + + with patch( + "libkernelbot.launchers.modal.modal.Function.from_name", + return_value=function, + ) as from_name: + result = await launcher.run_validation( + config, + get_gpu_by_name("B200"), + ) + + assert result == expected + from_name.assert_called_once_with( + "discord-bot-runner", + "run_validation_script_b200", + ) + function.remote.aio.assert_awaited_once_with(config=config) + + @pytest.mark.asyncio async def test_modal_queue_status_uses_function_stats(): launcher = ModalLauncher(add_include_dirs=[]) diff --git a/tests/test_task.py b/tests/test_task.py index 809a6907f..83130b34d 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -5,6 +5,7 @@ from libkernelbot.consts import SubmissionMode from libkernelbot.task import ( + ApplicationValidation, CudaTaskData, Language, LeaderboardDefinition, @@ -12,6 +13,7 @@ PythonTaskData, RankCriterion, build_task_config, + build_validation_config, make_task_definition, ) from libkernelbot.utils import KernelBotError @@ -250,3 +252,43 @@ def test_multi_gpu_task(task_directory): result = make_task_definition(task_directory / "multi-task.yml") assert result.task.multi_gpu is True + + +def test_application_validation_roundtrip_and_config(leaderboard_task): + leaderboard_task.validation = ApplicationValidation( + version="optimizer-v1", + source="print('validate')", + ) + + reconstructed = LeaderboardTask.from_str(leaderboard_task.to_str()) + assert reconstructed == leaderboard_task + assert build_validation_config( + reconstructed, + "def custom_kernel(x): return x", + ) == { + "version": "optimizer-v1", + "main": "validation.py", + "sources": { + "validation.py": "print('validate')", + "test.py": "code", + "main.py": "def custom_kernel(x): return x", + }, + "timeout": 900, + } + + +def test_make_task_definition_loads_validation_files(task_directory): + (task_directory / "validation.py").write_text("print('validation')") + task_yaml = (task_directory / "task.yml").read_text() + task_yaml += """ +validation: + version: optimizer-v1 + script: validation.py +""" + (task_directory / "task.yml").write_text(task_yaml) + + definition = make_task_definition(task_directory / "task.yml") + validation = definition.task.validation + assert validation is not None + assert validation.version == "optimizer-v1" + assert validation.source == "print('validation')" diff --git a/tests/test_validation_runtime.py b/tests/test_validation_runtime.py new file mode 100644 index 000000000..12cdd275f --- /dev/null +++ b/tests/test_validation_runtime.py @@ -0,0 +1,71 @@ +import json + +from libkernelbot.validation_runtime import run_validation_config + + +def test_validation_runtime_executes_problem_owned_entrypoint(): + config = { + "name": "optimizer", + "version": "optimizer-v1", + "main": "validation.py", + "sources": { + "validation.py": ( + "import json, os\n" + "config = json.loads(os.environ['KERNELBOT_VALIDATION_CONFIG'])\n" + "print(json.dumps({'contract_version': config['version'], " + "'passed_shapes': 1, 'total_shapes': 1}))\n" + ), + "submission.py": "def custom_kernel(value): return value\n", + }, + "shapes": [{"n": 32}], + "settings": {}, + "timeout": 10, + } + + result = run_validation_config(config) + + assert result == { + "status": "completed", + "result": { + "contract_version": "optimizer-v1", + "passed_shapes": 1, + "total_shapes": 1, + }, + } + + +def test_validation_runtime_does_not_return_untrusted_output(): + result = run_validation_config( + { + "name": "optimizer", + "version": "optimizer-v1", + "main": "validation.py", + "sources": { + "validation.py": "print('submission secret')\n", + "submission.py": "SECRET = 'must not leak'\n", + }, + "timeout": 10, + } + ) + + assert result["status"] == "failed" + assert "submission secret" not in json.dumps(result) + assert "must not leak" not in json.dumps(result) + + +def test_validation_runtime_rejects_path_traversal(): + result = run_validation_config( + { + "name": "optimizer", + "version": "optimizer-v1", + "main": "validation.py", + "sources": { + "validation.py": "print('{}')\n", + "../submission.py": "SECRET = True\n", + }, + "timeout": 10, + } + ) + + assert result["status"] == "failed" + assert "escapes workspace" in result["error"]