diff --git a/achievements.py b/achievements.py index 8f73122..33b6ef2 100644 --- a/achievements.py +++ b/achievements.py @@ -1,7 +1,6 @@ from datetime import datetime, timezone from typing import Dict, List, Optional - ACHIEVEMENTS = [ {"id": "first_solve", "title": "First Solve", "description": "Solve your first puzzle."}, {"id": "solve_5", "title": "Solver: 5", "description": "Solve 5 puzzles."}, @@ -44,20 +43,28 @@ def evaluate_achievements(state: Dict, levels: Optional[List[Dict]] = None) -> L for aid, cond in thresholds: if cond and aid not in unlocked: meta = _ach_by_id(aid) or {"id": aid} - unlocked[aid] = {"unlocked_at": _now_iso(), "title": meta.get("title"), "description": meta.get("description")} + unlocked[aid] = { + "unlocked_at": _now_iso(), + "title": meta.get("title"), + "description": meta.get("description"), + } newly.append({"id": aid, **unlocked[aid]}) # category completion achievements if levels provided if levels: # build category -> set(ids) cats = {} - for l in levels: - cid = l.get("category") or "uncategorized" - cats.setdefault(cid, set()).add(l.get("id")) + for level in levels: + cid = level.get("category") or "uncategorized" + cats.setdefault(cid, set()).add(level.get("id")) for cat, ids in cats.items(): if ids and ids.issubset(solved) and f"category_{cat}" not in unlocked: aid = f"category_{cat}" - unlocked[aid] = {"unlocked_at": _now_iso(), "title": f"Master: {cat}", "description": f"Solve all puzzles in {cat}."} + unlocked[aid] = { + "unlocked_at": _now_iso(), + "title": f"Master: {cat}", + "description": f"Solve all puzzles in {cat}.", + } newly.append({"id": aid, **unlocked[aid]}) return newly diff --git a/agent.py b/agent.py index 4b20067..8d0b8db 100644 --- a/agent.py +++ b/agent.py @@ -1,366 +1,382 @@ import os -SERVER = os.getenv("PUZZLE_SERVER", "http://127.0.0.1:5050") -HINT_MODELS_ENV = "LUX_OLLAMA_MODELS" -HINT_MODEL_ENV = "LUX_OLLAMA_MODEL" - import requests from requests.exceptions import RequestException -import storage -import cli + import achievements -import learning_paths +import cli import game_systems import leaderboard +import learning_paths +import storage + +SERVER = os.getenv("PUZZLE_SERVER", "http://127.0.0.1:5050") +HINT_MODELS_ENV = "LUX_OLLAMA_MODELS" +HINT_MODEL_ENV = "LUX_OLLAMA_MODEL" + def list_levels(): - try: - r = requests.get(f"{SERVER}/levels", timeout=5) - r.raise_for_status() - return r.json() - except RequestException as e: - print(f"Failed to fetch levels from {SERVER}: {e}") - return [] + try: + r = requests.get(f"{SERVER}/levels", timeout=5) + r.raise_for_status() + return r.json() + except RequestException as e: + print(f"Failed to fetch levels from {SERVER}: {e}") + return [] def get_level(level_id): - try: - r = requests.get(f"{SERVER}/level/{level_id}", timeout=5) - if r.status_code != 200: - return None - return r.json() - except RequestException as e: - print(f"Failed to fetch level {level_id}: {e}") - return None + try: + r = requests.get(f"{SERVER}/level/{level_id}", timeout=5) + if r.status_code != 200: + return None + return r.json() + except RequestException as e: + print(f"Failed to fetch level {level_id}: {e}") + return None def submit_attempt(level_id, attempt): - r = requests.post(f"{SERVER}/submit", json={"level_id": level_id, "attempt": attempt}, timeout=15) - r.raise_for_status() - return r.json() + r = requests.post( + f"{SERVER}/submit", json={"level_id": level_id, "attempt": attempt}, timeout=15 + ) + r.raise_for_status() + return r.json() def submit_files(level_id, files: dict): - """Submit a dict of filename->content to the server for script-based validation.""" - r = requests.post(f"{SERVER}/submit", json={"level_id": level_id, "files": files}, timeout=15) - r.raise_for_status() - return r.json() + """Submit a dict of filename->content to the server for script-based validation.""" + r = requests.post(f"{SERVER}/submit", json={"level_id": level_id, "files": files}, timeout=15) + r.raise_for_status() + return r.json() def _configured_hint_models(): - preferred = os.getenv(HINT_MODEL_ENV, "").strip() - configured = os.getenv(HINT_MODELS_ENV, "").strip() - defaults = ["llama3.2", "qwen2.5-coder:7b", "mistral", "phi3"] - models = [] - if preferred: - models.append(preferred) - if configured: - models.extend([item.strip() for item in configured.split(",") if item.strip()]) - if not models: - models = defaults - if preferred and preferred not in models: - models.insert(0, preferred) - # preserve order while removing duplicates - deduped = [] - for model in models: - if model not in deduped: - deduped.append(model) - return deduped + preferred = os.getenv(HINT_MODEL_ENV, "").strip() + configured = os.getenv(HINT_MODELS_ENV, "").strip() + defaults = ["llama3.2", "qwen2.5-coder:7b", "mistral", "phi3"] + models = [] + if preferred: + models.append(preferred) + if configured: + models.extend([item.strip() for item in configured.split(",") if item.strip()]) + if not models: + models = defaults + if preferred and preferred not in models: + models.insert(0, preferred) + # preserve order while removing duplicates + deduped = [] + for model in models: + if model not in deduped: + deduped.append(model) + return deduped def _safe_hint_text(text: str): - compact = " ".join((text or "").strip().split()) - if not compact: - return "" - sentences = [] - for chunk in compact.replace("!", ".").replace("?", ".").split("."): - chunk = chunk.strip() - if chunk: - sentences.append(chunk) - if len(sentences) >= 2: - break - hint = ". ".join(sentences) if sentences else compact - if len(hint) > 220: - hint = hint[:217].rstrip() + "..." - return hint + compact = " ".join((text or "").strip().split()) + if not compact: + return "" + sentences = [] + for chunk in compact.replace("!", ".").replace("?", ".").split("."): + chunk = chunk.strip() + if chunk: + sentences.append(chunk) + if len(sentences) >= 2: + break + hint = ". ".join(sentences) if sentences else compact + if len(hint) > 220: + hint = hint[:217].rstrip() + "..." + return hint def _attempt_summary_for_level(state, level_id): - stats = storage.get_level_attempt_stats(state, level_id) - if not stats: - return "No prior attempts recorded." - pieces = [f"{stats.get('attempts', 0)} attempts"] - if stats.get("incorrect"): - pieces.append(f"{stats['incorrect']} incorrect") - if stats.get("last_outcome"): - pieces.append(f"last outcome: {stats['last_outcome']}") - return ", ".join(pieces) + stats = storage.get_level_attempt_stats(state, level_id) + if not stats: + return "No prior attempts recorded." + pieces = [f"{stats.get('attempts', 0)} attempts"] + if stats.get("incorrect"): + pieces.append(f"{stats['incorrect']} incorrect") + if stats.get("last_outcome"): + pieces.append(f"last outcome: {stats['last_outcome']}") + return ", ".join(pieces) def _build_hint_prompt(level, state=None): - state = state or storage.load_state() - tags = ", ".join(level.get("tags", [])) - progress = storage.get_progress_summary(state) - attempt_summary = _attempt_summary_for_level(state, level.get("id")) - solved_sample = ", ".join(progress.get("recent_solved") or []) or "none" - return ( - "You are a careful tutoring assistant for Lux puzzle practice. " - "Give one short hint only. Do not reveal the solution, flag, exact command, or full code. " - "Prefer a next step, concept reminder, or debugging direction. " - "Keep the hint concise and practical.\n\n" - f"Title: {level['title']}\n" - f"Description: {level['description']}\n" - f"Category: {level.get('category', 'uncategorized')}\n" - f"Difficulty: {level.get('difficulty', 'unknown')}\n" - f"Tags: {tags or 'none'}\n" - f"Solved progress: {progress.get('solved_count', 0)} solved, current streak {progress.get('current_streak', 0)}, best streak {progress.get('longest_streak', 0)}\n" - f"Recent solved levels: {solved_sample}\n" - f"Attempt history for this level: {attempt_summary}\n" - ) + state = state or storage.load_state() + tags = ", ".join(level.get("tags", [])) + progress = storage.get_progress_summary(state) + attempt_summary = _attempt_summary_for_level(state, level.get("id")) + solved_sample = ", ".join(progress.get("recent_solved") or []) or "none" + return ( + "You are a careful tutoring assistant for Lux puzzle practice. " + "Give one short hint only. Do not reveal the solution, flag, exact command, or full code. " + "Prefer a next step, concept reminder, or debugging direction. " + "Keep the hint concise and practical.\n\n" + f"Title: {level['title']}\n" + f"Description: {level['description']}\n" + f"Category: {level.get('category', 'uncategorized')}\n" + f"Difficulty: {level.get('difficulty', 'unknown')}\n" + f"Tags: {tags or 'none'}\n" + f"Solved progress: {progress.get('solved_count', 0)} solved, " + f"current streak {progress.get('current_streak', 0)}, " + f"best streak {progress.get('longest_streak', 0)}\n" + f"Recent solved levels: {solved_sample}\n" + f"Attempt history for this level: {attempt_summary}\n" + ) def ask_hint_via_ollama(level, state=None): - try: - import ollama - except Exception as e: - raise RuntimeError("Ollama is not available: " + str(e)) - - prompt = _build_hint_prompt(level, state=state) - messages = [{"role": "user", "content": prompt}] - last_error = None - for model in _configured_hint_models(): try: - res = ollama.chat(model=model, messages=messages) - content = getattr(getattr(res, "message", None), "content", "") - hint = _safe_hint_text(content) - if hint: - return hint - except Exception as exc: - last_error = exc - continue - raise RuntimeError(f"Unable to generate a hint with Ollama: {last_error}") + import ollama + except Exception as e: + raise RuntimeError("Ollama is not available: " + str(e)) + + prompt = _build_hint_prompt(level, state=state) + messages = [{"role": "user", "content": prompt}] + last_error = None + for model in _configured_hint_models(): + try: + res = ollama.chat(model=model, messages=messages) + content = getattr(getattr(res, "message", None), "content", "") + hint = _safe_hint_text(content) + if hint: + return hint + except Exception as exc: + last_error = exc + continue + raise RuntimeError(f"Unable to generate a hint with Ollama: {last_error}") def print_levels(levels, solved_set=None, state=None): - print("\nAvailable levels:") - ordered = cli.sort_levels(levels, solved_set) - for lvl in ordered: - print(" ", cli.format_level_line(lvl, solved_set)) - if state is None: + print("\nAvailable levels:") + ordered = cli.sort_levels(levels, solved_set) + for lvl in ordered: + print(" ", cli.format_level_line(lvl, solved_set)) + if state is None: + try: + state = storage.load_state() + except Exception: + state = None + if state: + print("\nProgress:") + print(" " + cli.format_progress_summary(state, levels)) + # show achievement summary try: - state = storage.load_state() + ach = state.get("achievements", {}) + if ach: + print("\nAchievements:") + for k, v in ach.items(): + print(f" - {v.get('title', '?')} ({k}) unlocked: {v.get('unlocked_at')}") except Exception: - state = None - if state: - print("\nProgress:") - print(" " + cli.format_progress_summary(state, levels)) - # show achievement summary - try: - ach = state.get("achievements", {}) - if ach: - print("\nAchievements:") - for k, v in ach.items(): - print(f" - {v.get('title','?')} ({k}) unlocked: {v.get('unlocked_at')}") - except Exception: - pass + pass def read_files_from_paths(): - print("This level requires source files. Provide local file paths to upload.") - files = {} - while True: - path = input("Enter local path to a source file (or blank to finish): ").strip() - if not path: - break - try: - with open(path, "r") as f: - files[os.path.basename(path)] = f.read() - except Exception as e: - print("Failed to read file:", e) - return files + print("This level requires source files. Provide local file paths to upload.") + files = {} + while True: + path = input("Enter local path to a source file (or blank to finish): ").strip() + if not path: + break + try: + with open(path, "r") as f: + files[os.path.basename(path)] = f.read() + except Exception as e: + print("Failed to read file:", e) + return files def _persist_attempt_state(state, choice, lvl, attempt_preview, correct): - storage.record_attempt( - state, - choice, - correct=correct, - title=lvl.get("title"), - difficulty=lvl.get("difficulty"), - category=lvl.get("category"), - attempt_preview=attempt_preview, - ) - if correct: - stats = storage.get_level_attempt_stats(state, choice) - storage.mark_solved( - state, - choice, - attempts=stats.get("attempts", 1), - title=lvl.get("title"), - difficulty=lvl.get("difficulty"), - category=lvl.get("category"), + storage.record_attempt( + state, + choice, + correct=correct, + title=lvl.get("title"), + difficulty=lvl.get("difficulty"), + category=lvl.get("category"), + attempt_preview=attempt_preview, ) - try: - game_systems.on_level_solved(state, lvl) - except Exception: - pass - try: - levels = list_levels() - except Exception: - levels = [] - newly = achievements.evaluate_achievements(state, levels) - try: - leaderboard.upsert_local_entry(state, levels) - except Exception: - pass - if newly: - print("New achievements unlocked:") - for a in newly: - print(f" - {a.get('title')} ({a.get('id')})") - storage.save_state(state) + if correct: + stats = storage.get_level_attempt_stats(state, choice) + storage.mark_solved( + state, + choice, + attempts=stats.get("attempts", 1), + title=lvl.get("title"), + difficulty=lvl.get("difficulty"), + category=lvl.get("category"), + ) + try: + game_systems.on_level_solved(state, lvl) + except Exception: + pass + try: + levels = list_levels() + except Exception: + levels = [] + newly = achievements.evaluate_achievements(state, levels) + try: + leaderboard.upsert_local_entry(state, levels) + except Exception: + pass + if newly: + print("New achievements unlocked:") + for a in newly: + print(f" - {a.get('title')} ({a.get('id')})") + storage.save_state(state) def _submit_text_attempt(choice, lvl): - attempt = input("Enter your answer/command: ").strip() - try: - res = submit_attempt(choice, attempt) - except Exception as e: - print("Submission failed:", e) - return False - state = storage.load_state() - if res.get("correct"): - print("Correct! Level solved.") + attempt = input("Enter your answer/command: ").strip() try: - _persist_attempt_state(state, choice, lvl, attempt, True) + res = submit_attempt(choice, attempt) + except Exception as e: + print("Submission failed:", e) + return False + state = storage.load_state() + if res.get("correct"): + print("Correct! Level solved.") + try: + _persist_attempt_state(state, choice, lvl, attempt, True) + except Exception: + pass + return True + try: + _persist_attempt_state(state, choice, lvl, attempt, False) except Exception: - pass - return True - try: - _persist_attempt_state(state, choice, lvl, attempt, False) - except Exception: - pass - print("Incorrect or tests failed.") - if res.get("output"): - print(res["output"]) - return False + pass + print("Incorrect or tests failed.") + if res.get("output"): + print(res["output"]) + return False def _submit_script_attempt(choice, lvl): - files = read_files_from_paths() - if not files: - print("No files provided; canceling attempt.") - return False - try: - res = submit_files(choice, files) - except Exception as e: - print("Submission failed:", e) - return False - state = storage.load_state() - if res.get("correct"): - print("Correct! Level solved.") + files = read_files_from_paths() + if not files: + print("No files provided; canceling attempt.") + return False + try: + res = submit_files(choice, files) + except Exception as e: + print("Submission failed:", e) + return False + state = storage.load_state() + if res.get("correct"): + print("Correct! Level solved.") + try: + _persist_attempt_state(state, choice, lvl, None, True) + except Exception: + pass + return True try: - _persist_attempt_state(state, choice, lvl, None, True) + _persist_attempt_state(state, choice, lvl, None, False) except Exception: - pass - return True - try: - _persist_attempt_state(state, choice, lvl, None, False) - except Exception: - pass - print("Incorrect or tests failed.") - if res.get("output"): - print(res["output"]) - return False + pass + print("Incorrect or tests failed.") + if res.get("output"): + print(res["output"]) + return False def handle_attempt(choice, lvl): - if lvl.get("validator") == "script": - return _submit_script_attempt(choice, lvl) - return _submit_text_attempt(choice, lvl) + if lvl.get("validator") == "script": + return _submit_script_attempt(choice, lvl) + return _submit_text_attempt(choice, lvl) def handle_level(choice): - lvl = get_level(choice) - if not lvl: - print("Level not found.") - return - print(f"\n{lvl['title']}\n{lvl['description']}\n") - while True: - cmd = input("Options: (a)ttempt, (h)int, (b)ack: ").strip().lower() - if cmd in ("b", "back"): - return - if cmd in ("h", "hint"): - try: - print("Hint:\n", ask_hint_via_ollama(lvl, state=storage.load_state())) - except Exception as e: - print("Hint failed:", e) - continue - if cmd in ("a", "attempt"): - if handle_attempt(choice, lvl): + lvl = get_level(choice) + if not lvl: + print("Level not found.") return - continue - print("Unknown option — choose 'a', 'h', or 'b'.") + print(f"\n{lvl['title']}\n{lvl['description']}\n") + while True: + cmd = input("Options: (a)ttempt, (h)int, (b)ack: ").strip().lower() + if cmd in ("b", "back"): + return + if cmd in ("h", "hint"): + try: + print("Hint:\n", ask_hint_via_ollama(lvl, state=storage.load_state())) + except Exception as e: + print("Hint failed:", e) + continue + if cmd in ("a", "attempt"): + if handle_attempt(choice, lvl): + return + continue + print("Unknown option — choose 'a', 'h', or 'b'.") def _handle_menu_choice(choice, state, levels): - lowered = choice.lower() - if lowered in ("ach", "achievements"): - ach = state.get("achievements", {}) - if not ach: - print("No achievements unlocked yet.") - else: - print("Achievements:") - for k, v in ach.items(): - print(f" - {v.get('title','?')} ({k}) unlocked: {v.get('unlocked_at')}") - return True - if lowered in ("stats", "progress"): - print(cli.format_progress_summary(state, levels)) - return True - if lowered in ("reset-ach", "reset-achievements"): - confirm = input("Are you sure you want to reset all achievements? type 'yes' to confirm: ") - if confirm.strip().lower() == "yes": - state["achievements"] = {} - storage.save_state(state) - print("Achievements cleared.") - else: - print("Reset cancelled.") - return True - if lowered in ("path", "learning-path", "recommend"): - try: - path = learning_paths.build_learning_path(state, levels) - print(path.get("summary")) - print("Recommended levels:") - for item in path.get("recommended_levels", []): - print(f" - {item.get('id')}: {item.get('title')} ({item.get('category')} · {item.get('difficulty')})") - except Exception as exc: - print("Learning path unavailable:", exc) - return True - if lowered in ("daily",): - try: - daily = game_systems.daily_challenge_status(state, levels) - level = daily.get("level", {}) - print(f"Daily challenge ({daily.get('date')}): {level.get('id')} - {level.get('title')}") - print("Completed today:" if daily.get("completed_today") else "Not completed yet.") - except Exception as exc: - print("Daily challenge unavailable:", exc) - return True - return False + lowered = choice.lower() + if lowered in ("ach", "achievements"): + ach = state.get("achievements", {}) + if not ach: + print("No achievements unlocked yet.") + else: + print("Achievements:") + for k, v in ach.items(): + print(f" - {v.get('title', '?')} ({k}) unlocked: {v.get('unlocked_at')}") + return True + if lowered in ("stats", "progress"): + print(cli.format_progress_summary(state, levels)) + return True + if lowered in ("reset-ach", "reset-achievements"): + confirm = input("Are you sure you want to reset all achievements? type 'yes' to confirm: ") + if confirm.strip().lower() == "yes": + state["achievements"] = {} + storage.save_state(state) + print("Achievements cleared.") + else: + print("Reset cancelled.") + return True + if lowered in ("path", "learning-path", "recommend"): + try: + path = learning_paths.build_learning_path(state, levels) + print(path.get("summary")) + print("Recommended levels:") + for item in path.get("recommended_levels", []): + print( + f" - {item.get('id')}: {item.get('title')} " + f"({item.get('category')} · {item.get('difficulty')})" + ) + except Exception as exc: + print("Learning path unavailable:", exc) + return True + if lowered in ("daily",): + try: + daily = game_systems.daily_challenge_status(state, levels) + level = daily.get("level", {}) + print( + f"Daily challenge ({daily.get('date')}): {level.get('id')} - {level.get('title')}" + ) + print("Completed today:" if daily.get("completed_today") else "Not completed yet.") + except Exception as exc: + print("Daily challenge unavailable:", exc) + return True + return False def main(): - print("Interactive Puzzle Agent — connect to the puzzle server and request hints.") - while True: - state = storage.load_state() - levels = list_levels() - print_levels(levels, storage.get_solved_levels(state), state=state) - print("Options: enter a level id to open it, 'ach' to list achievements, 'stats' for progress, 'path' for recommendations, 'daily' for today's challenge, 'reset-ach' to clear achievements, or 'q' to quit.") - choice = input("Choose level id (or 'q' to quit): ").strip() - if choice.lower() in ("q", "quit", "exit"): - break - if _handle_menu_choice(choice, state, levels): - continue - handle_level(choice) + print("Interactive Puzzle Agent — connect to the puzzle server and request hints.") + while True: + state = storage.load_state() + levels = list_levels() + print_levels(levels, storage.get_solved_levels(state), state=state) + print( + "Options: enter a level id to open it, 'ach' to list achievements, " + "'stats' for progress, 'path' for recommendations, " + "'daily' for today's challenge, 'reset-ach' to clear achievements, " + "or 'q' to quit." + ) + choice = input("Choose level id (or 'q' to quit): ").strip() + if choice.lower() in ("q", "quit", "exit"): + break + if _handle_menu_choice(choice, state, levels): + continue + handle_level(choice) if __name__ == "__main__": - main() + main() diff --git a/cli.py b/cli.py index 2446394..b076b8d 100644 --- a/cli.py +++ b/cli.py @@ -20,13 +20,13 @@ def difficulty_key(d: str) -> int: def sort_levels(levels: List[Dict], solved: Set[str] = None) -> List[Dict]: solved = solved or set() - def key(l): + def key(level): return ( - l.get("id") in solved, - difficulty_key(l.get("difficulty")), - (l.get("category") or "").lower(), - (l.get("title") or "").lower(), - l.get("id") or "", + level.get("id") in solved, + difficulty_key(level.get("difficulty")), + (level.get("category") or "").lower(), + (level.get("title") or "").lower(), + level.get("id") or "", ) ordered = sorted(levels, key=key) @@ -35,7 +35,8 @@ def key(l): def _level_matches_text(level: Dict, query: str) -> bool: haystack = " ".join( - str(level.get(field, "")) for field in ("id", "title", "description", "category", "difficulty") + str(level.get(field, "")) + for field in ("id", "title", "description", "category", "difficulty") ).lower() return query in haystack @@ -98,7 +99,7 @@ def format_level_line(level: Dict, solved: Set[str] = None) -> str: attempt_text = f" · attempts {attempts}" if attempts else "" return ( f"{marker} {level.get('id')}: {level.get('title')} " - f"({level.get('difficulty','?')}) - {level.get('category','?')}" + f"({level.get('difficulty', '?')}) - {level.get('category', '?')}" f"{attempt_text}" f"{' - ' + tags if tags else ''}" ) @@ -112,8 +113,7 @@ def format_progress_summary(state: Dict, levels: Optional[List[Dict]] = None) -> if summary.get("percent_complete") is not None: pieces.append(f"({summary['percent_complete']}%)") pieces.append( - f"streak {summary.get('current_streak', 0)}" - f" / best {summary.get('longest_streak', 0)}" + f"streak {summary.get('current_streak', 0)} / best {summary.get('longest_streak', 0)}" ) pieces.append(f"attempts {summary.get('total_attempts', 0)}") recent = summary.get("recent_solved") or [] diff --git a/contributors.py b/contributors.py index ad0df76..31ab9ca 100644 --- a/contributors.py +++ b/contributors.py @@ -2,7 +2,6 @@ import os from typing import Any, Dict, List - DEFAULT_MANIFEST = os.path.join(os.path.dirname(__file__), "contributors.json") @@ -23,10 +22,14 @@ def load_contributors(path: str = DEFAULT_MANIFEST) -> List[Dict[str, Any]]: if not isinstance(entry, dict) or not entry.get("login"): continue badges = entry.get("badges", []) - valid.append({ - "login": str(entry["login"]), - "name": str(entry.get("name") or entry["login"]), - "contributions": [str(item) for item in entry.get("contributions", []) if item], - "badges": [str(item) for item in badges if item] if isinstance(badges, list) else [], - }) + valid.append( + { + "login": str(entry["login"]), + "name": str(entry.get("name") or entry["login"]), + "contributions": [str(item) for item in entry.get("contributions", []) if item], + "badges": [str(item) for item in badges if item] + if isinstance(badges, list) + else [], + } + ) return valid diff --git a/leaderboard.py b/leaderboard.py index 7d9c745..78ad08d 100644 --- a/leaderboard.py +++ b/leaderboard.py @@ -17,11 +17,13 @@ def _score_for_entry(state: Dict, entry: Dict, metric: str) -> float: return float(entry.get("solved_count", 0) or 0) -def get_leaderboard(state: Dict, levels: Optional[List[Dict]] = None, metric: str = "solved_count") -> Dict: +def get_leaderboard( + state: Dict, levels: Optional[List[Dict]] = None, metric: str = "solved_count" +) -> Dict: profile_name = _safe_profile_name(state) solved = state.get("solved") or {} streak = (state.get("meta") or {}).get("streak") or {} - game = (state.get("game") or {}) + game = state.get("game") or {} entry = { "name": profile_name, "solved_count": len(solved), diff --git a/puzzle_generator.py b/puzzle_generator.py index 05f1776..5dc2cf1 100644 --- a/puzzle_generator.py +++ b/puzzle_generator.py @@ -16,7 +16,14 @@ def _validate_generated_puzzle(puzzle: Dict) -> bool: return True -def generate_puzzle_with_ai(*, category: str, difficulty: str, topic: str, existing_ids: Iterable[str], next_id: Optional[str] = None) -> Dict: +def generate_puzzle_with_ai( + *, + category: str, + difficulty: str, + topic: str, + existing_ids: Iterable[str], + next_id: Optional[str] = None, +) -> Dict: category = (category or "Programming").strip() or "Programming" difficulty = (difficulty or "easy").strip().lower() or "easy" topic = (topic or "practice").strip() or "practice" diff --git a/sandbox.py b/sandbox.py index a451056..b79e5e8 100644 --- a/sandbox.py +++ b/sandbox.py @@ -9,7 +9,6 @@ import sandbox_audit - MAX_FILES = 16 MAX_FILENAME_BYTES = 255 MAX_FILE_BYTES = 256 * 1024 @@ -103,14 +102,16 @@ def run(self, files: Dict[str, str], script: str) -> Dict[str, Optional[object]] "command": cmd, "timed_out": True, } - _record_audit({ - "job_id": job_id, - "runtime": os.getenv("LUX_DOCKER_RUNTIME", "runc"), - "duration_ms": round((time.monotonic() - started) * 1000), - "exit_code": None, - "timed_out": True, - "passed": False, - }) + _record_audit( + { + "job_id": job_id, + "runtime": os.getenv("LUX_DOCKER_RUNTIME", "runc"), + "duration_ms": round((time.monotonic() - started) * 1000), + "exit_code": None, + "timed_out": True, + "passed": False, + } + ) return result stdout = completed.stdout.decode("utf-8", errors="ignore") stderr = completed.stderr.decode("utf-8", errors="ignore") @@ -122,14 +123,16 @@ def run(self, files: Dict[str, str], script: str) -> Dict[str, Optional[object]] "command": cmd, "timed_out": timed_out, } - _record_audit({ - "job_id": job_id, - "runtime": os.getenv("LUX_DOCKER_RUNTIME", "runc"), - "duration_ms": round((time.monotonic() - started) * 1000), - "exit_code": completed.returncode, - "timed_out": timed_out, - "passed": result["passed"], - }) + _record_audit( + { + "job_id": job_id, + "runtime": os.getenv("LUX_DOCKER_RUNTIME", "runc"), + "duration_ms": round((time.monotonic() - started) * 1000), + "exit_code": completed.returncode, + "timed_out": timed_out, + "passed": result["passed"], + } + ) return result finally: shutil.rmtree(workdir, ignore_errors=True) diff --git a/server.py b/server.py index 0eb3c37..4ad8d6c 100644 --- a/server.py +++ b/server.py @@ -1,13 +1,15 @@ -from flask import Flask, jsonify, request, render_template -from flask_wtf import CSRFProtect -from datetime import datetime import os -import storage +from datetime import datetime + +from flask import Flask, jsonify, render_template, request +from flask_wtf import CSRFProtect + +import contributors +import game_systems import leaderboard import learning_paths -import game_systems import puzzle_generator -import contributors +import storage from sandbox import DockerSandbox, get_runtime app = Flask(__name__) @@ -95,12 +97,14 @@ "description": "What one-liner prints each .txt file in the current directory?", "hint": "Use a for loop and echo", "validator": "contains", - "flag": "for f in *.txt; do echo \"$f\"; done", + "flag": 'for f in *.txt; do echo "$f"; done', }, { "id": "11", "title": "Linux: Permissions and ownership", - "description": "What commands set report.txt to rw-r----- and change its owner to alice:staff?", + "description": ( + "What commands set report.txt to rw-r----- and change its owner to alice:staff?" + ), "hint": "Use chmod and chown together", "validator": "contains", "flag": "chmod 640 report.txt && chown alice:staff report.txt", @@ -124,23 +128,36 @@ { "id": "14", "title": "Bash challenge: Count lines", - "description": "Write answer.sh so that sh answer.sh input.txt prints the number of lines in the file.", + "description": ( + "Write answer.sh so that sh answer.sh input.txt prints the number of lines in the file." + ), "hint": "wc -l can help", "validator": "script", - "test_script": "printf 'alpha\nbeta\ngamma\n' > input.txt && sh answer.sh input.txt | grep -xq '3'", + "test_script": ( + "printf 'alpha\nbeta\ngamma\n' > input.txt && sh answer.sh input.txt | grep -xq '3'" + ), }, { "id": "15", "title": "Bash challenge: Filter TODOs", - "description": "Write answer.sh so that sh answer.sh notes.txt prints only lines that start with TODO.", + "description": ( + "Write answer.sh so that sh answer.sh notes.txt prints only lines that start with TODO." + ), "hint": "grep with a start-of-line anchor is enough", "validator": "script", - "test_script": "printf 'TODO first\nskip\nTODO second\n' > notes.txt && sh answer.sh notes.txt | grep -xq 'TODO first' && sh answer.sh notes.txt | grep -xq 'TODO second'", + "test_script": ( + "printf 'TODO first\nskip\nTODO second\n' > notes.txt " + "&& sh answer.sh notes.txt | grep -xq 'TODO first' " + "&& sh answer.sh notes.txt | grep -xq 'TODO second'" + ), }, { "id": "16", "title": "C: Intermediate pointers", - "description": "Write answer.c so it prints the second value from the array {10, 20, 30} by dereferencing a pointer.", + "description": ( + "Write answer.c so it prints the second value from the array {10, 20, 30} " + "by dereferencing a pointer." + ), "hint": "Use pointer arithmetic or array indexing.", "validator": "script", "test_script": "gcc answer.c -o answer && ./answer | grep -xq '20'", @@ -148,7 +165,10 @@ { "id": "17", "title": "C: Memory management", - "description": "Write answer.c so it allocates space for two integers, stores 40 and 2, frees the memory, and prints their sum.", + "description": ( + "Write answer.c so it allocates space for two integers, stores 40 and 2, " + "frees the memory, and prints their sum." + ), "hint": "malloc and free both matter here.", "validator": "script", "test_script": "gcc answer.c -o answer && ./answer | grep -xq '42'", @@ -156,7 +176,10 @@ { "id": "18", "title": "C: Data structures", - "description": "Write answer.c so it builds a three-node linked list with values 1, 2, and 3, then prints the node count.", + "description": ( + "Write answer.c so it builds a three-node linked list with values 1, 2, and 3, " + "then prints the node count." + ), "hint": "A struct with a next pointer is enough.", "validator": "script", "test_script": "gcc answer.c -o answer && ./answer | grep -xq '3'", @@ -164,7 +187,10 @@ { "id": "19", "title": "C: Algorithms", - "description": "Write answer.c so it binary-searches the sorted array {5, 10, 15, 20, 25, 30, 35} for 25 and prints its index.", + "description": ( + "Write answer.c so it binary-searches the sorted array {5, 10, 15, 20, 25, 30, 35} " + "for 25 and prints its index." + ), "hint": "Divide the search space.", "validator": "script", "test_script": "gcc answer.c -o answer && ./answer | grep -xq '4'", @@ -172,23 +198,42 @@ { "id": "20", "title": "Python: List comprehension", - "description": "Write answer.py so it defines nums = [1, 2, 3] and uses a list comprehension to print [1, 4, 9].", + "description": ( + "Write answer.py so it defines nums = [1, 2, 3] and uses a list comprehension " + "to print [1, 4, 9]." + ), "hint": "Square each x inside a comprehension.", "validator": "script", - "test_script": "grep -Eq 'squares[[:space:]]*=[[:space:]]*\\[x\\*x[[:space:]]+for[[:space:]]+x[[:space:]]+in[[:space:]]+nums\\]' answer.py && grep -Eq 'print\\(squares\\)' answer.py", + "test_script": ( + "grep -Eq " + "'squares[[:space:]]*=[[:space:]]*\\[x\\*x[[:space:]]+for[[:space:]]+" + "x[[:space:]]+in[[:space:]]+nums\\]' answer.py " + "&& grep -Eq 'print\\(squares\\)' answer.py" + ), }, { "id": "21", "title": "Java: Entry point", - "description": "Write Main.java so it contains a standard public static void main(String[] args) and prints ready.", + "description": ( + "Write Main.java so it contains a standard public static void main(String[] args) " + "and prints ready." + ), "hint": "main is the entry point.", "validator": "script", - "test_script": "grep -Eq 'public[[:space:]]+static[[:space:]]+void[[:space:]]+main\\(String\\[\\][[:space:]]+args\\)' Main.java && grep -Eq 'System\\.out\\.println\\(\\\"ready\\\"\\);' Main.java", + "test_script": ( + "grep -Eq " + "'public[[:space:]]+static[[:space:]]+void[[:space:]]+main\\(String\\[\\]" + "[[:space:]]+args\\)' Main.java " + "&& grep -Eq 'System\\.out\\.println\\(\\\"ready\\\"\\);' Main.java" + ), }, { "id": "22", "title": "JavaScript: Strict equality", - "description": "Write answer.js so it compares left and right with strict equality and prints true when they match.", + "description": ( + "Write answer.js so it compares left and right with strict equality " + "and prints true when they match." + ), "hint": "Use three equals signs.", "validator": "script", "test_script": "grep -Eq '===' answer.js && grep -Eq 'console\\.log\\(true\\)' answer.js", @@ -196,18 +241,30 @@ { "id": "23", "title": "Web security: SQL injection", - "description": "Write answer.sh so it accepts only usernames from the allowlist alice, bob, or carol and prints ACCEPT when matched.", + "description": ( + "Write answer.sh so it accepts only usernames from the allowlist " + "alice, bob, or carol and prints ACCEPT when matched." + ), "hint": "Known-good inputs only.", "validator": "script", - "test_script": "printf 'alice\n' | sh answer.sh | grep -xq 'ACCEPT' && printf 'mallory\n' | sh answer.sh | grep -xq 'REJECT'", + "test_script": ( + "printf 'alice\n' | sh answer.sh | grep -xq 'ACCEPT' " + "&& printf 'mallory\n' | sh answer.sh | grep -xq 'REJECT'" + ), }, { "id": "24", "title": "Reverse engineering: Strings", - "description": "Write answer.sh so it uses strings on the input binary and prints any line containing FLAG.", + "description": ( + "Write answer.sh so it uses strings on the input binary " + "and prints any line containing FLAG." + ), "hint": "Extract printable text first.", "validator": "script", - "test_script": "printf 'abc\0FLAG{reverse_me}\0xyz' > sample.bin && sh answer.sh sample.bin | grep -xq 'FLAG{reverse_me}'", + "test_script": ( + "printf 'abc\0FLAG{reverse_me}\0xyz' > sample.bin " + "&& sh answer.sh sample.bin | grep -xq 'FLAG{reverse_me}'" + ), }, { "id": "25", @@ -215,7 +272,11 @@ "description": "Write answer.sh so it prints the SHA-256 checksum of evidence.bin.", "hint": "Use a checksum command from coreutils.", "validator": "script", - "test_script": "printf 'forensic data\n' > evidence.bin && expected=$(sha256sum evidence.bin | awk '{print $1}') && sh answer.sh evidence.bin | grep -xq \"$expected\"", + "test_script": ( + "printf 'forensic data\n' > evidence.bin " + "&& expected=$(sha256sum evidence.bin | awk '{print $1}') " + '&& sh answer.sh evidence.bin | grep -xq "$expected"' + ), }, { "id": "26", @@ -223,28 +284,43 @@ "description": "Write answer.sh so it base64-decodes encoded.txt and prints the plaintext.", "hint": "The encoding is reversible.", "validator": "script", - "test_script": "printf 'SGVsbG8=\n' > encoded.txt && sh answer.sh encoded.txt | grep -xq 'Hello'", + "test_script": ( + "printf 'SGVsbG8=\n' > encoded.txt && sh answer.sh encoded.txt | grep -xq 'Hello'" + ), }, { "id": "27", "title": "Secure coding: Input validation", - "description": "Write answer.sh so it only accepts usernames matching ^[a-z][a-z0-9_]*$ and rejects anything else.", + "description": ( + "Write answer.sh so it only accepts usernames matching ^[a-z][a-z0-9_]*$ " + "and rejects anything else." + ), "hint": "Validate input before using it.", "validator": "script", - "test_script": "printf 'alice1\n' | sh answer.sh | grep -xq 'VALID' && printf 'Bad-Name\n' | sh answer.sh | grep -xq 'INVALID'", + "test_script": ( + "printf 'alice1\n' | sh answer.sh | grep -xq 'VALID' " + "&& printf 'Bad-Name\n' | sh answer.sh | grep -xq 'INVALID'" + ), }, { "id": "28", "title": "CTF beginner: File signature", - "description": "Write answer.sh so it reports the file type of the input using the file command.", + "description": ( + "Write answer.sh so it reports the file type of the input using the file command." + ), "hint": "Magic bytes reveal the type.", "validator": "script", - "test_script": "printf 'hello world\n' > note.txt && sh answer.sh note.txt | grep -qi 'text'", + "test_script": ( + "printf 'hello world\n' > note.txt && sh answer.sh note.txt | grep -qi 'text'" + ), }, { "id": "29", "title": "Docker: Run container", - "description": "Write answer.sh so it prints the exact command to run an interactive Ubuntu container and remove it on exit.", + "description": ( + "Write answer.sh so it prints the exact command to run an interactive " + "Ubuntu container and remove it on exit." + ), "hint": "Combine --rm and -it.", "validator": "script", "test_script": "sh answer.sh | grep -xq 'docker run --rm -it ubuntu'", @@ -252,10 +328,14 @@ { "id": "30", "title": "Git: Branch creation", - "description": "Write answer.sh so it initializes a repo and creates a new branch named feature/auth.", + "description": ( + "Write answer.sh so it initializes a repo and creates a new branch named feature/auth." + ), "hint": "Use git checkout -b.", "validator": "script", - "test_script": "git init -q && sh answer.sh && git branch --show-current | grep -xq 'feature/auth'", + "test_script": ( + "git init -q && sh answer.sh && git branch --show-current | grep -xq 'feature/auth'" + ), }, { "id": "31", @@ -263,15 +343,24 @@ "description": "Write .github/workflows/ci.yml so the workflow triggers on push.", "hint": "The trigger key is short.", "validator": "script", - "test_script": "mkdir -p .github/workflows && grep -Eq '^on:' .github/workflows/ci.yml && grep -Eq 'push' .github/workflows/ci.yml", + "test_script": ( + "mkdir -p .github/workflows " + "&& grep -Eq '^on:' .github/workflows/ci.yml " + "&& grep -Eq 'push' .github/workflows/ci.yml" + ), }, { "id": "32", "title": "Cloud fundamentals: Shared model", - "description": "Write answer.sh so it prints IaaS when the prompt describes provider-managed hardware and customer-managed VMs.", + "description": ( + "Write answer.sh so it prints IaaS when the prompt describes " + "provider-managed hardware and customer-managed VMs." + ), "hint": "The answer is an acronym.", "validator": "script", - "test_script": "printf 'provider hardware and virtual machines\n' | sh answer.sh | grep -xq 'IaaS'", + "test_script": ( + "printf 'provider hardware and virtual machines\n' | sh answer.sh | grep -xq 'IaaS'" + ), }, ] @@ -279,31 +368,83 @@ "1": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "filesystem", "cli"]}, "2": {"category": "Programming", "difficulty": "easy", "tags": ["c", "build", "gcc"]}, "3": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "find", "filesystem"]}, - "4": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "permissions", "filesystem"]}, + "4": { + "category": "Linux", + "difficulty": "easy", + "tags": ["linux", "permissions", "filesystem"], + }, "5": {"category": "Programming", "difficulty": "easy", "tags": ["c", "stdio", "compile"]}, "6": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "navigation", "shell"]}, "7": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "process", "ps"]}, "8": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "networking", "sockets"]}, "9": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "packages", "apt"]}, "10": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "bash", "loops"]}, - "11": {"category": "Linux", "difficulty": "medium", "tags": ["linux", "permissions", "ownership"]}, + "11": { + "category": "Linux", + "difficulty": "medium", + "tags": ["linux", "permissions", "ownership"], + }, "12": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "logs", "systemd"]}, "13": {"category": "Linux", "difficulty": "easy", "tags": ["linux", "disk", "df"]}, "14": {"category": "Linux", "difficulty": "medium", "tags": ["bash", "scripting", "files"]}, "15": {"category": "Linux", "difficulty": "medium", "tags": ["bash", "filtering", "grep"]}, "16": {"category": "Programming", "difficulty": "medium", "tags": ["c", "pointers", "memory"]}, "17": {"category": "Programming", "difficulty": "medium", "tags": ["c", "malloc", "free"]}, - "18": {"category": "Programming", "difficulty": "medium", "tags": ["c", "structs", "linked-lists"]}, - "19": {"category": "Programming", "difficulty": "medium", "tags": ["c", "algorithms", "binary-search"]}, - "20": {"category": "Programming", "difficulty": "medium", "tags": ["python", "comprehensions", "lists"]}, - "21": {"category": "Programming", "difficulty": "easy", "tags": ["java", "entry-point", "syntax"]}, - "22": {"category": "Programming", "difficulty": "easy", "tags": ["javascript", "operators", "equality"]}, - "23": {"category": "Cybersecurity", "difficulty": "medium", "tags": ["web-security", "allowlist", "input-validation"]}, - "24": {"category": "Cybersecurity", "difficulty": "medium", "tags": ["reverse-engineering", "strings", "binary"]}, - "25": {"category": "Cybersecurity", "difficulty": "easy", "tags": ["forensics", "hashing", "sha256"]}, - "26": {"category": "Cybersecurity", "difficulty": "easy", "tags": ["cryptography", "base64", "encoding"]}, - "27": {"category": "Cybersecurity", "difficulty": "medium", "tags": ["secure-coding", "validation", "regex"]}, - "28": {"category": "Cybersecurity", "difficulty": "easy", "tags": ["ctf", "file-signatures", "file"]}, + "18": { + "category": "Programming", + "difficulty": "medium", + "tags": ["c", "structs", "linked-lists"], + }, + "19": { + "category": "Programming", + "difficulty": "medium", + "tags": ["c", "algorithms", "binary-search"], + }, + "20": { + "category": "Programming", + "difficulty": "medium", + "tags": ["python", "comprehensions", "lists"], + }, + "21": { + "category": "Programming", + "difficulty": "easy", + "tags": ["java", "entry-point", "syntax"], + }, + "22": { + "category": "Programming", + "difficulty": "easy", + "tags": ["javascript", "operators", "equality"], + }, + "23": { + "category": "Cybersecurity", + "difficulty": "medium", + "tags": ["web-security", "allowlist", "input-validation"], + }, + "24": { + "category": "Cybersecurity", + "difficulty": "medium", + "tags": ["reverse-engineering", "strings", "binary"], + }, + "25": { + "category": "Cybersecurity", + "difficulty": "easy", + "tags": ["forensics", "hashing", "sha256"], + }, + "26": { + "category": "Cybersecurity", + "difficulty": "easy", + "tags": ["cryptography", "base64", "encoding"], + }, + "27": { + "category": "Cybersecurity", + "difficulty": "medium", + "tags": ["secure-coding", "validation", "regex"], + }, + "28": { + "category": "Cybersecurity", + "difficulty": "easy", + "tags": ["ctf", "file-signatures", "file"], + }, "29": {"category": "DevOps", "difficulty": "easy", "tags": ["docker", "containers", "cli"]}, "30": {"category": "DevOps", "difficulty": "easy", "tags": ["git", "branches", "workflow"]}, "31": {"category": "DevOps", "difficulty": "easy", "tags": ["ci-cd", "github-actions", "yaml"]}, @@ -313,6 +454,7 @@ for puzzle in PUZZLES: puzzle.update(PUZZLE_METADATA.get(puzzle["id"], {})) + def get_puzzle(pid): return next((p for p in PUZZLES if p["id"] == pid), None) @@ -378,16 +520,18 @@ def ready(): @app.route("/levels", methods=["GET"]) def levels(): - return jsonify([ - { - "id": p["id"], - "title": p["title"], - "category": p["category"], - "difficulty": p["difficulty"], - "tags": p["tags"], - } - for p in PUZZLES - ]) + return jsonify( + [ + { + "id": p["id"], + "title": p["title"], + "category": p["category"], + "difficulty": p["difficulty"], + "tags": p["tags"], + } + for p in PUZZLES + ] + ) @app.route("/level/", methods=["GET"]) @@ -535,12 +679,14 @@ def api_level(pid): @app.route("/api/v1/progress", methods=["GET"]) def api_progress(): state = storage.load_state() - return jsonify({ - "progress": storage.get_progress_summary(state, catalog_levels()), - "solved": list(storage.get_solved_levels(state)), - "achievements": state.get("achievements", {}), - "profile": state.get("profile", {"display_name": "Learner", "preferences": {}}), - }) + return jsonify( + { + "progress": storage.get_progress_summary(state, catalog_levels()), + "solved": list(storage.get_solved_levels(state)), + "achievements": state.get("achievements", {}), + "profile": state.get("profile", {"display_name": "Learner", "preferences": {}}), + } + ) @app.route("/api/v1/achievements", methods=["GET"]) @@ -570,7 +716,15 @@ def web_puzzles(): solved = storage.get_solved_levels(state) if query: - levels = [lvl for lvl in levels if query in " ".join(str(lvl.get(k, "")) for k in ("id", "title", "description", "category", "difficulty")).lower()] + levels = [ + lvl + for lvl in levels + if query + in " ".join( + str(lvl.get(k, "")) + for k in ("id", "title", "description", "category", "difficulty") + ).lower() + ] if category: levels = [lvl for lvl in levels if lvl.get("category", "").lower() == category] if difficulty: @@ -595,7 +749,11 @@ def dashboard(): levels = catalog_levels() summary = storage.get_progress_summary(state, levels) solved = storage.get_solved_levels(state) - recent_levels = [get_puzzle(level_id) for level_id in state.get("meta", {}).get("recent_solved", []) if get_puzzle(level_id)] + recent_levels = [ + get_puzzle(level_id) + for level_id in state.get("meta", {}).get("recent_solved", []) + if get_puzzle(level_id) + ] return render_template( "dashboard.html", progress=summary, @@ -635,11 +793,17 @@ def web_submit(pid): if upload and upload.filename: files[upload.filename] = upload.read().decode(errors="ignore") if puzzle.get("validator") == "script" and not files: - return render_template("result.html", puzzle=puzzle_summary(puzzle), result={"ok": False, "error": "upload at least one source file"}, status=400) + return render_template( + "result.html", + puzzle=puzzle_summary(puzzle), + result={"ok": False, "error": "upload at least one source file"}, + status=400, + ) payload, status = _submission_response(puzzle, attempt, files) - return render_template("result.html", puzzle=puzzle_summary(puzzle), result=payload, status=status) - + return render_template( + "result.html", puzzle=puzzle_summary(puzzle), result=payload, status=status + ) def build_docker_command(source_dir): diff --git a/storage.py b/storage.py index 782a6cb..55da621 100644 --- a/storage.py +++ b/storage.py @@ -4,7 +4,6 @@ import tempfile from datetime import datetime, timezone - DEFAULT_STATE_ENV = "LUX_STATE" STATE_VERSION = 2 RECENT_ACTIVITY_LIMIT = 50 @@ -142,8 +141,7 @@ def normalize_state(state): elif not isinstance(solved, dict): solved = {} normalized["solved"] = { - str(level_id): _normalize_solved_entry(entry) - for level_id, entry in solved.items() + str(level_id): _normalize_solved_entry(entry) for level_id, entry in solved.items() } achievements = normalized.get("achievements", {}) @@ -191,7 +189,8 @@ def normalize_state(state): meta["recent_solved"] = [] meta["history"] = [ - item for item in (_normalize_activity_entry(entry) for entry in meta["history"]) + item + for item in (_normalize_activity_entry(entry) for entry in meta["history"]) if item is not None ][-RECENT_ACTIVITY_LIMIT:] @@ -279,15 +278,19 @@ def record_attempt( attempts[level_id] = entry history = meta.setdefault("history", []) - history.append(_normalize_activity_entry({ - "at": now, - "correct": correct, - "level_id": level_id, - "title": title, - "difficulty": difficulty, - "category": category, - "attempt_preview": attempt_preview[:120] if attempt_preview else None, - })) + history.append( + _normalize_activity_entry( + { + "at": now, + "correct": correct, + "level_id": level_id, + "title": title, + "difficulty": difficulty, + "category": category, + "attempt_preview": attempt_preview[:120] if attempt_preview else None, + } + ) + ) meta["history"] = [item for item in history if item is not None][-RECENT_ACTIVITY_LIMIT:] state.update(normalized) return state @@ -378,7 +381,9 @@ def update_profile(state: dict, *, display_name=None, preferences=None): normalized = normalize_state(state) profile = normalized.setdefault("profile", default_profile()) if display_name is not None: - profile["display_name"] = str(display_name).strip() or profile.get("display_name", "Learner") + profile["display_name"] = str(display_name).strip() or profile.get( + "display_name", "Learner" + ) if preferences is not None and isinstance(preferences, dict): prefs = profile.setdefault("preferences", {}) prefs.update(preferences) diff --git a/tests/test_achievements.py b/tests/test_achievements.py index 77fe3c3..14fcd2d 100644 --- a/tests/test_achievements.py +++ b/tests/test_achievements.py @@ -1,9 +1,8 @@ -import storage import achievements def test_threshold_achievements(tmp_path): - path = tmp_path / "state.json" + tmp_path / "state.json" state = {"version": 1, "solved": {}, "achievements": {}} # no achievements yet newly = achievements.evaluate_achievements(state) diff --git a/tests/test_agent_hints.py b/tests/test_agent_hints.py index 2e76ba4..c4d5856 100644 --- a/tests/test_agent_hints.py +++ b/tests/test_agent_hints.py @@ -12,14 +12,27 @@ def fake_chat(model, messages): calls.append(model) if model == "primary": raise RuntimeError("model unavailable") - return SimpleNamespace(message=SimpleNamespace(content="Use a smaller step. Avoid revealing the answer directly. And check the tags.")) + return SimpleNamespace( + message=SimpleNamespace( + content=( + "Use a smaller step. Avoid revealing the answer directly. And check the tags." + ) + ) + ) monkeypatch.setenv("LUX_OLLAMA_MODEL", "primary") monkeypatch.setenv("LUX_OLLAMA_MODELS", "fallback") monkeypatch.setitem(sys.modules, "ollama", SimpleNamespace(chat=fake_chat)) hint = agent.ask_hint_via_ollama( - {"id": "1", "title": "Test", "description": "Desc", "category": "Linux", "difficulty": "easy", "tags": ["shell"]}, + { + "id": "1", + "title": "Test", + "description": "Desc", + "category": "Linux", + "difficulty": "easy", + "tags": ["shell"], + }, state=storage.default_state(), ) @@ -38,7 +51,14 @@ def fake_chat(model, messages): try: agent.ask_hint_via_ollama( - {"id": "1", "title": "Test", "description": "Desc", "category": "Linux", "difficulty": "easy", "tags": []}, + { + "id": "1", + "title": "Test", + "description": "Desc", + "category": "Linux", + "difficulty": "easy", + "tags": [], + }, state=storage.default_state(), ) except RuntimeError as exc: diff --git a/tests/test_cli_state.py b/tests/test_cli_state.py index e421431..93c255b 100644 --- a/tests/test_cli_state.py +++ b/tests/test_cli_state.py @@ -1,10 +1,7 @@ -import os -import tempfile import json -from datetime import datetime, timezone -import storage import cli +import storage def test_storage_roundtrip(tmp_path): @@ -52,7 +49,9 @@ def test_progress_summary_tracks_streak(tmp_path): storage.mark_solved(state, "2", at="2026-08-07T10:00:00+00:00") storage.mark_solved(state, "3", at="2026-08-07T11:00:00+00:00") - summary = storage.get_progress_summary(state, [{"id": "1"}, {"id": "2"}, {"id": "3"}, {"id": "4"}]) + summary = storage.get_progress_summary( + state, [{"id": "1"}, {"id": "2"}, {"id": "3"}, {"id": "4"}] + ) assert summary["solved_count"] == 3 assert summary["current_streak"] == 2 diff --git a/tests/test_contributors.py b/tests/test_contributors.py index 6d65e73..d84b5db 100644 --- a/tests/test_contributors.py +++ b/tests/test_contributors.py @@ -6,23 +6,30 @@ def test_load_contributors_normalizes_entries(tmp_path): manifest = tmp_path / "contributors.json" - manifest.write_text(json.dumps({ - "contributors": [ + manifest.write_text( + json.dumps( { - "login": "ada", - "name": "Ada Lovelace", - "contributions": ["documentation"], - "badges": ["founding-contributor"], - }, - {"name": "missing login"}, - ]})) - - assert contributors.load_contributors(str(manifest)) == [{ - "login": "ada", - "name": "Ada Lovelace", - "contributions": ["documentation"], - "badges": ["founding-contributor"], - }] + "contributors": [ + { + "login": "ada", + "name": "Ada Lovelace", + "contributions": ["documentation"], + "badges": ["founding-contributor"], + }, + {"name": "missing login"}, + ] + } + ) + ) + + assert contributors.load_contributors(str(manifest)) == [ + { + "login": "ada", + "name": "Ada Lovelace", + "contributions": ["documentation"], + "badges": ["founding-contributor"], + } + ] def test_load_contributors_handles_malformed_manifest(tmp_path): @@ -34,12 +41,14 @@ def test_load_contributors_handles_malformed_manifest(tmp_path): def test_contributor_endpoints(monkeypatch): - entries = [{ - "login": "ada", - "name": "Ada Lovelace", - "contributions": ["documentation"], - "badges": ["founding-contributor"], - }] + entries = [ + { + "login": "ada", + "name": "Ada Lovelace", + "contributions": ["documentation"], + "badges": ["founding-contributor"], + } + ] monkeypatch.setattr(server.contributors, "load_contributors", lambda: entries) client = server.app.test_client() diff --git a/tests/test_server_sandbox.py b/tests/test_server_sandbox.py index 7f4ee5c..2063e72 100644 --- a/tests/test_server_sandbox.py +++ b/tests/test_server_sandbox.py @@ -1,9 +1,10 @@ import json -import server -import sandbox import pytest + +import sandbox import sandbox_audit +import server class _CompletedProcess: diff --git a/tests/test_smoke.py b/tests/test_smoke.py index acf1261..34f0c2c 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -46,8 +46,12 @@ def test_level_detail_and_submission_endpoints(monkeypatch): assert response.status_code == 200 assert response.get_json()["correct"] is True - monkeypatch.setattr(server, "run_docker", lambda files, script: {"passed": True, "exit_code": 0}) - response = client.post("/submit", json={"level_id": "5", "files": {"answer.c": "int main(void){return 0;}"}}) + monkeypatch.setattr( + server, "run_docker", lambda files, script: {"passed": True, "exit_code": 0} + ) + response = client.post( + "/submit", json={"level_id": "5", "files": {"answer.c": "int main(void){return 0;}"}} + ) assert response.status_code == 200 assert response.get_json()["correct"] is True diff --git a/tests/test_web_api.py b/tests/test_web_api.py index 6a8a183..dcd78b0 100644 --- a/tests/test_web_api.py +++ b/tests/test_web_api.py @@ -60,7 +60,9 @@ def test_versioned_api_endpoints(): def test_web_submission_flow(monkeypatch): client = server.app.test_client() - monkeypatch.setattr(server, "run_docker", lambda files, script: {"passed": True, "exit_code": 0}) + monkeypatch.setattr( + server, "run_docker", lambda files, script: {"passed": True, "exit_code": 0} + ) page = client.get("/puzzles/5") assert page.status_code == 200 @@ -69,7 +71,11 @@ def test_web_submission_flow(monkeypatch): response = client.post( "/puzzles/5/submit", - data={"csrf_token": token_match.group(1), "answer": "", "files": (BytesIO(b"int main(void){return 0;}"), "answer.c")}, + data={ + "csrf_token": token_match.group(1), + "answer": "", + "files": (BytesIO(b"int main(void){return 0;}"), "answer.c"), + }, ) assert response.status_code == 200 assert b"Submission accepted" in response.data