import json import sqlite3 import time import uuid from enum import Enum from pathlib import Path from typing import Dict, Optional, Any from dataclasses import dataclass, field from world.server_engine_loop import ServerEngineLoop from agent.agent_decision_core import AgentDecisionCore, TacticalStance from agent.llm_provider_client import LLMProviderConfig, LLMProviderType class AgentOrbTier(str, Enum): TIER_1 = "item_agent_orb_tier1" TIER_2 = "item_agent_orb_tier2" TIER_3 = "item_agent_orb_tier3" ORB_DURATIONS: Dict[str, float] = { AgentOrbTier.TIER_1.value: 7200.0, AgentOrbTier.TIER_2.value: 28800.0, AgentOrbTier.TIER_3.value: 86400.0, } ORB_NAMES: Dict[str, str] = { AgentOrbTier.TIER_1.value: "Chiến Hồn Tế Cốt - Sơ Cấp (2H)", AgentOrbTier.TIER_2.value: "Chiến Hồn Tế Cốt - Trung Cấp (8H)", AgentOrbTier.TIER_3.value: "Hung Ma Tế Cốt - Vô Song (24H)", } @dataclass class DelegationSession: session_id: str player_id: int orb_id: str start_timestamp_sec: float total_duration_sec: float remaining_duration_sec: float stance: TacticalStance llm_config: LLMProviderConfig decision_core: AgentDecisionCore is_active: bool = True is_offline: bool = False is_expired: bool = False exp_gained: int = 0 monsters_slain: int = 0 spirit_stones_looted: Dict[str, int] = field(default_factory=dict) potions_consumed: int = 0 actual_active_time_sec: float = 0.0 @dataclass(slots=True, frozen=True) class DelegationHarvestReport: session_id: str player_id: int total_duration_sec: float actual_active_time_sec: float exp_gained: int monsters_slain: int spirit_stones_looted: Dict[str, int] potions_consumed: int reclaimed_timestamp: float hmac_signature: str = "" class AgentOrbService: def __init__(self, engine: Optional[ServerEngineLoop] = None, db_path: str = "data/agent_delegations.db") -> None: self.engine = engine if db_path != ":memory:" and not Path(db_path).is_absolute(): base_dir = Path(__file__).resolve().parent.parent.parent self.db_path = str(base_dir / db_path) Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) else: self.db_path = db_path self.conn = sqlite3.connect(self.db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA journal_mode=WAL;") self.conn.execute("PRAGMA synchronous=NORMAL;") self._init_db() self.active_sessions: Dict[int, DelegationSession] = {} # Load active sessions from DB self._load_active_sessions() def _init_db(self): with self.conn: self.conn.execute(""" CREATE TABLE IF NOT EXISTS inventories ( player_id INTEGER, orb_id TEXT, count INTEGER, PRIMARY KEY (player_id, orb_id) ) """) self.conn.execute(""" CREATE TABLE IF NOT EXISTS active_sessions ( player_id INTEGER PRIMARY KEY, session_id TEXT, orb_id TEXT, start_timestamp_sec REAL, total_duration_sec REAL, remaining_duration_sec REAL, stance TEXT, llm_config TEXT, is_active INTEGER, is_offline INTEGER, is_expired INTEGER, exp_gained INTEGER, monsters_slain INTEGER, spirit_stones_looted TEXT, potions_consumed INTEGER, actual_active_time_sec REAL ) """) self.conn.execute(""" CREATE TABLE IF NOT EXISTS completed_reports ( session_id TEXT PRIMARY KEY, player_id INTEGER, report_json TEXT ) """) def _load_active_sessions(self): cur = self.conn.cursor() cur.execute("SELECT * FROM active_sessions WHERE is_active = 1") for row in cur.fetchall(): llm_conf_dict = json.loads(row["llm_config"]) llm_config = LLMProviderConfig( provider=LLMProviderType(llm_conf_dict["provider"]), model=llm_conf_dict.get("model", ""), api_key=llm_conf_dict.get("api_key", ""), temperature=llm_conf_dict.get("temperature", 0.7) ) stance = TacticalStance(row["stance"]) decision_core = AgentDecisionCore( player_id=row["player_id"], provider_config=llm_config, stance=stance ) sess = DelegationSession( session_id=row["session_id"], player_id=row["player_id"], orb_id=row["orb_id"], start_timestamp_sec=row["start_timestamp_sec"], total_duration_sec=row["total_duration_sec"], remaining_duration_sec=row["remaining_duration_sec"], stance=stance, llm_config=llm_config, decision_core=decision_core, is_active=bool(row["is_active"]), is_offline=bool(row["is_offline"]), is_expired=bool(row["is_expired"]), exp_gained=row["exp_gained"], monsters_slain=row["monsters_slain"], spirit_stones_looted=json.loads(row["spirit_stones_looted"]), potions_consumed=row["potions_consumed"], actual_active_time_sec=row["actual_active_time_sec"] ) self.active_sessions[row["player_id"]] = sess def _save_session(self, session: DelegationSession): with self.conn: llm_dict = { "provider": session.llm_config.provider.value, "model": session.llm_config.model, "api_key": session.llm_config.api_key, "temperature": session.llm_config.temperature } self.conn.execute(""" INSERT OR REPLACE INTO active_sessions (player_id, session_id, orb_id, start_timestamp_sec, total_duration_sec, remaining_duration_sec, stance, llm_config, is_active, is_offline, is_expired, exp_gained, monsters_slain, spirit_stones_looted, potions_consumed, actual_active_time_sec) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( session.player_id, session.session_id, session.orb_id, session.start_timestamp_sec, session.total_duration_sec, session.remaining_duration_sec, session.stance.value, json.dumps(llm_dict), int(session.is_active), int(session.is_offline), int(session.is_expired), session.exp_gained, session.monsters_slain, json.dumps(session.spirit_stones_looted), session.potions_consumed, session.actual_active_time_sec )) def grant_agent_orb(self, player_id: int, orb_id: str, count: int = 1) -> None: with self.conn: cur = self.conn.cursor() cur.execute("SELECT count FROM inventories WHERE player_id = ? AND orb_id = ?", (player_id, orb_id)) row = cur.fetchone() current = row["count"] if row else 0 new_count = current + count cur.execute("INSERT OR REPLACE INTO inventories (player_id, orb_id, count) VALUES (?, ?, ?)", (player_id, orb_id, new_count)) def get_player_orb_balance(self, player_id: int) -> Dict[str, int]: cur = self.conn.cursor() cur.execute("SELECT orb_id, count FROM inventories WHERE player_id = ?", (player_id,)) return {row["orb_id"]: row["count"] for row in cur.fetchall()} def activate_delegation( self, player_id: int, orb_id: str, stance: TacticalStance, llm_config: LLMProviderConfig ) -> Optional[DelegationSession]: balances = self.get_player_orb_balance(player_id) balance = balances.get(orb_id, 0) if balance <= 0: return None with self.conn: self.conn.execute("UPDATE inventories SET count = count - 1 WHERE player_id = ? AND orb_id = ?", (player_id, orb_id)) duration = ORB_DURATIONS.get(orb_id, 7200.0) now = time.time() session_id = f"del_{player_id}_{int(now)}_{uuid.uuid4().hex[:6]}" decision_core = AgentDecisionCore( player_id=player_id, provider_config=llm_config, stance=stance ) session = DelegationSession( session_id=session_id, player_id=player_id, orb_id=orb_id, start_timestamp_sec=now, total_duration_sec=duration, remaining_duration_sec=duration, stance=stance, llm_config=llm_config, decision_core=decision_core, is_active=True, is_offline=False ) self.active_sessions[player_id] = session self._save_session(session) return session def extend_delegation(self, player_id: int, orb_id: str) -> bool: session = self.active_sessions.get(player_id) if not session or not session.is_active: return False balances = self.get_player_orb_balance(player_id) balance = balances.get(orb_id, 0) if balance <= 0: return False with self.conn: self.conn.execute("UPDATE inventories SET count = count - 1 WHERE player_id = ? AND orb_id = ?", (player_id, orb_id)) extra_duration = ORB_DURATIONS.get(orb_id, 7200.0) session.total_duration_sec += extra_duration session.remaining_duration_sec += extra_duration self._save_session(session) return True def get_active_session(self, player_id: int) -> Optional[DelegationSession]: session = self.active_sessions.get(player_id) if session and session.is_active: return session return None def set_player_offline_state(self, player_id: int, is_offline: bool) -> None: session = self.active_sessions.get(player_id) if session and session.is_active: session.is_offline = is_offline self._save_session(session) def record_loot_harvest(self, player_id: int, item_id: str, count: int = 1) -> None: session = self.active_sessions.get(player_id) if session and session.is_active: current = session.spirit_stones_looted.get(item_id, 0) session.spirit_stones_looted[item_id] = current + count self._save_session(session) def record_monster_slain(self, player_id: int, exp: int = 100) -> None: session = self.active_sessions.get(player_id) if session and session.is_active: session.monsters_slain += 1 session.exp_gained += exp self._save_session(session) def record_potion_consumed(self, player_id: int, count: int = 1) -> None: session = self.active_sessions.get(player_id) if session and session.is_active: session.potions_consumed += count self._save_session(session) def step_tick(self, dt: float) -> None: dirty_sessions = [] for player_id, session in list(self.active_sessions.items()): if not session.is_active: continue session.remaining_duration_sec -= dt session.actual_active_time_sec += dt dirty_sessions.append(session) if session.remaining_duration_sec <= 0: session.remaining_duration_sec = 0 session.is_active = False session.is_expired = True continue if self.engine: perception = session.decision_core.build_perception(self.engine) old_thought = session.decision_core.last_thought plan = session.decision_core.evaluate_tick_action(perception) session.decision_core.dispatch_plan_to_server(plan, self.engine, dt=dt) new_thought = session.decision_core.last_thought if old_thought != new_thought: try: import gateway.ws_gateway_bridge as ws_gateway bridge = ws_gateway.get_bridge_instance() if bridge.is_running: bridge.broadcast_json({ "type": "agent-thought-stream", "player_id": player_id, "session_id": session.session_id, "thought": new_thought, "timestamp": int(time.time() * 1000) }) except ImportError: pass # Periodically flush or on every tick (this might be heavy, but it's okay for prototype) if dirty_sessions: with self.conn: update_data = [ ( session.remaining_duration_sec, session.actual_active_time_sec, int(session.is_active), int(session.is_expired), session.exp_gained, session.monsters_slain, json.dumps(session.spirit_stones_looted), session.potions_consumed, session.session_id ) for session in dirty_sessions ] self.conn.executemany(""" UPDATE active_sessions SET remaining_duration_sec = ?, actual_active_time_sec = ?, is_active = ?, is_expired = ?, exp_gained = ?, monsters_slain = ?, spirit_stones_looted = ?, potions_consumed = ? WHERE session_id = ? """, update_data) def reclaim_control(self, player_id: int) -> Optional[DelegationHarvestReport]: import hmac import hashlib import os session = self.active_sessions.pop(player_id, None) if not session: return None session.is_active = False reclaimed_time = time.time() secret_key = os.environ.get("AGENT_HMAC_SECRET", "default_freeexile_secret").encode("utf-8") drops_str = json.dumps(session.spirit_stones_looted.copy(), sort_keys=True) payload = f"{session.session_id}{session.exp_gained}{drops_str}{reclaimed_time}".encode("utf-8") signature = hmac.new(secret_key, payload, hashlib.sha256).hexdigest() report = DelegationHarvestReport( session_id=session.session_id, player_id=player_id, total_duration_sec=session.total_duration_sec, actual_active_time_sec=session.actual_active_time_sec, exp_gained=session.exp_gained, monsters_slain=session.monsters_slain, spirit_stones_looted=session.spirit_stones_looted.copy(), potions_consumed=session.potions_consumed, reclaimed_timestamp=reclaimed_time, hmac_signature=signature ) with self.conn: self.conn.execute("DELETE FROM active_sessions WHERE session_id = ?", (session.session_id,)) report_dict = { "session_id": report.session_id, "player_id": report.player_id, "total_duration_sec": report.total_duration_sec, "actual_active_time_sec": report.actual_active_time_sec, "exp_gained": report.exp_gained, "monsters_slain": report.monsters_slain, "spirit_stones_looted": report.spirit_stones_looted, "potions_consumed": report.potions_consumed, "reclaimed_timestamp": report.reclaimed_timestamp, "hmac_signature": report.hmac_signature } self.conn.execute("INSERT INTO completed_reports (session_id, player_id, report_json) VALUES (?, ?, ?)", (report.session_id, report.player_id, json.dumps(report_dict))) return report def validate_harvest_report(self, report: DelegationHarvestReport) -> bool: import hmac import hashlib import os secret_key = os.environ.get("AGENT_HMAC_SECRET", "default_freeexile_secret").encode("utf-8") drops_str = json.dumps(report.spirit_stones_looted, sort_keys=True) payload = f"{report.session_id}{report.exp_gained}{drops_str}{report.reclaimed_timestamp}".encode("utf-8") expected_sig = hmac.new(secret_key, payload, hashlib.sha256).hexdigest() return hmac.compare_digest(report.hmac_signature, expected_sig)