#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Browser setup entrypoint for PyCatan with per-agent OpenRouter models.""" import html as html_lib import json import os import ssl import sys import threading import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Dict, List, Optional from urllib.parse import parse_qs, urlparse try: import certifi os.environ.setdefault("SSL_CERT_FILE", certifi.where()) os.environ.setdefault("REQUESTS_CA_BUNDLE", certifi.where()) os.environ.setdefault("GRPC_DEFAULT_SSL_ROOTS_FILE_PATH", certifi.where()) ssl._create_default_https_context = ssl._create_unverified_context except Exception: pass sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from examples.ai_testing.play_with_ai import ( DEFAULT_PLAYER_NAMES, ELEVENLABS_TTS_MODELS, LOGS_DIR, PLAYER_COLORS, annotate_replay_session, create_game, group_replay_decisions, infer_players_from_decisions, infer_players_from_session, list_replay_marker_options, load_ai_config, load_env_file, load_replay_decision_chain, resolve_session_path, run_game, run_replay_viewer, ) from pycatan.ai.config import normalize_chat_language OPENROUTER_MODELS_URL = "https://openrouter.ai/api/v1/models" MIN_CONTEXT_LENGTH = 32000 MIN_COMPLETION_TOKENS = 4096 PLAYER_GENDERS = {1: "female", 2: "male", 3: "male", 4: "female"} GEMINI_TTS_MODELS = [ "gemini-2.5-flash-preview-tts", "gemini-2.5-pro-preview-tts", "gemini-3.1-flash-tts-preview", "gemini-3.1-pro-tts-preview", ] GEMINI_TTS_VOICES = [ "Kore", "Puck", "Zephyr", "Charon", "Fenrir", "Leda", "Orus", "Aoede", "Callirrhoe", "Autonoe", "Enceladus", "Iapetus", "Umbriel", "Algieba", "Despina", "Erinome", "Algenib", "Rasalgethi", "Laomedeia", "Achernar", "Alnilam", "Schedar", "Gacrux", "Pulcherrima", "Achird", "Zubenelgenubi", "Vindemiatrix", "Sadachbia", "Sadaltager", "Sulafat", ] FALLBACK_MODELS = [ { "id": "openai/gpt-4o", "name": "OpenAI: GPT-4o", "provider": "openai", "context_length": 128000, "max_completion_tokens": 16384, "output_modalities": ["text"], "supported_parameters": ["tools", "response_format", "temperature", "max_tokens"], "pricing": {"prompt": "", "completion": ""}, }, { "id": "openai/gpt-4o-mini", "name": "OpenAI: GPT-4o Mini", "provider": "openai", "context_length": 128000, "max_completion_tokens": 16384, "output_modalities": ["text"], "supported_parameters": ["tools", "response_format", "temperature", "max_tokens"], "pricing": {"prompt": "", "completion": ""}, }, { "id": "anthropic/claude-sonnet-4.5", "name": "Anthropic: Claude Sonnet 4.5", "provider": "anthropic", "context_length": 200000, "max_completion_tokens": 8192, "output_modalities": ["text"], "supported_parameters": ["tools", "response_format", "temperature", "max_tokens"], "pricing": {"prompt": "", "completion": ""}, }, { "id": "google/gemini-2.5-pro", "name": "Google: Gemini 2.5 Pro", "provider": "google", "context_length": 1048576, "max_completion_tokens": 65536, "output_modalities": ["text"], "supported_parameters": ["tools", "response_format", "temperature", "max_tokens"], "pricing": {"prompt": "", "completion": ""}, }, { "id": "google/gemini-2.5-flash", "name": "Google: Gemini 2.5 Flash", "provider": "google", "context_length": 1048576, "max_completion_tokens": 65536, "output_modalities": ["text"], "supported_parameters": ["tools", "response_format", "temperature", "max_tokens"], "pricing": {"prompt": "", "completion": ""}, }, ] def provider_from_model_id(model_id: str) -> str: return (model_id.split("/", 1)[0] if "/" in model_id else "custom").strip().lower() def normalize_model(raw: Dict[str, Any]) -> Dict[str, Any]: model_id = str(raw.get("id", "")).strip() pricing = raw.get("pricing") or {} architecture = raw.get("architecture") or {} top_provider = raw.get("top_provider") or {} return { "id": model_id, "name": raw.get("name") or model_id, "provider": provider_from_model_id(model_id), "context_length": raw.get("context_length") or 0, "max_completion_tokens": top_provider.get("max_completion_tokens") or 0, "output_modalities": architecture.get("output_modalities") or ["text"], "supported_parameters": raw.get("supported_parameters") or [], "pricing": { "prompt": pricing.get("prompt", ""), "completion": pricing.get("completion", ""), }, } def is_suitable_openrouter_model(model: Dict[str, Any]) -> bool: supported = set(model.get("supported_parameters") or []) output_modalities = set(model.get("output_modalities") or ["text"]) context_length = int(model.get("context_length") or 0) max_completion_tokens = int(model.get("max_completion_tokens") or 0) return ( bool(model.get("id")) and "tools" in supported and "response_format" in supported and "text" in output_modalities and context_length >= MIN_CONTEXT_LENGTH and (max_completion_tokens == 0 or max_completion_tokens >= MIN_COMPLETION_TOKENS) ) def fetch_openrouter_models(api_key: str = "") -> List[Dict[str, Any]]: try: import requests headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} request_kwargs = { "headers": headers, "params": {"output_modalities": "text"}, "timeout": 12, "verify": os.environ.get("REQUESTS_CA_BUNDLE") or True, } try: response = requests.get(OPENROUTER_MODELS_URL, **request_kwargs) except requests.exceptions.SSLError: requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined] response = requests.get(OPENROUTER_MODELS_URL, **{**request_kwargs, "verify": False}) response.raise_for_status() models = [normalize_model(item) for item in response.json().get("data", [])] except Exception as exc: print(f"[OPENROUTER] Could not fetch model list, using fallback models: {exc}") models = list(FALLBACK_MODELS) usable = [model for model in models if is_suitable_openrouter_model(model)] usable = usable or [model for model in FALLBACK_MODELS if is_suitable_openrouter_model(model)] usable.sort(key=lambda item: (item["provider"], item["name"].lower())) return usable def _env_has(name: str) -> bool: return bool(os.environ.get(name)) def _options(values: List[str], selected: str = "") -> str: return "".join( f'" for value in values ) def _tts_model_options(selected: str = "eleven_v3") -> str: return "".join( f'' for model in ELEVENLABS_TTS_MODELS ) def _recent_session_options() -> str: if not LOGS_DIR.exists(): return "" sessions = sorted( [path.name for path in LOGS_DIR.iterdir() if path.is_dir() and path.name.startswith("session_")], reverse=True, )[:30] return "".join(f'' for session in sessions) def _price_label(model: Dict[str, Any]) -> str: pricing = model.get("pricing") or {} prompt = pricing.get("prompt") or "?" completion = pricing.get("completion") or "?" return f"prompt {prompt} / completion {completion}" def _slot_llm_from_selected(selected: Dict[str, Any], slot: int, models: List[Dict[str, Any]]) -> Dict[str, str]: providers = sorted({model["provider"] for model in models}) provider = selected.get(f"provider_{slot}") or (providers[0] if providers else "openai") model_id = ( selected.get(f"manual_model_{slot}") or selected.get(f"model_{slot}") or next((model["id"] for model in models if model["provider"] == provider), models[0]["id"]) ) return { "provider": "openrouter", "model_provider": provider_from_model_id(model_id), "model_name": model_id, "api_key_env_var": "OPENROUTER_API_KEY", } def render_landing_page() -> bytes: recent_sessions = [] if LOGS_DIR.exists(): recent_sessions = sorted( [path.name for path in LOGS_DIR.iterdir() if path.is_dir() and path.name.startswith("session_")], reverse=True, )[:6] replay_links = "".join( "{html_lib.escape(session)}" for session in recent_sessions ) if not replay_links: replay_links = "

No public replay sessions are bundled yet.

" return f""" PyCatan AI
Public AI table

PyCatan AI

משחק קטאן עם שחקני AI, ריפליי של סשנים ציבוריים, וניתוח החלטות. אפשר לצפות בלי מפתח, או להתחיל משחק חי עם מפתח OpenRouter משלך.

התחל משחק או ריפליי צפה בריפליי
1. צופיםריפליי וניתוח החלטות זמינים בלי להזין מפתחות.
2. משחקיםמשחק חי עם LLM משתמש במפתח OpenRouter שהמשתמש מזין בעצמו.
3. משתפיםהסשנים נשמרים כתוכן ציבורי של הדמו, בהתאם לאחסון הזמין.
""".encode("utf-8") def render_settings_page( models: List[Dict[str, Any]], key_mode: str, errors: Optional[List[str]] = None, selected: Optional[Dict[str, Any]] = None, ) -> str: selected = selected or {} player_count = int(selected.get("player_count", 4) or 4) providers = sorted({model["provider"] for model in models}) provider_options = "".join( f'' for provider in providers ) errors_html = "" if errors: errors_html = "
" player_blocks = [] for index in range(4): slot = index + 1 slot_llm = _slot_llm_from_selected(selected, slot, models) name = selected.get(f"player_{slot}") or DEFAULT_PLAYER_NAMES[index] player_blocks.append(f"""
P{slot} - {PLAYER_COLORS[index]} - {PLAYER_GENDERS[slot]}
""") model_cards = [] for model in models[:28]: model_cards.append( "
  • " f"{html_lib.escape(model['id'])}" f"{html_lib.escape(model['name'])}" f"{int(model.get('context_length') or 0):,} ctx | {_price_label(model)}" "
    • " ) models_json = json.dumps(models, ensure_ascii=False) # Server-side validation below knows the selected run mode/TTS provider. # Avoid static browser "required" on fields that may be hidden by the UI. openrouter_required = "" gemini_required = "" eleven_key_required = "" eleven_voice_required = "" return f""" PyCatan OpenRouter Setup

    PyCatan OpenRouter Setup

    Full setup mode: run/replay, per-agent OpenRouter models, TTS, reactions, and player slots. TTS stays mapped to PLAYER_1 through PLAYER_4. P1 and P4 are female; P2 and P3 are male.

    {errors_html}
    Run Mode

    Replay markers are loaded from recorded game actions only.

    Execution

    Victory points are written into the game state and prompts. Leave random seed blank to use the current deterministic default: 0.

    Voice
    Players and models

    For replay/resume modes, player names are loaded from the selected session; only models remain editable.

    {''.join(player_blocks)}
    """ def render_starting_page(player_configs: List[Dict[str, Any]], run_mode: str) -> bytes: rows = "".join( f"
  • P{index + 1}: {html_lib.escape(player['name'])} - " f"{html_lib.escape(player.get('llm', {}).get('model_name', 'offline'))}
    • " for index, player in enumerate(player_configs) ) return f"""Starting PyCatan

    Starting game...

    Mode: {html_lib.escape(run_mode)}

    The board will open here as soon as the game server is ready.

    """.encode("utf-8") def _as_bool_field(fields: Dict[str, str], name: str) -> bool: return name in fields def _parse_optional_int(value: str, errors: List[str], label: str) -> Optional[int]: if not value: return None try: parsed = int(value) if parsed < 1: errors.append(f"{label} must be at least 1.") return parsed except ValueError: errors.append(f"{label} must be a number.") return None def _parse_float(value: str, default: float, errors: List[str], label: str) -> float: try: parsed = float(value or default) if parsed < 0: errors.append(f"{label} cannot be negative.") return parsed except ValueError: errors.append(f"{label} must be a number.") return default def _settings_selection_from_query(query: Dict[str, List[str]]) -> Dict[str, Any]: mode = (query.get("mode") or query.get("run_mode") or [""])[0].strip() session = (query.get("session") or query.get("replay_session") or [""])[0].strip() selected: Dict[str, Any] = {} if mode in {"new_game", "resume_session", "watch_replay", "analyse_game"}: selected["run_mode"] = mode if session: selected["replay_session"] = session if mode in {"watch_replay", "analyse_game"}: selected["tts_provider"] = "off" return selected def _infer_session_player_names(session_dir: Path) -> List[str]: """Infer locked player names for a replay/resume source session.""" names = infer_players_from_session(session_dir) if names: return names[:4] try: decisions = load_replay_decision_chain(session_dir) except Exception: return [] return infer_players_from_decisions(decisions)[:4] def collect_settings(port: int = 5000, key_mode: str = "env") -> Dict[str, Any]: settings: Dict[str, Any] = {} ready = threading.Event() models = fetch_openrouter_models(os.environ.get("OPENROUTER_API_KEY", "")) suitable_by_id = {model["id"]: model for model in models if is_suitable_openrouter_model(model)} valid_model_ids = set(suitable_by_id) valid_providers = {model["provider"] for model in models} use_env_keys = key_mode == "env" class ReusableThreadingHTTPServer(ThreadingHTTPServer): allow_reuse_address = True class SettingsHandler(BaseHTTPRequestHandler): def log_message(self, format, *args): return def _send_html(self, body: Any, status: int = 200) -> None: body_bytes = body if isinstance(body, bytes) else body.encode("utf-8") self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body_bytes))) self.end_headers() self.wfile.write(body_bytes) def _send_json(self, payload: Dict[str, Any], status: int = 200) -> None: body_bytes = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body_bytes))) self.end_headers() self.wfile.write(body_bytes) def do_GET(self): parsed_url = urlparse(self.path) if parsed_url.path == "/healthz": self._send_json({"ok": True}) return if parsed_url.path == "/replay-markers": query = parse_qs(parsed_url.query) session_ref = query.get("session", [""])[0].strip() if not session_ref: self._send_json({"markers": []}) return try: session_dir = resolve_session_path(session_ref) markers = list_replay_marker_options(session_dir) except Exception as exc: self._send_json({"error": str(exc), "markers": []}, status=400) return self._send_json({"markers": markers}) return if parsed_url.path == "/session-players": query = parse_qs(parsed_url.query) session_ref = query.get("session", [""])[0].strip() if not session_ref: self._send_json({"players": []}) return try: session_dir = resolve_session_path(session_ref) players = _infer_session_player_names(session_dir) except Exception as exc: self._send_json({"error": str(exc), "players": []}, status=400) return self._send_json({"players": players}) return if parsed_url.path == "/": self._send_html(render_landing_page()) return if parsed_url.path != "/settings": self.send_response(302) self.send_header("Location", "/") self.end_headers() return selected = _settings_selection_from_query(parse_qs(parsed_url.query)) self._send_html(render_settings_page(models, key_mode, selected=selected)) def do_POST(self): if self.path != "/start": self.send_error(404) return length = int(self.headers.get("Content-Length", "0")) raw_fields = parse_qs(self.rfile.read(length).decode("utf-8"), keep_blank_values=True) fields = {key: values[0].strip() for key, values in raw_fields.items()} errors: List[str] = [] run_mode = fields.get("run_mode", "new_game") valid_run_modes = {"new_game", "resume_session", "watch_replay", "analyse_game"} if run_mode not in valid_run_modes: errors.append("Choose a valid run mode.") live_mode = run_mode in {"new_game", "resume_session"} no_llm = _as_bool_field(fields, "no_llm") replay_speak = _as_bool_field(fields, "replay_speak") api_key = fields.get("openrouter_api_key") or (os.environ.get("OPENROUTER_API_KEY", "") if use_env_keys else "") if live_mode and not no_llm and not api_key: errors.append("Enter an OpenRouter API key or set OPENROUTER_API_KEY.") replay_session = fields.get("replay_session", "") if run_mode != "new_game" and not replay_session: errors.append("Choose a recorded session for replay/resume/analyse mode.") if fields.get("replay_through") and fields.get("replay_stop_before"): errors.append("Use either replay-through or replay-stop-before, not both.") replay_session_path_for_validation = None locked_player_names: List[str] = [] if run_mode != "new_game" and replay_session: try: replay_session_path_for_validation = resolve_session_path(replay_session) locked_player_names = _infer_session_player_names(replay_session_path_for_validation) except FileNotFoundError as exc: errors.append(str(exc)) if replay_session_path_for_validation and (fields.get("replay_through") or fields.get("replay_stop_before")): try: load_replay_decision_chain( replay_session_path_for_validation, replay_through=fields.get("replay_through") or None, replay_stop_before=fields.get("replay_stop_before") or None, ) except (TypeError, ValueError) as exc: errors.append( f"{exc}. Choose one of the suggested action markers; reaction-only table talk is not replayable as a marker." ) if fields.get("config_path") and not Path(fields["config_path"]).exists(): errors.append("Config file was not found.") player_count = _parse_optional_int(fields.get("player_count", "4"), errors, "Player count") or 4 if run_mode != "new_game" and locked_player_names: player_count = min(4, max(2, len(locked_player_names))) for index, name in enumerate(locked_player_names[:player_count]): fields[f"player_{index + 1}"] = name if player_count not in (2, 3, 4): errors.append("Choose 2, 3, or 4 players.") player_count = 4 fields["player_count"] = str(player_count) replay_max_decisions = _parse_optional_int(fields.get("replay_max_decisions", ""), errors, "Replay max decisions") replay_delay = _parse_float(fields.get("replay_delay", "2.5"), 2.5, errors, "Replay delay") replay_text_lead = _parse_float(fields.get("replay_text_lead", "0.25"), 0.25, errors, "Replay text lead") reaction_mode = fields.get("reaction_mode", "default") if reaction_mode not in {"default", "off", "sync", "async"}: errors.append("Choose a valid reaction mode.") reaction_batch_size = _parse_optional_int(fields.get("reaction_batch_size", ""), errors, "Reaction batch size") victory_points = _parse_optional_int(fields.get("victory_points", "5"), errors, "Victory points") or 5 fields["victory_points"] = str(victory_points) game_context = fields.get("game_context", "").strip() if len(game_context) > 4000: errors.append("Additional game context must be 4000 characters or less.") random_seed = 0 if fields.get("random_seed"): try: random_seed = int(fields["random_seed"]) except ValueError: errors.append("Random seed must be a whole number.") tts_provider = fields.get("tts_provider", "gemini") if tts_provider not in {"off", "gemini", "elevenlabs"}: errors.append("Choose a valid voice provider.") gemini_api_key = fields.get("gemini_api_key") or (os.environ.get("GEMINI_API_KEY", "") if use_env_keys else "") elevenlabs_api_key = fields.get("elevenlabs_api_key") or (os.environ.get("ELEVENLABS_API_KEY", "") if use_env_keys else "") elevenlabs_voice_id = fields.get("elevenlabs_default_voice_id") or (os.environ.get("ELEVENLABS_DEFAULT_VOICE_ID", "") if use_env_keys else "") needs_voice = live_mode or replay_speak if tts_provider == "gemini" and needs_voice and not gemini_api_key: errors.append("Enter a Gemini API key for Gemini TTS or set GEMINI_API_KEY.") if tts_provider == "elevenlabs" and needs_voice: if not elevenlabs_api_key: errors.append("Enter an ElevenLabs API key or set ELEVENLABS_API_KEY.") if not elevenlabs_voice_id: errors.append("Enter an ElevenLabs default voice ID or set ELEVENLABS_DEFAULT_VOICE_ID.") slot_llms: List[Dict[str, str]] = [] player_configs: List[Dict[str, Any]] = [] seen_names = set() for index in range(player_count): slot = index + 1 name = fields.get(f"player_{slot}") or DEFAULT_PLAYER_NAMES[index] model_id = fields.get(f"manual_model_{slot}") or fields.get(f"model_{slot}") or "" provider = fields.get(f"provider_{slot}") or provider_from_model_id(model_id) if run_mode == "new_game": if not name: errors.append(f"P{slot} needs a name.") if name.lower() in seen_names: errors.append("Player names must be unique.") seen_names.add(name.lower()) if provider and provider not in valid_providers: errors.append(f"Unknown provider for P{slot}: {provider}") if live_mode and not no_llm: if not model_id: errors.append(f"P{slot} needs a model.") elif model_id not in valid_model_ids: errors.append( f"Model for P{slot} is not suitable. Choose a listed model with tools, " f"response_format, text output, >= {MIN_CONTEXT_LENGTH} context and " f">= {MIN_COMPLETION_TOKENS} output tokens." ) llm = { "provider": "openrouter", "model_provider": provider_from_model_id(model_id), "model_name": model_id, "api_key_env_var": "OPENROUTER_API_KEY", } slot_llms.append(llm) player_configs.append({"name": name, "is_ai": True, "color": PLAYER_COLORS[index], "llm": llm}) if errors: self._send_html(render_settings_page(models, key_mode, errors=errors, selected=fields), status=400) return if api_key: os.environ["OPENROUTER_API_KEY"] = api_key if gemini_api_key: os.environ["GEMINI_API_KEY"] = gemini_api_key if tts_provider == "gemini": os.environ["TTS_PROVIDER"] = "gemini" os.environ["GEMINI_TTS_ENABLED"] = "true" os.environ["GEMINI_TTS_MODEL_ID"] = fields.get("gemini_tts_model", GEMINI_TTS_MODELS[0]) os.environ["GEMINI_TTS_VOICE_NAME"] = fields.get("gemini_tts_voice", "Kore") os.environ.setdefault("GEMINI_TTS_PLAY_AUDIO", "true") elif tts_provider == "elevenlabs": os.environ["TTS_PROVIDER"] = "elevenlabs" os.environ["ELEVENLABS_TTS_ENABLED"] = "true" os.environ["ELEVENLABS_API_KEY"] = elevenlabs_api_key os.environ["ELEVENLABS_DEFAULT_VOICE_ID"] = elevenlabs_voice_id os.environ["ELEVENLABS_TTS_MODEL_ID"] = fields.get("elevenlabs_tts_model", "eleven_v3") os.environ.setdefault("ELEVENLABS_TTS_OUTPUT_FORMAT", "pcm_16000") os.environ.setdefault("ELEVENLABS_TTS_PLAY_AUDIO", "true") else: os.environ["TTS_PROVIDER"] = "off" settings.update({ "run_mode": run_mode, "player_configs": player_configs if run_mode == "new_game" else [], "slot_llms": slot_llms, "chat_language": normalize_chat_language(fields.get("chat_language") or "hebrew"), "no_llm": no_llm, "reaction_mode": reaction_mode, "reaction_batch_size": reaction_batch_size, "victory_points": victory_points, "game_context": game_context, "random_seed": random_seed, "config_path": fields.get("config_path") or None, "replay_session": replay_session or None, "replay_max_decisions": replay_max_decisions, "replay_through": fields.get("replay_through") or None, "replay_stop_before": fields.get("replay_stop_before") or None, "replay_skip_chat": _as_bool_field(fields, "replay_skip_chat"), "replay_delay": replay_delay, "replay_text_lead": replay_text_lead, "replay_speak": replay_speak, }) starting_players = player_configs if run_mode == "new_game" else [ {"name": replay_session or "recorded session", "llm": {"model_name": run_mode}} ] self._send_html(render_starting_page(starting_players, run_mode)) ready.set() bind_host = os.environ.get("PYCATAN_BIND_HOST", "127.0.0.1") public_host = os.environ.get("PYCATAN_PUBLIC_HOST", "localhost") server = ReusableThreadingHTTPServer((bind_host, port), SettingsHandler) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() url = f"http://{public_host}:{port}/settings" print(f"[SETUP] OpenRouter settings page: {url}") if os.environ.get("PYCATAN_NO_BROWSER", "").lower() not in {"1", "true", "yes", "on"}: try: webbrowser.open(url) except Exception as exc: print(f"[SETUP] Could not open browser automatically: {exc}") print("[SETUP] Waiting for OpenRouter settings...") try: while not ready.wait(timeout=0.25): pass finally: server.shutdown() server.server_close() return settings def _player_configs_for_replay(player_names: List[str], slot_llms: List[Dict[str, str]]) -> List[Dict[str, Any]]: configs = [] for index, name in enumerate(player_names[:4]): llm = slot_llms[index] if index < len(slot_llms) else slot_llms[-1] configs.append({"name": name, "is_ai": True, "color": PLAYER_COLORS[index], "llm": llm}) return configs def _apply_reaction_settings(ai_config, settings: Dict[str, Any]) -> None: reaction_mode = settings["reaction_mode"] if reaction_mode == "off": ai_config.agent.enable_reactions = False elif reaction_mode == "async": ai_config.agent.enable_reactions = True ai_config.agent.async_reactions = True elif reaction_mode == "sync": ai_config.agent.enable_reactions = True ai_config.agent.async_reactions = False if settings["reaction_batch_size"] is not None: ai_config.agent.reaction_max_batch_messages = settings["reaction_batch_size"] def main() -> None: import argparse parser = argparse.ArgumentParser(description="Play Catan with per-agent OpenRouter models") parser.add_argument("--config", type=str, help="Optional AI config YAML") parser.add_argument("--port", type=int, default=5000, help="Setup and game web port") parser.add_argument("--use-env-keys", action="store_true", help="Use ENV/.env keys when browser fields are blank") parser.add_argument("--ask-api-keys", "--ask-keys", action="store_true", help="Require keys typed into the browser form") args = parser.parse_args() if args.use_env_keys and args.ask_api_keys: parser.error("--use-env-keys and --ask-api-keys cannot be used together") load_env_file() settings = collect_settings(port=args.port, key_mode=("ask" if args.ask_api_keys else "env")) ai_config = load_ai_config(settings.get("config_path") or args.config) ai_config.llm.provider = "openrouter" ai_config.llm.api_key_env_var = "OPENROUTER_API_KEY" ai_config.llm.enable_streaming = True ai_config.agent.chat_language = settings["chat_language"] _apply_reaction_settings(ai_config, settings) replay_session_path = resolve_session_path(settings["replay_session"]) if settings["replay_session"] else None watch_mode = settings["run_mode"] in {"watch_replay", "analyse_game"} analyse_mode = settings["run_mode"] == "analyse_game" if watch_mode and not replay_session_path: parser.error("watch/analyse mode requires a replay session") if ( watch_mode and replay_session_path and not os.environ.get("AI_TTS_CACHE_DIR") and not os.environ.get("PYCATAN_TTS_CACHE_DIR") ): os.environ["AI_TTS_CACHE_DIR"] = str(replay_session_path / "tts_cache") os.environ["AI_TTS_CACHE_DIR_AUTO"] = "replay_session_default" print(f"[REPLAY] TTS cache: {os.environ['AI_TTS_CACHE_DIR']}") elif not watch_mode and not os.environ.get("AI_TTS_CACHE_DIR") and not os.environ.get("PYCATAN_TTS_CACHE_DIR"): print("[TTS] Voice cache: per-session tts_cache/") replay_decision_list: List[Dict[str, Any]] = [] replay_decisions_by_player: Dict[str, List[Dict[str, Any]]] = {} if replay_session_path: replay_decision_list = load_replay_decision_chain( replay_session_path, max_decisions=settings["replay_max_decisions"], replay_through=settings["replay_through"], replay_stop_before=settings["replay_stop_before"], ) replay_decisions_by_player = group_replay_decisions(replay_decision_list) player_names = infer_players_from_session(replay_session_path) or infer_players_from_decisions(replay_decision_list) player_configs = _player_configs_for_replay(player_names, settings["slot_llms"]) print(f"[REPLAY] Source: {replay_session_path}") print(f"[REPLAY] Loaded {len(replay_decision_list)} parsed decisions") else: player_configs = settings["player_configs"] if not player_configs: parser.error("No players could be inferred for this run") if not settings["no_llm"]: ai_config.llm.model_name = player_configs[0]["llm"]["model_name"] send_to_llm = (not settings["no_llm"]) and not watch_mode manual_actions = False print(f"[MODE] LLM: {'ON' if send_to_llm else 'OFF'} | Actions: Auto") print(f"[CONFIG] Victory points to win: {settings['victory_points']}") if settings.get("game_context"): print("[CONFIG] Additional game context enabled") print("[CONFIG] OpenRouter per-agent models") game_manager, ai_manager, web_viz = create_game( player_configs, send_to_llm=send_to_llm, manual_actions=manual_actions, config=ai_config, replay_decisions=replay_decisions_by_player, replay_chat=not settings["replay_skip_chat"], replay_speak=(settings["replay_speak"] and not watch_mode), replay_only=watch_mode, web_port=args.port, random_seed=settings["random_seed"], game_config={ "victory_points": settings["victory_points"], "game_context": settings.get("game_context", ""), }, ) if send_to_llm: for player in player_configs: ai_manager.set_agent_llm_config( player_name=player["name"], provider=player["llm"]["provider"], model_name=player["llm"]["model_name"], api_key_env_var=player["llm"]["api_key_env_var"], ) print(f"[OPENROUTER] {player['name']}: {player['llm']['model_name']}") if replay_session_path: annotate_replay_session( ai_manager, replay_session_path, replay_decision_list, settings["replay_through"], settings["replay_stop_before"], mode=( "analyse_game_visual_playback" if analyse_mode else "watch_replay_visual_playback" if watch_mode else "fast_action_replay_then_live_ai" ), ) print(f"[REPLAY] New derived session: {ai_manager.get_session_path()}") if watch_mode: run_replay_viewer( game_manager, ai_manager, web_viz, delay_seconds=max(0.0, settings["replay_delay"]), source_session=replay_session_path, text_lead_seconds=max(0.0, settings["replay_text_lead"]), ) else: run_game(game_manager, ai_manager, web_viz) if __name__ == "__main__": main()