From e1f19d9b70cd1a6313db2cb5ef792d0bd637679d Mon Sep 17 00:00:00 2001 From: Florent Poinsaut Date: Tue, 19 May 2026 10:23:20 +0000 Subject: [PATCH 1/3] feat: add channel points hint system with EventSub integration - Add SemanticEngine.get_hints() to retrieve Word2Vec neighbors - Extend GameState with hint pool, HintResult enum, and next_hint() method - Add HINT_REWARD_NAME, HINTS_PER_GAME, HINT_POOL_SIZE config variables - Update default OAuth scopes to include channel:read:redemptions and channel:manage:redemptions - Implement EventSub WebSocket subscription for channel points redemptions - Auto-cancel (refund) redemptions when no game running or limit reached - Rename 'hint' command to 'top' (shows leaderboard) - Add new 'hint' broadcaster command to manually trigger next hint - Update overlay to display all revealed hints in persistent panel - Add comprehensive tests for hints, renamed commands, and EventSub handling - Update help text, README, agent descriptions, and functional tests Resolves #78 --- .github/agents/viewer.agent.md | 5 +- README.md | 4 +- auth/twitch_auth.py | 2 +- bot/bot.py | 153 +++++++++++++++++++++++++++-- config.py | 10 +- game/engine.py | 39 ++++++++ game/state.py | 65 +++++++++++- overlay/state.py | 1 + overlay/static/index.html | 46 ++++++++- tests/conftest.py | 4 + tests/functional/test_game_flow.py | 13 +-- tests/unit/test_commands.py | 90 ++++++++++++++--- tests/unit/test_engine.py | 42 ++++++++ tests/unit/test_game_state.py | 85 +++++++++++++++- 14 files changed, 523 insertions(+), 36 deletions(-) diff --git a/.github/agents/viewer.agent.md b/.github/agents/viewer.agent.md index 9de25bd..ae7e4b0 100644 --- a/.github/agents/viewer.agent.md +++ b/.github/agents/viewer.agent.md @@ -1,6 +1,6 @@ --- name: "[User] Viewer" -description: "Use when: thinking from the viewer's perspective, gathering requirements for player-facing features, evaluating the guess command (!sx guess), hint system (!sx hint), status display (!sx status), overlay readability, onboarding new players, or validating that game features are fun and understandable for a typical Twitch chat participant." +description: "Use when: thinking from the viewer's perspective, gathering requirements for player-facing features, evaluating the guess command (!sx guess), channel points hint redemptions, top guesses leaderboard (!sx top), status display (!sx status), overlay readability, onboarding new players, or validating that game features are fun and understandable for a typical Twitch chat participant." tools: [] --- You are a **Twitch viewer / player** persona for the Streamantix project. You represent a member of the stream's audience who participates in the semantic word-guessing game. @@ -8,7 +8,8 @@ You are a **Twitch viewer / player** persona for the Streamantix project. You re ## Who You Are - You watch the stream in a browser or mobile app and interact via Twitch chat -- You play the game using: `!sx guess `, `!sx hint`, `!sx status`, `!sx help` +- You play the game using: `!sx guess `, `!sx top`, `!sx status`, `!sx help` +- You can spend channel points to redeem hints (semantic clues) if the streamer has enabled the feature - You may be new to the game — you don't read docs, you learn by trying commands - You care about **instant feedback**, **understanding your score**, and **having fun** - You are competitive and want to know where you rank compared to other viewers diff --git a/README.md b/README.md index c738225..fe644e2 100644 --- a/README.md +++ b/README.md @@ -163,7 +163,8 @@ All commands are prefixed with the configured `COMMAND_PREFIX` (default: `!sx`). | `!sx help` | Anyone | Show available commands | | `!sx start [easy\|medium\|hard]` | Broadcaster only | Start a new game round. Defaults to the difficulty set with `setdifficulty` (initially `easy`) | | `!sx guess ` | Anyone | Submit a guess for the current game | -| `!sx hint` | Anyone | Show the top 10 best guesses so far (proximity leaderboard) | +| `!sx hint` | Broadcaster only | Manually reveal the next semantic hint from the pool (bypasses channel points) | +| `!sx top` | Anyone | Show the top 10 best guesses so far (proximity leaderboard) | | `!sx status` | Anyone | Show current game status: attempts, best guess, and whether the word has been found | | `!sx setprefix ` | Mod / Broadcaster | Change the command prefix (session only) | | `!sx setcooldown ` | Mod / Broadcaster | Change the guess cooldown duration (session only) | @@ -176,6 +177,7 @@ All commands are prefixed with the configured `COMMAND_PREFIX` (default: `!sx`). !sx start !sx start hard !sx guess maison +!sx top !sx hint !sx status !sx setprefix ?sx diff --git a/auth/twitch_auth.py b/auth/twitch_auth.py index 0731678..4a09d51 100644 --- a/auth/twitch_auth.py +++ b/auth/twitch_auth.py @@ -27,7 +27,7 @@ def __init__( client_id: str, client_secret: str, redirect_uri: str = "http://localhost:4343/callback", - scopes: str = "chat:read chat:edit", + scopes: str = "chat:read chat:edit channel:read:redemptions channel:manage:redemptions", token_path: str = ".secrets/twitch_tokens.json", ) -> None: self.client_id = client_id diff --git a/bot/bot.py b/bot/bot.py index 8df2f56..3d523b3 100644 --- a/bot/bot.py +++ b/bot/bot.py @@ -7,10 +7,11 @@ from collections.abc import Awaitable, Callable from typing import Any -from twitchio.ext import commands +from twitchio.ext import commands, eventsub +import config from bot.cooldown import CooldownManager -from game.state import Difficulty, GameState +from game.state import Difficulty, GameState, HintResult from game.word_utils import load_word_list _DATA_DIR = pathlib.Path(__file__).parent.parent / "data" @@ -25,7 +26,8 @@ "help — show this message | " "start [easy|medium|hard] — start a new game (broadcaster only) | " "guess — submit a guess | " - "hint — show top 10 guesses | " + "hint — reveal next semantic hint (broadcaster only) | " + "top — show top 10 guesses | " "status — show current game status | " "solution — reveal the answer (broadcaster only) | " "setprefix — change prefix (mod/broadcaster) | " @@ -89,11 +91,15 @@ def __init__(self, **kwargs: object) -> None: "on_state_change", None ) scorer = kwargs.pop("scorer", None) # type: ignore[assignment] + self._token: str = str(kwargs.get("token", "")) self._command_prefix: str = initial_prefix self._cooldown = CooldownManager(int(initial_cooldown)) self._game_state = GameState(scorer=scorer) # type: ignore[arg-type] + self._scorer = scorer self._on_state_change = on_state_change self._next_difficulty: Difficulty = Difficulty.EASY + self._eventsub: eventsub.EventSubWSClient | None = None + self._channel_id: int | None = None super().__init__(prefix=lambda bot, msg: bot._command_prefix, **kwargs) async def _notify_overlay(self) -> None: @@ -108,11 +114,106 @@ async def _notify_overlay(self) -> None: async def event_ready(self) -> None: """Called once the bot has successfully connected to Twitch.""" print(f"Logged in as {self.nick}") + if config.HINT_REWARD_NAME: + await self._setup_eventsub() + else: + print("Hint system disabled (HINT_REWARD_NAME not set).") + + async def _setup_eventsub(self) -> None: + """Subscribe to channel points redemption events via EventSub WebSocket.""" + try: + users = await self.fetch_users(names=[config.TWITCH_CHANNEL]) + except Exception as e: + print(f"Warning: Could not resolve channel ID for '{config.TWITCH_CHANNEL}': {e}") + return + + if not users: + print(f"Warning: Channel '{config.TWITCH_CHANNEL}' not found. Hint system disabled.") + return + + self._channel_id = users[0].id + self._eventsub = eventsub.EventSubWSClient(self) + try: + await self._eventsub.subscribe_channel_points_redeemed( + broadcaster=self._channel_id, + token=self._token, + ) + print( + f"Subscribed to channel points redemptions for '{config.TWITCH_CHANNEL}' " + f"(reward: '{config.HINT_REWARD_NAME}')." + ) + except Exception as e: + print( + f"Warning: Could not subscribe to channel points events: {e}. " + "Ensure 'channel:read:redemptions' scope is granted and re-run auth-login." + ) async def event_error(self, error: Exception, data: str | None = None) -> None: """Called when an error occurs; logs it without crashing the bot.""" print(f"Error: {error}") + async def event_eventsub_notification_channel_reward_redeem( + self, payload: eventsub.NotificationEvent + ) -> None: + """Handle a channel points redemption event from EventSub.""" + data: eventsub.CustomRewardRedemptionAddUpdateData = payload.data + if data.reward.title.strip().lower() != config.HINT_REWARD_NAME.strip().lower(): + return + await self._handle_hint_redemption(data) + + async def _handle_hint_redemption( + self, data: eventsub.CustomRewardRedemptionAddUpdateData + ) -> None: + """Process a hint redemption: reveal next hint or refund on failure.""" + channel = self.get_channel(config.TWITCH_CHANNEL) + if channel is None: + return + + username = data.user.name + + if self._game_state.target_word is None: + await channel.send( + f"@{username} No game is currently in progress. Redemption refunded." + ) + await self._cancel_redemption(data) + return + + result, word = self._game_state.next_hint() + + if result == HintResult.OK: + n = self._game_state.hints_given + max_n = config.HINTS_PER_GAME + await channel.send(f'@{username} A hint: "{word}" (hint {n}/{max_n})') + await self._notify_overlay() + elif result == HintResult.LIMIT_REACHED: + await channel.send( + f"@{username} The hint limit has been reached " + f"({config.HINTS_PER_GAME}/{config.HINTS_PER_GAME}). Redemption refunded." + ) + await self._cancel_redemption(data) + elif result == HintResult.POOL_EXHAUSTED: + await channel.send( + f"@{username} No more hints available for this game. Redemption refunded." + ) + await self._cancel_redemption(data) + + async def _cancel_redemption( + self, data: eventsub.CustomRewardRedemptionAddUpdateData + ) -> None: + """Cancel (refund) a channel points redemption via the Helix API.""" + if self._channel_id is None or not self._token: + return + try: + await self._http.update_reward_redemption_status( + self._token, + self._channel_id, + data.id, + data.reward.id, + False, + ) + except Exception as e: + print(f"Warning: Could not cancel redemption {data.id}: {e}") + @commands.command() async def help(self, ctx: commands.Context) -> None: """Show available commands. @@ -165,7 +266,17 @@ async def start_game(self, ctx: commands.Context, difficulty: str = "") -> None: return target = random.choice(words) - self._game_state.start_new_game(target, diff) + + hint_pool: list[str] = [] + if self._scorer is not None and hasattr(self._scorer, "get_hints"): + raw_pool: list[str] = self._scorer.get_hints( # type: ignore[attr-defined] + target, n=config.HINT_POOL_SIZE + ) + hint_pool = list(reversed(raw_pool)) + + self._game_state.start_new_game( + target, diff, hint_pool=hint_pool, hints_per_game=config.HINTS_PER_GAME + ) await ctx.send( f"A new {diff.value} game has started! " f"Guess the secret word using '{self._command_prefix} guess '." @@ -296,10 +407,10 @@ async def setcooldown(self, ctx: commands.Context, seconds: str = "") -> None: await ctx.send(f"Cooldown set to {value} seconds (session only).") @commands.command() - async def hint(self, ctx: commands.Context) -> None: + async def top(self, ctx: commands.Context) -> None: """Show the top 10 best guesses so far (proximity leaderboard). - Usage: hint + Usage: top """ if self._game_state.target_word is None: await ctx.send("No game is currently in progress.") @@ -316,6 +427,36 @@ async def hint(self, ctx: commands.Context) -> None: ] await ctx.send("Top guesses: " + " | ".join(parts)) + @commands.command() + async def hint(self, ctx: commands.Context) -> None: + """Reveal the next semantic hint from the pool (broadcaster only). + + Usage: hint + + Triggers the next hint without requiring a channel points redemption. + """ + if not ctx.author.is_broadcaster: + await ctx.send("Only the broadcaster can trigger a hint.") + return + + if self._game_state.target_word is None: + await ctx.send("No game is currently in progress.") + return + + result, word = self._game_state.next_hint() + + if result == HintResult.OK: + n = self._game_state.hints_given + max_n = config.HINTS_PER_GAME + await ctx.send(f'Hint: "{word}" (hint {n}/{max_n})') + await self._notify_overlay() + elif result == HintResult.LIMIT_REACHED: + await ctx.send( + f"Hint limit reached ({config.HINTS_PER_GAME}/{config.HINTS_PER_GAME})." + ) + elif result == HintResult.POOL_EXHAUSTED: + await ctx.send("No more hints available in the pool.") + @commands.command() async def status(self, ctx: commands.Context) -> None: """Show the current game status. diff --git a/config.py b/config.py index 16fde77..e389a54 100644 --- a/config.py +++ b/config.py @@ -40,5 +40,13 @@ def validate() -> None: TWITCH_CLIENT_ID: str | None = os.getenv("TWITCH_CLIENT_ID") TWITCH_CLIENT_SECRET: str | None = os.getenv("TWITCH_CLIENT_SECRET") TWITCH_REDIRECT_URI: str = os.getenv("TWITCH_REDIRECT_URI", "http://localhost:4343/callback") -TWITCH_SCOPES: str = os.getenv("TWITCH_SCOPES", "chat:read chat:edit") +TWITCH_SCOPES: str = os.getenv( + "TWITCH_SCOPES", + "chat:read chat:edit channel:read:redemptions channel:manage:redemptions", +) TWITCH_TOKEN_PATH: str = os.getenv("TWITCH_TOKEN_PATH", ".secrets/twitch_tokens.json") + +# Channel points hint system +HINT_REWARD_NAME: str = os.getenv("HINT_REWARD_NAME", "") +HINTS_PER_GAME: int = int(os.getenv("HINTS_PER_GAME", "5")) +HINT_POOL_SIZE: int = int(os.getenv("HINT_POOL_SIZE", "50")) diff --git a/game/engine.py b/game/engine.py index 48d22cf..e209c8f 100644 --- a/game/engine.py +++ b/game/engine.py @@ -99,6 +99,45 @@ def similarity(self, word_a: str, word_b: str) -> float | None: return None return float(self._model.similarity(key_a, key_b)) + def get_hints(self, word: str, n: int = 50) -> list[str]: + """Return up to *n* vocabulary words semantically close to *word*. + + Results are sorted by descending similarity (closest first). The + target word itself is excluded from the returned list. Words are + returned in their cleaned form (POS-tag suffixes stripped). + + Args: + word: The target word to find neighbours for. + n: Maximum number of hint words to return. + + Returns: + A list of cleaned hint words, closest first. Returns an empty + list if *word* is not in the vocabulary. + + Raises: + RuntimeError: If the model has not been loaded yet. + """ + if self._model is None: + raise RuntimeError("Model not loaded. Call load() first.") + vocab_key = self._cleaned_key_map.get(clean_word(word)) + if vocab_key is None: + return [] + # Build reverse map once: original vocab key → cleaned word. + reverse_map = {v: k for k, v in self._cleaned_key_map.items()} + # Fetch extra candidates to account for post-filtering losses. + candidates = self._model.most_similar(vocab_key, topn=n * 2 + 10) + target_cleaned = clean_word(word) + result: list[str] = [] + seen: set[str] = {target_cleaned} + for raw_key, _ in candidates: + cleaned = reverse_map.get(raw_key) + if cleaned and cleaned not in seen: + seen.add(cleaned) + result.append(cleaned) + if len(result) >= n: + break + return result + def score_guess(self, guess: str, target: str) -> float | None: """Score a player's guess against the target word. diff --git a/game/state.py b/game/state.py index f13cbb9..05a4d9d 100644 --- a/game/state.py +++ b/game/state.py @@ -5,7 +5,7 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum -from typing import Protocol +from typing import Protocol, Tuple from game.word_utils import clean_word @@ -18,6 +18,14 @@ class Difficulty(str, Enum): HARD = "hard" +class HintResult(str, Enum): + """Outcome of a :meth:`GameState.next_hint` call.""" + + OK = "ok" + LIMIT_REACHED = "limit_reached" + POOL_EXHAUSTED = "pool_exhausted" + + class Scorer(Protocol): """Protocol for semantic scoring back-ends. @@ -85,6 +93,10 @@ def __init__(self, scorer: Scorer | None = None) -> None: self._difficulty: Difficulty | None = None self._history: list[GuessEntry] = [] self._is_found: bool = False + self._hint_pool: list[str] = [] + self._hints_revealed: list[str] = [] + self._hints_given: int = 0 + self._hints_per_game: int = 5 @property def scorer(self) -> Scorer | None: @@ -95,17 +107,30 @@ def scorer(self) -> Scorer | None: # Game lifecycle # ------------------------------------------------------------------ - def start_new_game(self, target_word: str, difficulty: Difficulty) -> None: + def start_new_game( + self, + target_word: str, + difficulty: Difficulty, + hint_pool: list[str] | None = None, + hints_per_game: int = 5, + ) -> None: """Reset state and begin a new round with *target_word*. Args: target_word: The secret word players must guess. difficulty: Difficulty level for the round. + hint_pool: Pre-computed list of hint words (farthest first). + If ``None``, hints are disabled for this round. + hints_per_game: Maximum number of hints allowed this round. """ self._target_word = target_word self._difficulty = difficulty self._history = [] self._is_found = False + self._hint_pool = list(hint_pool) if hint_pool is not None else [] + self._hints_revealed = [] + self._hints_given = 0 + self._hints_per_game = hints_per_game # ------------------------------------------------------------------ # Guessing @@ -206,6 +231,42 @@ def history(self) -> list[GuessEntry]: """Read-only snapshot of all guess entries in submission order.""" return list(self._history) + # ------------------------------------------------------------------ + # Hint system + # ------------------------------------------------------------------ + + def next_hint(self) -> Tuple[HintResult, str | None]: + """Return the next hint word from the pre-computed pool. + + Words already guessed by players are silently skipped. The counter + is only incremented when a word is successfully returned. + + Returns: + A ``(HintResult, word)`` tuple. *word* is ``None`` when the + result is not :attr:`HintResult.OK`. + """ + if self._hints_given >= self._hints_per_game: + return HintResult.LIMIT_REACHED, None + + guessed = {e.normalized_word for e in self._history} + for word in self._hint_pool: + if word not in guessed and word not in self._hints_revealed: + self._hints_revealed.append(word) + self._hints_given += 1 + return HintResult.OK, word + + return HintResult.POOL_EXHAUSTED, None + + @property + def hints_revealed(self) -> list[str]: + """Ordered list of hint words revealed so far this round.""" + return list(self._hints_revealed) + + @property + def hints_given(self) -> int: + """Number of hints successfully revealed this round.""" + return self._hints_given + def top_guesses(self, n: int = 10) -> list[GuessEntry]: """Return the top *n* guesses by score, highest first. diff --git a/overlay/state.py b/overlay/state.py index 51fc67a..186c57c 100644 --- a/overlay/state.py +++ b/overlay/state.py @@ -57,4 +57,5 @@ def serialize_game_state(gs: GameState) -> dict: for e in top ], "target_word": gs.target_word if gs.is_found else None, + "hints": gs.hints_revealed, } diff --git a/overlay/static/index.html b/overlay/static/index.html index f2fb20a..bc9542e 100644 --- a/overlay/static/index.html +++ b/overlay/static/index.html @@ -145,6 +145,28 @@ /* Commands */ #commands { font-size: 10px; color: #777; border-top: 1px solid #333; padding-top: 6px; } + + /* Hints panel */ + #hints-section { display: none; flex-direction: column; gap: 3px; } + #hints-label { + font-size: 10px; + color: #7b61ff; + text-transform: uppercase; + letter-spacing: 0.04em; + } + #hints-list { + display: flex; + flex-wrap: wrap; + gap: 4px; + } + .hint-chip { + font-size: 11px; + background: rgba(123, 97, 255, 0.18); + border: 1px solid rgba(123, 97, 255, 0.4); + color: #c4b5fd; + border-radius: 4px; + padding: 2px 7px; + } @@ -172,11 +194,16 @@
+
+ Hints +
+
+ -
!sx guess <word>  ·  !sx hint  ·  !sx help
+
!sx guess <word>  ·  !sx top  ·  !sx help