diff --git a/src/kernelbot/cogs/top_three_cog.py b/src/kernelbot/cogs/top_three_cog.py new file mode 100644 index 000000000..60999ea9b --- /dev/null +++ b/src/kernelbot/cogs/top_three_cog.py @@ -0,0 +1,86 @@ +import asyncio +import datetime + +from discord.ext import commands, tasks + +from kernelbot.top_three import detect_podium_change, format_podium_change +from libkernelbot.db_types import LeaderboardRankedEntry +from libkernelbot.utils import setup_logging + +logger = setup_logging(__name__) + + +class TopThreeCog(commands.Cog): + """Announce leaderboard podium changes, including API-originated submissions.""" + + def __init__(self, bot): + self.bot = bot + self._standings: dict[tuple[str, str], list[LeaderboardRankedEntry]] = {} + self._poll_lock = asyncio.Lock() + self.watch_top_three.start() + + def cog_unload(self): + self.watch_top_three.cancel() + + def _read_standings(self): + standings = {} + now = datetime.datetime.now(datetime.timezone.utc) + with self.bot.leaderboard_db as db: + for leaderboard in db.get_leaderboards(): + if leaderboard["visibility"] != "public" or leaderboard["deadline"] <= now: + continue + for gpu_type in leaderboard["gpu_types"]: + key = (leaderboard["name"], gpu_type) + standings[key] = db.get_leaderboard_submissions( + leaderboard["name"], gpu_type, limit=3 + ) + return standings + + async def seed(self): + self._standings = self._read_standings() + logger.info("Seeded top-three watcher with %d leaderboard/GPU pairs", len(self._standings)) + + async def poll_once(self): + async with self._poll_lock: + current = self._read_standings() + changes = [] + for key, after in current.items(): + before = self._standings.get(key, []) + change = detect_podium_change(*key, before, after) + if change is not None: + changes.append((key, change)) + else: + self._standings[key] = after + + for removed_key in self._standings.keys() - current.keys(): + del self._standings[removed_key] + if not changes: + return + + channel = self.bot.get_channel(self.bot.leaderboard_submissions_id) + if channel is None: + logger.error("Could not find leaderboard submissions channel") + return + for key, change in changes: + await channel.send(format_podium_change(change)) + # Only acknowledge after Discord accepts it, so transient failures retry. + self._standings[key] = current[key] + + @tasks.loop(seconds=15) + async def watch_top_three(self): + try: + await self.poll_once() + except Exception: + # Keep the watcher alive through a transient DB or Discord failure. + logger.exception("Top-three poll failed") + + @watch_top_three.before_loop + async def before_watch_top_three(self): + await self.bot.wait_until_ready() + while not hasattr(self.bot, "leaderboard_submissions_id"): + await asyncio.sleep(0.1) + await self.seed() + + @watch_top_three.error + async def watch_top_three_error(self, error): + logger.exception("Top-three watcher failed", exc_info=error) diff --git a/src/kernelbot/main.py b/src/kernelbot/main.py index 517fad020..398d17166 100644 --- a/src/kernelbot/main.py +++ b/src/kernelbot/main.py @@ -8,6 +8,7 @@ from cogs.admin_cog import AdminCog from cogs.leaderboard_cog import LeaderboardCog from cogs.misc_cog import BotManagerCog +from cogs.top_three_cog import TopThreeCog from cogs.verify_run_cog import VerifyRunCog from discord import app_commands from discord.ext import commands @@ -87,6 +88,7 @@ async def setup_hook(self): await self.add_cog(LeaderboardCog(self)) await self.add_cog(VerifyRunCog(self)) await self.add_cog(AdminCog(self)) + await self.add_cog(TopThreeCog(self)) guild_id = ( env.DISCORD_CLUSTER_STAGING_ID diff --git a/src/kernelbot/top_three.py b/src/kernelbot/top_three.py new file mode 100644 index 000000000..87306440f --- /dev/null +++ b/src/kernelbot/top_three.py @@ -0,0 +1,125 @@ +import random +from dataclasses import dataclass +from typing import Callable, Sequence, TypeVar + +from libkernelbot.db_types import LeaderboardRankedEntry + +T = TypeVar("T") +Chooser = Callable[[Sequence[T]], T] + +WINNER_LINES = ( + "Beautiful kernel. Absolutely no notes.", + "Someone check the profiler; that run left scorch marks.", + "The leaderboard has been optimized against its will.", + "That kernel did not ask permission before going fast.", + "Fresh crown, fewer nanoseconds. You love to see it.", + "The compiler watched this happen and felt something.", +) + +ENTRANT_LINES = ( + "The podium has a new problem.", + "Three chairs, one newly occupied. Things are getting spicy.", + "A wild contender has entered the profiler trace.", + "The top three just got a little less comfortable.", + "Turns out the velvet rope was only a suggestion.", +) + +EVICTION_LINES = ( + "gpumode.com will remember the good times.", + "Please collect your registers on the way out.", + "The podium has reclaimed your chair for higher utilization.", + "Your top-three residency has been garbage-collected.", + "You have been preempted. No checkpoint was saved.", + "The leaderboard called: your free trial has expired.", + "Back to the profiler mines. The nanoseconds demand tribute.", + "Your kernel is now a historical benchmark. Very educational.", +) + +DETHRONED_LINES = ( + "The crown is in another CUDA stream now.", + "First place was apparently a temporary allocation.", + "The throne had an eviction policy. Awkward.", + "You are now the reference implementation for getting passed.", + "The leaderboard diff says `-1 crown`. Brutal.", + "Your reign had excellent throughput and terrible retention.", +) + + +@dataclass(frozen=True) +class PodiumChange: + leaderboard_name: str + gpu_type: str + winner: LeaderboardRankedEntry | None + entrants: tuple[LeaderboardRankedEntry, ...] + departures: tuple[LeaderboardRankedEntry, ...] + dethroned: LeaderboardRankedEntry | None + + +def detect_podium_change( + leaderboard_name: str, + gpu_type: str, + before: list[LeaderboardRankedEntry], + after: list[LeaderboardRankedEntry], +) -> PodiumChange | None: + """Describe a top-three membership or winner change.""" + before = before[:3] + after = after[:3] + before_ids = [str(entry["user_id"]) for entry in before] + after_ids = [str(entry["user_id"]) for entry in after] + + winner_changed = bool(after_ids) and (not before_ids or before_ids[0] != after_ids[0]) + entrants = tuple(entry for entry in after if str(entry["user_id"]) not in before_ids) + departures = tuple(entry for entry in before if str(entry["user_id"]) not in after_ids) + + if not winner_changed and not entrants and not departures: + return None + + return PodiumChange( + leaderboard_name=leaderboard_name, + gpu_type=gpu_type, + winner=after[0] if winner_changed else None, + entrants=entrants, + departures=departures, + dethroned=before[0] if winner_changed and before else None, + ) + + +def display_name(entry: LeaderboardRankedEntry) -> str: + """Use the leaderboard's public username without creating a Discord ping.""" + return f"**{entry['user_name']}**" + + +def format_podium_change( + change: PodiumChange, + choose: Chooser[str] = random.choice, +) -> str: + context = f"`{change.leaderboard_name}` on `{change.gpu_type}`" + lines: list[str] = [] + + if change.winner is not None: + lines.append( + f"🏆 {display_name(change.winner)} just took **#1** on {context}. " + f"{choose(WINNER_LINES)}" + ) + else: + for entrant in change.entrants: + lines.append( + f"🔥 {display_name(entrant)} just broke into the **top 3** on {context}. " + f"{choose(ENTRANT_LINES)}" + ) + + roasted_ids: set[str] = set() + for departure in change.departures: + roasted_ids.add(str(departure["user_id"])) + lines.append( + f"🗑️ {display_name(departure)} has been evicted from the top 3. " + f"{choose(EVICTION_LINES)}" + ) + + if change.dethroned is not None and str(change.dethroned["user_id"]) not in roasted_ids: + lines.append( + f"👑➡️🥈 {display_name(change.dethroned)} has been dethroned. " + f"{choose(DETHRONED_LINES)}" + ) + + return "\n".join(lines) diff --git a/tests/test_top_three.py b/tests/test_top_three.py new file mode 100644 index 000000000..eeaa75776 --- /dev/null +++ b/tests/test_top_three.py @@ -0,0 +1,205 @@ +import asyncio +import datetime +from unittest.mock import AsyncMock, Mock + +import pytest + +from kernelbot.cogs.top_three_cog import TopThreeCog +from kernelbot.top_three import ( + EVICTION_LINES, + detect_podium_change, + display_name, + format_podium_change, +) + + +def entry(user_id, name, rank): + return { + "submission_id": rank, + "rank": rank, + "submission_name": f"{name}.py", + "submission_time": None, + "submission_score": float(rank), + "leaderboard_name": "vector-add", + "user_id": user_id, + "user_name": name, + "gpu_type": "H100", + } + + +def first(options): + return options[0] + + +def test_new_winner_and_departure_are_both_called_out(): + before = [entry("1", "old-winner", 1), entry("2", "second", 2), entry("3", "third", 3)] + after = [entry("4", "new-winner", 1), entry("1", "old-winner", 2), entry("2", "second", 3)] + + change = detect_podium_change("vector-add", "H100", before, after) + message = format_podium_change(change, choose=first) + + assert change.winner["user_id"] == "4" + assert [item["user_id"] for item in change.departures] == ["3"] + assert "**new-winner** just took **#1**" in message + assert "**third** has been evicted" in message + assert "**old-winner** has been dethroned" in message + + +def test_new_third_place_without_winner_change(): + before = [entry("1", "first", 1), entry("2", "second", 2), entry("3", "third", 3)] + after = [entry("1", "first", 1), entry("2", "second", 2), entry("4", "entrant", 3)] + + change = detect_podium_change("vector-add", "H100", before, after) + message = format_podium_change(change, choose=first) + + assert change.winner is None + assert "**entrant** just broke into the **top 3**" in message + assert "**third** has been evicted" in message + + +def test_personal_best_with_same_podium_is_silent(): + before = [entry("1", "first", 1), entry("2", "second", 2), entry("3", "third", 3)] + after = [entry("1", "first", 1), entry("2", "second", 2), entry("3", "third", 3)] + after[0]["submission_score"] = 0.5 + + assert detect_podium_change("vector-add", "H100", before, after) is None + + +@pytest.mark.parametrize("existing_count", [0, 1, 2]) +def test_filling_first_three_slots_never_trashtalks(existing_count): + before = [entry(str(i), f"user-{i}", i) for i in range(1, existing_count + 1)] + newcomer = entry(str(existing_count + 1), f"user-{existing_count + 1}", existing_count + 1) + + change = detect_podium_change("vector-add", "H100", before, [*before, newcomer]) + message = format_podium_change(change, choose=first) + + assert change.departures == () + assert "evicted" not in message + assert "dethroned" not in message + + +def test_display_name_uses_public_leaderboard_username_without_ping(): + assert display_name(entry("12345", "octocat", 1)) == "**octocat**" + + +def test_trashtalk_has_variety(): + assert len(EVICTION_LINES) >= 8 + assert len(set(EVICTION_LINES)) == len(EVICTION_LINES) + + +@pytest.mark.asyncio +async def test_watcher_sends_api_visible_database_change_to_submissions_channel(): + before = [entry("1", "first", 1), entry("2", "second", 2), entry("3", "third", 3)] + after = [entry("4", "new-first", 1), entry("1", "first", 2), entry("2", "second", 3)] + channel = Mock(send=AsyncMock()) + bot = Mock(leaderboard_submissions_id=99) + bot.get_channel.return_value = channel + + watcher = object.__new__(TopThreeCog) + watcher.bot = bot + watcher._poll_lock = asyncio.Lock() + watcher._standings = {("vector-add", "H100"): before} + watcher._read_standings = Mock(return_value={("vector-add", "H100"): after}) + + await watcher.poll_once() + + channel.send.assert_awaited_once() + assert "**new-first** just took **#1**" in channel.send.await_args.args[0] + assert "**third** has been evicted" in channel.send.await_args.args[0] + assert watcher._standings[("vector-add", "H100")] == after + + +@pytest.mark.asyncio +async def test_end_to_end_seed_api_update_and_discord_send(): + before = [entry("1", "first", 1), entry("2", "second", 2), entry("3", "third", 3)] + after = [entry("4", "speed-demon", 1), entry("1", "first", 2), entry("2", "second", 3)] + + class FakeDB: + standings = before + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def get_leaderboards(self): + return [ + { + "name": "vector-add", + "gpu_types": ["H100"], + "visibility": "public", + "deadline": datetime.datetime.now(datetime.timezone.utc) + + datetime.timedelta(days=1), + } + ] + + def get_leaderboard_submissions(self, leaderboard, gpu, limit): + assert (leaderboard, gpu, limit) == ("vector-add", "H100", 3) + return self.standings + + db = FakeDB() + channel = Mock(send=AsyncMock()) + bot = Mock(leaderboard_db=db, leaderboard_submissions_id=99) + bot.get_channel.return_value = channel + watcher = object.__new__(TopThreeCog) + watcher.bot = bot + watcher._poll_lock = asyncio.Lock() + watcher._standings = {} + + await watcher.seed() + channel.send.assert_not_awaited() # Existing podium is silent at startup. + + db.standings = after # The API worker commits a newly ranked run. + await watcher.poll_once() + + channel.send.assert_awaited_once() + message = channel.send.await_args.args[0] + assert "**speed-demon** just took **#1**" in message + assert "**third** has been evicted" in message + assert "<@" not in message + + +def test_watcher_ignores_expired_and_closed_competitions(): + now = datetime.datetime.now(datetime.timezone.utc) + + class FakeDB: + queried = [] + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def get_leaderboards(self): + return [ + { + "name": "active-public", + "gpu_types": ["H100"], + "visibility": "public", + "deadline": now + datetime.timedelta(days=1), + }, + { + "name": "expired", + "gpu_types": ["H100"], + "visibility": "public", + "deadline": now - datetime.timedelta(days=1), + }, + { + "name": "closed", + "gpu_types": ["H100"], + "visibility": "closed", + "deadline": now + datetime.timedelta(days=1), + }, + ] + + def get_leaderboard_submissions(self, leaderboard, gpu, limit): + self.queried.append((leaderboard, gpu, limit)) + return [] + + watcher = object.__new__(TopThreeCog) + watcher.bot = Mock(leaderboard_db=FakeDB()) + + assert watcher._read_standings() == {("active-public", "H100"): []} + assert watcher.bot.leaderboard_db.queried == [("active-public", "H100", 3)]