217 lines
9.0 KiB
Python
217 lines
9.0 KiB
Python
```python
|
||
import re
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
import textarena as ta
|
||
|
||
|
||
class StellarTriadEnv(ta.Env):
|
||
"""
|
||
Stellar Triad — Turn-based two-player abstract alignment game-themed around cosmic architecture.
|
||
Implementation strictly follows the Stage 1 design document.
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""Initialize regex pattern and constants."""
|
||
super().__init__()
|
||
self.action_pattern = re.compile(r'^\[Channel:(?:[1-3])-(?:[1-3])\]$')
|
||
self.max_turns = 9 # Maximum number of turns (one per matrix cell)
|
||
self.state: Optional[ta.TwoPlayerState] = None
|
||
|
||
def _extract_answer_content(self, action: str) -> str:
|
||
"""
|
||
Extract the content within \boxed{{ ... }}. Return empty string if not found.
|
||
"""
|
||
match = re.search(r'\\boxed\{\{(.*?)\}\}', action, re.DOTALL)
|
||
if match:
|
||
return match.group(1).strip()
|
||
# Also support single brace variant if it may occur
|
||
match = re.search(r'\\boxed\{(.*?)\}', action, re.DOTALL)
|
||
if match:
|
||
return match.group(1).strip()
|
||
return ""
|
||
|
||
def reset(self, num_players: int, seed: Optional[int] = None):
|
||
"""
|
||
Resets the environment to an initial state.
|
||
|
||
Args:
|
||
num_players: Number of players (must be 2).
|
||
seed: Optional seed for deterministic starting player order.
|
||
|
||
Returns:
|
||
None
|
||
"""
|
||
if num_players != 2:
|
||
raise ValueError("Stellar Triad requires exactly 2 players.")
|
||
|
||
# Initialize state
|
||
self.state = ta.TwoPlayerState(num_players=num_players, seed=seed, max_turns=self.max_turns)
|
||
|
||
# Determine starting player based on seed parity (even seed => ArchitectA starts)
|
||
if seed is None:
|
||
seed = 0
|
||
starting_player = 0 if seed % 2 == 0 else 1
|
||
|
||
matrix_state = [[None for _ in range(3)] for _ in range(3)]
|
||
player_symbols = {"ArchitectA": "A", "ArchitectB": "B"}
|
||
active_player = "ArchitectA" if starting_player == 0 else "ArchitectB"
|
||
|
||
game_state = {
|
||
"matrix_state": matrix_state,
|
||
"player_symbols": player_symbols,
|
||
"turn_count": 0,
|
||
"active_player": active_player,
|
||
"last_action": None,
|
||
"move_history": [],
|
||
"game_result": None,
|
||
"winner": None,
|
||
"draw": False,
|
||
"seed": seed,
|
||
}
|
||
|
||
role_mapping = {0: "ArchitectA", 1: "ArchitectB"}
|
||
|
||
# Initialize the underlying textarena state
|
||
self.state.reset(game_state=game_state, player_prompt_function=self._generate_player_prompt, role_mapping=role_mapping)
|
||
|
||
# Manually set starting player based on seed
|
||
self.state.manually_set_current_player_id(starting_player)
|
||
|
||
# Add initial observations
|
||
self.state.add_observation("Welcome to Stellar Triad!", ta.ObservationType.GAME_MESSAGE)
|
||
self.state.add_observation("A 3x3 orbital matrix awaits your channeling commands.", ta.ObservationType.GAME_MESSAGE)
|
||
return None
|
||
|
||
def step(self, action: str) -> Tuple[bool, ta.Info]:
|
||
"""
|
||
Perform one environment step for the active player.
|
||
|
||
Args:
|
||
action: The response text from the agent containing a \boxed{{[Channel:X-Y]}} token.
|
||
|
||
Returns:
|
||
(done, info)
|
||
"""
|
||
player_id = self.state.current_player_id
|
||
player_role = "ArchitectA" if player_id == 0 else "ArchitectB"
|
||
|
||
# Log raw action as a player action observation
|
||
self.state.add_observation(message=action, observation_type=ta.ObservationType.PLAYER_ACTION,
|
||
from_id=player_id, to_id=-1)
|
||
|
||
extracted = self._extract_answer_content(action)
|
||
if not extracted:
|
||
self.state.set_invalid_move("Action missing or not boxed")
|
||
return self.state.step()
|
||
|
||
# Validate action format
|
||
if not self.action_pattern.match(extracted):
|
||
self.state.set_invalid_move("Malformed token: does not match [Channel:X-Y] pattern")
|
||
return self.state.step()
|
||
|
||
# Parse coordinates
|
||
try:
|
||
coords = extracted.strip("[]").split(":")[1].split("-")
|
||
x = int(coords[0]) - 1 # columns 1–3; convert to 0–2
|
||
y = int(coords[1]) - 1 # rows 1–3; convert to 0–2
|
||
except Exception:
|
||
self.state.set_invalid_move("Malformed token: cannot parse coordinates")
|
||
return self.state.step()
|
||
|
||
# Validate range
|
||
if not (0 <= x <= 2 and 0 <= y <= 2):
|
||
self.state.set_invalid_move("Coordinates out of range")
|
||
return self.state.step()
|
||
|
||
# Validate occupancy
|
||
matrix_state = self.state.game_state["matrix_state"]
|
||
if matrix_state[y][x] is not None:
|
||
self.state.set_invalid_move("Target cell occupied")
|
||
return self.state.step()
|
||
|
||
# Apply move
|
||
symbol = self.state.game_state["player_symbols"][player_role]
|
||
matrix_state[y][x] = symbol
|
||
|
||
# Update state-based info
|
||
self.state.game_state["matrix_state"] = matrix_state
|
||
self.state.game_state["last_action"] = extracted
|
||
self.state.game_state["turn_count"] += 1
|
||
self.state.game_state["move_history"].append(f"{player_role}:{extracted}")
|
||
|
||
# Check for win or draw
|
||
if self._check_alignment(matrix_state, symbol):
|
||
winner_id = 0 if player_role == "ArchitectA" else 1
|
||
self.state.game_state["game_result"] = f"{player_role}_won"
|
||
self.state.game_state["winner"] = player_role
|
||
self.state.state_done = True if hasattr(self.state, "state_done") else None
|
||
self.state.set_winner(winner_id, reason=f"{player_role} achieved Stellar Alignment.")
|
||
return self.state.step()
|
||
|
||
if self.state.game_state["turn_count"] >= 9:
|
||
self.state.game_state["draw"] = True
|
||
self.state.game_state["game_result"] = "Stellar_Collapse"
|
||
self.state.set_draw(reason="Orbital grid full without alignment (Stellar Collapse).")
|
||
return self.state.step()
|
||
|
||
# Switch active player
|
||
self.state.game_state["active_player"] = "ArchitectA" if player_role == "ArchitectB" else "ArchitectB"
|
||
|
||
return self.state.step()
|
||
|
||
def _check_alignment(self, board: List[List[Optional[str]]], symbol: str) -> bool:
|
||
"""
|
||
Check if the given symbol has achieved a Stellar Alignment (three-in-a-line).
|
||
"""
|
||
# Rows and columns
|
||
for i in range(3):
|
||
if all(cell == symbol for cell in board[i]):
|
||
return True
|
||
if all(board[row][i] == symbol for row in range(3)):
|
||
return True
|
||
# Diagonals
|
||
if all(board[i][i] == symbol for i in range(3)):
|
||
return True
|
||
if all(board[i][2 - i] == symbol for i in range(3)):
|
||
return True
|
||
return False
|
||
|
||
def _generate_player_prompt(self, player_id: int, game_state: Dict[str, Any]) -> str:
|
||
"""
|
||
Generates the textual game prompt for each Architect player.
|
||
"""
|
||
role = "ArchitectA" if player_id == 0 else "ArchitectB"
|
||
symbol = game_state["player_symbols"][role]
|
||
active_marker = " (You start.)" if game_state["active_player"] == role else ""
|
||
matrix_str = self._format_matrix(game_state["matrix_state"])
|
||
|
||
prompt = (
|
||
f"You are {role}, a cosmic architect channeling energy Nodes around a dying star.\n"
|
||
f"Your symbol: '{symbol}'.{active_marker}\n\n"
|
||
"Below is the current 3×3 orbital matrix. Empty slots are shown as '.' :\n"
|
||
f"{matrix_str}\n\n"
|
||
"Your goal: Achieve a *Stellar Alignment* — three of your Nodes in any straight line (horizontal, vertical, or diagonal).\n"
|
||
"If the matrix fills without alignment, the star collapses and both architects fail.\n\n"
|
||
"Each turn, choose one unoccupied cell to channel your energy Node into.\n"
|
||
"Use the exact format: [Channel:X-Y] (columns and rows from 1 to 3).\n"
|
||
"Example: [Channel:2-3] → channel into column 2, row 3.\n"
|
||
"Invalid formats include [Deploy:2-3] or [Channel:4-1].\n\n"
|
||
"Place your chosen command within \\boxed{{}} at the end of your response.\n\n"
|
||
"Example valid response:\n"
|
||
"I will project energy into the lower middle conduit.\n"
|
||
"\\boxed{{[Channel:2-3]}}\n\n"
|
||
"Example invalid response:\n"
|
||
"I channel energy south-east.\n"
|
||
"\\boxed{{Channel:SE}}\n"
|
||
)
|
||
return prompt
|
||
|
||
def _format_matrix(self, matrix: List[List[Optional[str]]]) -> str:
|
||
"""Return textual representation of the matrix."""
|
||
lines = []
|
||
for row in matrix:
|
||
line = " ".join(cell if cell is not None else "." for cell in row)
|
||
lines.append(line)
|
||
return "\n".join(lines)
|
||
``` |