Announce top-three leaderboard changes

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)

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

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)