""" detention_system.py — Detention tracking and enforcement. Detention triggers via two channels (K1: Hybrid): A. SCENE_DATA — LLM assigns/ends detention via detention_event block. B. Mechanical — trouble_level >= TROUBLE_AUTO_TRIGGER fires a fallback assign. Trouble level decay (K4: C = A+B): • Passive : -1 per turn when NOT in detention (apply_passive_decay). • On end : -DETENTION_TROUBLE_REDUCTION when detention concludes (end_detention). Detention ends (K5: B + safety): • Primary : LLM signals {"action": "end"} via SCENE_DATA. • Safety : If turns_remaining reaches 0, turn_engine calls end_detention. """ import logging from game.logging_config import event, get_logger, state_fields from systems.house_points_system import update_house_points logger = get_logger("detention") # ---------------------------- # CONSTANTS # ---------------------------- TROUBLE_AUTO_TRIGGER = 80 # trouble_level >= this auto-assigns detention (K1 channel B) PASSIVE_DECAY_PER_TURN = 1 # trouble_level drops by this each turn outside detention DETENTION_TROUBLE_REDUCTION = 20 # trouble_level drops by this when detention ends # ---------------------------- # INIT / BACKFILL # ---------------------------- def init_detention(state: dict) -> None: """ Ensures state["detention"] exists with all required fields. Safe to call every turn — idempotent. """ state.setdefault("detention", {}) det = state["detention"] det.setdefault("active", False) det.setdefault("assigned_by", None) det.setdefault("reason", None) det.setdefault("turns_remaining", 0) det.setdefault("total_count", 0) # lifetime count — for Chaos Legend aspiration # ---------------------------- # CORE OPERATIONS # ---------------------------- def assign_detention( state: dict, assigned_by: str, reason: str, turns: int, house_points_penalty: int = 0, ) -> dict: """ Assigns detention to the player. Applies trait modifiers (Lazy → turns ÷ 2, Organized → penalty ÷ 2). Teacher's Pet aspiration (if completed) grants full immunity. Returns the detention dict. """ init_detention(state) det = state["detention"] traits = state.get("traits", []) quests = state.get("quests", {}) if det.get("active"): event(logger, logging.WARNING, "detention.assignment_ignored", reason="already_active", **state_fields(state)) return det if isinstance(turns, bool) or not isinstance(turns, int): turns = 3 if isinstance(house_points_penalty, bool) or not isinstance(house_points_penalty, int): house_points_penalty = 0 turns = max(1, min(4, turns)) house_points_penalty = max(-30, min(0, house_points_penalty)) # --- Teacher's Pet immunity --- tp_quest = quests.get("teachers_pet") if tp_quest and tp_quest.get("status") == "completed": event(logger, logging.INFO, "detention.assignment_ignored", reason="teachers_pet_immunity", **state_fields(state)) return det # --- Trait modifiers --- actual_turns = turns actual_penalty = house_points_penalty if "Lazy" in traits: actual_turns = max(1, actual_turns // 2) event(logger, logging.DEBUG, "detention.trait_modifier", trait="Lazy", original_turns=turns, final_turns=actual_turns, **state_fields(state)) if "Organized" in traits and actual_penalty < 0: actual_penalty = actual_penalty // 2 event(logger, logging.DEBUG, "detention.trait_modifier", trait="Organized", original_penalty=house_points_penalty, final_penalty=actual_penalty, **state_fields(state)) # --- Apply --- det["active"] = True det["assigned_by"] = assigned_by det["reason"] = reason det["turns_remaining"] = actual_turns det["total_count"] += 1 if actual_penalty != 0: update_house_points(state, actual_penalty) event(logger, logging.WARNING, "detention.assigned", assigned_by=assigned_by, reason=reason, turns=actual_turns, total_count=det["total_count"], house_points_penalty=actual_penalty, **state_fields(state)) # TODO: Chaos Legend aspiration — fire completion hook when total_count >= 20 # (Aspiration completion system planned for a future phase) return det def end_detention(state: dict) -> dict: """ Ends active detention. Reduces trouble_level by DETENTION_TROUBLE_REDUCTION (K4 B part). Returns the detention dict. """ init_detention(state) det = state["detention"] if not det.get("active"): return det det["active"] = False det["assigned_by"] = None det["reason"] = None det["turns_remaining"] = 0 old_trouble = state.get("trouble_level", 0) state["trouble_level"] = max(0, old_trouble - DETENTION_TROUBLE_REDUCTION) event(logger, logging.INFO, "detention.ended", old_trouble=old_trouble, new_trouble=state["trouble_level"], **state_fields(state)) return det def tick_detention(state: dict) -> bool: """ Decrements turns_remaining by 1. Called once per turn. Returns True if turns_remaining just hit 0 (safety trigger). Does NOT call end_detention — caller does that if True is returned. """ init_detention(state) det = state["detention"] if not det.get("active"): return False det["turns_remaining"] = max(0, det["turns_remaining"] - 1) event(logger, logging.INFO, "detention.ticked", turns_remaining=det["turns_remaining"], **state_fields(state)) if det["turns_remaining"] <= 0: event(logger, logging.INFO, "detention.safety_end_triggered", **state_fields(state)) return True return False def apply_passive_decay(state: dict) -> None: """ Applies passive trouble_level decay (K4 A part): -PASSIVE_DECAY_PER_TURN per turn. Skipped when detention is active — end_detention handles that reduction separately. """ if state.get("detention", {}).get("active"): return # Don't decay during detention old = state.get("trouble_level", 0) if old > 0: state["trouble_level"] = max(0, old - PASSIVE_DECAY_PER_TURN) event(logger, logging.DEBUG, "trouble.passive_decay", old=old, new=state["trouble_level"], **state_fields(state)) def check_auto_trigger(state: dict) -> bool: """ Mechanical fallback (K1 channel B). Fires if trouble_level >= TROUBLE_AUTO_TRIGGER and no detention is currently active. Returns True if a detention was auto-assigned. """ init_detention(state) if state.get("trouble_level", 0) >= TROUBLE_AUTO_TRIGGER and not state["detention"]["active"]: assign_detention( state, assigned_by="a Hogwarts authority figure", reason="accumulated rule violations", turns=2, house_points_penalty=-10, ) event(logger, logging.WARNING, "detention.auto_triggered", threshold=TROUBLE_AUTO_TRIGGER, **state_fields(state)) return True return False # ---------------------------- # SCENE_DATA EVENT PROCESSOR # ---------------------------- def process_detention_event(state: dict, detention_event: dict) -> str | None: """ Processes a detention_event block from background_engine's SCENE_DATA. action="assign" → calls assign_detention. action="end" → calls end_detention. Returns the action string, or None if input was invalid. """ if not detention_event or not isinstance(detention_event, dict): return None action = detention_event.get("action") if action == "assign": if state.get("detention", {}).get("active"): return None assign_detention( state, assigned_by=detention_event.get("assigned_by", "a professor"), reason=detention_event.get("reason", "breaking school rules"), turns=detention_event.get("turns", 3), house_points_penalty=detention_event.get("house_points_penalty", 0), ) return "assign" elif action == "end": end_detention(state) return "end" event(logger, logging.WARNING, "detention.unknown_event", action=action, **state_fields(state)) return None # ---------------------------- # LLM CONTEXT # ---------------------------- def build_detention_message(state: dict) -> dict | None: """ Returns an OpenAI-style system message dict with mandatory detention instructions, or None if no detention is active. Injected into generate_llm_stream's messages list at high priority. """ init_detention(state) det = state["detention"] if not det.get("active"): return None assigned_by = det.get("assigned_by", "a professor") reason = det.get("reason", "breaking rules") turns_remaining = det.get("turns_remaining", 1) content = ( "[DETENTION ACTIVE — MANDATORY SCENE — READ BEFORE ALL OTHER INSTRUCTIONS]\n" f"The player is currently serving detention assigned by {assigned_by}.\n" f"Reason: {reason}. Estimated turns remaining: {turns_remaining}.\n\n" "DETENTION SCENE RULES:\n" "• This is a MANDATORY scene. The player cannot leave, skip, or avoid detention.\n" "• Write a vivid, specific detention scene with atmosphere and personality.\n" " Examples: scrubbing trophies in Filch's office, copying lines in Snape's cold\n" " dungeon, re-shelving books for Flitwick, polishing armour in the trophy room.\n" "• Avoid generic descriptions — give the location texture, smell, and feel.\n" "• NPCs may naturally pass through: a sympathetic friend peeking in, a mocking rival,\n" " a strict or surprisingly empathetic professor.\n" "• When the session concludes naturally (task complete, professor dismisses the player),\n" " end the narrative at that natural moment — the background system will detect this\n" " and signal the end of detention automatically.\n" "• Do NOT end detention abruptly — wait for a satisfying narrative conclusion." ) return {"role": "system", "content": content} def format_detention_for_llm(state: dict) -> str: """ Compact one-line summary for _build_dynamic_context injection. Returns empty string if no active detention (safe to inject unconditionally). """ init_detention(state) det = state["detention"] if not det.get("active"): return "" return ( f"[CURRENT STATUS: IN DETENTION — assigned by {det.get('assigned_by','?')} | " f"Reason: {det.get('reason','?')} | Turns remaining: {det.get('turns_remaining',0)}]" )