diff --git a/examples/experimental/werewolves/DEPLOYMENT.md b/examples/experimental/werewolves/DEPLOYMENT.md new file mode 100644 index 000000000..63baea223 --- /dev/null +++ b/examples/experimental/werewolves/DEPLOYMENT.md @@ -0,0 +1,274 @@ +# Werewolf Game Deployment Guide + +This guide covers deploying the Werewolf game using Option A (Hybrid architecture). + +## Architecture + +``` +Frontend (Next.js) on Vercel + ↓ HTTPS/WSS +Backend (FastAPI) on Railway/Render + ↓ +Redis (Upstash) - for future state management +``` + +## Prerequisites + +- Node.js 18+ and npm +- Python 3.10+ +- Railway or Render account (backend) +- Vercel account (frontend) +- Upstash account (optional, for Redis) + +## Local Development + +### 1. Backend Setup + +```bash +cd backend/ + +# If using conda (recommended if you already have sotopia-rl): +conda activate sotopia-rl + +# OR if using venv: +# python -m venv venv +# source venv/bin/activate # On Windows: venv\Scripts\activate + +# Install additional backend dependencies +pip install -r requirements.txt + +# Note: sotopia package should already be installed in your conda env +# If not, install it: +# cd ../../../.. (go to sotopia root) +# pip install -e . +# cd examples/experimental/werewolves/backend/ + +# Copy environment file +cp .env.example .env + +# Start backend +python main.py +# Backend runs on http://localhost:8000 +``` + +### 2. Frontend Setup + +```bash +cd frontend/ + +# Install dependencies +npm install + +# Copy environment file +cp .env.example .env.local + +# Update .env.local +# NEXT_PUBLIC_BACKEND_URL=http://localhost:8000 +# NEXT_PUBLIC_WS_URL=ws://localhost:8000 + +# Start frontend +npm run dev +# Frontend runs on http://localhost:3000 +``` + +### 3. Test Locally + +1. Open http://localhost:3000 +2. Enter your name and create a game +3. You'll be redirected to the game page +4. Check browser console and backend logs for WebSocket connection + +## Production Deployment + +### Backend (Railway) + +1. **Create Railway Project** + ```bash + cd backend/ + railway login + railway init + ``` + +2. **Set Environment Variables** (in Railway dashboard) + ``` + PORT=8000 + ALLOWED_ORIGINS=https://your-frontend.vercel.app,https://*.vercel.app + OPENAI_API_KEY=sk-... + REDIS_OM_URL=redis://:@localhost:6379 + ``` + +3. **Deploy** + ```bash + railway up + ``` + +4. **Note your Railway URL**: `https://your-backend.up.railway.app` + +### Backend (Alternative: Render) + +1. Create new Web Service on Render +2. Connect your GitHub repo +3. Set: + - Build Command: `pip install -r requirements.txt` + - Start Command: `uvicorn main:app --host 0.0.0.0 --port $PORT` +4. Add environment variables (same as Railway) +5. Deploy + +### Frontend (Vercel) + +1. **Install Vercel CLI** + ```bash + npm i -g vercel + ``` + +2. **Deploy** + ```bash + cd frontend/ + vercel + ``` + +3. **Set Environment Variables** (in Vercel dashboard or CLI) + ```bash + vercel env add NEXT_PUBLIC_BACKEND_URL + # Enter: https://your-backend.railway.app + + vercel env add NEXT_PUBLIC_WS_URL + # Enter: wss://your-backend.railway.app + ``` + +4. **Redeploy** to apply env vars + ```bash + vercel --prod + ``` + +### Update Backend CORS + +After deploying frontend, update backend's `ALLOWED_ORIGINS`: + +``` +ALLOWED_ORIGINS=https://your-frontend.vercel.app,https://*.vercel.app +``` + +## Testing Production + +1. Open `https://your-frontend.vercel.app` +2. Create a game +3. Check: + - Network tab shows WebSocket connection (`wss://...`) + - Backend logs show WebSocket accepts + - Game events appear in real-time + +## Troubleshooting + +### WebSocket Connection Fails + +**Issue**: Frontend can't connect to WebSocket + +**Solutions**: +- Check `NEXT_PUBLIC_WS_URL` uses `wss://` (not `ws://`) +- Verify backend CORS allows your frontend domain +- Check Railway/Render logs for connection errors + +### Backend Import Errors + +**Issue**: `ModuleNotFoundError: No module named 'sotopia'` + +**Solution**: Backend needs access to parent sotopia package + +```bash +# If using conda env sotopia-rl: +conda activate sotopia-rl +# Sotopia should already be installed. If not: +cd /path/to/sotopia (root directory) +pip install -e . + +# If using venv: +source venv/bin/activate +pip install -e ../../../.. # Install sotopia from parent directory + +# The game_manager.py already adds parent directory to sys.path as a fallback +``` + +### CORS Errors + +**Issue**: `Access to fetch at '...' from origin '...' has been blocked by CORS policy` + +**Solution**: Update backend `ALLOWED_ORIGINS` to include frontend URL + +### Game Doesn't Start + +**Issue**: Backend accepts connections but game doesn't progress + +**Solution**: Check backend logs for Python errors. Common issues: +- Missing OpenAI API key +- Redis connection issues (if using) +- Missing roster.json or role_actions.json + +## Environment Variables Reference + +### Backend (.env) + +```bash +# Required +PORT=8000 +ALLOWED_ORIGINS=https://your-frontend.vercel.app + +# Optional +REDIS_URL=redis://... +OPENAI_API_KEY=sk-... +``` + +### Frontend (.env.local) + +```bash +# Required +NEXT_PUBLIC_BACKEND_URL=https://your-backend.railway.app +NEXT_PUBLIC_WS_URL=wss://your-backend.railway.app +``` + +## Monitoring + +### Railway +- View logs: `railway logs` +- Check metrics in Railway dashboard + +### Vercel +- View logs in Vercel dashboard +- Check Analytics for frontend performance + +### Health Checks + +- Backend health: `https://your-backend.railway.app/health` +- Frontend: Visit homepage + +## Scaling Considerations + +Currently, the backend is a single long-running process. For higher scale: + +1. **Add Redis** for state persistence (allows backend restarts) +2. **Implement Option B** (refactor to stateless handlers) +3. **Use managed WebSocket** service (Pusher, Ably) + +## Cost Estimates + +- **Vercel**: Free tier (hobby projects) +- **Railway**: ~$5-10/month (after free credits) +- **Render**: $7/month (starter tier) +- **Upstash Redis**: Free tier or ~$0.20/100K commands + +## Next Steps + +- [ ] Add Redis integration for state persistence +- [ ] Implement proper error boundaries in frontend +- [ ] Add game lobby (multiple players can join) +- [ ] Implement spectator mode +- [ ] Add game history/replay +- [ ] Migrate to Option B (all-Vercel stateless) + +## Support + +For issues: +1. Check backend logs first +2. Check browser console +3. Verify environment variables +4. Test locally before blaming deployment diff --git a/examples/experimental/werewolves/QUICKSTART_CONDA.md b/examples/experimental/werewolves/QUICKSTART_CONDA.md new file mode 100644 index 000000000..bf2f21c7a --- /dev/null +++ b/examples/experimental/werewolves/QUICKSTART_CONDA.md @@ -0,0 +1,185 @@ +# Quick Start Guide (Conda Users) + +For users with the `sotopia-rl` conda environment already set up. + +## Prerequisites + +- ✅ Conda environment `sotopia-rl` activated +- ✅ Sotopia package installed (`pip install -e .` from repo root) +- ✅ OpenAI API key set (check with `echo $OPENAI_API_KEY`) +- ✅ Node.js 18+ installed + +## Step 1: Start Backend + +```bash +# Make sure you're in the werewolf directory +cd examples/experimental/werewolves + +# Activate conda environment +conda activate sotopia-rl + +# Go to backend folder +cd backend/ + +# Install backend-specific dependencies +pip install fastapi uvicorn[standard] websockets python-dotenv pydantic + +# Or install from requirements.txt +pip install -r requirements.txt + +# Copy environment template +cp .env.example .env + +# (Optional) Edit .env if needed - usually defaults are fine +# nano .env + +# Start the backend server +python main.py +``` + +You should see: +``` +INFO: Started server process +INFO: Waiting for application startup. +INFO: Application startup complete. +INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit) +``` + +**Keep this terminal running!** + +## Step 2: Start Frontend (New Terminal) + +```bash +# Navigate to frontend folder +cd examples/experimental/werewolves/frontend/ + +# Install Node dependencies (first time only) +npm install + +# Copy environment template +cp .env.example .env.local + +# The default values should work: +# NEXT_PUBLIC_BACKEND_URL=http://localhost:8000 +# NEXT_PUBLIC_WS_URL=ws://localhost:8000 + +# Start the frontend dev server +npm run dev +``` + +You should see: +``` + ▲ Next.js 14.1.0 + - Local: http://localhost:3000 + - Ready in 2.3s +``` + +## Step 3: Test the Game + +1. **Open your browser**: http://localhost:3000 + +2. **Create a game**: + - Enter your name (e.g., "Alice") + - Click "Create Game" + +3. **Check connections**: + - Open browser DevTools (F12) → Console tab + - You should see: `WebSocket connected` + - In the backend terminal, you should see: + ``` + INFO: Client connected to game {game_id} + ``` + +4. **Play the game**: + - Game events will appear in the event feed + - When it's your turn, the action panel will light up + - Select an action and submit + +## Troubleshooting + +### "ModuleNotFoundError: No module named 'sotopia'" + +```bash +# Make sure you're in the conda env +conda activate sotopia-rl + +# Install sotopia if not already installed +cd /path/to/sotopia/root +pip install -e . +``` + +### "WebSocket connection failed" + +- Check that backend is running on port 8000 +- Check browser console for errors +- Make sure `.env.local` has correct URLs + +### "CORS error" + +- Backend `.env` should have: + ``` + ALLOWED_ORIGINS=http://localhost:3000,https://*.vercel.app + ``` + +### Backend starts but game doesn't progress + +- Check that `OPENAI_API_KEY` is set: + ```bash + echo $OPENAI_API_KEY + ``` +- If not set, add to `.env`: + ```bash + OPENAI_API_KEY=sk-your-key-here + ``` + +## What's Happening Under the Hood + +1. **Backend (port 8000)**: + - Wraps your existing werewolf game logic + - Manages WebSocket connections + - Runs the game loop in the background + - Broadcasts events to connected clients + +2. **Frontend (port 3000)**: + - React UI for creating/joining games + - WebSocket client for real-time updates + - Action submission form + - Event feed display + +3. **Game Flow**: + ``` + User creates game + ↓ + Backend starts game session + ↓ + Game loop begins (async) + ↓ + Events broadcast via WebSocket + ↓ + Frontend receives & displays events + ↓ + User submits action + ↓ + Action queued for game loop + ↓ + Game loop processes action + ↓ + New events broadcast... + ``` + +## Next Steps + +- [ ] Test with multiple browser tabs (simulate multiple players) +- [ ] Check backend logs for game progression +- [ ] Try different actions (speak, vote, etc.) +- [ ] Once working locally, deploy to Railway + Vercel (see DEPLOYMENT.md) + +## Environment Summary + +**Your setup**: +- Python: conda environment `sotopia-rl` +- Node: system Node.js (v18+) +- Backend: FastAPI on port 8000 +- Frontend: Next.js on port 3000 + +**No venv needed** since you're using conda! 🎉 diff --git a/examples/experimental/werewolves/README_WEB.md b/examples/experimental/werewolves/README_WEB.md new file mode 100644 index 000000000..9c3de0d9b --- /dev/null +++ b/examples/experimental/werewolves/README_WEB.md @@ -0,0 +1,88 @@ +# Werewolf Web Game (Option A - Hybrid) + +Multi-player werewolf game with AI agents and real-time web interface. + +## Quick Start + +### Run Locally + +**Terminal 1 - Backend:** +```bash +# Activate your conda environment +conda activate sotopia-rl + +cd backend/ +pip install -r requirements.txt +cp .env.example .env +python main.py +``` + +**Terminal 2 - Frontend:** +```bash +cd frontend/ +npm install +cp .env.example .env.local +npm run dev +``` + +Open http://localhost:3000 + +## Project Structure + +``` +werewolves/ +├── backend/ # FastAPI backend +│ ├── main.py # API routes + WebSocket +│ ├── game_manager.py # Game loop wrapper +│ ├── ws_manager.py # WebSocket connections +│ ├── models.py # Pydantic models +│ └── requirements.txt +├── frontend/ # Next.js frontend +│ ├── app/ +│ │ ├── page.tsx # Home (create game) +│ │ └── game/[id]/page.tsx # Game UI +│ └── package.json +├── main_human.py # Original game logic +├── roster.json # Player configuration +├── role_actions.json # Game rules +└── DEPLOYMENT.md # Full deployment guide +``` + +## Key Features + +- ✅ Real-time WebSocket updates +- ✅ Beautiful UI with Tailwind CSS +- ✅ Single-player with AI agents +- ✅ Mobile-responsive design +- ✅ Production-ready (Railway + Vercel) + +## API Endpoints + +### HTTP +- `POST /api/game/create` - Create new game +- `GET /api/game/{id}/state` - Get game state +- `POST /api/game/{id}/action` - Submit player action + +### WebSocket +- `WS /ws/{id}` - Real-time game events + +## Development + +See [DEPLOYMENT.md](./DEPLOYMENT.md) for detailed setup and deployment instructions. + +## Tech Stack + +- **Frontend**: Next.js 14, TypeScript, Tailwind CSS +- **Backend**: FastAPI, Python 3.10+ +- **Game Logic**: Sotopia framework +- **LLM**: GPT-4o-mini (OpenAI) +- **Deployment**: Vercel (frontend) + Railway (backend) + +## Roadmap + +- [ ] Multi-player lobby +- [ ] Spectator mode +- [ ] Game replay/history +- [ ] Voice chat integration +- [ ] Mobile app (React Native) +- [ ] Option B migration (all-Vercel stateless) diff --git a/examples/experimental/werewolves/backend/.env.example b/examples/experimental/werewolves/backend/.env.example new file mode 100644 index 000000000..a63c69413 --- /dev/null +++ b/examples/experimental/werewolves/backend/.env.example @@ -0,0 +1,17 @@ +# Server Configuration +PORT=8000 + +# CORS - Comma-separated list of allowed origins +ALLOWED_ORIGINS=http://localhost:3000,https://*.vercel.app + +# OpenAI API Key +# If using conda env sotopia-rl, this should already be set as an environment variable +# If not, uncomment and add your key: +# OPENAI_API_KEY=sk-... + +# Redis for Sotopia (should already be configured in your conda env) +# If not set in environment, uncomment: +# REDIS_OM_URL=redis://:@localhost:6379 + +# Redis for game state (for future use with Option B) +# REDIS_URL=redis://localhost:6379 diff --git a/examples/experimental/werewolves/backend/.gitignore b/examples/experimental/werewolves/backend/.gitignore new file mode 100644 index 000000000..f9f593df4 --- /dev/null +++ b/examples/experimental/werewolves/backend/.gitignore @@ -0,0 +1,27 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +venv/ +env/ +ENV/ +.venv + +# Environment +.env +.env.local + +# IDEs +.vscode/ +.idea/ +*.swp +*.swo + +# Logs +*.log + +# OS +.DS_Store +Thumbs.db diff --git a/examples/experimental/werewolves/backend/Procfile b/examples/experimental/werewolves/backend/Procfile new file mode 100644 index 000000000..0e048402e --- /dev/null +++ b/examples/experimental/werewolves/backend/Procfile @@ -0,0 +1 @@ +web: uvicorn main:app --host 0.0.0.0 --port $PORT diff --git a/examples/experimental/werewolves/backend/game_manager.py b/examples/experimental/werewolves/backend/game_manager.py new file mode 100644 index 000000000..1800d28f7 --- /dev/null +++ b/examples/experimental/werewolves/backend/game_manager.py @@ -0,0 +1,388 @@ +"""Game manager that wraps the existing werewolf game logic.""" + +import asyncio +import sys +from pathlib import Path +from typing import Dict, Any, Optional, Callable +from datetime import datetime +import logging + +# Add parent directory to path to import existing game code +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from main_human import ( + prepare_scenario, + build_environment, + create_agents, + PlayerViewHumanAgent, +) +from sotopia.server import arun_one_episode + +logger = logging.getLogger(__name__) + + +class GameSession: + """Manages a single game session.""" + + def __init__( + self, + game_id: str, + player_name: str, + broadcast_callback: Callable[[dict], None], + ): + self.game_id = game_id + self.player_name = player_name + self.broadcast_callback = broadcast_callback + self.status = "initializing" # initializing, playing, finished + self.events = [] + self.player_role = None + self.all_player_names = [] + self.current_phase = None + self.input_queue = asyncio.Queue() + self.waiting_for_input = False + self.available_actions = [] + + async def broadcast_event(self, event: dict): + """Broadcast an event to all connected clients.""" + event["timestamp"] = datetime.now().isoformat() + event["game_id"] = self.game_id + self.events.append(event) + + # Call the callback (will be the WebSocket broadcast) + if self.broadcast_callback: + await self.broadcast_callback(event) + + async def start_game(self): + """Start the game loop in background.""" + try: + self.status = "playing" + await self.broadcast_event( + { + "type": "phase_change", + "content": "Game starting...", + "event_class": "phase", + } + ) + + # Run the game + await self._run_game() + + except Exception as e: + logger.error(f"Game {self.game_id} crashed: {e}", exc_info=True) + self.status = "error" + await self.broadcast_event( + { + "type": "error", + "content": f"Game error: {str(e)}", + "event_class": "phase", + } + ) + + async def _run_game(self): + """Run the game loop (adapted from main_human.py).""" + try: + # Prepare scenario + logger.info(f"Game {self.game_id}: Preparing scenario...") + env_profile, agent_profiles, role_assignments = prepare_scenario() + logger.info(f"Game {self.game_id}: Scenario prepared successfully") + except Exception as e: + logger.error( + f"Game {self.game_id}: Failed to prepare scenario: {e}", exc_info=True + ) + raise + + # Store player info + self.all_player_names = [ + f"{p.first_name} {p.last_name}" for p in agent_profiles + ] + self.player_role = list(role_assignments.values())[0] + + await self.broadcast_event( + { + "type": "game_info", + "content": f"You are playing as {self.player_name}. Role: {self.player_role}", + "event_class": "phase", + } + ) + + # Build environment + env_model = "gpt-5" + agent_model_list = ["gpt-5"] * len(agent_profiles) + + env = build_environment(env_profile, role_assignments, env_model) + agents = create_agents(agent_profiles, env_profile, agent_model_list) + + # Create custom PlayerView that broadcasts via WebSocket + class WebSocketPlayerView: + """Player view that broadcasts to WebSocket instead of writing HTML.""" + + def __init__(self, game_session): + self.game_session = game_session + self.events = [] + + async def add_event(self, event_type: str, content: str, speaker: str = ""): + event = { + "type": event_type, + "content": content, + "speaker": speaker, + "event_class": event_type, + } + self.events.append(event) + await self.game_session.broadcast_event(event) + + def enable_input(self, available_actions): + self.game_session.waiting_for_input = True + self.game_session.available_actions = available_actions + asyncio.create_task( + self.game_session.broadcast_event( + { + "type": "input_request", + "content": "Your turn!", + "available_actions": available_actions, + "event_class": "phase", + } + ) + ) + + async def wait_for_input(self): + """Wait for player input from the queue.""" + data = await self.game_session.input_queue.get() + self.game_session.waiting_for_input = False + self.game_session.available_actions = [] + return data + + player_view = WebSocketPlayerView(self) + + # Replace first agent with human player + class WebSocketHumanAgent(PlayerViewHumanAgent): + """Human agent that uses WebSocket player view with full observation parsing.""" + + async def aact(self, obs): + from sotopia.messages import AgentAction + + self.recv_message("Environment", obs) + + # Parse observation to extract player-visible information + if hasattr(obs, "to_natural_language"): + obs_text = obs.to_natural_language() + + # Parse line by line to properly categorize events + lines = obs_text.split("\n") + for line in lines: + line = line.strip() + if not line: + continue + + # Skip system prompts and metadata + if ( + line.startswith("Scenario:") + or " goal:" in line + or line.startswith("GAME RULES:") + or line.startswith("You are ") + or line.startswith("Primary directives:") + or line.startswith("Role guidance:") + or line.startswith("System constraints:") + ): + continue + + # Check for game over + if ( + "GAME OVER" in line.upper() + or ( + "Winner:" in line + and ("Werewolves" in line or "Villagers" in line) + ) + or ("[God] Werewolves win;" in line) + or ("[God] Villagers win;" in line) + ): + clean_line = line.replace("[God]", "").strip() + await player_view.add_event("phase", f"🎮 {clean_line}") + + # Check for death announcements + elif ( + "was found dead" in line + or " died" in line + or "was killed" in line + ) and (" said:" not in line and " says:" not in line): + clean_death = line.replace("[God]", "").strip() + if clean_death: # Only add if not empty + await player_view.add_event( + "death", f"💀 {clean_death}" + ) + + # Check for phase announcements + elif ( + "Night phase begins" in line + or "Phase: 'night'" in line.lower() + ): + await player_view.add_event( + "phase", "🌙 Night phase begins" + ) + + elif ( + "Day discussion starts" in line + or "Phase: 'day_discussion' begins" in line + ): + await player_view.add_event( + "phase", "☀️ Day breaks. Time to discuss!" + ) + + elif ( + "Voting phase" in line + or "Phase: 'voting' begins" in line + or "Phase 'day_vote' begins" in line + ): + await player_view.add_event("phase", "🗳️ Voting phase") + + elif ( + "twilight_execution" in line or "Execution results" in line + ): + await player_view.add_event("phase", "⚖️ Execution results") + + elif "Night returns" in line: + await player_view.add_event("phase", "🌙 Night returns") + + # Check for player speech + elif ( + " said:" in line or " says:" in line + ) and "[God]" not in line: + parts = line.split( + " said:" if " said:" in line else " says:" + ) + if len(parts) == 2: + speaker = parts[0].strip() + message = parts[1].strip().strip('"') + await player_view.add_event("speak", message, speaker) + + # Check for voting and eliminations + elif ( + "voted for" in line + or "was executed" in line + or "was eliminated" in line + or "Votes are tallied" in line + or "Majority condemns" in line + ): + clean_action = line.replace("[God]", "").strip() + if "was executed" in clean_action: + await player_view.add_event( + "death", f"⚖️ {clean_action}" + ) + elif "Majority condemns" in clean_action: + await player_view.add_event( + "action", f"🗳️ {clean_action}" + ) + elif clean_action: + await player_view.add_event("action", clean_action) + + # Get available actions + available_actions = getattr(obs, "available_actions", ["none"]) + + if available_actions != ["none"]: + player_view.enable_input(available_actions) + logger.info(f"Waiting for {self.agent_name}'s input...") + + # Wait for input from queue + input_data = await player_view.wait_for_input() + action_type = input_data.get("action_type", "none") + argument = input_data.get("argument", "") + + return AgentAction(action_type=action_type, argument=argument) + else: + return AgentAction(action_type="none", argument="") + + human_agent = WebSocketHumanAgent( + agent_profile=agent_profiles[0], + available_agent_names=self.all_player_names, + player_view=player_view, + ) + human_agent.goal = env_profile.agent_goals[0] + agents[0] = human_agent + + # Run the episode + await arun_one_episode( + env=env, + agent_list=agents, + omniscient=False, + script_like=False, + json_in_script=False, + tag=None, + push_to_db=False, + ) + + # Game finished - check for winner + self.status = "finished" + + # Check if there's a winner payload + if hasattr(env, "_winner_payload") and env._winner_payload: + winner = env._winner_payload.get("winner", "Unknown") + reason = env._winner_payload.get("message", "") + + logger.info(f"Game {self.game_id}: Winner={winner}, Reason={reason}") + + await self.broadcast_event( + { + "type": "game_over", + "content": f"🎮 Game Over! Winner: {winner}. {reason}", + "event_class": "phase", + } + ) + else: + logger.warning(f"Game {self.game_id}: No winner payload found") + await self.broadcast_event( + { + "type": "game_over", + "content": "Game finished!", + "event_class": "phase", + } + ) + + async def submit_action(self, action: Dict[str, Any]): + """Submit a player action.""" + if not self.waiting_for_input: + return {"error": "Not waiting for input"} + + # Put action in queue for game loop to process + await self.input_queue.put(action) + return {"status": "received"} + + +class GameManager: + """Manages all active game sessions.""" + + def __init__(self): + self.games: Dict[str, GameSession] = {} + + async def create_game( + self, + game_id: str, + player_name: str, + broadcast_callback: Callable[[dict], None], + ) -> GameSession: + """Create and start a new game.""" + game = GameSession(game_id, player_name, broadcast_callback) + self.games[game_id] = game + + # Start game in background + asyncio.create_task(game.start_game()) + + return game + + def get_game(self, game_id: str) -> Optional[GameSession]: + """Get a game session by ID.""" + return self.games.get(game_id) + + def get_game_state(self, game_id: str) -> Optional[Dict[str, Any]]: + """Get current game state.""" + game = self.get_game(game_id) + if not game: + return None + + return { + "game_id": game_id, + "status": game.status, + "phase": game.current_phase, + "players": game.all_player_names, + "events": game.events[-50:], # Last 50 events + "your_turn": game.waiting_for_input, + "available_actions": game.available_actions, + } diff --git a/examples/experimental/werewolves/backend/main.py b/examples/experimental/werewolves/backend/main.py new file mode 100644 index 000000000..a565be53c --- /dev/null +++ b/examples/experimental/werewolves/backend/main.py @@ -0,0 +1,187 @@ +"""FastAPI backend for Werewolf game.""" + +from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +import uvicorn +import os +import uuid +import logging +from dotenv import load_dotenv + +from models import ( + CreateGameRequest, + CreateGameResponse, + PlayerAction, + GameStateResponse, +) +from ws_manager import ConnectionManager +from game_manager import GameManager + +# Load environment variables +load_dotenv() + +# Configure logging +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + +# Initialize FastAPI app +app = FastAPI( + title="Werewolf Game API", + description="Backend for Duskmire Werewolves social deduction game", + version="1.0.0", +) + +# CORS configuration +ALLOWED_ORIGINS = os.getenv( + "ALLOWED_ORIGINS", "http://localhost:3000,https://*.vercel.app" +).split(",") + +app.add_middleware( + CORSMiddleware, + allow_origins=ALLOWED_ORIGINS, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# Initialize managers +ws_manager = ConnectionManager() +game_manager = GameManager() + + +@app.get("/") +async def root(): + """Health check endpoint.""" + return {"status": "ok", "service": "werewolf-game-api", "version": "1.0.0"} + + +@app.get("/health") +async def health(): + """Detailed health check.""" + return { + "status": "healthy", + "active_games": len(game_manager.games), + "websocket_connections": sum( + ws_manager.get_connection_count(gid) for gid in game_manager.games.keys() + ), + } + + +@app.post("/api/game/create", response_model=CreateGameResponse) +async def create_game(request: CreateGameRequest): + """Create a new game session.""" + try: + game_id = str(uuid.uuid4()) + logger.info(f"Creating game {game_id} for player {request.player_name}") + + # Define broadcast callback + async def broadcast(event: dict): + await ws_manager.broadcast(event, game_id) + + # Create game + game = await game_manager.create_game( + game_id=game_id, + player_name=request.player_name, + broadcast_callback=broadcast, + ) + + return CreateGameResponse( + game_id=game_id, + player_role=game.player_role or "Unknown", + all_players=game.all_player_names, + ) + + except Exception as e: + logger.error(f"Failed to create game: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/game/{game_id}/state", response_model=GameStateResponse) +async def get_game_state(game_id: str): + """Get current game state.""" + state = game_manager.get_game_state(game_id) + if not state: + raise HTTPException(status_code=404, detail="Game not found") + + return GameStateResponse(**state) + + +@app.post("/api/game/{game_id}/action") +async def submit_action(game_id: str, action: PlayerAction): + """Submit a player action.""" + game = game_manager.get_game(game_id) + if not game: + raise HTTPException(status_code=404, detail="Game not found") + + try: + result = await game.submit_action(action.model_dump()) + + # Broadcast action acknowledgment + await ws_manager.broadcast( + { + "type": "action_received", + "content": f"Action received: {action.action_type}", + "timestamp": action.timestamp, + }, + game_id, + ) + + return result + + except Exception as e: + logger.error(f"Failed to submit action: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) + + +@app.websocket("/ws/{game_id}") +async def websocket_endpoint(websocket: WebSocket, game_id: str): + """WebSocket endpoint for real-time game events.""" + await ws_manager.connect(websocket, game_id) + + try: + # Send initial connection confirmation + await ws_manager.send_personal_message( + { + "type": "connected", + "content": f"Connected to game {game_id}", + "game_id": game_id, + }, + websocket, + ) + + # Keep connection alive and listen for messages + while True: + # Client can send messages (e.g., ping/pong) + data = await websocket.receive_text() + logger.debug(f"Received WebSocket message: {data}") + + # Echo back (optional) + # await ws_manager.send_personal_message( + # {"type": "echo", "content": data}, + # websocket + # ) + + except WebSocketDisconnect: + ws_manager.disconnect(websocket, game_id) + logger.info(f"WebSocket disconnected from game {game_id}") + except Exception as e: + logger.error(f"WebSocket error: {e}", exc_info=True) + ws_manager.disconnect(websocket, game_id) + + +@app.exception_handler(Exception) +async def global_exception_handler(_request, exc): + """Global exception handler.""" + logger.error(f"Unhandled exception: {exc}", exc_info=True) + return JSONResponse( + status_code=500, content={"error": "Internal server error", "detail": str(exc)} + ) + + +if __name__ == "__main__": + port = int(os.getenv("PORT", 8000)) + # Run app object directly to avoid reload import issues + uvicorn.run(app, host="0.0.0.0", port=port, log_level="info") diff --git a/examples/experimental/werewolves/backend/models.py b/examples/experimental/werewolves/backend/models.py new file mode 100644 index 000000000..6aceb09f9 --- /dev/null +++ b/examples/experimental/werewolves/backend/models.py @@ -0,0 +1,61 @@ +"""Pydantic models for API requests and responses.""" + +from pydantic import BaseModel, Field +from typing import Optional, List, Dict, Any + + +class CreateGameRequest(BaseModel): + """Request to create a new game.""" + + player_name: str = Field(..., min_length=1, max_length=50) + + +class CreateGameResponse(BaseModel): + """Response with game ID.""" + + game_id: str + player_role: str + all_players: List[str] + + +class PlayerAction(BaseModel): + """Player action submitted during game.""" + + action_type: str = Field(..., description="Type of action: speak, action, none") + argument: str = Field( + default="", description="Action argument (e.g., message text, vote target)" + ) + timestamp: Optional[str] = None + + +class GameStateResponse(BaseModel): + """Current game state snapshot.""" + + game_id: str + status: str # "lobby", "playing", "finished" + phase: Optional[str] = None + players: List[Dict[str, Any]] + events: List[Dict[str, Any]] + your_turn: bool = False + available_actions: List[str] = [] + + +class GameEvent(BaseModel): + """Real-time game event pushed via WebSocket.""" + + type: str = Field( + ..., description="Event type: phase_change, speak, action, death, game_over" + ) + timestamp: str + speaker: Optional[str] = None + content: str + event_class: Optional[str] = Field( + None, description="CSS class: phase, speak, action, death" + ) + + +class ErrorResponse(BaseModel): + """Error response.""" + + error: str + detail: Optional[str] = None diff --git a/examples/experimental/werewolves/backend/railway.json b/examples/experimental/werewolves/backend/railway.json new file mode 100644 index 000000000..c75bb286c --- /dev/null +++ b/examples/experimental/werewolves/backend/railway.json @@ -0,0 +1,13 @@ +{ + "$schema": "https://railway.app/railway.schema.json", + "build": { + "builder": "NIXPACKS" + }, + "deploy": { + "startCommand": "uvicorn main:app --host 0.0.0.0 --port $PORT", + "healthcheckPath": "/health", + "healthcheckTimeout": 100, + "restartPolicyType": "ON_FAILURE", + "restartPolicyMaxRetries": 10 + } +} diff --git a/examples/experimental/werewolves/backend/requirements.txt b/examples/experimental/werewolves/backend/requirements.txt new file mode 100644 index 000000000..7b78afea1 --- /dev/null +++ b/examples/experimental/werewolves/backend/requirements.txt @@ -0,0 +1,6 @@ +fastapi==0.109.0 +uvicorn[standard]==0.27.0 +websockets==12.0 +redis==5.0.1 +python-dotenv==1.0.0 +pydantic==2.5.3 diff --git a/examples/experimental/werewolves/backend/ws_manager.py b/examples/experimental/werewolves/backend/ws_manager.py new file mode 100644 index 000000000..0c417e358 --- /dev/null +++ b/examples/experimental/werewolves/backend/ws_manager.py @@ -0,0 +1,67 @@ +"""WebSocket connection manager for real-time game events.""" + +from typing import Dict, Set +from fastapi import WebSocket +import json +import logging + +logger = logging.getLogger(__name__) + + +class ConnectionManager: + """Manages WebSocket connections for game rooms.""" + + def __init__(self): + # game_id -> set of WebSocket connections + self.active_connections: Dict[str, Set[WebSocket]] = {} + + async def connect(self, websocket: WebSocket, game_id: str): + """Accept a new WebSocket connection for a game.""" + await websocket.accept() + if game_id not in self.active_connections: + self.active_connections[game_id] = set() + self.active_connections[game_id].add(websocket) + logger.info( + f"Client connected to game {game_id}. Total: {len(self.active_connections[game_id])}" + ) + + def disconnect(self, websocket: WebSocket, game_id: str): + """Remove a WebSocket connection.""" + if game_id in self.active_connections: + self.active_connections[game_id].discard(websocket) + logger.info( + f"Client disconnected from game {game_id}. Remaining: {len(self.active_connections[game_id])}" + ) + + # Clean up empty game rooms + if not self.active_connections[game_id]: + del self.active_connections[game_id] + + async def send_personal_message(self, message: dict, websocket: WebSocket): + """Send a message to a specific client.""" + try: + await websocket.send_text(json.dumps(message)) + except Exception as e: + logger.error(f"Failed to send personal message: {e}") + + async def broadcast(self, message: dict, game_id: str): + """Broadcast a message to all clients in a game room.""" + if game_id not in self.active_connections: + logger.warning(f"No connections for game {game_id}") + return + + dead_connections = [] + for connection in list(self.active_connections[game_id]): + try: + await connection.send_text(json.dumps(message)) + except Exception as e: + logger.error(f"Failed to send to connection: {e}") + dead_connections.append(connection) + + # Clean up dead connections + for connection in dead_connections: + self.disconnect(connection, game_id) + + def get_connection_count(self, game_id: str) -> int: + """Get the number of active connections for a game.""" + return len(self.active_connections.get(game_id, set())) diff --git a/examples/experimental/werewolves/frontend/.env.example b/examples/experimental/werewolves/frontend/.env.example new file mode 100644 index 000000000..2a86785ef --- /dev/null +++ b/examples/experimental/werewolves/frontend/.env.example @@ -0,0 +1,9 @@ +# Backend API URL (HTTP) +NEXT_PUBLIC_BACKEND_URL=http://localhost:8000 + +# WebSocket URL +NEXT_PUBLIC_WS_URL=ws://localhost:8000 + +# For production: +# NEXT_PUBLIC_BACKEND_URL=https://your-backend.railway.app +# NEXT_PUBLIC_WS_URL=wss://your-backend.railway.app diff --git a/examples/experimental/werewolves/frontend/.gitignore b/examples/experimental/werewolves/frontend/.gitignore new file mode 100644 index 000000000..5f3ff3157 --- /dev/null +++ b/examples/experimental/werewolves/frontend/.gitignore @@ -0,0 +1,34 @@ +# Dependencies +/node_modules +/.pnp +.pnp.js + +# Testing +/coverage + +# Next.js +/.next/ +/out/ + +# Production +/build + +# Misc +.DS_Store +*.pem + +# Debug +npm-debug.log* +yarn-debug.log* +yarn-error.log* + +# Local env files +.env*.local +.env + +# Vercel +.vercel + +# Typescript +*.tsbuildinfo +next-env.d.ts diff --git a/examples/experimental/werewolves/frontend/app/game/[id]/page.tsx b/examples/experimental/werewolves/frontend/app/game/[id]/page.tsx new file mode 100644 index 000000000..6109806ab --- /dev/null +++ b/examples/experimental/werewolves/frontend/app/game/[id]/page.tsx @@ -0,0 +1,240 @@ +'use client'; + +import { useEffect, useRef, useState } from 'react'; +import { useParams } from 'next/navigation'; + +interface GameEvent { + type: string; + speaker?: string; + content: string; + timestamp: string; + event_class?: string; + available_actions?: string[]; +} + +export default function GamePage() { + const { id } = useParams<{ id: string }>(); + const [events, setEvents] = useState([]); + const [message, setMessage] = useState(''); + const [selectedAction, setSelectedAction] = useState(''); + const [availableActions, setAvailableActions] = useState([]); + const [yourTurn, setYourTurn] = useState(false); + const [connected, setConnected] = useState(false); + const wsRef = useRef(null); + const eventsEndRef = useRef(null); + + // Auto-scroll to bottom + useEffect(() => { + eventsEndRef.current?.scrollIntoView({ behavior: 'smooth' }); + }, [events]); + + // WebSocket connection + useEffect(() => { + const wsUrl = process.env.NEXT_PUBLIC_WS_URL || 'ws://localhost:8000'; + const ws = new WebSocket(`${wsUrl}/ws/${id}`); + wsRef.current = ws; + + ws.onopen = () => { + console.log('WebSocket connected'); + setConnected(true); + }; + + ws.onmessage = (event) => { + const data = JSON.parse(event.data) as GameEvent; + console.log('Received:', data); + + // Update events + setEvents((prev) => [...prev, data]); + + // Check if it's your turn + if (data.type === 'input_request') { + setYourTurn(true); + setAvailableActions(data.available_actions || []); + if (data.available_actions && data.available_actions.length > 0) { + setSelectedAction(data.available_actions[0]); + } + } + + // Reset turn when action is received + if (data.type === 'action_received') { + setYourTurn(false); + setAvailableActions([]); + setSelectedAction(''); + setMessage(''); + } + }; + + ws.onerror = (error) => { + console.error('WebSocket error:', error); + setConnected(false); + }; + + ws.onclose = () => { + console.log('WebSocket disconnected'); + setConnected(false); + }; + + return () => { + ws.close(); + }; + }, [id]); + + const submitAction = async () => { + if (!selectedAction || !yourTurn) return; + + try { + const backendUrl = process.env.NEXT_PUBLIC_BACKEND_URL || 'http://localhost:8000'; + const response = await fetch(`${backendUrl}/api/game/${id}/action`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + action_type: selectedAction, + argument: message, + timestamp: new Date().toISOString(), + }), + }); + + if (response.ok) { + console.log('Action submitted successfully'); + } + } catch (error) { + console.error('Failed to submit action:', error); + } + }; + + const getEventClassName = (eventClass?: string) => { + const baseClasses = 'border-l-4 p-4 mb-3 rounded-r'; + switch (eventClass) { + case 'phase': + return `${baseClasses} border-orange-500 bg-slate-700`; + case 'death': + return `${baseClasses} border-red-500 bg-red-950`; + case 'speak': + return `${baseClasses} border-blue-500 bg-slate-800`; + case 'action': + return `${baseClasses} border-yellow-500 bg-slate-800`; + default: + return `${baseClasses} border-gray-500 bg-slate-800`; + } + }; + + return ( +
+
+ {/* Header */} +
+

🌕 Duskmire Werewolves

+
+ Game ID: {id} + + {connected ? '● Connected' : '○ Disconnected'} + + {yourTurn && ( + + 🎮 YOUR TURN! + + )} +
+
+ +
+ {/* Events Feed */} +
+
+

Game Events

+
+ {events.length === 0 && ( +

Waiting for game to start...

+ )} + {events.map((event, i) => ( +
+
+ {new Date(event.timestamp).toLocaleTimeString()} +
+ {event.speaker && ( +
{event.speaker}
+ )} +
{event.content}
+
+ ))} +
+
+
+
+ + {/* Action Panel */} +
+
+

Your Action

+ + {!yourTurn ? ( +

Waiting for your turn...

+ ) : ( +
+ {/* Action Type Selector */} +
+ +
+ {availableActions.map((action) => ( + + ))} +
+
+ + {/* Message Input */} + {selectedAction && selectedAction !== 'none' && ( +
+ + + +
+
+
+ +
+
+

🌕 Duskmire Werewolves

+
+ You are: {self.player_name} | Role: {self.role} +
+
+
+
+
{datetime.now().strftime('%H:%M:%S')}
+
Game starting...
+
+
+
+ + + +""" + self.output_path.write_text(html) + + def add_event(self, event_type: str, content: str, speaker: str = "") -> None: + """Add a new event to the player view.""" + # De-duplicate identical events across live updates and post-game backfill + content_norm = content.strip() + speaker_norm = speaker.strip() + event_key = f"{event_type}|{speaker_norm}|{content_norm}" + if event_key in self._seen_event_keys: + return + self._seen_event_keys.add(event_key) + + timestamp = datetime.now().strftime("%H:%M:%S") + + speaker_html = f'
{speaker}
' if speaker else "" + + event_html = f""" +
+
{timestamp}
+ {speaker_html} +
{content}
+
""" + + self.events.append(event_html) + self._update_html() + + def enable_input(self, available_actions: List[str]) -> None: + """Enable the input controls with available actions.""" + self.waiting_for_input = True + self.available_actions = available_actions + # Clear any previous input + if self.input_file.exists(): + self.input_file.unlink() + self._update_html() + + def wait_for_input(self) -> dict[str, str]: + """Wait for player input from the HTML interface.""" + import time + + while not self.input_file.exists(): + time.sleep(0.5) + + try: + data = json.loads(self.input_file.read_text()) + self.waiting_for_input = False + self._update_html() + return cast(Dict[str, str], data) + except (json.JSONDecodeError, KeyError): + # If file is corrupt, wait and try again + time.sleep(0.5) + return self.wait_for_input() + + def _update_html(self) -> None: + """Update the HTML file with all events and dynamic input state.""" + events_html = "\n".join(self.events) + player_list = "\n".join([f"
  • {name}
    • " for name in self.all_player_names]) + + # Generate action buttons HTML based on available actions + action_buttons_html = "" + input_display = "none" + status_text = "Waiting for your turn..." + input_box_class = "" + + if self.waiting_for_input and self.available_actions: + input_display = "block" + status_text = "🎮 YOUR TURN! Select an action:" + input_box_class = "waiting" + action_buttons = [] + for action in self.available_actions: + action_label = action.replace("_", " ").title() + action_buttons.append( + f'' + ) + action_buttons_html = "\n".join(action_buttons) + + html = f""" + + + + Duskmire Werewolves - {self.player_name} + + + +
    +
    +

    👥 Players

    +
      +{player_list} +
    +
    + +
    +

    🎮 Your Action

    +
    {status_text}
    +
    +
    +{action_buttons_html} +
    + + +
    +
    +
    + +
    +
    +

    🌕 Duskmire Werewolves

    +
    + You are: {self.player_name} | Role: {self.role} +
    +
    +
    +{events_html} +
    +
    + + + +""" + self.output_path.write_text(html) + + +def _emit_entry_to_player_view( + entry: Dict[str, Any], player_view: "PlayerView" +) -> None: + """Mirror a single structured phase_log entry into the player view (non-vote items).""" + # Public announcements + for msg in entry.get("public", []): + text = msg.replace("[God]", "").strip() + if "Majority condemns" in text: + condemned = text.split("Majority condemns", 1)[1].strip() + if "." in condemned: + condemned = condemned.split(".", 1)[0].strip() + player_view.add_event( + "action", f"🗳️ Majority condemns {condemned}. Execution at twilight." + ) + elif "Votes are tallied" in text: + player_view.add_event("phase", "🗳️ Votes are tallied.") + elif "was executed" in text: + player_view.add_event("death", text) + elif ("was found dead" in text or " died" in text) and ( + " said:" not in text and " says:" not in text + ): + player_view.add_event("death", text) + elif "Phase 'day_vote' begins" in text or "Voting phase" in text: + player_view.add_event("phase", "🗳️ Voting phase. Time to make your choice.") + elif "Night returns" in text: + player_view.add_event("phase", "🌙 Night returns.") + + +async def stream_phase_log(env: Any, player_view: "PlayerView") -> None: + """Stream structured events; buffer during vote and flush in strict order.""" + from collections import Counter + + last_index = 0 + in_vote_phase = False + current_votes: Dict[str, str] = {} + vote_order: list[str] = [] + buffered_phase_after_tally: list[str] = [] + buffered_action_after_tally: list[str] = [] + buffered_deaths_after_tally: list[str] = [] + saw_tally_marker = False + + def record_vote(actor: str, target: str) -> None: + nonlocal current_votes, vote_order + if actor not in current_votes: + vote_order.append(actor) + current_votes[actor] = target + + async def flush_votes_and_buffered() -> None: + nonlocal in_vote_phase, current_votes, vote_order + nonlocal \ + buffered_phase_after_tally, \ + buffered_action_after_tally, \ + buffered_deaths_after_tally + nonlocal saw_tally_marker + + await asyncio.sleep(0.3) + player_view.add_event("phase", "🗳️ Votes are tallied.") + for actor in vote_order: + target = current_votes.get(actor, "none") + player_view.add_event("action", f"🗳️ {actor} voted for {target}") + if current_votes: + tally = Counter(current_votes.values()) + ordered_targets: list[str] = [] + for actor in vote_order: + t = current_votes.get(actor, "none") + if t not in ordered_targets: + ordered_targets.append(t) + parts = [f"{t}: {tally[t]}" for t in ordered_targets] + player_view.add_event("phase", "🗳️ Vote summary: " + ", ".join(parts)) + for text in buffered_action_after_tally: + player_view.add_event("action", text) + for text in buffered_phase_after_tally: + player_view.add_event("phase", text) + for text in buffered_deaths_after_tally: + player_view.add_event("death", text) + in_vote_phase = False + current_votes.clear() + vote_order.clear() + buffered_phase_after_tally.clear() + buffered_action_after_tally.clear() + buffered_deaths_after_tally.clear() + saw_tally_marker = False + + while True: + try: + phase_log = getattr(env, "phase_log", []) + while last_index < len(phase_log): + entry = phase_log[last_index] + last_index += 1 + public_msgs = [ + m.replace("[God]", "").strip() for m in entry.get("public", []) + ] + + if any("Phase 'day_vote' begins" in m for m in public_msgs): + in_vote_phase = True + current_votes.clear() + vote_order.clear() + buffered_phase_after_tally.clear() + buffered_action_after_tally.clear() + buffered_deaths_after_tally.clear() + saw_tally_marker = False + player_view.add_event( + "phase", "🗳️ Voting phase. Time to make your choice." + ) + continue + + if in_vote_phase: + for actor, action in entry.get("actions", {}).items(): + if action.get("action_type") == "action" and str( + action.get("argument", "") + ).startswith("vote"): + raw = str(action.get("argument", "")) + target = raw[5:].strip() or "none" + record_vote(actor, target) + for text in public_msgs: + if "Votes are tallied" in text: + saw_tally_marker = True + elif "Majority condemns" in text: + condemned = text.split("Majority condemns", 1)[1].strip() + if "." in condemned: + condemned = condemned.split(".", 1)[0].strip() + buffered_action_after_tally.append( + f"🗳️ Majority condemns {condemned}. Execution at twilight." + ) + elif ( + "twilight_execution" in text or "Execution results" in text + ): + buffered_phase_after_tally.append( + "⚖️ Twilight execution results." + ) + elif "was executed" in text: + buffered_deaths_after_tally.append(text) + elif "Night returns" in text: + buffered_phase_after_tally.append("🌙 Night returns.") + if saw_tally_marker: + await flush_votes_and_buffered() + continue + + _emit_entry_to_player_view(entry, player_view) + + if getattr(env, "_winner_payload", None) and last_index >= len(phase_log): + break + except Exception: + pass + await asyncio.sleep(0.5) + + +class PlayerViewHumanAgent(HumanAgent): + """HumanAgent that also writes to PlayerView HTML and reads input from it.""" + + def __init__( + self, + agent_name: str | None = None, + uuid_str: str | None = None, + agent_profile: Any | None = None, + available_agent_names: list[str] | None = None, + player_view: PlayerView | None = None, + ) -> None: + super().__init__( + agent_name=agent_name, + uuid_str=uuid_str, + agent_profile=agent_profile, + available_agent_names=available_agent_names, + ) + self.player_view = player_view + + async def aact(self, obs: Any) -> Any: + """Act and update player view with relevant information.""" + from sotopia.messages import AgentAction + + self.recv_message("Environment", obs) + + # Parse observation to extract player-visible information + if self.player_view and hasattr(obs, "to_natural_language"): + obs_text = obs.to_natural_language() + + # Parse line by line to avoid duplicates and properly categorize events + lines = obs_text.split("\n") + for line in lines: + line = line.strip() + if not line: + continue + + # Skip system prompts and metadata (scenario, goals, rules) + if ( + line.startswith("Scenario:") + or " goal:" in line + or line.startswith("GAME RULES:") + or line.startswith("You are ") + or line.startswith("Primary directives:") + or line.startswith("Role guidance:") + or line.startswith("System constraints:") + ): + continue + + # Check for game over / winner announcement + # Use more specific patterns to avoid matching game rules text + if ( + "GAME OVER" in line.upper() + or ( + "Winner:" in line + and ("Werewolves" in line or "Villagers" in line) + ) + or ("[God] Werewolves win;" in line) # Actual game result message + or ("[God] Villagers win;" in line) # Actual game result message + ): + clean_line = line.replace("[God]", "").strip() + # Normalize some common variants + if "Winner:" in clean_line: + self.player_view.add_event("phase", f"🎮 {clean_line}") + elif "Werewolves win;" in clean_line: + self.player_view.add_event( + "phase", "🎮 Game Over! Winner: Werewolves" + ) + elif "Villagers win;" in clean_line: + self.player_view.add_event( + "phase", "🎮 Game Over! Winner: Villagers" + ) + else: + self.player_view.add_event( + "phase", f"🎮 GAME OVER: {clean_line}" + ) + + # Check for voting results and eliminations + elif ( + "voted for" in line + or "has been eliminated" in line + or "was eliminated" in line + or "was executed" in line + or "Votes are tallied" in line + or "Majority condemns" in line + or "Execution will happen" in line + ): + # Normalize and add as action/phase/death events accordingly + clean_line = line.replace("[God]", "").strip() + + # Capture vote summaries from structured logs + if ( + clean_line.startswith("Action logged:") + and "-> action vote" in clean_line + ): + try: + payload = clean_line.split("Action logged:", 1)[1].strip() + actor, rest = payload.split("-> action vote", 1) + actor = actor.strip() + target = rest.strip() + if actor and target: + self.player_view.add_event( + "action", f"🗳️ {actor} voted for {target}" + ) + continue + except Exception: + pass + + # Majority condemnation announcement + if "Majority condemns" in clean_line: + condemned_name = clean_line.split("Majority condemns", 1)[ + 1 + ].strip() + if "." in condemned_name: + condemned_name = condemned_name.split(".", 1)[0].strip() + self.player_view.add_event( + "action", + f"🗳️ Majority condemns {condemned_name}. Execution at twilight.", + ) + continue + + # Execution result + if "was executed" in clean_line: + self.player_view.add_event("death", clean_line) + continue + + # Vote tally marker + if "Votes are tallied" in clean_line: + self.player_view.add_event("phase", "🗳️ Votes are tallied.") + continue + + # Fallback: add as action event + self.player_view.add_event("action", clean_line) + + # Check for death announcements + elif ("was found dead" in line or " died" in line) and ( + " said:" not in line and " says:" not in line + ): + self.player_view.add_event( + "death", line.replace("[God]", "").strip() + ) + + # Check for phase announcements + elif "Night phase begins" in line: + if not ( + self.player_view.events + and "Night phase begins" in self.player_view.events[-1] + ): + self.player_view.add_event( + "phase", "🌙 Night phase begins. Stay quiet..." + ) + + elif "Day discussion starts" in line or ( + "Phase: 'day_discussion' begins" in line + ): + if not ( + self.player_view.events + and "Day breaks" in self.player_view.events[-1] + ): + self.player_view.add_event( + "phase", "☀️ Day breaks. Time to discuss!" + ) + + elif ( + "Voting phase" in line + or ("Phase: 'voting' begins" in line) + or ("Phase 'day_vote' begins" in line) + ): + if not ( + self.player_view.events + and "Voting phase" in self.player_view.events[-1] + ): + self.player_view.add_event( + "phase", "🗳️ Voting phase. Time to make your choice." + ) + elif "twilight_execution" in line or "Execution results" in line: + self.player_view.add_event("phase", "⚖️ Twilight execution results.") + elif "Night returns" in line: + self.player_view.add_event("phase", "🌙 Night returns.") + + # Check for speech from players (avoid God messages and duplicates) + elif (" said:" in line or " says:" in line) and "[God]" not in line: + parts = line.split(" said:" if " said:" in line else " says:") + if len(parts) == 2: + speaker = parts[0].strip() + message = parts[1].strip().strip('"') + # Check if not duplicate + if not ( + self.player_view.events + and speaker in self.player_view.events[-1] + and message in self.player_view.events[-1] + ): + self.player_view.add_event("speak", message, speaker) + + # Get available actions from observation + available_actions = ( + obs.available_actions if hasattr(obs, "available_actions") else ["none"] + ) + + if available_actions != ["none"] and self.player_view: + # Enable HTML input and wait for player response + self.player_view.enable_input(available_actions) + print( + f"\n🎮 Waiting for {self.agent_name}'s input in the HTML interface..." + ) + + # Wait for input from HTML + input_data = self.player_view.wait_for_input() + action_type = input_data.get("action_type", "none") + argument = input_data.get("argument", "") + + # Enhanced voting support + if action_type == "action" and argument.lower().startswith("vote"): + name_part = argument[4:].strip() + if name_part and self.available_agent_names: + matched_name = self._find_matching_name(name_part) + if matched_name: + argument = f"vote {matched_name}" + print(f"✓ Voting for: {matched_name}") + + result = AgentAction(action_type=action_type, argument=argument) + else: + result = AgentAction(action_type="none", argument="") + + # Log player's own action to HTML + if self.player_view and result.action_type in ["speak", "action"]: + if result.action_type == "speak": + self.player_view.add_event( + "speak", result.argument, f"{self.agent_name} (You)" + ) + elif result.action_type == "action": + self.player_view.add_event( + "action", f"You performed action: {result.argument}" + ) + + return result + + +def load_json(path: Path) -> Dict[str, Any]: + return cast(Dict[str, Any], json.loads(path.read_text())) + + +def ensure_agent(player: Dict[str, Any]) -> AgentProfile: + try: + profile = AgentProfile.find( + AgentProfile.first_name == player["first_name"], + AgentProfile.last_name == player["last_name"], + ).all()[0] + return profile # type: ignore[return-value] + except IndexError: + profile = AgentProfile( + first_name=player["first_name"], + last_name=player["last_name"], + age=player.get("age", 30), + occupation="", + gender="", + gender_pronoun=player.get("pronouns", "they/them"), + public_info="", + personality_and_values="", + decision_making_style="", + secret=player.get("secret", ""), + ) + profile.save() + return profile + + +def build_agent_goal(player: Dict[str, Any], role_name: str, role_prompt: str) -> str: + # Build role description based on actual role + if role_name == "Villager": + role_desc = f"You are {player['first_name']} {player['last_name']}, a Villager." + else: + role_desc = f"You are {player['first_name']} {player['last_name']}. Your true role is {role_name}. Other players see you as a villager." + + return ( + f"{role_desc}\n" + f"Primary directives: {player['goal']}\n" + f"Role guidance: {role_prompt}\n" + f"System constraints: {COMMON_GUIDANCE}" + ) + + +def prepare_scenario() -> tuple[EnvironmentProfile, List[AgentProfile], Dict[str, str]]: + role_actions = load_json(ROLE_ACTIONS_PATH) + roster = load_json(ROSTER_PATH) + + agents: List[AgentProfile] = [] + agent_goals: List[str] = [] + role_assignments: Dict[str, str] = {} + + for player in roster["players"]: + profile = ensure_agent(player) + agents.append(profile) + full_name = f"{player['first_name']} {player['last_name']}" + role = player["role"] + role_prompt = role_actions["roles"][role]["goal_prompt"] + agent_goals.append(build_agent_goal(player, role, role_prompt)) + role_assignments[full_name] = role + + scenario_text = ( + roster["scenario"] + + " Werewolves must be eliminated before they achieve parity with villagers." + ) + + env_profile = EnvironmentProfile( + scenario=scenario_text, + agent_goals=agent_goals, + relationship=RelationshipType.acquaintance, + game_metadata={ + "mode": "social_game", + "rulebook_path": str(RULEBOOK_PATH), + "actions_path": str(ROLE_ACTIONS_PATH), + "role_assignments": role_assignments, + }, + tag="werewolves", + ) + env_profile.save() + return env_profile, agents, role_assignments + + +def build_environment( + env_profile: EnvironmentProfile, + role_assignments: Dict[str, str], + model_name: str, +) -> SocialGameEnv: + return SocialGameEnv( + env_profile=env_profile, + rulebook_path=str(RULEBOOK_PATH), + actions_path=str(ROLE_ACTIONS_PATH), + role_assignments=role_assignments, + model_name=model_name, + action_order="round-robin", + evaluators=[RuleBasedTerminatedEvaluator(max_turn_number=40, max_stale_turn=2)], + terminal_evaluators=[ + EpisodeLLMEvaluator( + model_name, + EvaluationForAgents[SotopiaDimensions], + ) + ], + ) + + +def create_agents( + agent_profiles: List[AgentProfile], + env_profile: EnvironmentProfile, + model_names: List[str], +) -> List[Union[LLMAgent, HumanAgent]]: + agents: List[Union[LLMAgent, HumanAgent]] = [] + for profile, model_name, goal in zip( + agent_profiles, + model_names, + env_profile.agent_goals, + strict=True, + ): + agent = LLMAgent(agent_profile=profile, model_name=model_name) + agent.goal = goal + agents.append(agent) + return agents + + +def summarize_phase_log(phase_log: List[Dict[str, Any]]) -> None: + if not phase_log: + print("\nNo structured events recorded.") + return + + print("\nTimeline by Phase") + print("=" * 60) + + last_label: str | None = None + for entry in phase_log: + phase_name = entry["phase"] + meta = entry.get("meta", {}) + group = meta.get("group") + cycle = meta.get("group_cycle") + stage = meta.get("group_stage") + title = phase_name.replace("_", " ").title() + if group: + group_label = group.replace("_", " ").title() + if cycle and stage: + label = f"{group_label} {cycle}.{stage} – {title}" + elif cycle: + label = f"{group_label} {cycle} – {title}" + else: + label = f"{group_label}: {title}" + else: + label = title + + if label != last_label: + print(f"\n[{label}]") + last_label = label + instructions = entry.get("instructions", []) + for info_line in instructions: + print(f" Info: {info_line}") + role_instr = entry.get("role_instructions", {}) + for role, lines in role_instr.items(): + for line in lines: + print(f" Role {role}: {line}") + + for msg in entry.get("public", []): + print(f" Public: {msg}") + for team, messages in entry.get("team", {}).items(): + for msg in messages: + print(f" Team ({team}) private: {msg}") + for agent, messages in entry.get("private", {}).items(): + for msg in messages: + print(f" Private to {agent}: {msg}") + for actor, action in entry.get("actions", {}).items(): + print( + f" Action logged: {actor} -> {action['action_type']} {action['argument']}" + ) + + +def print_roster(role_assignments: Dict[str, str]) -> None: + print("Participants & roles:") + for name, role in role_assignments.items(): + print(f" - {name}: {role}") + + +def start_http_server(port: int = 8000) -> Any: + """Start a simple HTTP server to handle player input and return the server instance.""" + from http.server import HTTPServer, SimpleHTTPRequestHandler + import threading + + class PlayerInputHandler(SimpleHTTPRequestHandler): + def do_POST(self) -> None: + if self.path == "/save_action": + content_length = int(self.headers["Content-Length"]) + post_data = self.rfile.read(content_length) + try: + data = json.loads(post_data.decode("utf-8")) + # Write to player_input.json in the same directory + input_file = BASE_DIR / "player_input.json" + input_file.write_text(json.dumps(data)) + self.send_response(200) + self.send_header("Content-type", "application/json") + self.send_header("Access-Control-Allow-Origin", "*") + self.end_headers() + self.wfile.write(b'{"status": "success"}') + except Exception as e: + self.send_response(500) + self.send_header("Content-type", "application/json") + self.end_headers() + self.wfile.write( + json.dumps({"status": "error", "message": str(e)}).encode() + ) + else: + self.send_response(404) + self.end_headers() + + def do_OPTIONS(self) -> None: + self.send_response(200) + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS") + self.send_header("Access-Control-Allow-Headers", "Content-Type") + self.end_headers() + + def log_message(self, _format: str, *_args: Any) -> None: + # Suppress log messages + pass + + os.chdir(BASE_DIR) + server = HTTPServer(("localhost", port), PlayerInputHandler) + thread = threading.Thread(target=server.serve_forever, daemon=False) + thread.start() + print(f"✓ HTTP server started on http://localhost:{port}") + return server + + +async def main() -> None: + # Start HTTP server for handling player input from HTML and keep it alive + # Start and keep the HTTP server alive for the browser UI + _ = start_http_server(8000) + + env_profile, agent_profiles, role_assignments = prepare_scenario() + env_model = "gpt-5" + agent_model_list = [ + "gpt-5", + "gpt-5", + "gpt-5", + "gpt-5", + "gpt-5", + "gpt-5", + ] + + env = build_environment(env_profile, role_assignments, env_model) + agents = create_agents(agent_profiles, env_profile, agent_model_list) + + # Get all agent names for voting support + all_agent_names = [f"{p.first_name} {p.last_name}" for p in agent_profiles] + + # Randomize which roster entry will be the human player each run + human_idx = random.randrange(len(agent_profiles)) + + # Get player info + player_name = ( + f"{agent_profiles[human_idx].first_name} {agent_profiles[human_idx].last_name}" + ) + player_role = role_assignments.get(player_name, "Villager") + + # Create PlayerView HTML for clean player-visible information + player_view = PlayerView( + PLAYER_VIEW_HTML, player_name, player_role, all_agent_names + ) + + # Replace the chosen agent with human player that writes to PlayerView + human_agent = PlayerViewHumanAgent( + agent_profile=agent_profiles[human_idx], + available_agent_names=all_agent_names, + player_view=player_view, + ) + human_agent.goal = env_profile.agent_goals[human_idx] + agents[human_idx] = human_agent + + print("\n🌕 Duskmire Werewolves — Interactive Social Game") + print("=" * 60) + print(f"You are playing as: {player_name}") + print(f"Your role: {player_role}") + print("=" * 60) + print( + "\n📖 PLAYER VIEW: Opens in your browser at http://localhost:8000/player_view.html" + ) + print(" This shows only what your character can see + interactive input.") + print("\n🔮 TERMINAL: Shows the full omniscient game state") + print(" (all agent actions and decisions for debugging)") + print("=" * 60) + + # Auto-open the HTML file in browser via HTTP server + try: + webbrowser.open("http://localhost:8000/player_view.html") + print("✓ Player view opened in your browser") + except Exception as e: + print(f"⚠ Could not auto-open browser: {e}") + print(" Please manually open: http://localhost:8000/player_view.html") + + print("=" * 60) + print("Other participants:") + for name in role_assignments.keys(): + if name != player_name: + print(f" - {name}") + print("=" * 60) + + # Start phase-log streaming task so UI reflects structured events in real time + stream_task = asyncio.create_task(stream_phase_log(env, player_view)) + + await arun_one_episode( + env=env, + agent_list=agents, + omniscient=False, + script_like=False, + json_in_script=False, + tag=None, + push_to_db=False, + ) + + # Ensure all streamed entries are flushed before proceeding + try: + await asyncio.wait_for(stream_task, timeout=2.0) + except Exception: + pass + + summarize_phase_log(env.phase_log) + + # Post-game: walk phase_log to ensure final votes/execution/winner are reflected in the UI + try: + # Add any vote actions logged + for entry in env.phase_log: + actions = entry.get("actions", {}) + for actor, action in actions.items(): + if action.get("action_type") == "action" and action.get( + "argument", "" + ).startswith("vote"): + target = action.get("argument", "")[5:].strip() + if target: + player_view.add_event("action", f"🗳️ {actor} voted for {target}") + # Execution results or majority announcements may be in public messages + for msg in entry.get("public", []): + if "Majority condemns" in msg: + condemned = msg.split("Majority condemns", 1)[1].strip() + if "." in condemned: + condemned = condemned.split(".", 1)[0].strip() + player_view.add_event( + "action", + f"🗳️ Majority condemns {condemned}. Execution at twilight.", + ) + if "was executed" in msg: + player_view.add_event("death", msg.replace("[God]", "").strip()) + if "Votes are tallied" in msg: + player_view.add_event("phase", "🗳️ Votes are tallied.") + + if env._winner_payload: # noqa: SLF001 (internal inspection for demo) + winner = env._winner_payload.get("winner", "Unknown") + reason = env._winner_payload.get("message", "") + print("\n" + "=" * 60) + print("GAME RESULT") + print("=" * 60) + print(f"Winner: {winner}") + print(f"Reason: {reason}") + player_view.add_event( + "phase", f"🎮 Game Over! Winner: {winner}. Reason: {reason}" + ) + except Exception as e: + print(f"Post-game UI update failed: {e}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/experimental/werewolves/player_input.json b/examples/experimental/werewolves/player_input.json new file mode 100644 index 000000000..13f71105e --- /dev/null +++ b/examples/experimental/werewolves/player_input.json @@ -0,0 +1 @@ +{"action_type": "none", "argument": "", "timestamp": "2025-10-30T03:32:40.695Z"} diff --git a/examples/experimental/werewolves/player_view.html b/examples/experimental/werewolves/player_view.html new file mode 100644 index 000000000..479541890 --- /dev/null +++ b/examples/experimental/werewolves/player_view.html @@ -0,0 +1,484 @@ + + + + + Duskmire Werewolves - Aurora Harper + + + +
    +
    +

    👥 Players

    +
      +
    • Aurora Harper
      • +
    • Bram Nightshade
      • +
    • Celeste Moonseer
      • +
    • Dorian Blackwood
      • +
    • Elise Farrow
      • +
    • Finn Alder
      • +
    +
    + +
    +

    🎮 Your Action

    +
    Waiting for your turn...
    + +
    +
    + +
    +
    +

    🌕 Duskmire Werewolves

    +
    + You are: Aurora Harper | + Role: Villager +
    +
    +
    +
    +
    23:31:46
    + +
    Elise Farrow was found dead at dawn.
    +
    + +
    +
    23:31:46
    + +
    ☀️ Day breaks. Time to discuss!
    +
    + +
    +
    23:32:15
    +
    Bram Nightshade
    +
    + Well, this is unfortunate. Elise was a valuable member of our group, + and her loss leaves us in a precarious position. I think we should + carefully consider who we suspect might be behind this. I honestly + didn't notice anything suspicious from Elise last night. She seemed + pretty much like herself, but perhaps one of us had a keen eye on + her? What do you all think? +
    +
    + +
    +
    23:32:17
    +
    Celeste Moonseer
    +
    + It's truly tragic to lose Elise so early in the game. We must be + vigilant. I sensed tension among some players last night, but it's + hard to identify without more clues. Let's pool our observations and + see if we can deduce who the werewolves might be. Any thoughts on + who looked suspicious to you all? +
    +
    + +
    +
    23:32:19
    +
    Dorian Blackwood
    +
    + I agree with both Bram and Celeste. We need to be cautious and + analyze the behavior of each player. It's always puzzling when + someone like Elise is taken out early. I didn't notice anything + suspicious about her either, but I did find that Finn seemed a bit + uneasy during our previous discussions. Perhaps we should keep an + eye on him as we continue the game. What does everyone else think? +
    +
    + +
    +
    23:32:21
    + +
    🗳️ Voting phase. Time to make your choice.
    +
    + +
    +
    23:32:21
    +
    Finn Alder
    +
    + I agree with Dorian about needing more observation. While my unease + may have been noted, I think we should really focus on those who + might benefit from Elise's death. We need to consider the dynamics + of our interactions last night. Has anyone noticed any patterns in + the way people reacted or spoke? It might give us clues about + possible werewolves. +
    +
    + +
    +
    23:32:26
    + +
    + 🗳️ Majority condemns Dorian Blackwood. Execution at twilight. +
    +
    + +
    +
    23:32:26
    + +
    🗳️ Votes are tallied.
    +
    + +
    +
    23:32:26
    + +
    ⚖️ Twilight execution results.
    +
    + +
    +
    23:32:26
    + +
    + Dorian Blackwood was executed. They belonged to team Werewolves. +
    +
    + +
    +
    23:32:26
    + +
    🌙 Night returns.
    +
    + +
    +
    23:32:26
    + +
    🗳️ Dorian Blackwood voted for Finn
    +
    + +
    +
    23:32:26
    + +
    🗳️ Vote summary: Finn: 1
    +
    + +
    +
    23:32:27
    + +
    Finn Alder was found dead at dawn.
    +
    + +
    +
    23:32:34
    +
    Bram Nightshade
    +
    + It’s shocking to face another loss. Finn's death is quite + distressing. I saw potential in both Dorian and Finn; they seemed so + invested in the game. Now, without them, we must question who + benefits from their departures. I can't help but feel that we have + to scrutinize those who remained. Perhaps a focus on how they + interacted could shed some light on our next move. What are your + thoughts, Aurora and Celeste? +
    +
    + +
    +
    23:32:37
    +
    Celeste Moonseer
    +
    + I share your shock, Bram. Losing Finn so soon is disheartening. We + must look closely at the dynamics among the remaining players. I + noticed a certain eagerness from both you and Aurora defending + Dorian—could that have been protective instincts or mere gameplay + strategy? Let's not let emotion cloud our judgement; we need to + ensure we identify any hidden threats before they eliminate us. What + do you think? +
    +
    + +
    +
    23:32:45
    + +
    Aurora Harper was found dead at dawn.
    +
    + +
    +
    23:32:45
    + +
    + 🎮 Game Over! Winner: Werewolves. Reason: [God] Werewolves win; they + now match the village. +
    +
    +
    +
    + + + + diff --git a/examples/experimental/werewolves/role_actions.json b/examples/experimental/werewolves/role_actions.json new file mode 100644 index 000000000..2c88851a9 --- /dev/null +++ b/examples/experimental/werewolves/role_actions.json @@ -0,0 +1,75 @@ +{ + "roles": { + "Villager": { + "name": "Villager", + "team": "Villagers", + "description": "Ordinary resident with no night power but vital voice in daytime debates.", + "goal_prompt": "Keep sharp notes about player behaviour and vote to execute suspected werewolves each day.", + "default_actions": ["speak"], + "phase_actions": { + "night_werewolves": ["none"], + "night_seer": ["none"], + "night_witch": ["none"], + "dawn_report": ["none"], + "day_discussion": ["speak"], + "day_vote": ["action"], + "twilight_execution": ["none"] + }, + "initial_state": {} + }, + "Seer": { + "name": "Seer", + "team": "Villagers", + "description": "Mystic who divines alignments during the night.", + "goal_prompt": "Inspect one player each night using an action like 'inspect NAME'; leak findings strategically without exposing yourself too early.", + "default_actions": ["speak"], + "phase_actions": { + "night_werewolves": ["none"], + "night_seer": ["action"], + "night_witch": ["none"], + "dawn_report": ["none"], + "day_discussion": ["speak"], + "day_vote": ["action"], + "twilight_execution": ["none"] + }, + "initial_state": {} + }, + "Witch": { + "name": "Witch", + "team": "Villagers", + "description": "Potion expert who may save one player per game and poison one player per game during the night.", + "goal_prompt": "During your witch phase, decide whether to 'save NAME', 'poison NAME', or pass. Use your limited potions wisely to keep villagers alive and remove wolves when confident.", + "default_actions": ["speak"], + "phase_actions": { + "night_werewolves": ["none"], + "night_seer": ["none"], + "night_witch": ["action"], + "dawn_report": ["none"], + "day_discussion": ["speak"], + "day_vote": ["action"], + "twilight_execution": ["none"] + }, + "initial_state": { + "save_available": true, + "poison_available": true + } + }, + "Werewolf": { + "name": "Werewolf", + "team": "Werewolves", + "description": "Predator hiding among villagers, coordinating nightly kills and sowing mistrust by day.", + "goal_prompt": "Confer quietly with fellow wolves at night. Use actions like 'kill NAME' to propose a victim. During the day, blend in while pushing misdirection.", + "default_actions": ["speak"], + "phase_actions": { + "night_werewolves": ["speak", "action"], + "night_seer": ["none"], + "night_witch": ["none"], + "dawn_report": ["none"], + "day_discussion": ["speak"], + "day_vote": ["action"], + "twilight_execution": ["none"] + }, + "initial_state": {} + } + } +} diff --git a/examples/experimental/werewolves/roster.json b/examples/experimental/werewolves/roster.json new file mode 100644 index 000000000..6abb27ea3 --- /dev/null +++ b/examples/experimental/werewolves/roster.json @@ -0,0 +1,61 @@ +{ + "scenario": "This game is called Werewolves (also known as Mafia), which is a social deduction game. GAME STRUCTURE: This game contains six players: two villagers, two werewolves, one seer, and one witch. Each cycle consists of Night phases followed by Day phases. WIN CONDITIONS: Villagers win by eliminating all Werewolves. Werewolves win when they equal or outnumber the Villagers. NIGHT PHASES: (1) Werewolves wake and privately choose one Villager to kill using 'kill NAME'. (2) The Seer wakes and inspects one player using 'inspect NAME' to learn if they are a Werewolf or Villager. (3) The Witch wakes and may use their one-time save potion with 'save NAME' (if someone died tonight) or their one-time poison potion with 'poison NAME' to kill someone. DAY PHASES: (1) Dawn: All players learn who died during the night (if any). (2) Discussion: All living players openly discuss and debate who might be a Werewolf. Use 'speak' to share your thoughts and suspicions. (3) Voting: Each player votes to eliminate one suspected Werewolf using 'vote NAME'. The player with the most votes is executed. (4) Twilight: The execution result is announced and night returns. COMMUNICATION RULES: All day discussions are public. Dead players cannot speak or act. STRATEGY: Villagers must use logic, observation, and deduction to identify Werewolves through their behavior and voting patterns. Werewolves must deceive others, blend in as Villagers, and create confusion. Special roles (Seer, Witch) should use their powers strategically without revealing themselves too early. Trust is earned through consistent behavior and alignment of words with actions.", + "players": [ + { + "first_name": "Aurora", + "last_name": "Harper", + "role": "Villager", + "public_role": "Villager", + "age": 54, + "pronouns": "she/her", + "goal": "Keep discussion orderly and support executions only when evidence feels solid." + }, + { + "first_name": "Bram", + "last_name": "Nightshade", + "role": "Werewolf", + "public_role": "Villager", + "age": 33, + "pronouns": "he/him", + "goal": "Blend in with confident speech while steering suspicion toward ordinary villagers.", + "secret": "You are a werewolf working with Dorian. Coordinate night kills." + }, + { + "first_name": "Celeste", + "last_name": "Moonseer", + "role": "Seer", + "public_role": "Villager", + "age": 29, + "pronouns": "she/her", + "goal": "Inspect one player per night and nudge the village toward the wolves." + }, + { + "first_name": "Dorian", + "last_name": "Blackwood", + "role": "Werewolf", + "public_role": "Villager", + "age": 38, + "pronouns": "he/him", + "goal": "Support Bram's stories and pressure outspoken villagers into missteps.", + "secret": "You are a werewolf working with Bram. Coordinate night kills." + }, + { + "first_name": "Elise", + "last_name": "Farrow", + "role": "Witch", + "public_role": "Villager", + "age": 41, + "pronouns": "she/her", + "goal": "Use your save and poison sparingly; protect confirmed villagers and strike when a wolf is exposed." + }, + { + "first_name": "Finn", + "last_name": "Alder", + "role": "Villager", + "public_role": "Villager", + "age": 36, + "pronouns": "he/him", + "goal": "Track inconsistencies and rally the town to execute the most suspicious player each day." + } + ] +} diff --git a/sotopia/agents/llm_agent.py b/sotopia/agents/llm_agent.py index 497954d7b..41315307d 100644 --- a/sotopia/agents/llm_agent.py +++ b/sotopia/agents/llm_agent.py @@ -140,6 +140,7 @@ def __init__( agent_name: str | None = None, uuid_str: str | None = None, agent_profile: AgentProfile | None = None, + available_agent_names: list[str] | None = None, ) -> None: super().__init__( agent_name=agent_name, @@ -147,6 +148,7 @@ def __init__( agent_profile=agent_profile, ) self.model_name = "human" + self.available_agent_names = available_agent_names or [] @property def goal(self) -> str: @@ -171,14 +173,53 @@ def act(self, obs: Observation) -> AgentAction: return AgentAction(action_type=action_type, argument=argument) + def _find_matching_name(self, user_input: str) -> str | None: + """Find a matching agent name from partial input (case-insensitive).""" + user_input_lower = user_input.lower().strip() + + # Try exact match first + for name in self.available_agent_names: + if name.lower() == user_input_lower: + return name + + # Try partial match on first name or last name + matches = [] + for name in self.available_agent_names: + name_parts = name.lower().split() + if any(part.startswith(user_input_lower) for part in name_parts): + matches.append(name) + + if len(matches) == 1: + return matches[0] + elif len(matches) > 1: + print("Ambiguous name. Did you mean one of these?") + for i, match in enumerate(matches): + print(f" {i}: {match}") + return None + + return None + async def aact(self, obs: Observation) -> AgentAction: self.recv_message("Environment", obs) - print("Available actions:") - for i, action in enumerate(obs.available_actions): - print(f"{i}: {action}") + # Only print if last_turn changed (avoid duplicate prompts) + should_prompt = True + if len(self.inbox) >= 2: + last_obs = self.inbox[-2][1] + if ( + isinstance(last_obs, Observation) + and last_obs.last_turn == obs.last_turn + ): + should_prompt = False if obs.available_actions != ["none"]: + if should_prompt: + print("\n" + "=" * 60) + print("YOUR TURN") + print("=" * 60) + print("Available actions:") + for i, action in enumerate(obs.available_actions): + print(f"{i}: {action}") action_type_number = await ainput( "Action type (Please only input the number): " ) @@ -194,8 +235,38 @@ async def aact(self, obs: Observation) -> AgentAction: action_type = obs.available_actions[action_type_number] else: action_type = "none" - if action_type in ["speak", "non-verbal communication"]: + + if action_type in ["speak", "non-verbal communication", "action"]: argument = await ainput("Argument: ") + + # Enhanced voting support + if action_type == "action" and argument.lower().startswith("vote"): + # Extract the name part after "vote" + name_part = argument[4:].strip() + if name_part and self.available_agent_names: + matched_name = self._find_matching_name(name_part) + if matched_name: + argument = f"vote {matched_name}" + print(f"✓ Voting for: {matched_name}") + else: + print(f"⚠ Could not find player matching '{name_part}'") + print("Available players:") + for i, name in enumerate(self.available_agent_names): + print(f" {i}: {name}") + retry = await ainput( + "Enter player number or name to vote for: " + ) + try: + idx = int(retry) + if 0 <= idx < len(self.available_agent_names): + matched_name = self.available_agent_names[idx] + argument = f"vote {matched_name}" + print(f"✓ Voting for: {matched_name}") + except ValueError: + matched_name = self._find_matching_name(retry) + if matched_name: + argument = f"vote {matched_name}" + print(f"✓ Voting for: {matched_name}") else: argument = "" diff --git a/sotopia/database/persistent_profile.py b/sotopia/database/persistent_profile.py index ab2f78fcb..23e1871e9 100644 --- a/sotopia/database/persistent_profile.py +++ b/sotopia/database/persistent_profile.py @@ -88,6 +88,10 @@ class BaseEnvironmentProfile(BaseModel): agent_constraint: list[list[str]] | None = Field( default_factory=lambda: None, ) + game_metadata: dict[str, Any] | None = Field( + default_factory=lambda: None, + description="Optional metadata for structured social games (rulebooks, config paths, etc.).", + ) tag: str = Field( index=True, default_factory=lambda: "", diff --git a/sotopia/envs/__init__.py b/sotopia/envs/__init__.py index fa56ad757..30b8d8a37 100644 --- a/sotopia/envs/__init__.py +++ b/sotopia/envs/__init__.py @@ -1,3 +1,4 @@ from .parallel import ParallelSotopiaEnv +from .social_game import SocialGameEnv -__all__ = ["ParallelSotopiaEnv"] +__all__ = ["ParallelSotopiaEnv", "SocialGameEnv"] diff --git a/sotopia/envs/social_game.py b/sotopia/envs/social_game.py new file mode 100644 index 000000000..a0b2b29bc --- /dev/null +++ b/sotopia/envs/social_game.py @@ -0,0 +1,935 @@ +"""Social game environment that reads its rulebook and action space from JSON.""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterable, Optional, Sequence, cast + +from pydantic import BaseModel, Field, RootModel, ValidationError + +from sotopia.envs.parallel import ParallelSotopiaEnv, render_text_for_agent +from sotopia.agents.llm_agent import Agents +from sotopia.database import EnvironmentProfile +from sotopia.messages import AgentAction, Observation, SimpleMessage + + +class RoleActionConfig(BaseModel): + """Declared abilities and messaging semantics for a specific role.""" + + name: str + team: str + description: str = "" + goal_prompt: str = "" + default_actions: list[str] = Field(default_factory=lambda: ["speak", "action"]) + phase_actions: dict[str, list[str]] = Field(default_factory=dict) + initial_state: dict[str, Any] = Field(default_factory=dict) + allow_team_private_speech: bool = False + allow_role_private_speech: bool = False + + +class RoleActionLibrary(RootModel[dict[str, RoleActionConfig]]): + """Pydantic wrapper for mapping roles to role metadata.""" + + def team_for_role(self, role: str) -> str: + return self.root[role].team + + +class PhaseResolution(BaseModel): + operation: str = Field( + default="noop", + description="Name of the builtin resolution handler to invoke at phase end.", + ) + state_key: str | None = None + visibility: str = Field( + default="public", + description="Default visibility for resolution feedback.", + ) + + +class PhaseDefinition(BaseModel): + name: str + kind: str = Field( + default="discussion", + description="Macro describing how the phase behaves (discussion, team_target, vote, single_target, announcement).", + ) + group: str | None = Field( + default=None, + description="Optional label used to cluster phases into higher-level cycles (e.g., 'night', 'day').", + ) + turn_mode: str = Field( + default="round-robin", + description="round-robin => sequential actors, simultaneous => everyone at once, single => one actor only.", + ) + acting_roles: list[str] | None = None + acting_teams: list[str] | None = None + max_cycles: int = Field( + default=1, + description="Number of complete round-robin passes required before the phase advances.", + ) + max_turns: int | None = Field( + default=None, + description="Optional cap on total turns inside the phase (overrides max_cycles when smaller).", + ) + speech_visibility: str = Field( + default="public", + description="Where speech is visible ('public', 'team', 'private', 'hidden').", + ) + action_visibility: str = Field( + default="public", + description="Where action outcomes are visible ('public', 'team', 'private', 'hidden').", + ) + instructions: list[str] = Field( + default_factory=list, + description="General prompts injected into agent observations for this phase.", + ) + role_instructions: dict[str, list[str]] = Field( + default_factory=dict, + description="Optional role-specific prompts keyed by role name.", + ) + resolution: PhaseResolution | None = None + entry_messages: list[str] = Field(default_factory=list) + exit_messages: list[str] = Field(default_factory=list) + description: str = "" + + +class EndConditionDefinition(BaseModel): + operation: str + team: str | None = None + other_team: str | None = None + winner: str | None = None + message: str | None = None + + +class RulebookConfig(BaseModel): + initial_phase: str + phases: list[PhaseDefinition] + phase_transitions: dict[str, str] + end_conditions: list[EndConditionDefinition] = Field(default_factory=list) + max_cycles: int | None = Field( + default=None, + description="Optional safety bound on day/night cycles to prevent infinite games.", + ) + + +@dataclass +class AgentState: + name: str + role: str + team: str + alive: bool = True + attributes: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class PhaseEvents: + public: list[str] = field(default_factory=list) + team: dict[str, list[str]] = field(default_factory=dict) + private: dict[str, list[str]] = field(default_factory=dict) + system: list[str] = field(default_factory=list) + + def extend(self, other: "PhaseEvents") -> None: + self.public.extend(other.public) + for team, messages in other.team.items(): + self.team.setdefault(team, []).extend(messages) + for agent, messages in other.private.items(): + self.private.setdefault(agent, []).extend(messages) + self.system.extend(other.system) + + @classmethod + def phase_entry(cls, phase_name: str, messages: list[str]) -> "PhaseEvents": + events = cls() + for msg in messages: + events.public.append(f"[God] Phase '{phase_name}' begins: {msg}") + if not messages: + events.public.append(f"[God] Phase '{phase_name}' begins.") + return events + + +class GameRulebook: + """Runtime state machine that enforces the JSON described social game.""" + + def __init__(self, rules: RulebookConfig, roles: RoleActionLibrary) -> None: + self.rules = rules + self.roles = roles + self.phase_lookup = {phase.name: phase for phase in rules.phases} + self.agent_states: dict[str, AgentState] = {} + self.agent_name_lookup: dict[str, str] = {} + self.current_phase: str = rules.initial_phase + self.phase_cycle_progress: int = 0 + self.turns_in_phase: int = 0 + self.current_actor_index: int = 0 + self.state_flags: dict[str, Any] = {} + self.group_cycle: dict[str, int] = {} + self.group_stage: dict[str, int] = {} + self.current_phase_meta: dict[str, Any] = {} + self.pending_events: PhaseEvents = PhaseEvents() + + # ------------------------------------------------------------------ + # Initialisation + # ------------------------------------------------------------------ + def assign_agents( + self, + agents: Sequence[str], + role_assignments: dict[str, str], + ) -> None: + self.agent_states = {} + self.agent_name_lookup = {} + for name in agents: + role = role_assignments[name] + role_cfg = self.roles.root.get(role) + if role_cfg is None: + raise ValueError(f"Unknown role '{role}' for agent '{name}'") + attrs = dict(role_cfg.initial_state) + state = AgentState( + name=name, + role=role, + team=role_cfg.team, + alive=True, + attributes=attrs, + ) + self.agent_states[name] = state + self.agent_name_lookup[name.lower()] = name + self.agent_name_lookup[name.split()[0].lower()] = name + + self.current_phase = self.rules.initial_phase + self.phase_cycle_progress = 0 + self.turns_in_phase = 0 + self.current_actor_index = 0 + self.state_flags = { + "day_execution": None, + "night_target": None, + "witch_saved": None, + "witch_poisoned": None, + "seer_result": "", + } + self.group_cycle.clear() + self.group_stage.clear() + self.current_phase_meta = {} + self._register_phase_entry(self.current_phase) + entry_phase = self.phase_lookup[self.current_phase] + self.pending_events = PhaseEvents.phase_entry( + self.current_phase, entry_phase.entry_messages + ) + + # ------------------------------------------------------------------ + # Accessors used by the environment + # ------------------------------------------------------------------ + def alive_agents(self) -> list[str]: + return [name for name, state in self.agent_states.items() if state.alive] + + def active_agents_for_phase(self) -> list[str]: + phase = self.phase_lookup[self.current_phase] + eligible = self._eligible_candidates(phase) + if not eligible: + return [] + if phase.turn_mode == "round-robin": + idx = self.current_actor_index + if idx >= len(eligible): + idx = len(eligible) - 1 + if idx < 0: + idx = 0 + return [eligible[idx]] + return eligible + + def available_actions(self, agent_name: str) -> list[str]: + agent_state = self.agent_states[agent_name] + if not agent_state.alive: + return ["none"] + role_cfg = self.roles.root[agent_state.role] + actions = role_cfg.phase_actions.get( + self.current_phase, role_cfg.default_actions + ) + if "none" not in actions: + actions = list(actions) + ["none"] + return actions + + def collect_pending_events(self) -> PhaseEvents: + events = self.pending_events + self.pending_events = PhaseEvents() + return events + + # ------------------------------------------------------------------ + # Core update logic + # ------------------------------------------------------------------ + def process_actions( + self, actions: dict[str, AgentAction] + ) -> tuple[PhaseEvents, bool, Optional[dict[str, str]]]: + phase = self.phase_lookup[self.current_phase] + acting_agents = self.active_agents_for_phase() + events = PhaseEvents() + + if phase.kind == "announcement": + events.extend(self._resolve_phase(phase, {})) + winner = self._check_end_conditions() + self._schedule_phase_exit(phase) + return events, True, winner + + if not acting_agents: + events.extend(self._resolve_phase(phase, {})) + winner = self._check_end_conditions() + self._schedule_phase_exit(phase) + return events, True, winner + + relevant = { + name: actions.get(name, AgentAction(action_type="none", argument="")) + for name in acting_agents + } + + if phase.turn_mode == "round-robin": + actor = acting_agents[0] + events.extend(self._record_speech(actor, relevant[actor], phase)) + events.extend(self._resolve_phase(phase, {actor: relevant[actor]})) + self._advance_round_robin(phase) + advance = self._should_advance(phase) + else: + for actor, action in relevant.items(): + events.extend(self._record_speech(actor, action, phase)) + events.extend(self._resolve_phase(phase, relevant)) + advance = True + + winner = self._check_end_conditions() + if winner: + self._schedule_phase_exit(phase) + return events, True, winner + + if advance: + self._schedule_phase_exit(phase) + return events, advance, winner + + def start_next_phase(self) -> PhaseEvents: + next_phase = self.rules.phase_transitions.get(self.current_phase) + if next_phase is None: + raise ValueError( + f"No transition defined after phase '{self.current_phase}'" + ) + self.current_phase = next_phase + self.phase_cycle_progress = 0 + self.turns_in_phase = 0 + self.current_actor_index = 0 + self._register_phase_entry(next_phase) + phase_def = self.phase_lookup[next_phase] + entry = PhaseEvents.phase_entry(next_phase, phase_def.entry_messages) + return entry + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + def _phase_group(self, phase: PhaseDefinition) -> str: + if phase.group: + return phase.group + return phase.name + + def _register_phase_entry(self, phase_name: str) -> None: + phase = self.phase_lookup[phase_name] + group = self._phase_group(phase) + previous_group = ( + self.current_phase_meta.get("group") if self.current_phase_meta else None + ) + cycle = self.group_cycle.get(group, 0) + stage = self.group_stage.get(group, 0) + if previous_group != group: + cycle += 1 + stage = 1 + else: + stage += 1 + self.group_cycle[group] = cycle + self.group_stage[group] = stage + self.current_phase_meta = { + "phase": phase_name, + "group": group, + "group_cycle": cycle, + "group_stage": stage, + "display_name": phase.name.replace("_", " ").title(), + } + + def current_phase_metadata(self) -> dict[str, Any]: + return dict(self.current_phase_meta) if self.current_phase_meta else {} + + def _eligible_candidates(self, phase: PhaseDefinition) -> list[str]: + names = [name for name, state in self.agent_states.items() if state.alive] + if phase.acting_roles: + names = [ + name + for name in names + if self.agent_states[name].role in phase.acting_roles + ] + if phase.acting_teams: + names = [ + name + for name in names + if self.agent_states[name].team in phase.acting_teams + ] + return names + + def _record_speech( + self, actor: str, action: AgentAction, phase: PhaseDefinition + ) -> PhaseEvents: + events = PhaseEvents() + if action.action_type not in {"speak", "non-verbal communication"}: + return events + utterance = action.argument.strip() + if not utterance: + return events + line = f'{actor} said: "{utterance}"' + if phase.speech_visibility == "team": + team = self.agent_states[actor].team + events.team.setdefault(team, []).append(line) + elif phase.speech_visibility == "private": + events.private.setdefault(actor, []).append(line) + elif phase.speech_visibility == "hidden": + pass + else: + events.public.append(line) + return events + + def _resolve_phase( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + ) -> PhaseEvents: + if phase.resolution is None: + return PhaseEvents() + handler = getattr(self, f"_resolve_{phase.resolution.operation}", None) + if handler is None: + raise ValueError( + f"Unsupported resolution operation '{phase.resolution.operation}'" + ) + return cast(PhaseEvents, handler(phase, actions, phase.resolution)) + + def _resolve_noop( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + resolution: PhaseResolution, + ) -> PhaseEvents: + return PhaseEvents() + + def _resolve_store_target( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + resolution: PhaseResolution, + ) -> PhaseEvents: + events = PhaseEvents() + target = self._extract_target(actions.values()) + if target: + self.state_flags[resolution.state_key or "night_target"] = target + teams = phase.acting_teams or [self.agent_states[a].team for a in actions] + for team in teams: + events.team.setdefault(team, []).append( + f"[God] Target locked: {target}." + ) + return events + + def _resolve_seer_inspect( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + resolution: PhaseResolution, + ) -> PhaseEvents: + events = PhaseEvents() + if not actions: + return events + actor, action = next(iter(actions.items())) + target = self._extract_target([action]) + if not target: + events.private.setdefault(actor, []).append( + "[God] Vision failed: unable to interpret your target." + ) + return events + team = self.agent_states[target].team + message = f"[God] Vision reveals {target} serves team {team}." + events.private.setdefault(actor, []).append(message) + self.state_flags["seer_result"] = message + return events + + def _resolve_witch_phase( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + resolution: PhaseResolution, + ) -> PhaseEvents: + events = PhaseEvents() + if not actions: + return events + actor, action = next(iter(actions.items())) + state = self.agent_states[actor] + text = action.argument.lower() + if "save" in text and state.attributes.get("save_available", True): + target = self._extract_target([action]) or self.state_flags.get( + "night_target" + ) + if target: + self.state_flags["witch_saved"] = target + state.attributes["save_available"] = False + events.private.setdefault(actor, []).append( + f"[God] You secretly saved {target} tonight." + ) + if "poison" in text and state.attributes.get("poison_available", True): + target = self._extract_target([action]) + if target: + self.state_flags["witch_poisoned"] = target + state.attributes["poison_available"] = False + events.private.setdefault(actor, []).append( + f"[God] You poisoned {target}." + ) + if not text.strip() or "pass" in text: + events.private.setdefault(actor, []).append( + "[God] You chose to remain idle." + ) + return events + + def _resolve_resolve_night( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + resolution: PhaseResolution, + ) -> PhaseEvents: + events = PhaseEvents() + saved = self.state_flags.get("witch_saved") + target = self.state_flags.get("night_target") + poison = self.state_flags.get("witch_poisoned") + casualties: list[str] = [] + if target and target != saved: + casualties.append(target) + if poison and poison not in casualties: + casualties.append(poison) + if not casualties: + events.public.append("[God] Dawn breaks peacefully. No one died.") + for victim in casualties: + if victim in self.agent_states and self.agent_states[victim].alive: + self.agent_states[victim].alive = False + events.public.append(f"[God] {victim} was found dead at dawn.") + self.state_flags["night_target"] = None + self.state_flags["witch_saved"] = None + self.state_flags["witch_poisoned"] = None + self.state_flags["seer_result"] = "" + return events + + def _resolve_vote( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + resolution: PhaseResolution, + ) -> PhaseEvents: + events = PhaseEvents() + tally: dict[str, int] = {} + for action in actions.values(): + target = self._extract_target([action]) + if target: + tally[target] = tally.get(target, 0) + 1 + elif "none" in action.argument.lower(): + tally.setdefault("none", 0) + tally["none"] += 1 + if not tally: + events.public.append("[God] No valid votes were cast.") + self.state_flags["day_execution"] = None + return events + winner, votes = max(tally.items(), key=lambda kv: kv[1]) + if winner == "none": + events.public.append("[God] The town decided to stay their hand.") + self.state_flags["day_execution"] = None + return events + if list(tally.values()).count(votes) > 1: + events.public.append("[God] The vote is tied. No execution today.") + self.state_flags["day_execution"] = None + return events + self.state_flags["day_execution"] = winner + events.public.append( + f"[God] Majority condemns {winner}. Execution will happen at twilight." + ) + return events + + def _resolve_post_vote_cleanup( + self, + phase: PhaseDefinition, + actions: dict[str, AgentAction], + resolution: PhaseResolution, + ) -> PhaseEvents: + events = PhaseEvents() + target = self.state_flags.get("day_execution") + if target and target in self.agent_states and self.agent_states[target].alive: + self.agent_states[target].alive = False + team = self.agent_states[target].team + events.public.append( + f"[God] {target} was executed. They belonged to team {team}." + ) + self.state_flags["day_execution"] = None + return events + + def _extract_target(self, actions: Iterable[AgentAction]) -> str | None: + for action in actions: + corpus = f"{action.action_type} {action.argument}".lower() + for name in self.agent_states: + if name.lower() in corpus: + return name + for name in self.agent_states: + first = name.split()[0].lower() + if first in corpus: + return name + return None + + def _advance_round_robin(self, phase: PhaseDefinition) -> None: + base = self._eligible_candidates(phase) + self.turns_in_phase += 1 + if not base: + self.current_actor_index = 0 + return + self.current_actor_index += 1 + if self.current_actor_index >= len(base): + self.phase_cycle_progress += 1 + self.current_actor_index = 0 + + def _should_advance(self, phase: PhaseDefinition) -> bool: + if phase.turn_mode != "round-robin": + return True + base = self._eligible_candidates(phase) + if not base: + return True + if phase.max_turns is not None and self.turns_in_phase >= phase.max_turns: + return True + if self.phase_cycle_progress >= phase.max_cycles: + return True + return False + + def _schedule_phase_exit(self, phase: PhaseDefinition) -> None: + exit_events = PhaseEvents() + for msg in phase.exit_messages: + exit_events.public.append(f"[God] {msg}") + self.pending_events.extend(exit_events) + + def _check_end_conditions(self) -> Optional[dict[str, str]]: + for cond in self.rules.end_conditions: + if cond.operation == "team_eliminated" and cond.team: + alive = sum( + 1 + for state in self.agent_states.values() + if state.alive and state.team == cond.team + ) + if alive == 0: + message = ( + cond.message or f"[God] Team {cond.team} has been eliminated." + ) + return { + "winner": cond.winner or cond.other_team or cond.team, + "message": message, + } + if cond.operation == "parity" and cond.team and cond.other_team: + team_count = sum( + 1 + for state in self.agent_states.values() + if state.alive and state.team == cond.team + ) + other_count = sum( + 1 + for state in self.agent_states.values() + if state.alive and state.team == cond.other_team + ) + if team_count >= other_count: + message = cond.message or ( + f"[God] Parity reached: {cond.team} now matches or exceeds {cond.other_team}." + ) + return { + "winner": cond.winner or cond.team, + "message": message, + } + return None + + +class SocialGameEnv(ParallelSotopiaEnv): + """Environment subclass that enforces multi-phase social game mechanics.""" + + def __init__( + self, + env_profile: EnvironmentProfile, + *, + rulebook_path: str, + actions_path: str, + role_assignments: dict[str, str], + **kwargs: Any, + ) -> None: + super().__init__(env_profile=env_profile, **kwargs) + self._rulebook_path = Path(rulebook_path) + self._actions_path = Path(actions_path) + self._role_assignments = role_assignments + self.game_rulebook: GameRulebook | None = None + self._last_events: PhaseEvents = PhaseEvents() + self._winner_payload: dict[str, str] | None = None + self.phase_log: list[dict[str, Any]] = [] + + # ------------------------------------------------------------------ + # Config loading helpers + # ------------------------------------------------------------------ + def _load_configs(self) -> tuple[RulebookConfig, RoleActionLibrary]: + try: + rules = RulebookConfig.model_validate_json(self._rulebook_path.read_text()) + except ValidationError as exc: + raise ValueError(f"Invalid rulebook config: {exc}") from exc + actions_raw = json.loads(self._actions_path.read_text()) + try: + roles = RoleActionLibrary.model_validate(actions_raw["roles"]) + except (KeyError, ValidationError) as exc: + raise ValueError(f"Invalid action-space config: {exc}") from exc + return rules, roles + + # ------------------------------------------------------------------ + # Overrides + # ------------------------------------------------------------------ + def reset( + self, + seed: int | None = None, + options: dict[str, str] | None = None, + agents: Agents | None = None, + omniscient: bool = False, + lite: bool = False, + ) -> dict[str, Observation]: + base_obs = super().reset( + seed=seed, + options=options, + agents=agents, + omniscient=omniscient, + lite=lite, + ) + rules, role_actions = self._load_configs() + self.game_rulebook = GameRulebook(rules, role_actions) + self.game_rulebook.assign_agents(self.agents, self._role_assignments) + self.phase_log = [] + self._apply_action_mask() + self._last_events = self.game_rulebook.collect_pending_events() + self._winner_payload = None + self._record_phase_history( + phase_name=self.game_rulebook.current_phase, + actions={}, + events=self._last_events, + ) + return self._augment_observations(base_obs, append_to_existing=True) + + def _phase_prompt_lines( + self, + *, + agent_name: str, + phase: PhaseDefinition, + acting: bool, + available: list[str], + ) -> list[str]: + assert self.game_rulebook is not None + meta = self.game_rulebook.current_phase_metadata() + group = meta.get("group") + cycle = meta.get("group_cycle") + stage = meta.get("group_stage") + title = phase.name.replace("_", " ").title() + if group: + group_label = group.replace("_", " ").title() + if cycle and stage: + label = f"{group_label} {cycle}.{stage} – {title}" + elif cycle: + label = f"{group_label} {cycle} – {title}" + else: + label = f"{group_label}: {title}" + else: + label = title + lines = [f"[God] Phase: {label}"] + if acting: + lines.append("[God] It is your turn to act in this phase.") + else: + lines.append("[God] You are observing while others act.") + lines.append(f"[God] Available actions right now: {', '.join(available)}") + lines.extend(f"[God] {text}" for text in phase.instructions) + role = self.game_rulebook.agent_states[agent_name].role + for text in phase.role_instructions.get(role, []): + lines.append(f"[God] {text}") + return lines + + def _record_phase_history( + self, + *, + phase_name: str, + actions: dict[str, AgentAction], + events: PhaseEvents, + ) -> None: + if self.game_rulebook is None: + return + if not (events.public or events.team or events.private): + if any(a.action_type != "none" for a in actions.values()): + pass + else: + return + action_summary = { + agent: {"action_type": action.action_type, "argument": action.argument} + for agent, action in actions.items() + if action.action_type != "none" + } + phase_def = ( + self.game_rulebook.phase_lookup.get(phase_name) + if self.game_rulebook + else None + ) + snapshot = { + "phase": phase_name, + "turn": self.turn_number, + "public": list(events.public), + "team": {team: list(msgs) for team, msgs in events.team.items()}, + "private": {agent: list(msgs) for agent, msgs in events.private.items()}, + "actions": action_summary, + "meta": self.game_rulebook.current_phase_metadata() + if self.game_rulebook + else {}, + "instructions": phase_def.instructions if phase_def else [], + "role_instructions": phase_def.role_instructions if phase_def else {}, + } + self.phase_log.append(snapshot) + + def _augment_observations( + self, + baseline: dict[str, Observation], + *, + append_to_existing: bool, + ) -> dict[str, Observation]: + assert self.game_rulebook is not None + acting = set(self.game_rulebook.active_agents_for_phase()) + events = self._last_events + phase_name = self.game_rulebook.current_phase + phase_def = self.game_rulebook.phase_lookup[phase_name] + new_obs: dict[str, Observation] = {} + for idx, agent_name in enumerate(self.agents): + current = baseline[agent_name] + available = ( + self.game_rulebook.available_actions(agent_name) + if agent_name in acting + else ["none"] + ) + phase_lines = self._phase_prompt_lines( + agent_name=agent_name, + phase=phase_def, + acting=agent_name in acting, + available=available, + ) + messages: list[str] = [] + messages.extend(events.public) + team = self.game_rulebook.agent_states[agent_name].team + messages.extend(events.team.get(team, [])) + messages.extend(events.private.get(agent_name, [])) + if not messages: + messages.append("[God] Await instructions from the host.") + segments: list[str] = [] + if append_to_existing: + prefix = current.last_turn.strip() + if prefix: + segments.append(prefix) + segments.extend(phase_lines) + segments.extend(messages) + combined = "\n".join(segment for segment in segments if segment) + new_obs[agent_name] = Observation( + last_turn=render_text_for_agent(combined, agent_id=idx), + turn_number=current.turn_number, + available_actions=available, + ) + return new_obs + + def _create_blank_observations(self) -> dict[str, Observation]: + assert self.game_rulebook is not None + acting = set(self.game_rulebook.active_agents_for_phase()) + blank: dict[str, Observation] = {} + for agent_name in self.agents: + available = ( + self.game_rulebook.available_actions(agent_name) + if agent_name in acting + else ["none"] + ) + blank[agent_name] = Observation( + last_turn="", + turn_number=self.turn_number, + available_actions=available, + ) + return blank + + def _apply_action_mask(self) -> None: + assert self.game_rulebook is not None + acting = set(self.game_rulebook.active_agents_for_phase()) + self.action_mask = [ + agent in acting and self.game_rulebook.agent_states[agent].alive + for agent in self.agents + ] + + async def astep( + self, actions: dict[str, AgentAction] | dict[str, dict[str, int | str]] + ) -> tuple[ + dict[str, Observation], + dict[str, float], + dict[str, bool], + dict[str, bool], + dict[str, dict[Any, Any]], + ]: + assert self.game_rulebook is not None + self._apply_action_mask() + self.turn_number += 1 + prepared = self._coerce_actions(actions) + self.recv_message( + "Environment", SimpleMessage(message=f"Turn #{self.turn_number}") + ) + for agent, action in prepared.items(): + self.recv_message(agent, action) + phase_name = self.game_rulebook.current_phase + events, advance, winner = self.game_rulebook.process_actions(prepared) + exit_events = self.game_rulebook.collect_pending_events() + events.extend(exit_events) + self._record_phase_history( + phase_name=phase_name, + actions=prepared, + events=events, + ) + self._last_events = events + if advance: + next_events = self.game_rulebook.start_next_phase() + self._record_phase_history( + phase_name=self.game_rulebook.current_phase, + actions={}, + events=next_events, + ) + self._last_events.extend(next_events) + self._apply_action_mask() + baseline = self._create_blank_observations() + observations = self._augment_observations(baseline, append_to_existing=False) + rewards = {agent_name: 0.0 for agent_name in self.agents} + terminated = {agent_name: bool(winner) for agent_name in self.agents} + truncations = {agent_name: False for agent_name in self.agents} + info = { + agent_name: { + "comments": winner["message"] if winner else "", + "complete_rating": 0, + } + for agent_name in self.agents + } + if winner: + self._winner_payload = winner + return observations, rewards, terminated, truncations, info + + def _coerce_actions( + self, actions: dict[str, AgentAction] | dict[str, dict[str, int | str]] + ) -> dict[str, AgentAction]: + prepared: dict[str, AgentAction] = {} + for agent, raw in actions.items(): + if isinstance(raw, AgentAction): + prepared[agent] = raw + else: + idx = int(raw.get("action_type", 0)) + action_type = self.available_action_types[idx] + prepared[agent] = AgentAction( + action_type=action_type, + argument=str(raw.get("argument", "")), + ) + return prepared + + def step( + self, actions: dict[str, AgentAction] | dict[str, dict[str, int | str]] + ) -> tuple[ + dict[str, Observation], + dict[str, float], + dict[str, bool], + dict[str, bool], + dict[str, dict[Any, Any]], + ]: + return asyncio.run(self.astep(actions)) diff --git a/sotopia/generation_utils/generate.py b/sotopia/generation_utils/generate.py index 5abce7ace..d33ef31b2 100644 --- a/sotopia/generation_utils/generate.py +++ b/sotopia/generation_utils/generate.py @@ -224,6 +224,13 @@ async def _call_with_retry(completion_kwargs: dict[str, Any]) -> Any: ), "response_schema is not supported in this model" messages = [{"role": "user", "content": template}] + # Log the full prompt being sent to LLM + log.info(f"\n{'='*80}") + log.info(f"LLM PROMPT (model={model_name}):") + log.info(f"{'='*80}") + log.info(f"{template}") + log.info(f"{'='*80}\n") + assert isinstance( output_parser, PydanticOutputParser ), "structured output only supported in PydanticOutputParser" diff --git a/sotopia/samplers/uniform_sampler.py b/sotopia/samplers/uniform_sampler.py index d519eee0d..38d1585f7 100644 --- a/sotopia/samplers/uniform_sampler.py +++ b/sotopia/samplers/uniform_sampler.py @@ -65,8 +65,21 @@ def sample( env_profile = random.choice(self.env_candidates) if isinstance(env_profile, str): env_profile = EnvironmentProfile.get(env_profile) - logger.info("Creating ParallelSotopiaEnv with %s agents", n_agent) - env = ParallelSotopiaEnv(env_profile=env_profile, **env_params) + logger.info("Creating environment with %s agents", n_agent) + game_meta = getattr(env_profile, "game_metadata", None) or {} + env: ParallelSotopiaEnv + if game_meta.get("mode") == "social_game": + from sotopia.envs import SocialGameEnv + + env = SocialGameEnv( + env_profile=env_profile, + rulebook_path=game_meta["rulebook_path"], + actions_path=game_meta["actions_path"], + role_assignments=game_meta["role_assignments"], + **env_params, + ) + else: + env = ParallelSotopiaEnv(env_profile=env_profile, **env_params) agent_profile_candidates = self.agent_candidates if len(agent_profile_candidates) == n_agent: diff --git a/sotopia/server.py b/sotopia/server.py index 844e98794..bb603e72a 100644 --- a/sotopia/server.py +++ b/sotopia/server.py @@ -312,11 +312,6 @@ def get_agent_class( ), ], } - # agents_model_dict = { - # agent_name: model_name - # for agent_name, model_name in model_dict.items() - # if agent_name.startswith("agent") - # } agent_keys = sorted(k for k in model_dict if re.fullmatch(r"agent\d+", k)) agent_models = [model_dict[k] for k in agent_keys] @@ -382,7 +377,6 @@ async def arun_one_script( env.reset(agents=agents, omniscient=omniscient) agent_names = [agent.agent_name for agent in agent_list] - # assert len(agent_names) == 2, f"only support 2 agents, current: {agent_names}" assert ( agents and len(agents) >= 2 ), "At least two agents required, current: {agent_names}" diff --git a/uv.lock b/uv.lock index a0d147290..2e38d6dd9 100644 --- a/uv.lock +++ b/uv.lock @@ -3163,7 +3163,7 @@ requires-dist = [ { name = "google-generativeai", marker = "extra == 'google-generativeai'" }, { name = "groq", marker = "extra == 'groq'" }, { name = "hiredis", specifier = ">=3.0.0" }, - { name = "json-repair", specifier = ">=0.35.0,<0.45.0" }, + { name = "json-repair", specifier = ">=0.35.0,<0.49.0" }, { name = "litellm", specifier = ">=1.65.0" }, { name = "lxml", specifier = ">=4.9.3,<6.0.0" }, { name = "modal", marker = "extra == 'api'" },