From a339f34bb3b9947c747e1f88688a804473fba109 Mon Sep 17 00:00:00 2001 From: Mark Saroufim Date: Sun, 26 Jul 2026 08:43:20 -0700 Subject: [PATCH 1/3] feat: validate top kernels in nightly application runs --- README.md | 6 + docs/application-validation.md | 110 +++++ src/kernelbot/api/main.py | 37 ++ src/kernelbot/env.py | 8 + src/kernelbot/main.py | 41 +- src/libkernelbot/application_validation.py | 382 ++++++++++++++++++ src/libkernelbot/db_types.py | 7 + src/libkernelbot/launchers/launcher.py | 3 + src/libkernelbot/launchers/modal.py | 35 +- src/libkernelbot/leaderboard_db.py | 184 +++++++++ src/libkernelbot/task.py | 83 ++++ src/libkernelbot/validation_runtime.py | 122 ++++++ .../20260726_01_application-validation.py | 56 +++ src/runners/modal_runner_archs.py | 10 + tests/conftest.py | 3 +- tests/test_admin_api.py | 63 ++- tests/test_application_validation.py | 198 +++++++++ tests/test_leaderboard_db.py | 105 +++++ tests/test_modal.py | 50 +++ tests/test_task.py | 75 ++++ tests/test_validation_runtime.py | 71 ++++ 21 files changed, 1635 insertions(+), 14 deletions(-) create mode 100644 docs/application-validation.md create mode 100644 src/libkernelbot/application_validation.py create mode 100644 src/libkernelbot/validation_runtime.py create mode 100644 src/migrations/20260726_01_application-validation.py create mode 100644 tests/test_application_validation.py create mode 100644 tests/test_validation_runtime.py diff --git a/README.md b/README.md index a75703fb3..2dc90f28b 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,10 @@ 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 +KERNELGUARD_ENABLED=1 # Required for application validation +APPLICATION_VALIDATION_ENABLED=true +APPLICATION_VALIDATION_POLL_SECONDS=60 +MODAL_ENVIRONMENT= # Set to an isolated Modal environment for local debug # 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..e5ea6f635 --- /dev/null +++ b/docs/application-validation.md @@ -0,0 +1,110 @@ +# Application validation + +KernelBot can re-run the current top submissions inside a small, +problem-owned application workload. This complements the leaderboard's +operator correctness checks with an end-to-end signal such as training +convergence. + +## Contract + +The validation contract lives beside the problem in `reference-kernels`. +KernelBot stores the resolved contract in the leaderboard task JSON, so every +result is tied to an explicit version. + +```yaml +validation: + name: natural-gradient-training + version: cholesky-natural-gradient-v1 + main: validation.py + files: + - {name: submission.py, source: "@SUBMISSION@"} + - {name: validation.py, source: validation.py} + timeout: 900 + top_k: 10 + max_concurrency: 2 + schedule: + hour: 22 + minute: 0 + timezone: America/Los_Angeles + settings: + require_no_torch_fallback: true + min_speedup: 1.0 + shapes: + - {batch: 4096, n: 32, steps: 12} +``` + +The problem entrypoint receives `name`, `version`, `settings`, and `shapes` as +JSON in `KERNELBOT_VALIDATION_CONFIG`. It must print one JSON object as its last +stdout line. Aggregate fields are: + +```json +{ + "passed_shapes": 8, + "total_shapes": 8, + "fully_validated": true, + "geomean_sync_wall_speedup": 1.42, + "results": [] +} +``` + +`fully_validated` is accepted only when the result shape count matches the +versioned contract and every shape passes. Changing the workload or a gate +requires a new contract version. + +## Nightly flow + +At the contract's local scheduled time, each KernelBot replica tries to claim +one `(leaderboard, GPU, contract version, local date)` sweep. The database +unique constraint lets exactly one replica proceed. + +The owner: + +1. snapshots the current best submission for each of the top `top_k` users; +2. runs KernelGuard on every selected source; +3. launches at most `max_concurrency` isolated Modal jobs; +4. stores a versioned summary for each exact submission ID; and +5. marks the sweep complete. + +Raw submitted source, stdout, and stderr are never stored in validation rows or +returned by the admin endpoint. Only whitelisted aggregate and per-shape +metrics are persisted. + +The scheduler is enabled by default when a task has a validation contract. +Set `APPLICATION_VALIDATION_ENABLED=false` for an emergency stop. Polling +defaults to 60 seconds and can be changed with +`APPLICATION_VALIDATION_POLL_SECONDS`. + +Application validation fails closed when `KERNELGUARD_ENABLED` is not enabled +or KernelGuard is unavailable. + +## Local debug + +Deploy the Modal functions into a non-production environment: + +```bash +modal environment create cholesky-validation-debug +modal deploy --env cholesky-validation-debug src/runners/modal_runner_archs.py +``` + +Run a local API instance against a migrated development database: + +```bash +MODAL_ENVIRONMENT=cholesky-validation-debug \ +APPLICATION_VALIDATION_ENABLED=false \ +KERNELGUARD_ENABLED=1 \ +python src/kernelbot/main.py --api-only --debug +``` + +Trigger one synchronous sweep: + +```bash +curl -X POST \ + -H "Authorization: Bearer ${ADMIN_TOKEN}" \ + "http://localhost:8000/admin/application-validations/cholesky/B200?wait=true" +``` + +Omit `wait=true` to enqueue the manual sweep and return immediately. + +The rollout order is KernelBot migration and runner support, then the +`reference-kernels` contract, then the Kernelboard badge. Kernelboard only +shows results whose contract version matches the leaderboard's current task. 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..0b338ed25 100644 --- a/src/kernelbot/env.py +++ b/src/kernelbot/env.py @@ -35,6 +35,7 @@ env.GITHUB_REPO = os.getenv("GITHUB_REPO") env.GITHUB_WORKFLOW_BRANCH = os.getenv("GITHUB_WORKFLOW_BRANCH", get_github_branch_name()) env.PROBLEMS_REPO = os.getenv("PROBLEMS_REPO") +env.MODAL_ENVIRONMENT = os.getenv("MODAL_ENVIRONMENT") # Directory that will be used for local problem development. env.PROBLEM_DEV_DIR = os.getenv("PROBLEM_DEV_DIR", "examples") @@ -42,6 +43,13 @@ # 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"} +env.APPLICATION_VALIDATION_POLL_SECONDS = int( + os.getenv("APPLICATION_VALIDATION_POLL_SECONDS", "60") +) def init_environment(skip_discord: bool = False): diff --git a/src/kernelbot/main.py b/src/kernelbot/main.py index 398d17166..cbff71e1d 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 @@ -26,7 +32,12 @@ def create_backend(debug_mode: bool = False) -> KernelBackend: """Create and configure a KernelBackend with launchers.""" backend = KernelBackend(env=env, debug_mode=debug_mode) - backend.register_launcher(ModalLauncher(consts.MODAL_CUDA_INCLUDE_DIRS)) + backend.register_launcher( + ModalLauncher( + consts.MODAL_CUDA_INCLUDE_DIRS, + environment_name=env.MODAL_ENVIRONMENT, + ) + ) backend.register_launcher( GitHubLauncher(env.GITHUB_REPO, env.GITHUB_TOKEN, env.GITHUB_WORKFLOW_BRANCH) ) @@ -50,11 +61,20 @@ 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, + poll_seconds=env.APPLICATION_VALIDATION_POLL_SECONDS, + ) + ) + await validation_service.start() server = create_uvicorn_server() try: await server.serve() finally: + await validation_service.stop() await manager.stop() @@ -241,9 +261,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 +278,14 @@ 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, + poll_seconds=env.APPLICATION_VALIDATION_POLL_SECONDS, + ) + ) + await validation_service.start() server = create_uvicorn_server() try: @@ -266,6 +294,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 +312,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..b7204edb3 --- /dev/null +++ b/src/libkernelbot/application_validation.py @@ -0,0 +1,382 @@ +"""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 SubmissionMode, get_gpu_by_name +from libkernelbot.kernelguard import ( + enforce_submission_precheck, + should_precheck_submission, +) +from libkernelbot.task import ( + ApplicationValidation, + LeaderboardTask, + build_validation_config, +) +from libkernelbot.utils import KernelBotError, setup_logging + +logger = setup_logging(__name__) + + +def _kernelguard_precheck(source: str, file_name: str) -> None: + if not should_precheck_submission(SubmissionMode.LEADERBOARD): + raise KernelBotError( + "KernelGuard must be enabled before application validation can run" + ) + enforce_submission_precheck(source, file_name) + + +def _due_date( + validation: ApplicationValidation, + now: datetime.datetime, +) -> datetime.date | None: + timezone = ZoneInfo(validation.schedule.timezone) + if now.tzinfo is None: + now = now.replace(tzinfo=datetime.timezone.utc) + local_now = now.astimezone(timezone) + scheduled_time = datetime.time( + validation.schedule.hour, + validation.schedule.minute, + tzinfo=timezone, + ) + if local_now < datetime.datetime.combine( + local_now.date(), + scheduled_time, + ): + return None + return local_now.date() + + +def _safe_result(result: dict[str, Any]) -> dict[str, Any]: + """Whitelist aggregate and per-shape metrics safe to keep in the database.""" + public: dict[str, Any] = {} + for key in ( + "passed_shapes", + "total_shapes", + "fully_validated", + "geomean_sync_wall_speedup", + ): + if key in result: + public[key] = result[key] + + public_shapes = [] + for shape in result.get("results", []): + if not isinstance(shape, dict): + continue + public_shapes.append( + { + key: shape[key] + for key in ( + "batch", + "n", + "steps", + "passed", + "numerically_stable", + "converged", + "route", + "torch_fallback_calls", + "sync_wall_speedup", + "factor_residual", + "baseline_final_loss", + "candidate_final_loss", + ) + if key in shape + } + ) + if public_shapes: + public["results"] = public_shapes + return public + + +class ApplicationValidationService: + def __init__( + self, + backend, + *, + enabled: bool = False, + poll_seconds: int = 60, + precheck=_kernelguard_precheck, + ): + self.backend = backend + self.enabled = enabled + self.poll_seconds = poll_seconds + self.precheck = precheck + 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]]: + now = now or datetime.datetime.now(datetime.timezone.utc) + with self.backend.db as db: + leaderboards = db.get_leaderboards() + + summaries = [] + for leaderboard in leaderboards: + validation = leaderboard["task"].validation + if validation is None: + continue + scheduled_for = _due_date(validation, now) + if scheduled_for 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, + validation: ApplicationValidation, + gpu_type: str, + semaphore: asyncio.Semaphore, + ) -> dict[str, Any]: + submission_id = entry["submission_id"] + async with semaphore: + try: + with self.backend.db as db: + source = db.get_submission_code_for_validation(submission_id) + self.precheck(source, entry["submission_name"]) + config = build_validation_config(task, source) + del 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) + del config + + safe_result = _safe_result(response.get("result", {})) + status = ( + "completed" + if response.get("status") == "completed" + else "failed" + ) + expected_total = len(validation.shapes) + passed_shapes = int(safe_result.get("passed_shapes", 0)) + reported_total = int( + safe_result.get("total_shapes", expected_total) + ) + if status == "completed" and reported_total != expected_total: + raise ValueError( + "validator result shape count does not match contract" + ) + if not 0 <= passed_shapes <= expected_total: + raise ValueError( + "validator result has an invalid passed shape count" + ) + fully_validated = bool( + status == "completed" + and passed_shapes == expected_total + and safe_result.get("fully_validated", False) + ) + speedup = safe_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 result has an invalid speedup") + error = response.get("error") + with self.backend.db as db: + db.upsert_submission_validation( + submission_id=submission_id, + gpu_type=gpu_type, + contract_name=validation.name, + contract_version=validation.version, + status=status, + passed_shapes=passed_shapes, + total_shapes=expected_total, + fully_validated=fully_validated, + geomean_sync_wall_speedup=speedup, + result=safe_result, + error=str(error)[:1000] if error else None, + ) + return { + "submission_id": submission_id, + "status": status, + "passed_shapes": passed_shapes, + "total_shapes": expected_total, + "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_name=validation.name, + contract_version=validation.version, + status="failed", + passed_shapes=0, + total_shapes=len(validation.shapes), + fully_validated=False, + geomean_sync_wall_speedup=None, + result={}, + error=error, + ) + return { + "submission_id": submission_id, + "status": "failed", + "passed_shapes": 0, + "total_shapes": len(validation.shapes), + "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, + top_k=validation.top_k, + ) + 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=validation.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(validation.max_concurrency) + + try: + results = await asyncio.gather( + *( + self._validate_submission( + entry, + task=leaderboard["task"], + validation=validation, + 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/db_types.py b/src/libkernelbot/db_types.py index 75acdfa3d..eea48b8bf 100644 --- a/src/libkernelbot/db_types.py +++ b/src/libkernelbot/db_types.py @@ -34,6 +34,13 @@ class LeaderboardRankedEntry(TypedDict): user_id: int user_name: str gpu_type: str + validation_status: NotRequired[Optional[str]] + validation_shapes_passed: NotRequired[Optional[int]] + validation_shapes_total: NotRequired[Optional[int]] + validation_fully_validated: NotRequired[Optional[bool]] + validation_geomean_speedup: NotRequired[Optional[float]] + validation_contract_version: NotRequired[Optional[str]] + validation_checked_at: NotRequired[Optional[datetime.datetime]] class RunItem(TypedDict): 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..5b802407d 100644 --- a/src/libkernelbot/launchers/modal.py +++ b/src/libkernelbot/launchers/modal.py @@ -13,9 +13,26 @@ class ModalLauncher(Launcher): - def __init__(self, add_include_dirs: list): + def __init__( + self, + add_include_dirs: list, + environment_name: str | None = None, + ): super().__init__("Modal", gpus=ModalGPU) self.additional_include_dirs = add_include_dirs + self.environment_name = environment_name + + def _lookup_function(self, func_name: str): + kwargs = ( + {"environment_name": self.environment_name} + if self.environment_name + else {} + ) + return modal.Function.from_name( + "discord-bot-runner", + func_name, + **kwargs, + ) async def run_submission( self, config: dict, gpu_type: GPU, status: RunProgressReporter @@ -28,13 +45,23 @@ async def run_submission( await status.push("⏳ Waiting for Modal run to finish...") - function = modal.Function.from_name("discord-bot-runner", func_name) + function = self._lookup_function(func_name) result = await function.remote.aio(config=config) await status.update("✅ Waiting for modal run to finish... Done") 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 = self._lookup_function(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()}" @@ -48,9 +75,7 @@ async def get_queue_status( try: stats = await loop.run_in_executor( None, - lambda: modal.Function.from_name( - "discord-bot-runner", func_name - ).get_current_stats(), + lambda: self._lookup_function(func_name).get_current_stats(), ) except Exception as e: logger.warning("Could not get Modal queue stats for %s", func_name, exc_info=e) diff --git a/src/libkernelbot/leaderboard_db.py b/src/libkernelbot/leaderboard_db.py index 2ed15acb3..1f0bc41bc 100644 --- a/src/libkernelbot/leaderboard_db.py +++ b/src/libkernelbot/leaderboard_db.py @@ -880,6 +880,183 @@ 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], + top_k: int, + ) -> 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, top_k + ) + VALUES (%s, %s, %s, %s, 'running', %s) + ON CONFLICT (leaderboard_id, gpu_type, contract_version, scheduled_for) + DO NOTHING + RETURNING id + """, + (leaderboard_id, gpu_type, contract_version, scheduled_for, top_k), + ) + 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_name: 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_name, 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, %s::jsonb, %s) + ON CONFLICT (submission_id, gpu_type, contract_version) + DO UPDATE SET + contract_name = EXCLUDED.contract_name, + 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_name, + 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_submission_validation_statuses( + self, + submission_ids: list[int], + gpu_type: str, + ) -> dict[int, dict]: + """Return only public validation summaries, never raw validator output.""" + if not submission_ids: + return {} + self.cursor.execute( + """ + SELECT DISTINCT ON (submission_id) + submission_id, status, passed_shapes, total_shapes, + fully_validated, geomean_sync_wall_speedup, + contract_version, checked_at + FROM leaderboard.submission_validation + WHERE submission_id = ANY(%s) + AND gpu_type = %s + AND contract_version = ( + SELECT l.task->'validation'->>'version' + FROM leaderboard.submission s + JOIN leaderboard.leaderboard l ON l.id = s.leaderboard_id + WHERE s.id = submission_validation.submission_id + ) + ORDER BY submission_id, checked_at DESC + """, + (submission_ids, gpu_type), + ) + return { + row[0]: { + "validation_status": row[1], + "validation_shapes_passed": row[2], + "validation_shapes_total": row[3], + "validation_fully_validated": row[4], + "validation_geomean_speedup": row[5], + "validation_contract_version": row[6], + "validation_checked_at": row[7], + } + for row in self.cursor.fetchall() + } + def get_leaderboard_submissions( self, leaderboard_name: str, @@ -1025,6 +1202,13 @@ def get_leaderboard_submissions( raise KernelBotError( f"Invalid GPU type '{gpu_name}' for leaderboard '{leaderboard_name}'" ) + else: + validation_statuses = self.get_submission_validation_statuses( + [entry["submission_id"] for entry in result], + gpu_name, + ) + for entry in result: + entry.update(validation_statuses.get(entry["submission_id"], {})) return result diff --git a/src/libkernelbot/task.py b/src/libkernelbot/task.py index e0fb0a410..4149b11f0 100644 --- a/src/libkernelbot/task.py +++ b/src/libkernelbot/task.py @@ -27,6 +27,51 @@ class PythonTaskData: TestCaseType = Dict[str, Union[int, str]] +@dataclasses.dataclass +class ValidationSchedule: + hour: int + minute: int = 0 + timezone: str = "America/Los_Angeles" + + def __post_init__(self): + if not 0 <= self.hour <= 23: + raise ValueError("validation schedule hour must be between 0 and 23") + if not 0 <= self.minute <= 59: + raise ValueError("validation schedule minute must be between 0 and 59") + + +@dataclasses.dataclass +class ApplicationValidation: + name: str + version: str + main: str + files: dict[str, str] + shapes: list[TestCaseType] + settings: dict[str, Union[bool, float, int, str]] + schedule: ValidationSchedule + timeout: int = 900 + top_k: int = 10 + max_concurrency: int = 2 + + def __post_init__(self): + if not self.name or not self.version: + raise ValueError("validation name and version must be non-empty") + if self.main not in self.files: + raise ValueError(f"validation main file {self.main!r} is not in validation files") + if "@SUBMISSION@" not in self.files.values(): + raise ValueError("validation files must contain one @SUBMISSION@ source") + if not self.shapes: + raise ValueError("validation shapes must not be empty") + if self.timeout <= 0 or self.top_k <= 0 or self.max_concurrency <= 0: + raise ValueError("validation timeout, top_k, and max_concurrency must be positive") + + @classmethod + def from_dict(cls, data: dict) -> "ApplicationValidation": + values = copy.deepcopy(data) + values["schedule"] = ValidationSchedule(**values["schedule"]) + return cls(**values) + + @dataclasses.dataclass class LeaderboardTask: """ @@ -62,6 +107,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 +123,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.from_dict(data["validation"]) if lang == Language.Python: data_["config"] = PythonTaskData(**data["config"]) else: @@ -143,6 +191,19 @@ def make_task_definition(yaml_file: str | Path) -> LeaderboardDefinition: # noq raw["files"] = file_dict + validation = raw.get("validation") + if validation is not None: + validation_files = {} + for file_spec in validation["files"]: + name = file_spec["name"] + source = file_spec["source"] + validation_files[name] = ( + "@SUBMISSION@" + if source == "@SUBMISSION@" + else (root / source).read_text() + ) + validation["files"] = validation_files + # load template files templates = {} for lang, source in raw.get("templates", {}).items(): @@ -219,3 +280,25 @@ 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 validation.files.items() + } + return { + "name": validation.name, + "version": validation.version, + "main": validation.main, + "sources": sources, + "shapes": validation.shapes, + "settings": validation.settings, + "timeout": validation.timeout, + } 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..d36813ec8 --- /dev/null +++ b/src/migrations/20260726_01_application-validation.py @@ -0,0 +1,56 @@ +""" +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')), + top_k INTEGER NOT NULL CHECK (top_k > 0), + 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_name 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_application_validation.py b/tests/test_application_validation.py new file mode 100644 index 000000000..ab9b4c7ce --- /dev/null +++ b/tests/test_application_validation.py @@ -0,0 +1,198 @@ +import asyncio +import datetime +from types import SimpleNamespace + +import pytest + +from libkernelbot.application_validation import ( + ApplicationValidationService, + _due_date, + _kernelguard_precheck, +) +from libkernelbot.task import LeaderboardTask +from libkernelbot.utils import KernelBotError + + +def _task() -> LeaderboardTask: + return LeaderboardTask.from_dict( + { + "lang": "py", + "files": {"submission.py": "@SUBMISSION@"}, + "config": {"main": "submission.py"}, + "validation": { + "name": "toy-training", + "version": "v1", + "main": "validation.py", + "files": { + "submission.py": "@SUBMISSION@", + "validation.py": "print('validator')", + }, + "shapes": [{"n": 1}, {"n": 2}], + "settings": {"min_speedup": 1.0}, + "schedule": { + "hour": 22, + "minute": 0, + "timezone": "America/Los_Angeles", + }, + "top_k": 10, + "max_concurrency": 2, + }, + } + ) + + +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, + top_k, + ): + 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(): + validation = _task().validation + assert validation is not None + + assert _due_date( + validation, + datetime.datetime(2026, 7, 27, 4, 59, tzinfo=datetime.timezone.utc), + ) is None + assert _due_date( + validation, + datetime.datetime(2026, 7, 27, 5, 0, tzinfo=datetime.timezone.utc), + ) == datetime.date(2026, 7, 26) + + +def test_application_validation_requires_kernelguard(monkeypatch): + monkeypatch.delenv("KERNELGUARD_ENABLED", raising=False) + + with pytest.raises(KernelBotError, match="KernelGuard must be enabled"): + _kernelguard_precheck("def custom_kernel(): pass", "submission.py") + + +@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}) + checked = [] + service = ApplicationValidationService( + backend, + precheck=lambda source, name: checked.append((source, name)), + ) + + 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 len(checked) == 10 + assert database.sweeps == [(1, "completed", None)] + assert database.saved[0]["fully_validated"] is True + assert "unexpected_private_field" not in database.saved[0]["result"]["results"][0] + + +@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, + precheck=lambda _source, _name: None, + ) + 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..56fe419e6 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,110 @@ 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( + { + "name": "toy-training", + "version": "v1", + "main": "validation.py", + "files": { + "submission.py": "@SUBMISSION@", + "validation.py": "print('ok')", + }, + "shapes": [{"n": 1}], + "settings": {}, + "schedule": {"hour": 22}, + } + ), + 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, + top_k=10, + ) + 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, + top_k=10, + ) is None + + db.upsert_submission_validation( + submission_id=submission_id, + gpu_type="B200", + contract_name="toy-training", + 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}, + ) + statuses = db.get_submission_validation_statuses( + [submission_id], + "B200", + ) + assert statuses[submission_id]["validation_shapes_passed"] == 7 + assert statuses[submission_id]["validation_shapes_total"] == 8 + assert statuses[submission_id]["validation_fully_validated"] is False + assert statuses[submission_id]["validation_geomean_speedup"] == 1.2 + + db.upsert_submission_validation( + submission_id=submission_id, + gpu_type="B200", + contract_name="toy-training", + 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}, + ) + statuses = db.get_submission_validation_statuses( + [submission_id], + "B200", + ) + assert statuses[submission_id]["validation_fully_validated"] is True + assert statuses[submission_id]["validation_geomean_speedup"] == 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..9a5a40819 100644 --- a/tests/test_modal.py +++ b/tests/test_modal.py @@ -49,6 +49,56 @@ 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_validation_can_target_debug_environment(): + launcher = ModalLauncher( + add_include_dirs=[], + environment_name="cholesky-validation-debug", + ) + function = MagicMock() + function.remote.aio = AsyncMock(return_value={"status": "completed"}) + + with patch( + "libkernelbot.launchers.modal.modal.Function.from_name", + return_value=function, + ) as from_name: + await launcher.run_validation( + {"version": "v1"}, + get_gpu_by_name("B200"), + ) + + from_name.assert_called_once_with( + "discord-bot-runner", + "run_validation_script_b200", + environment_name="cholesky-validation-debug", + ) + + @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..ee9d51e78 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -5,13 +5,16 @@ from libkernelbot.consts import SubmissionMode from libkernelbot.task import ( + ApplicationValidation, CudaTaskData, Language, LeaderboardDefinition, LeaderboardTask, PythonTaskData, RankCriterion, + ValidationSchedule, build_task_config, + build_validation_config, make_task_definition, ) from libkernelbot.utils import KernelBotError @@ -250,3 +253,75 @@ 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( + name="optimizer", + version="optimizer-v1", + main="validate.py", + files={ + "validate.py": "print('validate')", + "submission.py": "@SUBMISSION@", + }, + shapes=[{"batch": 2, "n": 32, "steps": 4}], + settings={"min_speedup": 1.0, "require_no_torch_fallback": True}, + schedule=ValidationSchedule( + hour=22, + minute=0, + timezone="America/Los_Angeles", + ), + ) + + reconstructed = LeaderboardTask.from_str(leaderboard_task.to_str()) + assert reconstructed == leaderboard_task + assert build_validation_config( + reconstructed, + "def custom_kernel(x): return x", + ) == { + "name": "optimizer", + "version": "optimizer-v1", + "main": "validate.py", + "sources": { + "validate.py": "print('validate')", + "submission.py": "def custom_kernel(x): return x", + }, + "shapes": [{"batch": 2, "n": 32, "steps": 4}], + "settings": { + "min_speedup": 1.0, + "require_no_torch_fallback": True, + }, + "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: + name: optimizer + version: optimizer-v1 + main: validation.py + files: + - {name: submission.py, source: "@SUBMISSION@"} + - {name: validation.py, source: validation.py} + shapes: + - {batch: 2, n: 32, steps: 4} + settings: + min_speedup: 1.0 + schedule: + hour: 22 + timezone: America/Los_Angeles +""" + (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.files == { + "submission.py": "@SUBMISSION@", + "validation.py": "print('validation')", + } + assert validation.schedule.hour == 22 + assert validation.schedule.timezone == "America/Los_Angeles" 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"] From 2b5ae1703828a1d70dc4311bf70dedd2c9384547 Mon Sep 17 00:00:00 2001 From: Mark Saroufim Date: Sun, 26 Jul 2026 09:05:08 -0700 Subject: [PATCH 2/3] refactor: simplify application validation --- README.md | 3 - docs/application-validation.md | 91 +++------ src/kernelbot/env.py | 4 - src/kernelbot/main.py | 9 +- src/libkernelbot/application_validation.py | 177 +++++------------- src/libkernelbot/db_types.py | 7 - src/libkernelbot/launchers/modal.py | 27 +-- src/libkernelbot/leaderboard_db.py | 62 +----- src/libkernelbot/task.py | 64 +------ .../20260726_01_application-validation.py | 2 - tests/test_application_validation.py | 45 +---- tests/test_leaderboard_db.py | 43 ++--- tests/test_modal.py | 25 --- tests/test_task.py | 49 +---- 14 files changed, 121 insertions(+), 487 deletions(-) diff --git a/README.md b/README.md index 2dc90f28b..aff59cb10 100644 --- a/README.md +++ b/README.md @@ -54,10 +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 -KERNELGUARD_ENABLED=1 # Required for application validation APPLICATION_VALIDATION_ENABLED=true -APPLICATION_VALIDATION_POLL_SECONDS=60 -MODAL_ENVIRONMENT= # Set to an isolated Modal environment for local debug # 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 index e5ea6f635..5c2cb8fe3 100644 --- a/docs/application-validation.md +++ b/docs/application-validation.md @@ -1,41 +1,20 @@ # Application validation -KernelBot can re-run the current top submissions inside a small, -problem-owned application workload. This complements the leaderboard's -operator correctness checks with an end-to-end signal such as training -convergence. - -## Contract - -The validation contract lives beside the problem in `reference-kernels`. -KernelBot stores the resolved contract in the leaderboard task JSON, so every -result is tied to an explicit version. +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: - name: natural-gradient-training version: cholesky-natural-gradient-v1 - main: validation.py - files: - - {name: submission.py, source: "@SUBMISSION@"} - - {name: validation.py, source: validation.py} - timeout: 900 - top_k: 10 - max_concurrency: 2 - schedule: - hour: 22 - minute: 0 - timezone: America/Los_Angeles - settings: - require_no_torch_fallback: true - min_speedup: 1.0 - shapes: - - {batch: 4096, n: 32, steps: 12} + script: validation.py ``` -The problem entrypoint receives `name`, `version`, `settings`, and `shapes` as -JSON in `KERNELBOT_VALIDATION_CONFIG`. It must print one JSON object as its last -stdout line. Aggregate fields are: +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 { @@ -47,55 +26,32 @@ stdout line. Aggregate fields are: } ``` -`fully_validated` is accepted only when the result shape count matches the -versioned contract and every shape passes. Changing the workload or a gate -requires a new contract version. - -## Nightly flow - -At the contract's local scheduled time, each KernelBot replica tries to claim -one `(leaderboard, GPU, contract version, local date)` sweep. The database -unique constraint lets exactly one replica proceed. - -The owner: +## Schedule -1. snapshots the current best submission for each of the top `top_k` users; -2. runs KernelGuard on every selected source; -3. launches at most `max_concurrency` isolated Modal jobs; -4. stores a versioned summary for each exact submission ID; and -5. marks the sweep complete. +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. -Raw submitted source, stdout, and stderr are never stored in validation rows or -returned by the admin endpoint. Only whitelisted aggregate and per-shape -metrics are persisted. - -The scheduler is enabled by default when a task has a validation contract. -Set `APPLICATION_VALIDATION_ENABLED=false` for an emergency stop. Polling -defaults to 60 seconds and can be changed with -`APPLICATION_VALIDATION_POLL_SECONDS`. - -Application validation fails closed when `KERNELGUARD_ENABLED` is not enabled -or KernelGuard is unavailable. +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 -Deploy the Modal functions into a non-production environment: +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 -``` -Run a local API instance against a migrated development database: - -```bash MODAL_ENVIRONMENT=cholesky-validation-debug \ APPLICATION_VALIDATION_ENABLED=false \ -KERNELGUARD_ENABLED=1 \ python src/kernelbot/main.py --api-only --debug ``` -Trigger one synchronous sweep: +Run one sweep synchronously: ```bash curl -X POST \ @@ -103,8 +59,5 @@ curl -X POST \ "http://localhost:8000/admin/application-validations/cholesky/B200?wait=true" ``` -Omit `wait=true` to enqueue the manual sweep and return immediately. - -The rollout order is KernelBot migration and runner support, then the -`reference-kernels` contract, then the Kernelboard badge. Kernelboard only -shows results whose contract version matches the leaderboard's current task. +Roll out KernelBot's migration and runner first, then the reference-kernels +contract, then the Kernelboard badge. diff --git a/src/kernelbot/env.py b/src/kernelbot/env.py index 0b338ed25..1d9456609 100644 --- a/src/kernelbot/env.py +++ b/src/kernelbot/env.py @@ -35,7 +35,6 @@ env.GITHUB_REPO = os.getenv("GITHUB_REPO") env.GITHUB_WORKFLOW_BRANCH = os.getenv("GITHUB_WORKFLOW_BRANCH", get_github_branch_name()) env.PROBLEMS_REPO = os.getenv("PROBLEMS_REPO") -env.MODAL_ENVIRONMENT = os.getenv("MODAL_ENVIRONMENT") # Directory that will be used for local problem development. env.PROBLEM_DEV_DIR = os.getenv("PROBLEM_DEV_DIR", "examples") @@ -47,9 +46,6 @@ "APPLICATION_VALIDATION_ENABLED", "true", ).lower() in {"1", "true", "yes"} -env.APPLICATION_VALIDATION_POLL_SECONDS = int( - os.getenv("APPLICATION_VALIDATION_POLL_SECONDS", "60") -) def init_environment(skip_discord: bool = False): diff --git a/src/kernelbot/main.py b/src/kernelbot/main.py index cbff71e1d..64de6b262 100644 --- a/src/kernelbot/main.py +++ b/src/kernelbot/main.py @@ -32,12 +32,7 @@ def create_backend(debug_mode: bool = False) -> KernelBackend: """Create and configure a KernelBackend with launchers.""" backend = KernelBackend(env=env, debug_mode=debug_mode) - backend.register_launcher( - ModalLauncher( - consts.MODAL_CUDA_INCLUDE_DIRS, - environment_name=env.MODAL_ENVIRONMENT, - ) - ) + backend.register_launcher(ModalLauncher(consts.MODAL_CUDA_INCLUDE_DIRS)) backend.register_launcher( GitHubLauncher(env.GITHUB_REPO, env.GITHUB_TOKEN, env.GITHUB_WORKFLOW_BRANCH) ) @@ -65,7 +60,6 @@ async def run_api_server(backend: KernelBackend): ApplicationValidationService( backend, enabled=env.APPLICATION_VALIDATION_ENABLED, - poll_seconds=env.APPLICATION_VALIDATION_POLL_SECONDS, ) ) await validation_service.start() @@ -282,7 +276,6 @@ async def start_bot_and_api(debug_mode: bool): ApplicationValidationService( bot_instance.backend, enabled=env.APPLICATION_VALIDATION_ENABLED, - poll_seconds=env.APPLICATION_VALIDATION_POLL_SECONDS, ) ) await validation_service.start() diff --git a/src/libkernelbot/application_validation.py b/src/libkernelbot/application_validation.py index b7204edb3..203a23dfe 100644 --- a/src/libkernelbot/application_validation.py +++ b/src/libkernelbot/application_validation.py @@ -9,104 +9,32 @@ from typing import Any from zoneinfo import ZoneInfo -from libkernelbot.consts import SubmissionMode, get_gpu_by_name -from libkernelbot.kernelguard import ( - enforce_submission_precheck, - should_precheck_submission, -) -from libkernelbot.task import ( - ApplicationValidation, - LeaderboardTask, - build_validation_config, -) +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__) - -def _kernelguard_precheck(source: str, file_name: str) -> None: - if not should_precheck_submission(SubmissionMode.LEADERBOARD): - raise KernelBotError( - "KernelGuard must be enabled before application validation can run" - ) - enforce_submission_precheck(source, file_name) +SCHEDULE = datetime.time(22, 0) +TIMEZONE = ZoneInfo("America/Los_Angeles") +TOP_K = 10 +MAX_CONCURRENCY = 2 -def _due_date( - validation: ApplicationValidation, - now: datetime.datetime, -) -> datetime.date | None: - timezone = ZoneInfo(validation.schedule.timezone) +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) - scheduled_time = datetime.time( - validation.schedule.hour, - validation.schedule.minute, - tzinfo=timezone, - ) - if local_now < datetime.datetime.combine( - local_now.date(), - scheduled_time, - ): + local_now = now.astimezone(TIMEZONE) + if local_now.time() < SCHEDULE: return None return local_now.date() -def _safe_result(result: dict[str, Any]) -> dict[str, Any]: - """Whitelist aggregate and per-shape metrics safe to keep in the database.""" - public: dict[str, Any] = {} - for key in ( - "passed_shapes", - "total_shapes", - "fully_validated", - "geomean_sync_wall_speedup", - ): - if key in result: - public[key] = result[key] - - public_shapes = [] - for shape in result.get("results", []): - if not isinstance(shape, dict): - continue - public_shapes.append( - { - key: shape[key] - for key in ( - "batch", - "n", - "steps", - "passed", - "numerically_stable", - "converged", - "route", - "torch_fallback_calls", - "sync_wall_speedup", - "factor_residual", - "baseline_final_loss", - "candidate_final_loss", - ) - if key in shape - } - ) - if public_shapes: - public["results"] = public_shapes - return public - - class ApplicationValidationService: - def __init__( - self, - backend, - *, - enabled: bool = False, - poll_seconds: int = 60, - precheck=_kernelguard_precheck, - ): + def __init__(self, backend, *, enabled: bool = False, poll_seconds: int = 60): self.backend = backend self.enabled = enabled self.poll_seconds = poll_seconds - self.precheck = precheck self._scheduler_task: asyncio.Task | None = None self._manual_tasks: set[asyncio.Task] = set() @@ -146,17 +74,18 @@ async def run_due_once( self, now: datetime.datetime | None = None, ) -> list[dict[str, Any]]: - now = now or datetime.datetime.now(datetime.timezone.utc) + 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: - validation = leaderboard["task"].validation - if validation is None: - continue - scheduled_for = _due_date(validation, now) - if scheduled_for is None: + if leaderboard["task"].validation is None: continue for gpu_type in leaderboard["gpu_types"]: summaries.append( @@ -193,18 +122,16 @@ async def _validate_submission( entry: dict, *, task: LeaderboardTask, - validation: ApplicationValidation, 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) - self.precheck(source, entry["submission_name"]) config = build_validation_config(task, source) - del source gpu = get_gpu_by_name(gpu_type) if gpu is None: @@ -215,57 +142,52 @@ async def _validate_submission( f"No validation launcher is registered for {gpu_type!r}" ) response = await launcher.run_validation(config, gpu) - del config - - safe_result = _safe_result(response.get("result", {})) status = ( "completed" if response.get("status") == "completed" else "failed" ) - expected_total = len(validation.shapes) - passed_shapes = int(safe_result.get("passed_shapes", 0)) - reported_total = int( - safe_result.get("total_shapes", expected_total) - ) - if status == "completed" and reported_total != expected_total: - raise ValueError( - "validator result shape count does not match contract" - ) - if not 0 <= passed_shapes <= expected_total: - raise ValueError( - "validator result has an invalid passed shape count" - ) - fully_validated = bool( - status == "completed" - and passed_shapes == expected_total - and safe_result.get("fully_validated", False) - ) - speedup = safe_result.get("geomean_sync_wall_speedup") + 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 result has an invalid speedup") - error = response.get("error") + 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_name=validation.name, - contract_version=validation.version, + contract_version=contract_version, status=status, passed_shapes=passed_shapes, - total_shapes=expected_total, + total_shapes=total_shapes, fully_validated=fully_validated, geomean_sync_wall_speedup=speedup, - result=safe_result, - error=str(error)[:1000] if error else None, + result=result, + error=response.get("error"), ) return { "submission_id": submission_id, "status": status, "passed_shapes": passed_shapes, - "total_shapes": expected_total, + "total_shapes": total_shapes, "fully_validated": fully_validated, "geomean_sync_wall_speedup": speedup, } @@ -280,11 +202,10 @@ async def _validate_submission( db.upsert_submission_validation( submission_id=submission_id, gpu_type=gpu_type, - contract_name=validation.name, - contract_version=validation.version, + contract_version=contract_version, status="failed", passed_shapes=0, - total_shapes=len(validation.shapes), + total_shapes=0, fully_validated=False, geomean_sync_wall_speedup=None, result={}, @@ -294,7 +215,7 @@ async def _validate_submission( "submission_id": submission_id, "status": "failed", "passed_shapes": 0, - "total_shapes": len(validation.shapes), + "total_shapes": 0, "fully_validated": False, } @@ -323,7 +244,6 @@ async def run_sweep( gpu_type=gpu_type, contract_version=validation.version, scheduled_for=scheduled_for, - top_k=validation.top_k, ) if sweep_id is None: return { @@ -335,7 +255,7 @@ async def run_sweep( submissions = db.get_leaderboard_submissions( leaderboard_name, gpu_type, - limit=validation.top_k, + limit=TOP_K, ) logger.info( @@ -345,7 +265,7 @@ async def run_sweep( gpu_type, len(submissions), ) - semaphore = asyncio.Semaphore(validation.max_concurrency) + semaphore = asyncio.Semaphore(MAX_CONCURRENCY) try: results = await asyncio.gather( @@ -353,7 +273,6 @@ async def run_sweep( self._validate_submission( entry, task=leaderboard["task"], - validation=validation, gpu_type=gpu_type, semaphore=semaphore, ) diff --git a/src/libkernelbot/db_types.py b/src/libkernelbot/db_types.py index eea48b8bf..75acdfa3d 100644 --- a/src/libkernelbot/db_types.py +++ b/src/libkernelbot/db_types.py @@ -34,13 +34,6 @@ class LeaderboardRankedEntry(TypedDict): user_id: int user_name: str gpu_type: str - validation_status: NotRequired[Optional[str]] - validation_shapes_passed: NotRequired[Optional[int]] - validation_shapes_total: NotRequired[Optional[int]] - validation_fully_validated: NotRequired[Optional[bool]] - validation_geomean_speedup: NotRequired[Optional[float]] - validation_contract_version: NotRequired[Optional[str]] - validation_checked_at: NotRequired[Optional[datetime.datetime]] class RunItem(TypedDict): diff --git a/src/libkernelbot/launchers/modal.py b/src/libkernelbot/launchers/modal.py index 5b802407d..ce7ae06ee 100644 --- a/src/libkernelbot/launchers/modal.py +++ b/src/libkernelbot/launchers/modal.py @@ -13,26 +13,9 @@ class ModalLauncher(Launcher): - def __init__( - self, - add_include_dirs: list, - environment_name: str | None = None, - ): + def __init__(self, add_include_dirs: list): super().__init__("Modal", gpus=ModalGPU) self.additional_include_dirs = add_include_dirs - self.environment_name = environment_name - - def _lookup_function(self, func_name: str): - kwargs = ( - {"environment_name": self.environment_name} - if self.environment_name - else {} - ) - return modal.Function.from_name( - "discord-bot-runner", - func_name, - **kwargs, - ) async def run_submission( self, config: dict, gpu_type: GPU, status: RunProgressReporter @@ -45,7 +28,7 @@ async def run_submission( await status.push("⏳ Waiting for Modal run to finish...") - function = self._lookup_function(func_name) + function = modal.Function.from_name("discord-bot-runner", func_name) result = await function.remote.aio(config=config) await status.update("✅ Waiting for modal run to finish... Done") @@ -59,7 +42,7 @@ async def run_validation(self, config: dict, gpu_type: GPU) -> dict: func_name, config.get("version"), ) - function = self._lookup_function(func_name) + 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: @@ -75,7 +58,9 @@ async def get_queue_status( try: stats = await loop.run_in_executor( None, - lambda: self._lookup_function(func_name).get_current_stats(), + lambda: modal.Function.from_name( + "discord-bot-runner", func_name + ).get_current_stats(), ) except Exception as e: logger.warning("Could not get Modal queue stats for %s", func_name, exc_info=e) diff --git a/src/libkernelbot/leaderboard_db.py b/src/libkernelbot/leaderboard_db.py index 1f0bc41bc..06357e7b3 100644 --- a/src/libkernelbot/leaderboard_db.py +++ b/src/libkernelbot/leaderboard_db.py @@ -887,7 +887,6 @@ def claim_validation_sweep( gpu_type: str, contract_version: str, scheduled_for: Optional[datetime.date], - top_k: int, ) -> Optional[int]: """Atomically claim a nightly validation sweep. @@ -899,14 +898,14 @@ def claim_validation_sweep( """ INSERT INTO leaderboard.validation_sweep ( leaderboard_id, gpu_type, contract_version, scheduled_for, - status, top_k + status ) - VALUES (%s, %s, %s, %s, 'running', %s) + 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, top_k), + (leaderboard_id, gpu_type, contract_version, scheduled_for), ) row = self.cursor.fetchone() self.connection.commit() @@ -964,7 +963,6 @@ def upsert_submission_validation( *, submission_id: int, gpu_type: str, - contract_name: str, contract_version: str, status: str, passed_shapes: int, @@ -980,14 +978,13 @@ def upsert_submission_validation( self.cursor.execute( """ INSERT INTO leaderboard.submission_validation ( - submission_id, gpu_type, contract_name, contract_version, + 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, %s::jsonb, %s) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s) ON CONFLICT (submission_id, gpu_type, contract_version) DO UPDATE SET - contract_name = EXCLUDED.contract_name, status = EXCLUDED.status, passed_shapes = EXCLUDED.passed_shapes, total_shapes = EXCLUDED.total_shapes, @@ -1000,7 +997,6 @@ def upsert_submission_validation( ( submission_id, gpu_type, - contract_name, contract_version, status, passed_shapes, @@ -1017,46 +1013,6 @@ def upsert_submission_validation( logger.exception("Error saving submission application validation", exc_info=e) raise KernelBotError("Could not save submission application validation") from e - def get_submission_validation_statuses( - self, - submission_ids: list[int], - gpu_type: str, - ) -> dict[int, dict]: - """Return only public validation summaries, never raw validator output.""" - if not submission_ids: - return {} - self.cursor.execute( - """ - SELECT DISTINCT ON (submission_id) - submission_id, status, passed_shapes, total_shapes, - fully_validated, geomean_sync_wall_speedup, - contract_version, checked_at - FROM leaderboard.submission_validation - WHERE submission_id = ANY(%s) - AND gpu_type = %s - AND contract_version = ( - SELECT l.task->'validation'->>'version' - FROM leaderboard.submission s - JOIN leaderboard.leaderboard l ON l.id = s.leaderboard_id - WHERE s.id = submission_validation.submission_id - ) - ORDER BY submission_id, checked_at DESC - """, - (submission_ids, gpu_type), - ) - return { - row[0]: { - "validation_status": row[1], - "validation_shapes_passed": row[2], - "validation_shapes_total": row[3], - "validation_fully_validated": row[4], - "validation_geomean_speedup": row[5], - "validation_contract_version": row[6], - "validation_checked_at": row[7], - } - for row in self.cursor.fetchall() - } - def get_leaderboard_submissions( self, leaderboard_name: str, @@ -1202,14 +1158,6 @@ def get_leaderboard_submissions( raise KernelBotError( f"Invalid GPU type '{gpu_name}' for leaderboard '{leaderboard_name}'" ) - else: - validation_statuses = self.get_submission_validation_statuses( - [entry["submission_id"] for entry in result], - gpu_name, - ) - for entry in result: - entry.update(validation_statuses.get(entry["submission_id"], {})) - 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 4149b11f0..ddd819451 100644 --- a/src/libkernelbot/task.py +++ b/src/libkernelbot/task.py @@ -27,49 +27,14 @@ class PythonTaskData: TestCaseType = Dict[str, Union[int, str]] -@dataclasses.dataclass -class ValidationSchedule: - hour: int - minute: int = 0 - timezone: str = "America/Los_Angeles" - - def __post_init__(self): - if not 0 <= self.hour <= 23: - raise ValueError("validation schedule hour must be between 0 and 23") - if not 0 <= self.minute <= 59: - raise ValueError("validation schedule minute must be between 0 and 59") - - @dataclasses.dataclass class ApplicationValidation: - name: str version: str - main: str - files: dict[str, str] - shapes: list[TestCaseType] - settings: dict[str, Union[bool, float, int, str]] - schedule: ValidationSchedule - timeout: int = 900 - top_k: int = 10 - max_concurrency: int = 2 + source: str def __post_init__(self): - if not self.name or not self.version: - raise ValueError("validation name and version must be non-empty") - if self.main not in self.files: - raise ValueError(f"validation main file {self.main!r} is not in validation files") - if "@SUBMISSION@" not in self.files.values(): - raise ValueError("validation files must contain one @SUBMISSION@ source") - if not self.shapes: - raise ValueError("validation shapes must not be empty") - if self.timeout <= 0 or self.top_k <= 0 or self.max_concurrency <= 0: - raise ValueError("validation timeout, top_k, and max_concurrency must be positive") - - @classmethod - def from_dict(cls, data: dict) -> "ApplicationValidation": - values = copy.deepcopy(data) - values["schedule"] = ValidationSchedule(**values["schedule"]) - return cls(**values) + if not self.version or not self.source: + raise ValueError("validation version and source must be non-empty") @dataclasses.dataclass @@ -124,7 +89,7 @@ def from_dict(cls, data: dict): data_["ranking_by"] = criterion data_["multi_gpu"] = data.get("multi_gpu", False) if data.get("validation") is not None: - data_["validation"] = ApplicationValidation.from_dict(data["validation"]) + data_["validation"] = ApplicationValidation(**data["validation"]) if lang == Language.Python: data_["config"] = PythonTaskData(**data["config"]) else: @@ -193,16 +158,7 @@ def make_task_definition(yaml_file: str | Path) -> LeaderboardDefinition: # noq validation = raw.get("validation") if validation is not None: - validation_files = {} - for file_spec in validation["files"]: - name = file_spec["name"] - source = file_spec["source"] - validation_files[name] = ( - "@SUBMISSION@" - if source == "@SUBMISSION@" - else (root / source).read_text() - ) - validation["files"] = validation_files + validation["source"] = (root / validation.pop("script")).read_text() # load template files templates = {} @@ -291,14 +247,12 @@ def build_validation_config( raise KernelBotError("leaderboard does not define application validation") sources = { name: submission_content if content == "@SUBMISSION@" else content - for name, content in validation.files.items() + for name, content in task.files.items() } + sources["validation.py"] = validation.source return { - "name": validation.name, "version": validation.version, - "main": validation.main, + "main": "validation.py", "sources": sources, - "shapes": validation.shapes, - "settings": validation.settings, - "timeout": validation.timeout, + "timeout": 900, } diff --git a/src/migrations/20260726_01_application-validation.py b/src/migrations/20260726_01_application-validation.py index d36813ec8..6d9c45725 100644 --- a/src/migrations/20260726_01_application-validation.py +++ b/src/migrations/20260726_01_application-validation.py @@ -18,7 +18,6 @@ scheduled_for DATE, status TEXT NOT NULL CHECK (status IN ('running', 'completed', 'failed')), - top_k INTEGER NOT NULL CHECK (top_k > 0), started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), completed_at TIMESTAMPTZ, error TEXT, @@ -30,7 +29,6 @@ submission_id INTEGER NOT NULL REFERENCES leaderboard.submission(id) ON DELETE CASCADE, gpu_type TEXT NOT NULL, - contract_name TEXT NOT NULL, contract_version TEXT NOT NULL, status TEXT NOT NULL CHECK (status IN ('completed', 'failed')), diff --git a/tests/test_application_validation.py b/tests/test_application_validation.py index ab9b4c7ce..1002be7b9 100644 --- a/tests/test_application_validation.py +++ b/tests/test_application_validation.py @@ -7,10 +7,8 @@ from libkernelbot.application_validation import ( ApplicationValidationService, _due_date, - _kernelguard_precheck, ) from libkernelbot.task import LeaderboardTask -from libkernelbot.utils import KernelBotError def _task() -> LeaderboardTask: @@ -20,22 +18,8 @@ def _task() -> LeaderboardTask: "files": {"submission.py": "@SUBMISSION@"}, "config": {"main": "submission.py"}, "validation": { - "name": "toy-training", "version": "v1", - "main": "validation.py", - "files": { - "submission.py": "@SUBMISSION@", - "validation.py": "print('validator')", - }, - "shapes": [{"n": 1}, {"n": 2}], - "settings": {"min_speedup": 1.0}, - "schedule": { - "hour": 22, - "minute": 0, - "timezone": "America/Los_Angeles", - }, - "top_k": 10, - "max_concurrency": 2, + "source": "print('validator')", }, } ) @@ -81,7 +65,6 @@ def claim_validation_sweep( gpu_type, contract_version, scheduled_for, - top_k, ): key = (leaderboard_id, gpu_type, contract_version, scheduled_for) if scheduled_for is not None and key in self.claims: @@ -132,36 +115,20 @@ async def run_validation(self, config, _gpu): def test_due_date_uses_problem_timezone(): - validation = _task().validation - assert validation is not None - assert _due_date( - validation, datetime.datetime(2026, 7, 27, 4, 59, tzinfo=datetime.timezone.utc), ) is None assert _due_date( - validation, datetime.datetime(2026, 7, 27, 5, 0, tzinfo=datetime.timezone.utc), ) == datetime.date(2026, 7, 26) -def test_application_validation_requires_kernelguard(monkeypatch): - monkeypatch.delenv("KERNELGUARD_ENABLED", raising=False) - - with pytest.raises(KernelBotError, match="KernelGuard must be enabled"): - _kernelguard_precheck("def custom_kernel(): pass", "submission.py") - - @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}) - checked = [] - service = ApplicationValidationService( - backend, - precheck=lambda source, name: checked.append((source, name)), - ) + service = ApplicationValidationService(backend) summary = await service.run_sweep( "cholesky", @@ -173,10 +140,9 @@ async def test_sweep_validates_top_ten_with_bounded_concurrency(): assert len(summary["results"]) == 10 assert len(database.saved) == 10 assert launcher.max_active == 2 - assert len(checked) == 10 assert database.sweeps == [(1, "completed", None)] assert database.saved[0]["fully_validated"] is True - assert "unexpected_private_field" not in database.saved[0]["result"]["results"][0] + assert database.saved[0]["result"]["results"][0]["unexpected_private_field"] == "drop me" @pytest.mark.asyncio @@ -184,10 +150,7 @@ async def test_nightly_sweep_is_claimed_once(): database = FakeDB() launcher = FakeLauncher() backend = SimpleNamespace(db=database, launcher_map={"B200": launcher}) - service = ApplicationValidationService( - backend, - precheck=lambda _source, _name: None, - ) + service = ApplicationValidationService(backend) now = datetime.datetime(2026, 7, 27, 5, 1, tzinfo=datetime.timezone.utc) first = await service.run_due_once(now) diff --git a/tests/test_leaderboard_db.py b/tests/test_leaderboard_db.py index 56fe419e6..063f66edc 100644 --- a/tests/test_leaderboard_db.py +++ b/tests/test_leaderboard_db.py @@ -109,16 +109,8 @@ def test_application_validation_persistence(database, submit_leaderboard): ( json.dumps( { - "name": "toy-training", "version": "v1", - "main": "validation.py", - "files": { - "submission.py": "@SUBMISSION@", - "validation.py": "print('ok')", - }, - "shapes": [{"n": 1}], - "settings": {}, - "schedule": {"hour": 22}, + "source": "print('ok')", } ), submit_leaderboard, @@ -141,7 +133,6 @@ def test_application_validation_persistence(database, submit_leaderboard): gpu_type="B200", contract_version="v1", scheduled_for=scheduled_for, - top_k=10, ) assert sweep_id is not None assert db.claim_validation_sweep( @@ -149,13 +140,11 @@ def test_application_validation_persistence(database, submit_leaderboard): gpu_type="B200", contract_version="v1", scheduled_for=scheduled_for, - top_k=10, ) is None db.upsert_submission_validation( submission_id=submission_id, gpu_type="B200", - contract_name="toy-training", contract_version="v1", status="completed", passed_shapes=7, @@ -164,19 +153,20 @@ def test_application_validation_persistence(database, submit_leaderboard): geomean_sync_wall_speedup=1.2, result={"passed_shapes": 7, "total_shapes": 8}, ) - statuses = db.get_submission_validation_statuses( - [submission_id], - "B200", + 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 statuses[submission_id]["validation_shapes_passed"] == 7 - assert statuses[submission_id]["validation_shapes_total"] == 8 - assert statuses[submission_id]["validation_fully_validated"] is False - assert statuses[submission_id]["validation_geomean_speedup"] == 1.2 + assert db.cursor.fetchone() == (7, 8, False, 1.2) db.upsert_submission_validation( submission_id=submission_id, gpu_type="B200", - contract_name="toy-training", contract_version="v1", status="completed", passed_shapes=8, @@ -185,12 +175,15 @@ def test_application_validation_persistence(database, submit_leaderboard): geomean_sync_wall_speedup=1.3, result={"passed_shapes": 8, "total_shapes": 8}, ) - statuses = db.get_submission_validation_statuses( - [submission_id], - "B200", + 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 statuses[submission_id]["validation_fully_validated"] is True - assert statuses[submission_id]["validation_geomean_speedup"] == 1.3 + assert db.cursor.fetchone() == (True, 1.3) db.complete_validation_sweep(sweep_id, status="completed") diff --git a/tests/test_modal.py b/tests/test_modal.py index 9a5a40819..f3ac827b2 100644 --- a/tests/test_modal.py +++ b/tests/test_modal.py @@ -74,31 +74,6 @@ async def test_modal_validation_uses_b200_validation_function(): function.remote.aio.assert_awaited_once_with(config=config) -@pytest.mark.asyncio -async def test_modal_validation_can_target_debug_environment(): - launcher = ModalLauncher( - add_include_dirs=[], - environment_name="cholesky-validation-debug", - ) - function = MagicMock() - function.remote.aio = AsyncMock(return_value={"status": "completed"}) - - with patch( - "libkernelbot.launchers.modal.modal.Function.from_name", - return_value=function, - ) as from_name: - await launcher.run_validation( - {"version": "v1"}, - get_gpu_by_name("B200"), - ) - - from_name.assert_called_once_with( - "discord-bot-runner", - "run_validation_script_b200", - environment_name="cholesky-validation-debug", - ) - - @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 ee9d51e78..83130b34d 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -12,7 +12,6 @@ LeaderboardTask, PythonTaskData, RankCriterion, - ValidationSchedule, build_task_config, build_validation_config, make_task_definition, @@ -257,20 +256,8 @@ def test_multi_gpu_task(task_directory): def test_application_validation_roundtrip_and_config(leaderboard_task): leaderboard_task.validation = ApplicationValidation( - name="optimizer", version="optimizer-v1", - main="validate.py", - files={ - "validate.py": "print('validate')", - "submission.py": "@SUBMISSION@", - }, - shapes=[{"batch": 2, "n": 32, "steps": 4}], - settings={"min_speedup": 1.0, "require_no_torch_fallback": True}, - schedule=ValidationSchedule( - hour=22, - minute=0, - timezone="America/Los_Angeles", - ), + source="print('validate')", ) reconstructed = LeaderboardTask.from_str(leaderboard_task.to_str()) @@ -279,17 +266,12 @@ def test_application_validation_roundtrip_and_config(leaderboard_task): reconstructed, "def custom_kernel(x): return x", ) == { - "name": "optimizer", "version": "optimizer-v1", - "main": "validate.py", + "main": "validation.py", "sources": { - "validate.py": "print('validate')", - "submission.py": "def custom_kernel(x): return x", - }, - "shapes": [{"batch": 2, "n": 32, "steps": 4}], - "settings": { - "min_speedup": 1.0, - "require_no_torch_fallback": True, + "validation.py": "print('validate')", + "test.py": "code", + "main.py": "def custom_kernel(x): return x", }, "timeout": 900, } @@ -300,28 +282,13 @@ def test_make_task_definition_loads_validation_files(task_directory): task_yaml = (task_directory / "task.yml").read_text() task_yaml += """ validation: - name: optimizer version: optimizer-v1 - main: validation.py - files: - - {name: submission.py, source: "@SUBMISSION@"} - - {name: validation.py, source: validation.py} - shapes: - - {batch: 2, n: 32, steps: 4} - settings: - min_speedup: 1.0 - schedule: - hour: 22 - timezone: America/Los_Angeles + 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.files == { - "submission.py": "@SUBMISSION@", - "validation.py": "print('validation')", - } - assert validation.schedule.hour == 22 - assert validation.schedule.timezone == "America/Los_Angeles" + assert validation.version == "optimizer-v1" + assert validation.source == "print('validation')" From ed9df99996be24be1ec5b0b16cf1fa4f93f3808e Mon Sep 17 00:00:00 2001 From: Mark Saroufim Date: Sun, 26 Jul 2026 09:28:50 -0700 Subject: [PATCH 3/3] feat: add manual top-ten validation command --- docs/application-validation.md | 9 ++-- pyproject.toml | 3 ++ src/kernelbot/admin_cli.py | 62 +++++++++++++++++++++++++++ tests/test_admin_cli.py | 78 ++++++++++++++++++++++++++++++++++ 4 files changed, 149 insertions(+), 3 deletions(-) create mode 100644 src/kernelbot/admin_cli.py create mode 100644 tests/test_admin_cli.py diff --git a/docs/application-validation.md b/docs/application-validation.md index 5c2cb8fe3..46a9b9f4e 100644 --- a/docs/application-validation.md +++ b/docs/application-validation.md @@ -54,10 +54,13 @@ python src/kernelbot/main.py --api-only --debug Run one sweep synchronously: ```bash -curl -X POST \ - -H "Authorization: Bearer ${ADMIN_TOKEN}" \ - "http://localhost:8000/admin/application-validations/cholesky/B200?wait=true" +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/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()