botWebWars/trollagent_ai/bot.py

"""AI-driven Troll Agent for botWebWars, powered by an Ollama LLM.

Implements the unique rules and tactical gameplay of the Troll character type:
- Solitary hunter: Never joins or forms parties.
- 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 local/remote Ollama LLM:
  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 time
from typing import Any, Dict, List, Optional, Tuple

import requests

DEFAULT_SERVER_URL = os.getenv("TROLL_SERVER_URL", os.getenv("BOT_SERVER_URL", "http://localhost:8000/api"))
DEFAULT_TROLL_NAME = os.getenv("TROLL_NAME", "GorgonAITroll")
DEFAULT_TROLL_COLOR = os.getenv("TROLL_COLOR", "#15803d")
DEFAULT_TROLL_STRENGTH = float(os.getenv("TROLL_STRENGTH", "3.0"))
DEFAULT_TROLL_HEALTH = float(os.getenv("TROLL_HEALTH", "10.0"))

OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://192.168.1.220:11434")
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gemma4:12b")

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.
- You CANNOT duel or interact with Gary the Wizard.
- SLEEP RESTORATION: On any turn when you are not in adjacent 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 output."""
    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 OllamaClient:
    def __init__(self, base_url: str, model: str):
        self.base_url = base_url.rstrip("/")
        self.model = model

    def ask_json(self, prompt: str) -> Optional[Dict[str, Any]]:
        """Query Ollama with a JSON formatting constraint."""
        url = f"{self.base_url}/api/generate"
        payload = {
            "model": self.model,
            "prompt": prompt,
            "format": "json",
            "stream": False,
            "options": {"temperature": 0.3},
        }
        try:
            res = requests.post(url, json=payload, timeout=45)
            res.raise_for_status()
            raw = res.json().get("response", "")
            return extract_json(raw)
        except requests.exceptions.RequestException as e:
            print(f"⚠️ [OLLAMA ERROR] {e}")
            return None


class AITrollAgent:
    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,
        ollama_url: str = OLLAMA_BASE_URL,
        ollama_model: str = OLLAMA_MODEL,
        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.llm = OllamaClient(ollama_url, ollama_model)
        self.loop_delay = loop_delay
        self.bot_id: Optional[str] = None

    def register(self):
        """Register the troll avatar on the grid or reconnect if existing."""
        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 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:
        """Rest for 1 turn to regenerate +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} rested 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 when no actions are possible."""
        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 checking 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 tactical D20 confrontation!")
        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 the LLM to decide whether to sleep or move, and if moving, 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")

        # Format available movement choices
        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 (all blocked by obstacles/bounds)."

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

Available Passable Moves:
{choices_str}

OPTIONS:
1. "sleep" - Take a rest turn to regenerate +0.1 HP. (Recommended if wounded or trapped).
2. "move" - Move in one of the passable directions toward the nearest player.

Decide which action to take. If action is "move", pick the best passable direction from the list.
Respond ONLY with a JSON object:
{{"action": "sleep"|"move", "direction": "UP"|"DOWN"|"LEFT"|"RIGHT"|"UP_LEFT"|"UP_RIGHT"|"DOWN_LEFT"|"DOWN_RIGHT"|null, "reasoning": "short tactical rationale"}}
"""
        decision = self.llm.ask_json(prompt) or {}
        action = str(decision.get("action", "move")).lower().strip()
        direction = decision.get("direction")
        reasoning = decision.get("reasoning", "")

        # Fallback validation
        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):
        """Turn execution pipeline: combat check -> LLM reasoning -> act."""
        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. Immediate 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 sensor data
        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. LLM strategic decision
        action, direction, reasoning = self._decide_action_and_direction(my_info, radar_res, available_moves)
        print(f"🧠 [LLM STRATEGY] Action: {action.upper()}{f' -> {direction}' if direction else ''}. Reasoning: {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 game loop."""
        self.register()
        try:
            while True:
                # Check life status
                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 marked 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] AI 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👋 AI Troll {self.name} disconnecting gracefully.")


def main():
    parser = argparse.ArgumentParser(description="Ollama LLM-Assisted 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("--ollama-url", default=OLLAMA_BASE_URL, help="Ollama API base URL")
    parser.add_argument("--ollama-model", default=OLLAMA_MODEL, help="Ollama model name (default: gemma4:12b)")
    parser.add_argument("--loop-delay", type=float, default=1.0, help="Polling interval in seconds")

    args = parser.parse_args()

    agent = AITrollAgent(
        name=args.name,
        color=args.color,
        strength=args.strength,
        health=args.health,
        server_url=args.url,
        ollama_url=args.ollama_url,
        ollama_model=args.ollama_model,
        loop_delay=args.loop_delay,
    )
    agent.run()


if __name__ == "__main__":
    main()