botWebWars/trollagent_gear/bot.py

"""AI-driven Troll Agent for botWebWars, powered by Google Cloud Vertex AI (Gemini).

Implements the unique rules and tactical gameplay of the Troll character type:
- Solitary brute: Never joins or forms parties/alliances.
- Troll truce: Never battles other trolls.
- Relentless hunter: Mandatory tactical combat against players and squads.
- Strategic healing: Can take turns sleeping to regenerate +0.1 HP.
- Cannot challenge or duel Gary the Wizard.
- Strategic decision making via Vertex AI / Gemini:
  1. Action Choice: Evaluates health and distance to decide whether to SLEEP (recover health) or HUNT (move).
  2. Directional Navigation: Chooses the optimal passable path around obstacles towards target players.
"""

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import time
from typing import Any, Dict, List, Optional, Tuple

import requests

# Optional google-auth integration
try:
    import google.auth
    import google.auth.transport.requests
    HAVE_GOOGLE_AUTH = True
except ImportError:
    HAVE_GOOGLE_AUTH = False

DEFAULT_SERVER_URL = os.getenv("TROLL_SERVER_URL", os.getenv("BOT_SERVER_URL", "http://localhost:8000/api"))
DEFAULT_TROLL_NAME = os.getenv("TROLL_NAME", "GeminiTroll")
DEFAULT_TROLL_COLOR = os.getenv("TROLL_COLOR", "#047857")
DEFAULT_TROLL_STRENGTH = float(os.getenv("TROLL_STRENGTH", "3.0"))
DEFAULT_TROLL_HEALTH = float(os.getenv("TROLL_HEALTH", "10.0"))
DEFAULT_MODEL = os.getenv("VERTEX_MODEL", os.getenv("GEMINI_MODEL", "gemini-2.5-flash"))
DEFAULT_LOCATION = os.getenv("VERTEX_LOCATION", "global")

TROLL_RULES_SUMMARY = """
You are an autonomous tactical Troll in botWebWars!
Troll Rules of Engagement:
- You NEVER form alliances or parties with anyone. You are a solitary brute hunter.
- You NEVER fight other trolls. Trolls maintain an instinctive truce.
- Your sole mission is to hunt down human/bot players and player squads and crush them in 3-bout D20 battles.
- When victorious in battle, you earn +2 victory points (score). You do NOT gain strength or absorb squad members.
- If defeated in battle, you take 1 to 3 health points damage. At 0 HP, you die and a gravestone is placed on the board.
- You CANNOT duel or interact with Gary the Wizard.
- SLEEP RESTORATION: On any turn when you are not locked in combat, you may choose to SLEEP. Sleeping skips movement but regenerates +0.1 HP!
- Squeezing diagonally between obstacle corners costs -0.1 strength penalty.
- The game concludes when all surviving entities are resolved. You can win the game on the final scoreboards!
"""


def normalize_url(url: str) -> str:
    """Ensure the API URL ends with /api without trailing slashes."""
    cleaned = url.rstrip("/")
    if not cleaned.endswith("/api"):
        cleaned = f"{cleaned}/api"
    return cleaned


def extract_json(text: str) -> Optional[Dict[str, Any]]:
    """Extract first valid JSON object from model response."""
    if not text:
        return None
    cleaned = text.strip()
    if cleaned.startswith("```"):
        cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned)
        cleaned = re.sub(r"\s*```$", "", cleaned)
    match = re.search(r"\{.*\}", cleaned, re.DOTALL)
    if not match:
        return None
    try:
        return json.loads(match.group(0))
    except json.JSONDecodeError:
        return None


class VertexGeminiClient:
    """Client for querying Gemini models on Google Cloud Vertex AI or Google AI Studio."""

    def __init__(
        self,
        project_id: Optional[str] = None,
        location: str = DEFAULT_LOCATION,
        model: str = DEFAULT_MODEL,
        api_key: Optional[str] = None,
    ):
        self.location = location or DEFAULT_LOCATION
        self.model = model or DEFAULT_MODEL
        self.api_key = api_key or os.getenv("VERTEX_API_KEY") or os.getenv("GEMINI_API_KEY")
        self.project_id = project_id or os.getenv("VERTEX_PROJECT_ID") or os.getenv("GCP_PROJECT") or os.getenv("GOOGLE_CLOUD_PROJECT")

        self._cached_token: Optional[str] = None
        self._token_expiry: float = 0.0

        if not self.project_id and not self.api_key:
            self.project_id = self._detect_project()

    def _detect_project(self) -> Optional[str]:
        """Attempt to determine the GCP project from environment or gcloud config."""
        if HAVE_GOOGLE_AUTH:
            try:
                _, proj = google.auth.default()
                if proj:
                    return proj
            except Exception:
                pass

        if shutil.which("gcloud"):
            try:
                res = subprocess.check_output(
                    ["gcloud", "config", "get-value", "project"],
                    stderr=subprocess.DEVNULL,
                    text=True,
                ).strip()
                if res and res != "(unset)":
                    return res
            except Exception:
                pass

        return None

    def _get_access_token(self) -> Optional[str]:
        """Obtain a valid OAuth 2.0 Bearer access token for Vertex AI."""
        env_token = os.getenv("VERTEX_ACCESS_TOKEN") or os.getenv("GOOGLE_OAUTH_ACCESS_TOKEN")
        if env_token:
            return env_token

        now = time.time()
        if self._cached_token and now < self._token_expiry - 60:
            return self._cached_token

        if HAVE_GOOGLE_AUTH:
            try:
                credentials, _ = google.auth.default(
                    scopes=["https://www.googleapis.com/auth/cloud-platform"]
                )
                auth_req = google.auth.transport.requests.Request()
                credentials.refresh(auth_req)
                self._cached_token = credentials.token
                self._token_expiry = now + 3000
                return self._cached_token
            except Exception:
                pass

        if shutil.which("gcloud"):
            try:
                token = subprocess.check_output(
                    ["gcloud", "auth", "print-access-token"],
                    stderr=subprocess.PIPE,
                    text=True,
                ).strip()
                if token:
                    self._cached_token = token
                    self._token_expiry = now + 3000
                    return token
            except subprocess.CalledProcessError:
                pass

        return None

    def check_auth(self) -> Tuple[bool, str]:
        """Validate whether authentication is ready."""
        if self.api_key:
            return True, f"Using direct API key (model: {self.model})"

        if not self.project_id:
            return False, (
                "No Google Cloud project ID detected.\n"
                "Please set VERTEX_PROJECT_ID=<your-project-id> or run:\n"
                "  gcloud config set project <your-project-id>"
            )

        token = self._get_access_token()
        if not token:
            return False, (
                "Unable to obtain Google Cloud authentication token.\n"
                "Please authenticate using:\n"
                "  1. gcloud auth application-default login\n"
                "  2. Or set GEMINI_API_KEY / VERTEX_API_KEY"
            )

        return True, f"Authenticated to GCP Project '{self.project_id}' in region '{self.location}' (model: {self.model})"

    def _build_vertex_url(self, location: Optional[str] = None) -> str:
        loc = location or self.location or "global"
        if loc == "global":
            endpoint = "aiplatform.googleapis.com"
        else:
            endpoint = f"{loc}-aiplatform.googleapis.com"
        return f"https://{endpoint}/v1/projects/{self.project_id}/locations/{loc}/publishers/google/models/{self.model}:generateContent"

    def ask_json(self, prompt: str, system_instruction: Optional[str] = None) -> Optional[Dict[str, Any]]:
        """Query Gemini model requesting a structured JSON response."""
        # Method 1: Google AI Studio API key
        if self.api_key:
            url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent?key={self.api_key}"
            headers = {"Content-Type": "application/json"}
            body: Dict[str, Any] = {
                "contents": [{"role": "user", "parts": [{"text": prompt}]}],
                "generationConfig": {
                    "temperature": 0.3,
                    "responseMimeType": "application/json",
                },
            }
            if system_instruction:
                body["systemInstruction"] = {"parts": [{"text": system_instruction}]}

            try:
                res = requests.post(url, headers=headers, json=body, timeout=45)
                res.raise_for_status()
                data = res.json()
                text = (
                    data.get("candidates", [{}])[0]
                    .get("content", {})
                    .get("parts", [{}])[0]
                    .get("text", "")
                )
                return extract_json(text)
            except Exception as e:
                print(f"⚠️ [GEMINI API KEY ERROR] {e}")
                return None

        # Method 2: Vertex AI endpoint
        token = self._get_access_token()
        if not token:
            print("⚠️ [AUTH ERROR] No Google Cloud access token available.")
            return None

        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        }
        body = {
            "contents": [{"role": "user", "parts": [{"text": prompt}]}],
            "generationConfig": {
                "temperature": 0.3,
                "responseMimeType": "application/json",
            },
        }
        if system_instruction:
            body["systemInstruction"] = {"parts": [{"text": system_instruction}]}

        url = self._build_vertex_url()
        try:
            res = requests.post(url, headers=headers, json=body, timeout=45)
            if res.status_code == 404 and self.location == "global":
                url_fallback = self._build_vertex_url("us-central1")
                res = requests.post(url_fallback, headers=headers, json=body, timeout=45)
            res.raise_for_status()
            data = res.json()
            text = (
                data.get("candidates", [{}])[0]
                .get("content", {})
                .get("parts", [{}])[0]
                .get("text", "")
            )
            return extract_json(text)
        except Exception as e:
            print(f"⚠️ [VERTEX AI ERROR] {e}")
            return None


class GearTrollAgent:
    def __init__(
        self,
        name: str = DEFAULT_TROLL_NAME,
        color: str = DEFAULT_TROLL_COLOR,
        strength: float = DEFAULT_TROLL_STRENGTH,
        health: float = DEFAULT_TROLL_HEALTH,
        server_url: str = DEFAULT_SERVER_URL,
        project_id: Optional[str] = None,
        location: str = DEFAULT_LOCATION,
        model: str = DEFAULT_MODEL,
        api_key: Optional[str] = None,
        loop_delay: float = 1.0,
    ):
        self.name = name
        self.color = color
        self.strength = strength
        self.health = health
        self.base_url = normalize_url(server_url)
        self.gemini = VertexGeminiClient(
            project_id=project_id,
            location=location,
            model=model,
            api_key=api_key,
        )
        self.loop_delay = loop_delay
        self.bot_id: Optional[str] = None

    def register(self):
        """Register the troll avatar or reconnect."""
        try:
            players = requests.get(f"{self.base_url}/players").json()
            for p in players:
                if p.get("name") == self.name:
                    self.bot_id = p["id"]
                    print(
                        f"🔄 [RECONNECT] Reconnected to existing Troll {self.name} "
                        f"(ID: {self.bot_id}, Str: {p.get('strength', self.strength)}, "
                        f"HP: {p.get('health', self.health)}) at ({p.get('x')}, {p.get('y')})"
                    )
                    return
        except Exception:
            pass

        payload = {
            "name": self.name,
            "color": self.color,
            "strength": self.strength,
            "health": self.health,
            "piece_type": "troll",
            "character_type": "troll",
        }

        res = requests.post(f"{self.base_url}/players", json=payload)
        if res.status_code == 400 and "already registered" in res.text:
            players = requests.get(f"{self.base_url}/players").json()
            for p in players:
                if p.get("name") == self.name:
                    self.bot_id = p["id"]
                    print(
                        f"🔄 [RECONNECT] Reconnected to existing Troll {self.name} "
                        f"(ID: {self.bot_id}, HP: {p.get('health', self.health)}) at ({p.get('x')}, {p.get('y')})"
                    )
                    return

        res.raise_for_status()
        data = res.json()
        self.bot_id = data["id"]
        print(
            f"👹 [REGISTER] Spawned Vertex AI Troll {self.name} (ID: {self.bot_id}, Str: {self.strength}, "
            f"HP: {data.get('health', self.health)}) at ({data['x']}, {data['y']})"
        )

    def refresh_status(self) -> Optional[Dict[str, Any]]:
        """Get live troll state."""
        try:
            res = requests.get(f"{self.base_url}/players/{self.bot_id}")
            if res.status_code == 200:
                return res.json()
        except Exception as e:
            print(f"Status refresh error: {e}")
        return None

    def sleep_and_heal(self) -> bool:
        """Execute sleep turn to recover +0.1 HP."""
        try:
            res = requests.post(f"{self.base_url}/players/{self.bot_id}/sleep")
            if res.status_code == 200:
                data = res.json()
                print(
                    f"💤 [SLEEP] {self.name} curled up and slept for 1 turn. "
                    f"(+{data.get('health_gained', 0.1)} HP -> ❤️ {data.get('new_health')} HP)"
                )
                return True
            else:
                print(f"Sleep failed ({res.status_code}): {res.text}")
                requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
                return False
        except Exception as e:
            print(f"Error sleeping: {e}")
            requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
            return False

    def pass_turn(self):
        """Pass turn."""
        try:
            requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
            print(f"⏩ [PASS] {self.name} passed turn.")
        except Exception as e:
            print(f"Error passing turn: {e}")

    def _find_adjacent_player(self, my_x: int, my_y: int) -> Optional[Dict[str, Any]]:
        """Detect living adjacent player or player squad (ignores other trolls)."""
        try:
            players: List[Dict[str, Any]] = requests.get(f"{self.base_url}/players").json()
            for p in players:
                if p.get("id") == self.bot_id:
                    continue
                if p.get("character_type") == "troll":
                    continue
                if p.get("is_alive") is False or p.get("health", 10) <= 0:
                    continue
                chebyshev = max(abs(p["x"] - my_x), abs(p["y"] - my_y))
                if chebyshev <= 1:
                    return p
        except Exception as e:
            print(f"Error scanning adjacent players: {e}")
        return None

    def attack_player(self, defender: Dict[str, Any]):
        """Initiate mandatory 3-bout D20 battle against adjacent player."""
        target_label = f"Squad '{defender.get('party_id')}'" if defender.get("party_id") else f"Player {defender.get('name')}"
        print(f"⚔️ [BATTLE CLASH] Troll {self.name} ambushes {target_label} in 3-bout D20 combat!")
        try:
            res = requests.post(
                f"{self.base_url}/battles/fight",
                json={"challenger_id": self.bot_id, "defender_id": defender["id"]},
            )
            if res.status_code == 200:
                b = res.json()
                won = b.get("winner_leader_name") == self.name or b.get("winner_party_name", "").startswith(self.name)
                outcome = "VICTORY (+2 Victory Points!)" if won else "DEFEAT (HP damage taken)"
                print(f"⚔️ [RESULT] {outcome}: {b.get('winner_party_name')} defeated {b.get('defeated_party_name')}")
                for bout in b.get("bouts", []):
                    print(
                        f"  Bout #{bout['bout_number']}: "
                        f"{bout['party1_name']} D20({bout['party1_roll']})×Str({bout['party1_strength']})={bout['party1_score']:.1f} vs "
                        f"{bout['party2_name']} D20({bout['party2_roll']})×Str({bout['party2_strength']})={bout['party2_score']:.1f} "
                        f"-> Winner: {bout['winner']}"
                    )
            else:
                print(f"Battle failed ({res.status_code}): {res.text}")
                self.pass_turn()
        except Exception as e:
            print(f"Error fighting battle: {e}")
            self.pass_turn()

    def _decide_action_and_direction(
        self,
        my_info: Dict[str, Any],
        radar_res: Dict[str, Any],
        available_moves: Dict[str, Any],
    ) -> Tuple[str, Optional[str], str]:
        """Ask Vertex AI Gemini whether to sleep or move, and which direction."""
        my_hp = float(my_info.get("health", 10.0))
        max_hp = float(my_info.get("max_health", 10.0))
        nearest = radar_res.get("nearest_target")
        rec_dir = radar_res.get("recommended_direction")

        choices_desc = []
        for d, chk in available_moves.items():
            if chk.get("available"):
                pen = " [squeeze penalty -0.1 STR]" if chk.get("strength_penalty", 0.0) > 0 else ""
                choices_desc.append(f"- {d}: Target ({chk.get('target_x')}, {chk.get('target_y')}){pen}")

        choices_str = "\n".join(choices_desc) if choices_desc else "No open moves available."

        prompt = f"""Current Troll Status:
- Name: "{self.name}" | HP: {my_hp:.1f} / {max_hp:.1f} | Strength: {self.strength:.1f} | Score: {my_info['score']}
- Position: ({my_info['x']}, {my_info['y']})
- Closest Detected Player: {nearest.get('name') if nearest else 'None'} (Distance: {nearest.get('distance') if nearest else 'N/A'}, Pos: {nearest.get('x') if nearest else '?'},{nearest.get('y') if nearest else '?'})
- Radar Pathfinder Recommendation: {rec_dir or 'None'}

Available Passable Moves:
{choices_str}

OPTIONS:
1. "sleep" - Rest for 1 turn to heal +0.1 HP (best if injured and not in direct pursuit).
2. "move" - Advance in one of the passable directions toward the nearest player.

Decide which action to take. If moving, select the best direction from the available moves.
Respond ONLY with JSON:
{{"action": "sleep"|"move", "direction": "UP"|"DOWN"|"LEFT"|"RIGHT"|"UP_LEFT"|"UP_RIGHT"|"DOWN_LEFT"|"DOWN_RIGHT"|null, "reasoning": "concise strategic reasoning"}}
"""
        decision = self.gemini.ask_json(prompt, system_instruction=TROLL_RULES_SUMMARY) or {}
        action = str(decision.get("action", "move")).lower().strip()
        direction = decision.get("direction")
        reasoning = decision.get("reasoning", "")

        if action not in ("sleep", "move"):
            action = "sleep" if my_hp < 6.0 and my_hp < max_hp else "move"

        valid_dirs = [d for d, chk in available_moves.items() if chk.get("available")]
        if action == "move":
            if direction not in valid_dirs:
                direction = rec_dir if rec_dir in valid_dirs else (valid_dirs[0] if valid_dirs else None)
            if not direction:
                action = "sleep" if my_hp < max_hp else "pass"

        return action, direction, reasoning

    def decide_and_act(self):
        """Main turn decision logic."""
        my_info = self.refresh_status()
        if not my_info:
            self.pass_turn()
            return

        my_x = my_info.get("x", 0)
        my_y = my_info.get("y", 0)

        # 1. Combat check: Trolls always attack adjacent players
        adjacent_player = self._find_adjacent_player(my_x, my_y)
        if adjacent_player:
            self.attack_player(adjacent_player)
            return

        # 2. Get sensors
        try:
            radar_res = requests.get(f"{self.base_url}/players/{self.bot_id}/radar").json()
        except Exception:
            radar_res = {}

        try:
            moves_res = requests.get(f"{self.base_url}/players/{self.bot_id}/available-moves").json()
            available_moves = moves_res.get("moves", {})
        except Exception:
            available_moves = {}

        # 3. Gemini reasoning
        action, direction, reasoning = self._decide_action_and_direction(my_info, radar_res, available_moves)
        print(f"✨ [GEMINI REASONING] Action: {action.upper()}{f' -> {direction}' if direction else ''}. Rationale: {reasoning}")

        if action == "sleep":
            self.sleep_and_heal()
        elif action == "move" and direction:
            try:
                res = requests.post(
                    f"{self.base_url}/players/{self.bot_id}/move",
                    json={"direction": direction},
                ).json()
                if res.get("battle_triggered") and res.get("battle_result"):
                    b = res["battle_result"]
                    print(f"⚔️ Move clash! Winner: {b.get('winner_party_name')} (Defeated: {b.get('defeated_party_name')})")
            except Exception as e:
                print(f"Move failed: {e}")
                self.pass_turn()
        else:
            self.pass_turn()

    def run(self):
        """Main loop."""
        self.register()
        try:
            while True:
                my_status = self.refresh_status()
                if my_status and (my_status.get("is_alive") is False or my_status.get("health", 10) <= 0):
                    print(f"\n🪦 [FALLEN TROLL] {self.name} has fallen (0 HP)! Gravestone on board.")
                    print(f"Final Score: {my_status.get('score', 0)} pts preserved on scoreboard. Spectating...")
                    while True:
                        try:
                            conc = requests.get(f"{self.base_url}/game/conclusion").json()
                            if conc.get("concluded"):
                                print(f"\n🎉 [GAME CONCLUDED] Arena concluded! Winner: '{conc.get('winning_party_name')}'")
                                return
                        except Exception:
                            pass
                        time.sleep(2.0)

                turn_info = requests.get(f"{self.base_url}/turn").json()
                if not turn_info.get("game_started", False):
                    if self.bot_id:
                        res = requests.get(f"{self.base_url}/players/{self.bot_id}")
                        if res.status_code == 404:
                            print("\n⚠️ [RESET] Board regenerated. Rejoining lobby...")
                            self.register()

                    print("⏳ [LOBBY] Waiting for game to start via 'Start Game' in UI...      ", end="\r", flush=True)
                    time.sleep(1.0)
                    continue

                curr_player_id = turn_info.get("current_player_id")

                if curr_player_id == self.bot_id:
                    print(f"\n⚡ [MY TURN] Vertex Gemini Troll {self.name} (Round {turn_info.get('round_number')}, Turn {turn_info.get('turn_number')})")
                    self.decide_and_act()

                    conc = requests.get(f"{self.base_url}/game/conclusion").json()
                    if conc.get("concluded"):
                        print(f"\n🎉 [GAME CONCLUDED] Arena concluded! Winner: '{conc.get('winning_party_name')}'!")
                        break

                time.sleep(self.loop_delay)

        except KeyboardInterrupt:
            print(f"\n👋 Vertex Gemini Troll {self.name} disconnecting gracefully.")


def main():
    parser = argparse.ArgumentParser(description="Vertex AI (Gemini) Autonomous Troll Agent for botWebWars")
    parser.add_argument("-u", "--url", default=DEFAULT_SERVER_URL, help="botWebWars API base URL")
    parser.add_argument("-n", "--name", default=DEFAULT_TROLL_NAME, help="Troll name")
    parser.add_argument("-c", "--color", default=DEFAULT_TROLL_COLOR, help="Troll color hex code")
    parser.add_argument("-s", "--strength", type=float, default=DEFAULT_TROLL_STRENGTH, help="Starting strength (1-10)")
    parser.add_argument("-H", "--health", type=float, default=DEFAULT_TROLL_HEALTH, help="Starting health points (default: 10.0)")
    parser.add_argument("--project-id", default=None, help="Google Cloud project ID (or VERTEX_PROJECT_ID)")
    parser.add_argument("--location", default=DEFAULT_LOCATION, help="Vertex AI region (default: global)")
    parser.add_argument("--model", default=DEFAULT_MODEL, help="Gemini model (default: gemini-2.5-flash)")
    parser.add_argument("--api-key", default=None, help="Google AI Studio or Vertex API key")
    parser.add_argument("--loop-delay", type=float, default=1.0, help="Polling interval in seconds")

    args = parser.parse_args()

    agent = GearTrollAgent(
        name=args.name,
        color=args.color,
        strength=args.strength,
        health=args.health,
        server_url=args.url,
        project_id=args.project_id,
        location=args.location,
        model=args.model,
        api_key=args.api_key,
        loop_delay=args.loop_delay,
    )

    auth_ok, auth_msg = agent.gemini.check_auth()
    print(f"🔐 [AUTH CHECK] {auth_msg}")
    if not auth_ok:
        print("⚠️ Warning: Gemini calls will fail without valid credentials.")

    agent.run()


if __name__ == "__main__":
    main()