""" game_engine.py — Core Game State & Turn Processing =================================================== Contains the GameState class, stat management, hidden scenario triggers, and the main turn-processing pipeline for the ORV Scenario Simulator. Part of the ORV (Omniscient Reader's Viewpoint) Scenario Simulator. Build Small Hackathon 2026. """ import copy import json from typing import Any from scenarios import SCENARIOS # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # CONSTELLATION NAME MAPPING # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ _CONSTELLATION_DISPLAY_NAMES: dict[str, str] = { "abyssal_black_flame_dragon": "Abyssal Black Flame Dragon", "secretive_plotter": "Secretive Plotter", "demon_like_judge": "Demon-like Judge of Fire", "maritime_war_god": "Maritime War God", "prisoner_golden_headband": "Prisoner of the Golden Headband", } _TRUST_DISPLAY_NAMES: dict[str, str] = { "yoo_sangah": "Yoo Sangah", "lee_hyunsung": "Lee Hyunsung", "jung_heewon": "Jung Heewon", "han_sooyoung": "Han Sooyoung", } # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # GAME STATE CLASS # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ class GameState: """ Encapsulates the full mutable state of an ORV Scenario Simulator session. The internal ``state`` dict is the canonical representation and is serialisable to / from plain dicts for Gradio ``gr.State`` transport. """ def __init__(self) -> None: self.state: dict[str, Any] = { "player_name": "Kim Dokja", "hp": 100, "max_hp": 100, "coins": 0, "scenario_num": 1, "turn": 1, "total_turns": 0, "current_phase": "Exploration", "active_debuffs": [], "meta_exposure": 0, "prob_stability": 100, "entertainment": 0, "titles": [], "attributes": ["Omniscient Reader's Viewpoint (Hidden)"], "skills": { "Fourth Wall": {"level": 1, "cooldown": 0, "max_cooldown": 2}, "Bookmark": {"level": 1, "cooldown": 0, "max_cooldown": 3}, "Omniscient Reader's Viewpoint": {"level": 1, "cooldown": 0, "max_cooldown": 4} }, "inventory": [], "sponsor": None, "trust": { "yoo_sangah": 0, "lee_hyunsung": 0, "jung_heewon": 0, "han_sooyoung": 0, }, "constellation_affinity": { "abyssal_black_flame_dragon": 0, "secretive_plotter": 0, "demon_like_judge": 0, "maritime_war_god": 0, "prisoner_golden_headband": 0, }, "unique_constellations": [], "action_history": [], "action_tags": [], "hidden_scenarios_triggered": [], "scenario_rankings": [], "star_stream_history": [], "game_over": False, "game_over_reason": None, } # ------------------------------------------------------------------ # Stat Management # ------------------------------------------------------------------ def apply_stat_changes(self, changes: dict[str, Any]) -> None: """ Apply a ``stat_changes`` dict returned by the AI. Handles top-level numeric fields (hp, coins, meta_exposure, prob_stability) as well as nested ``trust`` and ``constellation_affinity`` sub-dicts. All values are clamped to their valid ranges after application. """ s = self.state # --- Top-level numeric stats --- s["hp"] = max(0, min(s["max_hp"], s["hp"] + changes.get("hp", 0))) s["coins"] = max(0, s["coins"] + changes.get("coins", 0)) s["meta_exposure"] = max( 0, min(100, s["meta_exposure"] + changes.get("meta_exposure", 0)) ) s["prob_stability"] = max( 0, min(100, s["prob_stability"] + changes.get("prob_stability", 0)) ) # --- Trust sub-dict --- trust_changes = changes.get("trust", {}) for key, delta in trust_changes.items(): if key in s["trust"]: s["trust"][key] = max(0, s["trust"][key] + delta) # --- Constellation affinity sub-dict --- affinity_changes = changes.get("constellation_affinity", {}) for key, delta in affinity_changes.items(): if key in s["constellation_affinity"]: s["constellation_affinity"][key] = max( 0, s["constellation_affinity"][key] + delta ) def add_coins_from_entertainment(self, score: int) -> None: """Add entertainment-based coin reward (score × 10).""" self.state["coins"] += score * 10 self.state["entertainment"] += score def add_coins_from_constellations(self, reactions: list[dict[str, Any]]) -> None: """Sum coin donations from all constellation reactions.""" for reaction in reactions: coins = reaction.get("coins", 0) if isinstance(coins, (int, float)) and coins > 0: self.state["coins"] += int(coins) def buy_item(self, item_name: str, cost: int) -> bool: """Attempt to buy an item from the Dokkaebi Bag.""" if self.state["coins"] >= cost: self.state["coins"] -= cost self.state["inventory"].append(item_name) return True return False # ------------------------------------------------------------------ # History # ------------------------------------------------------------------ def add_to_history(self, action: str, tags: list[str]) -> None: """ Record a player action and associated tags. ``action_history`` is capped at the 10 most recent entries. ``action_tags`` grows unbounded (used for pattern detection). """ self.state["action_history"].append(action) if len(self.state["action_history"]) > 10: self.state["action_history"] = self.state["action_history"][-10:] self.state["action_tags"].extend(tags) # ------------------------------------------------------------------ # Hidden Scenario Triggers # ------------------------------------------------------------------ def check_hidden_scenarios(self) -> str | None: """ Evaluate behavioural patterns against hidden scenario triggers. Returns the scenario name if a new hidden scenario should fire, or ``None`` if nothing triggers. Trigger rules ------------- - **The Last Good Human**: 5+ ``'save'`` tags in action_tags. - **The Reader's Trial**: ``meta_exposure >= 75``. - **A Throne Built on Corpses**: 3+ ``'betray'`` tags. - **One Who Rejects Fate**: 3+ ``'reject'`` tags. - **The Untouchable**: completed 3+ scenarios with HP still at max. """ tags = self.state["action_tags"] triggered = self.state["hidden_scenarios_triggered"] # The Reader's Trial — highest priority if ( self.state["meta_exposure"] >= 75 and "The Reader's Trial" not in triggered ): return "The Reader's Trial" # The Last Good Human if tags.count("save") >= 5 and "The Last Good Human" not in triggered: return "The Last Good Human" # A Throne Built on Corpses if tags.count("betray") >= 3 and "A Throne Built on Corpses" not in triggered: return "A Throne Built on Corpses" # One Who Rejects Fate if tags.count("reject") >= 3 and "One Who Rejects Fate" not in triggered: return "One Who Rejects Fate" # The Untouchable if ( len(self.state["scenario_rankings"]) >= 3 and self.state["hp"] == self.state["max_hp"] and "The Untouchable" not in triggered ): return "The Untouchable" return None # ------------------------------------------------------------------ # Game-Over & Scenario Advancement # ------------------------------------------------------------------ def check_game_over(self) -> bool: """Check if the player is dead. Sets game_over flags if so.""" if self.state["hp"] <= 0: self.state["game_over"] = True self.state["game_over_reason"] = "Death" return True return False def advance_scenario(self, rank: str | None) -> None: """ Complete the current scenario, store its rank, and advance. Parameters ---------- rank : str | None The rank the player earned (e.g. "S", "A", "B", "C", "F"). """ self.state["scenario_rankings"].append(rank or "C") # Determine next scenario current = self.state["scenario_num"] if current == 1: next_s = 1.5 elif current == 1.5: next_s = 2 elif current == 2: next_s = 3 else: next_s = current + 1 self.state["scenario_num"] = next_s self.state["turn"] = 1 self.state["current_phase"] = "Exploration" # ------------------------------------------------------------------ # Meta Level # ------------------------------------------------------------------ def get_meta_level(self) -> str: """ Return a human-readable meta-exposure tier. Tiers: 0-24 → ``'normal'`` 25-49 → ``'suspicious'`` 50-74 → ``'unstable'`` 75-99 → ``'breaking'`` 100 → ``'noticed'`` """ me = self.state["meta_exposure"] if me >= 100: return "noticed" if me >= 75: return "breaking" if me >= 50: return "unstable" if me >= 25: return "suspicious" return "normal" # ------------------------------------------------------------------ # Sponsor System # ------------------------------------------------------------------ def check_sponsor_threshold(self) -> str | None: """ Check if any constellation's affinity has reached the sponsor threshold (≥ 50). Returns the display name of the first qualifying constellation, or ``None``. """ for key, value in self.state["constellation_affinity"].items(): if value >= 50: return _CONSTELLATION_DISPLAY_NAMES.get( key, key.replace("_", " ").title() ) return None # ------------------------------------------------------------------ # Serialisation # ------------------------------------------------------------------ def to_dict(self) -> dict[str, Any]: """Return a deep copy of the state dict (safe for gr.State).""" return copy.deepcopy(self.state) @classmethod def from_dict(cls, d: dict[str, Any]) -> "GameState": """ Reconstruct a GameState from a plain dict. Parameters ---------- d : dict A state dict previously produced by ``to_dict()``. Returns ------- GameState A new GameState instance with the restored state. """ gs = cls() gs.state = copy.deepcopy(d) return gs # ------------------------------------------------------------------ # Prompt Context Builder # ------------------------------------------------------------------ def to_prompt_context(self) -> str: """ Return a formatted, human-readable summary of the game state suitable for injection into the Dokkaebi system prompt. Only includes non-zero trust / affinity values to keep the context window efficient. """ s = self.state lines: list[str] = [ f"Player: {s['player_name']}", f"HP: {s['hp']}/{s['max_hp']}", f"Coins: {s['coins']}", f"Scenario: {s['scenario_num']} | Turn: {s['turn']} | Total Turns: {s['total_turns']}", f"Current Phase: {s['current_phase']}", f"Meta Level: {self.get_meta_level()} ({s['meta_exposure']}%)", f"Probability Stability: {s['prob_stability']}%", ] if s.get("active_debuffs"): lines.append(f"Active Debuffs: {', '.join(s['active_debuffs'])}") if s["titles"]: lines.append(f"Titles: {', '.join(s['titles'])}") if s["skills"]: skills_str = ", ".join([f"{name} (Lv.{data.get('level', 1)})" for name, data in s["skills"].items()]) lines.append(f"Skills: {skills_str}") if s["inventory"]: lines.append(f"Inventory: {', '.join(s['inventory'])}") if s["sponsor"]: lines.append(f"Sponsor: {s['sponsor']}") # Trust — only non-zero non_zero_trust = { _TRUST_DISPLAY_NAMES.get(k, k): v for k, v in s["trust"].items() if v != 0 } if non_zero_trust: trust_str = ", ".join(f"{name}: {val}" for name, val in non_zero_trust.items()) lines.append(f"Trust: {trust_str}") # Constellation affinity — only non-zero non_zero_aff = { _CONSTELLATION_DISPLAY_NAMES.get(k, k): v for k, v in s["constellation_affinity"].items() if v != 0 } if non_zero_aff: aff_str = ", ".join(f"{name}: {val}" for name, val in non_zero_aff.items()) lines.append(f"Constellation Affinity: {aff_str}") # Recent actions (last 3) recent = s["action_history"][-3:] if recent: lines.append(f"Recent Actions: {' → '.join(recent)}") return "\n".join(lines) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # STANDALONE FUNCTIONS # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ def infer_action_tags(player_action: str, ai_response: dict[str, Any], stance: str = "") -> list[str]: """ Derive behavioural tags from the player's action text and AI response. Uses simple keyword matching on the lowercased action string. Multiple tags can apply to a single action. Tag rules --------- - ``'save'`` — action contains *save*, *help*, or *protect* - ``'fight'`` — action contains *kill*, *attack*, or *fight* - ``'betray'`` — action contains *betray*, *abandon*, or *steal* - ``'avoid'`` — action contains *run*, *hide*, or *avoid* - ``'meta'`` — AI flagged ``meta_detected`` as truthy - ``'reject'`` — action contains *reject*, *refuse*, or *decline* Parameters ---------- player_action : str The raw text the player typed. ai_response : dict The parsed AI response dict. Returns ------- list[str] A (possibly empty) list of inferred tags. """ tags: list[str] = [] action_lower = player_action.lower() _TAG_KEYWORDS: dict[str, list[str]] = { "save": ["save", "help", "protect"], "fight": ["kill", "attack", "fight"], "betray": ["betray", "abandon", "steal"], "avoid": ["run", "hide", "avoid"], "reject": ["reject", "refuse", "decline"], } for tag, keywords in _TAG_KEYWORDS.items(): if any(kw in action_lower for kw in keywords): tags.append(tag) if stance: # map stance to specific tags stance_lower = stance.lower() if "aggressive" in stance_lower: tags.append("fight") elif "deceptive" in stance_lower: tags.append("betray") elif "empathetic" in stance_lower: tags.append("save") elif "observant" in stance_lower: tags.append("meta") if ai_response.get("meta_detected"): tags.append("meta") return tags def process_turn( state_dict: dict[str, Any], player_action: str, ai_response: dict[str, Any], stance: str = "", ) -> dict[str, Any]: """ Execute the full turn-processing pipeline. This is the main entry point called after the AI has responded. It mutates the game state based on the AI's output, handles all side-effects (coins, sponsors, hidden scenarios, game-over), and returns the updated state dict. Pipeline -------- 1. Reconstruct ``GameState`` from ``state_dict``. 2. Apply ``stat_changes`` from the AI response. 3. Add entertainment coins (``entertainment_score × 10``). 4. Add constellation donation coins. 5. Apply meta-exposure threshold penalties to ``prob_stability``. 6. If ``meta_detected``, decrease ``prob_stability`` by 5. 7. Handle ``new_title`` (append to titles list). 8. Check for hidden scenario triggers. 9. Handle scenario completion (``scenario_complete`` + ``scenario_rank``). 10. Check sponsor threshold. 11. Record action + inferred tags in history. 12. Check game-over condition. 13. Increment turn counters. 14. Return updated state dict. Parameters ---------- state_dict : dict The serialised game state (from ``GameState.to_dict()``). player_action : str The text the player typed. ai_response : dict The parsed JSON response from the Dokkaebi AI. Returns ------- dict The updated game state dict. """ gs = GameState.from_dict(state_dict) # 1-2. Apply stat changes stat_changes = ai_response.get("stat_changes", {}) gs.apply_stat_changes(stat_changes) # Decrement active skill cooldowns for skill_name, skill_data in gs.state.get("skills", {}).items(): if skill_data.get("cooldown", 0) > 0: skill_data["cooldown"] -= 1 # 3. Entertainment coins entertainment_score = ai_response.get("entertainment_score", 0) if isinstance(entertainment_score, (int, float)): gs.add_coins_from_entertainment(int(entertainment_score)) # 4. Constellation donation coins constellation_reactions = ai_response.get("constellation_reactions", []) if isinstance(constellation_reactions, list): gs.add_coins_from_constellations(constellation_reactions) # 5. Meta-exposure threshold penalties (only applied once when crossing threshold) # Track prev meta to detect crossing prev_meta = state_dict.get("meta_exposure", 0) # Use original before stat changes me = gs.state["meta_exposure"] if me >= 75 and prev_meta < 75: # Just crossed into "breaking" tier — one-time penalty gs.state["prob_stability"] = max(0, gs.state["prob_stability"] - 20) elif me >= 50 and prev_meta < 50: # Just crossed into "unstable" tier — one-time penalty gs.state["prob_stability"] = max(0, gs.state["prob_stability"] - 10) elif me >= 25 and prev_meta < 25: # Just crossed into "suspicious" tier — one-time penalty gs.state["prob_stability"] = max(0, gs.state["prob_stability"] - 5) # 6. Additional penalty for detected meta-gaming if ai_response.get("meta_detected"): gs.state["prob_stability"] = max(0, gs.state["prob_stability"] - 5) # 7. New titles new_title = ai_response.get("new_title") if new_title and new_title not in gs.state["titles"]: gs.state["titles"].append(new_title) # 8. Hidden scenario triggers hidden = gs.check_hidden_scenarios() if hidden and hidden not in gs.state["hidden_scenarios_triggered"]: gs.state["hidden_scenarios_triggered"].append(hidden) # 9. Scenario completion scenario_def = SCENARIOS.get(gs.state["scenario_num"]) max_turns = scenario_def.get("turns", 8) if scenario_def else 8 current_turn = gs.state["turn"] if ai_response.get("scenario_complete"): # AI explicitly flagged completion rank = ai_response.get("scenario_rank", "C") gs.advance_scenario(rank) gs.state["turn"] = 1 # Reset cleanly (advance_scenario sets to 1, skip the +1 below) scenario_advanced = True elif current_turn >= max_turns: # Hard enforcement: force scenario completion when turns are exhausted # Rank based on player HP and entertainment score entertainment = ai_response.get("entertainment_score", 3) hp_pct = gs.state["hp"] / max(gs.state["max_hp"], 1) if entertainment >= 7 and hp_pct > 0.7: forced_rank = "S" elif entertainment >= 5 and hp_pct > 0.4: forced_rank = "A" elif entertainment >= 3 and hp_pct > 0.2: forced_rank = "B" else: forced_rank = "C" print(f"[Engine] Forcing scenario completion at turn {current_turn}/{max_turns} — Rank {forced_rank}") gs.advance_scenario(forced_rank) gs.state["turn"] = 1 # Reset cleanly scenario_advanced = True else: scenario_advanced = False # 10. Sponsor threshold sponsor_name = gs.check_sponsor_threshold() if sponsor_name and gs.state["sponsor"] is None: gs.state["sponsor"] = sponsor_name # 11. Record action + tags tags = infer_action_tags(player_action, ai_response, stance) gs.add_to_history(player_action, tags) # 12. Game-over check gs.check_game_over() # 12.5 Probability Storms and Phase Shifts if gs.state["prob_stability"] < 50: import random if random.random() < 0.2: debuffs = ["Skill Lock", "Dokkaebi Greed", "Constellation Silence"] new_debuff = random.choice(debuffs) if new_debuff not in gs.state.setdefault("active_debuffs", []): gs.state["active_debuffs"].append(new_debuff) phase_shift = ai_response.get("phase_shift") if phase_shift in ["Exploration", "Combat", "Safe Zone"]: gs.state["current_phase"] = phase_shift # 13. Increment turn counters (skip turn if scenario just advanced — already reset to 1) if not scenario_advanced: gs.state["turn"] += 1 gs.state["total_turns"] += 1 # 14. Add constellation reactions to star stream history if constellation_reactions: for reaction in constellation_reactions[-5:]: gs.state["star_stream_history"].append(reaction) # Keep only the last 20 entries gs.state["star_stream_history"] = gs.state["star_stream_history"][-20:] return gs.to_dict()