deploy.py

"""正式世界部署适配器：把进化的 HeuristicStrategy 接入官方 arena-hero SDK。

用法：
    export ARENA_HERO_API_KEY=your-key
    .venv/bin/python deploy.py --genes results/evolve_final.json

--genes 指向 run_evolve.py 输出 JSON 的 "history"（取 fitness 最高者），
或直接传一个 gene JSON 文件（形如 {"name": value, ...}）。

本地试运行（不连正式世界）：
    .venv/bin/python deploy.py --local --genes results/evolve_final.json
"""

import argparse
import json
import os
import sys
import threading
import time
from collections import Counter, deque
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from arena_hero import ArenaHeroClient, Direction, UnitType
from arena_hero.models import CoreView

from ahsim.observation import Observation
from ahsim.vision import visible_cells
from status import LIVE_STATUS, atomic_write_json, write_status
from strategies.heuristic import HeuristicStrategy, normalize_genes

VISION = {"WORKER": 3, "VANGUARD": 4, "RANGER": 5, "CORE": 5}
VIEW_HALF = 64          # 监控地图视窗半径（以 Core 为中心）
HISTORY_PATH = "results/agent_history.jsonl"   # 线上运行统计（供后续进化分析）
EVENTS_PATH = "results/agent_events.jsonl"     # 每 tick 一行结构化事件（供后端解析层）
KEY_EVENTS_PATH = "results/agent_key_events.log"  # 人类可读关键事件日志（采集/交付/进攻/击杀）

# 战斗 tag（用于"出发进攻"检测）
COMBAT_TAGS = {"attack", "track", "raid", "sweep", "harass", "kite"}


def _uid_short(uid, n=6):
    """UUID/对象 → 短标识（前 n 位）。"""
    if uid is None:
        return "?"
    s = str(uid)
    return s[:n]


# ----------------------------------------------------------------------
# 基因加载
# ----------------------------------------------------------------------
def load_genes(path):
    """Load a current-rules genome from an evolution artifact.

    New artifacts carry ``best_genes`` selected with the same metric that was
    used for holdout/rolling seeds.  For old artifacts we retain the history
    fallback, but normalize it through the current v0.14 bounds so legacy
    population/self-destruct genes cannot silently override policy freezes.
    """
    data = json.load(open(path))
    if isinstance(data.get("best_genes"), dict):
        genes = data["best_genes"]
        metric = data.get("best_metric", data.get("best"))
        print(f"[deploy] 使用 gen {data.get('best_gen', '?')} 的已选冠军"
              f" (metric {float(metric):.1f})" if metric is not None
              else "[deploy] 使用结果文件中的已选冠军")
    elif "history" in data:
        history = [h for h in data["history"] if isinstance(h, dict)
                   and isinstance(h.get("genes"), dict)]
        if not history:
            raise ValueError("evolution result has no usable history")
        metric_key = "holdout" if any(h.get("holdout") is not None for h in history) \
            else "best"
        best = max(history, key=lambda h: h.get(metric_key)
                   if h.get(metric_key) is not None else -float("inf"))
        genes = best["genes"]
        metric = best.get(metric_key, best.get("best"))
        print(f"[deploy] 使用 gen {best['gen']} 的基因 ({metric_key} {metric:.1f})")
    else:
        genes = data
    return normalize_genes(genes)


# ----------------------------------------------------------------------
# Observation 构建（模拟器格式 → 正式世界 state）
# ----------------------------------------------------------------------
def _ev_type(e):
    """事件类型：兼容 SDK ResolutionEvent（.event_type）与 dict（.get）。"""
    if hasattr(e, "event_type"):
        return e.event_type
    if isinstance(e, dict):
        return e.get("type", "")
    return ""


def _unit_costs(population):
    """下一单位造价（动态涨价：>20 人口每 5 个 +30%）。供面板展示生产规划。"""
    try:
        from arena_hero.rules import unit_cost, UNIT_BASE_COSTS
        return {ut.value: unit_cost(ut, population) for ut in UNIT_BASE_COSTS}
    except Exception:
        return None


class LiveAdapter:
    """把 SDK 的 Turn 转成模拟器 Observation，并回填策略决策。"""

    # 官方事件名 → 模拟器事件名（Memory 只消费移动类事件）
    MOVE_FAILED = ("UNIT_MOVE_FAILED", "CORE_MOVE_FAILED",
                   "CORE_MOVE_START_FAILED")
    MOVE_OK = ("UNIT_MOVE_SUCCEEDED", "CORE_MOVE_SUCCEEDED")

    def __init__(self, strategy):
        self.strategy = strategy
        self._last_tick = None
        self._sent_dirs = {}     # uid -> 上一 tick 提交的 MOVE 方向
        self._core_dir = None    # Core 最近一次已知的迁移方向
        self._sent_core_dir = None  # 最近实际提交的 START_MOVE 方向
        self.reconnects = 0      # 检测到的断线次数
        self.missed_ticks = 0    # 累计错过的 tick 数
        self.recent_events = deque(maxlen=150)   # 监控用：跨 tick 的事件滚动窗口
        self._last_combat_tag = {}   # uid -> 上 tick 决策 tag（出发进攻检测）
        self._known_own_ids = set()  # 上 Tick IDs；死亡目标不在当前 observation 中
        self.submit_ms = 0.0
        self.started_at = time.time()
        # 运行统计（供后续进化/策略分析；心跳时落盘 agent_history.jsonl）
        self.stats = {
            "harvests": 0,          # 成功采集次数
            "harvest_amount": 0,    # 采集资源总量
            "deposits": 0,          # 成功存放次数
            "deposited": 0,         # 存入 Core 资源总量
            "loot_captured": 0,     # 击败敌方 Core 抢资源次数
            "loot_amount": 0,
            "cores_destroyed": 0,   # 摧毁敌方 Core 次数
            "units_destroyed": 0,   # 摧毁敌方单位次数
            "heals": 0,             # 治疗成功次数
            "event_types": Counter(),  # 全部事件类型分布（学习用）
        }

    def _log_key_event(self, ev, et, my_ids, obs=None):
        """人类可读关键事件日志（采集/交付/进攻/击杀/生产/受伤）。
        旁路：异常绝不影响对局。"""
        try:
            msg = self._fmt_key_event(ev, et, my_ids, obs)
            if not msg:
                return
            ts = time.strftime("%H:%M:%S")
            line = f"[{ts} tick {ev.tick}] {msg}"
            with open(KEY_EVENTS_PATH, "a") as f:
                f.write(line + "\n")
            print(line, flush=True)
        except Exception:
            pass

    def _fmt_key_event(self, ev, et, my_ids, obs=None):
        """格式化关键事件；非关键事件返回 None。

        成功类用 emoji 高亮；失败/异常类用 ⚠ 灰字也进流，便于调试
        "为什么这 tick 没采到/没生出来/没打中"。
        """
        vals = ev.values or {}
        pos = f"@({ev.position[0]},{ev.position[1]})" if ev.position else ""
        who = _uid_short(ev.actor_id)
        tgt = _uid_short(ev.target_id)
        # ---- 成功类 ----
        if et == "HARVEST_SUCCEEDED":
            src = vals.get("source")
            tag = "掉落" if src == "DROPPED_CARGO" else "资源"
            return f"🟢 采集{tag} +{vals.get('amount', 1)} {pos}（{who}）"
        if et == "DEPOSIT_SUCCEEDED":
            return f"📦 交付 +{vals.get('amount', 1)} → Core（{who}）"
        if et == "CORE_SPAWN_SUCCEEDED":
            return f"🏭 生产单位 {pos}（{who}）"
        if et == "SHOT_HIT":
            return f"🎯 命中 {tgt} -{vals.get('damage', 1)} {pos}（{who}）"
        if et == "UNIT_DAMAGED" and vals.get("hp") == 0:
            if ev.target_id in my_ids:
                return f"☠️ 我方单位阵亡 {tgt} {pos}"
            owner = ""
            if obs is not None:
                for e in obs.enemies or []:
                    if e["uid"] == ev.target_id:
                        owner = f"（{e.get('owner') or '?'}）"
                        break
            return f"💥 摧毁敌方单位 {tgt}{owner} {pos}"
        if et == "DESTRUCTION_PARTICIPATION":
            kind = "Core" if ev.reason_code == "CORE" else "单位"
            return f"🏆 参与摧毁敌{kind} {tgt} {pos}"
        if et == "CORE_DAMAGED":
            return f"🚨 我方 Core 受伤 -{vals.get('damage', 1)} {pos}"
        if et == "CORE_DESTROYED":
            if ev.target_id in my_ids:
                return f"☠️ 我方 Core 被摧毁 {pos}（{ev.reason_code}）"
            return f"🏆 摧毁敌 Core {tgt} {pos}"
        if et == "CORE_RESOURCES_CAPTURED":
            cap = ev.core_resource_capture
            if cap:
                return f"💰 抢敌 Core 资源 +{cap.amount}（毁 {cap.destroyed}/余 {cap.available}）{pos}"
            return f"💰 抢敌 Core 资源 +{vals.get('amount', 1)} {pos}"
        if et == "CORE_RESOURCE_OVERFLOW_DESTROYED":
            return f"⚠ Core 资源溢出销毁 -{vals.get('amount', 1)} {pos}"
        if et == "WORKER_CARGO_DROPPED":
            return f"⚠ Worker 货物掉落 +{vals.get('amount', 1)} {pos}（{who}）"
        if et == "BEACON_PICKED_UP":
            return f"🚩 拾取信标 {pos}（{who}）"
        if et == "BEACON_DROPPED":
            return f"🚩 信标落地 {pos}"
        if et == "CORE_MOVE_STARTED":
            return f"🚚 Core 开始迁移 {pos}"
        if et == "CORE_MOVE_SUCCEEDED":
            return f"🚚 Core 迁移完成 {pos}"
        if et == "CORE_MOVE_CANCELLED":
            return f"⚠ Core 取消迁移"
        if et == "UNIT_HEALED":
            h = ev.healing
            if h:
                return f"💚 治疗 {who} +{h.amount}HP（耗 {h.cost} 资源）{pos}"
            return f"💚 治疗 {who} {pos}"
        if et == "CORE_HEAL_SUCCEEDED":
            h = ev.healing
            if h:
                return f"💚 Core 治疗 +{h.amount}HP（耗 {h.cost} 资源）"
            return f"💚 Core 治疗"
        if et == "CORE_RESPAWNED":
            return f"♻ Core 重生 {pos}"
        if et == "RESOURCE_DEPLETED":
            return f"⚠ 资源点耗尽 {pos}"
        # ---- 失败/异常类（灰字也进流，便于调试）----
        FAIL = {
            "HARVEST_FAILED": "采集失败",
            "DEPOSIT_FAILED": "交付失败",
            "SPAWN_FAILED": "生产失败",
            "RESPAWN_FAILED": "重生失败",
            "SHOT_MISSED": "射偏",
        }
        if et in FAIL:
            return f"⚠ {FAIL[et]} {pos}（{ev.reason_code or ''}）"
        if et.startswith("MOVE_BLOCKED"):
            return f"⚠ 移动被挡 {pos}（{ev.reason_code or et}）"
        if et in ("CORE_MOVE_FAILED", "CORE_MOVE_START_FAILED"):
            return f"⚠ Core 迁移受阻 {pos}（{ev.reason_code or et}）"
        return None

    def _log_attack_launch(self, tick, decisions):
        """出发进攻日志：单位 tag 从非战斗变为战斗（attack/track/raid 等）。"""
        try:
            for uid, d in (decisions or {}).items():
                tag = d.get("tag")
                prev = self._last_combat_tag.get(uid)
                if tag in COMBAT_TAGS and prev not in COMBAT_TAGS:
                    goal = d.get("goal")
                    gpos = f"→ @({goal[0]},{goal[1]})" if goal else ""
                    ts = time.strftime("%H:%M:%S")
                    line = (f"[{ts} tick {tick}] ⚔️ 出发进攻 {gpos} "
                            f"（{_uid_short(uid)} {tag}）")
                    with open(KEY_EVENTS_PATH, "a") as f:
                        f.write(line + "\n")
                    print(line, flush=True)
                self._last_combat_tag[uid] = tag
        except Exception:
            pass

    def _tick_stats(self, ev):
        """从官方事件更新运行统计（宽松匹配事件名，不中断主流程）。"""
        st = self.stats
        et = ev.event_type
        st["event_types"][et] += 1
        amt = ev.resource_amount
        if et == "HARVEST_SUCCEEDED":
            st["harvests"] += 1
            if amt: st["harvest_amount"] += amt
        elif et == "DEPOSIT_SUCCEEDED":
            st["deposits"] += 1
            if amt: st["deposited"] += amt
        elif et == "CORE_RESOURCES_CAPTURED":
            st["loot_captured"] += 1
            if amt: st["loot_amount"] += amt
        elif "DESTROYED" in et or "KILLED" in et:
            if "CORE" in et:
                st["cores_destroyed"] += 1
            else:
                st["units_destroyed"] += 1
        elif "HEAL_SUCCEEDED" in et:
            st["heals"] += 1

    def write_event(self, tick, obs, plan, events=None):
        """每 tick 一行结构化事件日志（agent_events.jsonl）。

        含单位级决策历史（decisions: uid8 -> {tag, goal}）——便于离线
        回放分析"某单位某 tick 为什么往那走"（结合 units 位置轨迹）。
        后端解析层读这个文件做查询/聚合（/api/events、/api/stats）。
        轻量 ~500B/行，一次 append，不影响对局。
        """
        try:
            dec = self.strategy._decisions if hasattr(self.strategy, "_decisions") else {}
            row = {
                "tick": tick, "ts": round(time.time(), 1),
                "pop": obs.population,
                "res": obs.core["resources"] if obs.core else None,
                "core": [obs.core["pos"][0], obs.core["pos"][1],
                         obs.core["hp"], obs.core["shield"]] if obs.core else None,
                "units": [[str(u["uid"])[:8], u["utype"], u["pos"][0], u["pos"][1],
                           u["hp"], u["cargo"], 1 if u["carries_beacon"] else 0]
                          for u in obs.units],
                "decisions": {str(u["uid"])[:8]: {
                    "tag": (dec.get(u["uid"]) or {}).get("tag"),
                    "goal": (dec.get(u["uid"]) or {}).get("goal")}
                    for u in obs.units},
                "enemies": len(obs.enemies),
                "events": [_ev_type(e) for e in (events or [])],
                "plan": ",".join(a for a, _ in (plan or {}).get("units", {}).values())
                        if plan else "",
            }
            with open(EVENTS_PATH, "a") as f:
                f.write(json.dumps(row, ensure_ascii=False) + "\n")
        except Exception as exc:
            print(f"[deploy] 事件日志写入失败（忽略）: {exc}", flush=True)

    def write_history(self, tick, obs):
        """心跳时把运行统计追加到 agent_history.jsonl（每 25 tick 一行）。

        数据供后续进化改进用：采集/经济/战斗/探索的时间序列。
        """
        try:
            st = self.stats
            row = {
                "tick": tick, "ts": round(time.time(), 1),
                "uptime": round(time.time() - self.started_at, 1),
                "pop": obs.population,
                "res": obs.core["resources"] if obs.core else None,
                "explored": len(self.strategy.mem.visited),
                "beacon": obs.beacon.get("status") if obs.beacon else None,
                "harvests": st["harvests"], "harvest_amount": st["harvest_amount"],
                "deposits": st["deposits"], "deposited": st["deposited"],
                "loot": st["loot_captured"], "loot_amount": st["loot_amount"],
                "cores_destroyed": st["cores_destroyed"],
                "units_destroyed": st["units_destroyed"],
                "heals": st["heals"],
                "reconnects": self.reconnects, "missed_ticks": self.missed_ticks,
            }
            with open(HISTORY_PATH, "a") as f:
                f.write(json.dumps(row, ensure_ascii=False) + "\n")
        except Exception as exc:
            print(f"[deploy] 历史写入失败（忽略）: {exc}", flush=True)

    def detect_gap(self, turn):
        """检测 tick 不连续（断线错过）：返回错过的 tick 数；首次进入返回 0。"""
        if self._last_tick is None:
            self._last_tick = turn.tick
            return 0
        gap = turn.tick - self._last_tick - 1
        self._last_tick = turn.tick
        if gap > 0:
            self.reconnects += 1
            self.missed_ticks += gap
            # 断线期间世界推进：清临时目标（保留永久障碍/资源记忆——
            # 障碍地形永不变，断线不使它失效）
            self.strategy.reset_transient()
            self._sent_dirs.clear()   # 中间 tick 的指令结果已丢失，方向不再可信
            self._core_dir = None
            self._sent_core_dir = None
            print(f"[deploy] ⚠ 检测到断线（错过 {gap} tick，累计 {self.missed_ticks}）"
                  f"，已清临时目标（保留世界记忆）", flush=True)
        return gap

    def translate_events(self, turn, obs=None):
        """turn.events 就是上一 Tick 的解析结果，直接转成模拟器格式。

        官方移动事件只带 actor_id + reason_code，不带方向，所以方向要从
        我们上一 tick 实际提交的指令里反查——否则 Memory 学不到障碍。
        obs: 可选（我方观察，用于关键事件日志的敌我判断）。
        """
        out = []
        current_ids = set()
        if obs is not None:
            for u in obs.units or []:
                current_ids.add(u["uid"])
            if obs.core:
                current_ids.add(obs.core["uid"])
        my_ids = current_ids | self._known_own_ids
        for ev in turn.events:
            et = ev.event_type
            actor = ev.actor_id
            self._tick_stats(ev)
            self._log_key_event(ev, et, my_ids, obs)
            self.recent_events.append({
                "tick": ev.tick, "type": et, "reason": ev.reason_code,
                "actor": str(ev.actor_id) if ev.actor_id is not None else None,
                "target": str(ev.target_id) if ev.target_id is not None else None,
                "target_is_own": ev.target_id in my_ids if ev.target_id else None,
                "pos": list(ev.position) if ev.position else None,
                "values": ev.values or None,
            })
            if et in self.MOVE_FAILED:
                direction = ((self._sent_core_dir or self._core_dir)
                             if et.startswith("CORE")
                             else self._sent_dirs.get(actor))
                if direction is None:
                    continue      # 方向不明就跳过：宁可不学，也不能学错格子
                blocked = {"type": "MOVE_BLOCKED", "obj_id": actor,
                           "direction": direction,
                           "reason": ev.reason_code,
                           "source_type": et}
                if ev.position is not None:
                    dx, dy = {"UP": (0, -1), "DOWN": (0, 1),
                              "LEFT": (-1, 0), "RIGHT": (1, 0)}[direction]
                    blocked["blocked_cell"] = (
                        ev.position[0] + dx, ev.position[1] + dy)
                out.append(blocked)
                if et.startswith("CORE"):
                    self._sent_core_dir = None
            elif et in self.MOVE_OK:
                d = {"type": "MOVED", "obj_id": actor}
                if ev.position is not None:
                    d["to"] = (ev.position[0], ev.position[1])
                out.append(d)
            else:
                # 其余事件 Memory 目前不消费，原样保留（含官方字段名）
                d = {"type": et, "reason": ev.reason_code}
                if actor is not None:
                    d["actor_id"] = actor
                if ev.target_id is not None:
                    d["target_id"] = ev.target_id
                out.append(d)
        if obs is not None:
            self._known_own_ids = current_ids
        return out

    def build_observation(self, turn):
        state = turn.state
        carrier_id = turn.beacon.carrier_id
        # 自己单位
        units = []
        vis_cells = set()
        obstacles = set(turn.obstacle_cells)
        for w in list(turn.workers) + list(turn.vanguards) + list(turn.rangers):
            v = w.view
            pos = (v.position[0], v.position[1])
            units.append({
                "uid": v.id, "utype": v.unit_type.value, "pos": pos,
                "hp": v.hp, "cargo": v.cargo or 0,
                # UnitView 没有 carries_beacon 字段，由 Beacon 的 carrier_id 推出
                "carries_beacon": carrier_id is not None and v.id == carrier_id,
            })
            r = VISION[v.unit_type.value]
            # 遮挡同时纳入记忆障碍：服务端本 tick 只报告"当前可见"障碍，
            # 口袋内圈障碍被外圈挡住时不在 turn.obstacle_cells 里；若只用
            # 它做遮挡，本地视野会"穿透"外圈看到内圈格 → 反证逻辑把刚记住
            # 的内圈障碍误删（webui 表现：离开视野就消失）。记忆障碍当遮挡
            # 后，内圈格不再被算作可见 → 不会被反证删除。
            vis_cells.update(visible_cells(pos[0], pos[1], r,
                                           lambda x, y: (x, y) in obstacles
                                           or (x, y) in self.strategy.mem.obstacles))
        # 自己 Core
        core_obs = None
        if turn.core is not None:
            c = turn.core.view
            pos = (c.position[0], c.position[1])
            core_obs = {"uid": c.id, "pos": pos, "hp": c.hp, "shield": c.shield,
                        "resources": turn.resources,
                        "state": c.state.value,
                        "migration": (c.move_direction.value, c.move_progress)
                        if c.move_direction is not None else None,
                        "destination": [c.destination[0], c.destination[1]]
                        if c.destination is not None else None,
                        "move_required_ticks": c.move_required_ticks,
                        "capacity": turn.resource_capacity,
                        "resource_space": turn.resource_space}
            vis_cells.update(visible_cells(pos[0], pos[1], VISION["CORE"],
                                           lambda x, y: (x, y) in obstacles
                                           or (x, y) in self.strategy.mem.obstacles))
        # 敌人（可见对象）
        enemies = []
        enemy_cores = []
        for obj in turn.visible_enemies:
            if isinstance(obj, CoreView):
                enemy_cores.append({"uid": obj.id, "pos": (obj.position[0], obj.position[1]),
                                    "hp": obj.hp, "shield": obj.shield,
                                    "owner": obj.owner_username,
                                    "state": obj.state.value,
                                    "destination": [obj.destination[0], obj.destination[1]]
                                    if obj.destination is not None else None,
                                    "move_required_ticks": obj.move_required_ticks})
            else:
                enemies.append({"uid": obj.id, "utype": obj.unit_type.value,
                                "pos": (obj.position[0], obj.position[1]), "hp": obj.hp,
                                "owner": getattr(obj, "owner_username", None)})
        # Beacon
        b = turn.beacon
        beacon = {"position": [b.position[0], b.position[1]]}
        if b.status is not None:
            beacon["status"] = b.status  # "GROUND" / "CARRIED"
            if b.carrier_id is not None:
                beacon["carrier_id"] = b.carrier_id
        else:
            beacon["status"] = "UNKNOWN"
        # 上一 tick 事件（先翻译，再刷新方向记录，保证反查用的是当时提交的方向）
        obs = Observation(
            player_id=0, tick=turn.tick,
            core=core_obs, units=units,
            enemies=enemies, enemy_cores=enemy_cores,
            resources=set((p[0], p[1]) for p in turn.resource_cells),
            obstacles=obstacles, beacon=beacon,
            population=state.population,
            visible_cells=vis_cells, prev_events=[],
            status=state.status.value if hasattr(state, "status")
            else ("RESPAWNING" if core_obs is None else "ACTIVE"),
            respawn_at_tick=getattr(state, "respawn_at_tick", None),
        )
        prev = self.translate_events(turn, obs=obs)
        obs.prev_events = prev
        if turn.core is not None and turn.core.view.move_direction is not None:
            self._core_dir = turn.core.view.move_direction.value
        return obs

    def apply_plan(self, turn, plan):
        # 记录本 tick 提交的移动方向：官方移动失败事件不带方向，下一 tick
        # 要靠这份记录才能推出被挡的格子。
        self._sent_dirs = {}
        for uid, (atype, args) in (plan.get("units") or {}).items():
            u = turn.unit(uid)
            if atype == "MOVE":
                self._sent_dirs[uid] = args["direction"]
                u.move(Direction(args["direction"]))
            elif atype == "HARVEST":
                u.harvest()
            elif atype == "DEPOSIT":
                u.deposit()
            elif atype == "SWEEP":
                u.sweep(Direction(args["direction"]))
            elif atype == "SHOOT":
                u.shoot_cell((args["expected_cell"][0], args["expected_cell"][1]))
            elif atype == "HEAL":
                u.heal()
            elif atype == "SELF_DESTRUCT":
                u.self_destruct()
            elif atype == "PICKUP_BEACON":
                u.pickup_beacon()
            elif atype == "DROP_BEACON":
                u.drop_beacon()
        ca = plan.get("core")
        if ca and turn.core is not None:
            atype, args = ca
            if atype == "SPAWN":
                turn.core.spawn(UnitType(args["unit_type"]))
            elif atype == "HEAL":
                turn.core.heal()
            elif atype == "REPAIR_SHIELD":
                turn.core.repair_shield()
            elif atype == "START_MOVE":
                self._core_dir = args["direction"]
                self._sent_core_dir = args["direction"]
                turn.core.start_move(Direction(args["direction"]))
            elif atype == "CANCEL_MOVE":
                self._sent_core_dir = None
                turn.core.cancel_move()
            elif atype == "PICKUP_BEACON":
                turn.core.pickup_beacon()
            elif atype == "DROP_BEACON":
                turn.core.drop_beacon()
            elif atype == "SELF_DESTRUCT":
                turn.core.self_destruct()

    # ------------------------------------------------------------------
    # 监控快照
    # ------------------------------------------------------------------
    def build_status(self, tick, obs, plan, player_status="ACTIVE",
                     carrier_known=False):
        """给 monitor.py 用的运行快照。只读内存，不做任何阻塞操作。

        故意不接收 SDK 的 Turn：这样本地模拟试运行也能产出同样格式的快照，
        面板不用等上线就能验证。
        """
        mem = self.strategy.mem
        core = obs.core
        cx, cy = core["pos"] if core else (obs.units[0]["pos"] if obs.units else (0, 0))
        x0, y0 = cx - VIEW_HALF, cy - VIEW_HALF
        span = VIEW_HALF * 2 + 1

        def inview(pos):
            return x0 <= pos[0] < x0 + span and y0 <= pos[1] < y0 + span

        # 地图用 Agent 自己的记忆（线上没有上帝视角，这就是它真实"知道"的世界）
        view = {
            "origin": [x0, y0], "size": span,
            "obstacles": [[c[0], c[1]] for c in mem.obstacles if inview(c)],
            "resources": [[c[0], c[1]] for c in mem.resources if inview(c)],
        }
        acts = Counter(a for a, _ in plan.get("units", {}).values()) if plan else Counter()
        p_units = (plan or {}).get("units") or {}

        def _unit_dir(u):
            """本 tick 提交的方向：从策略 plan 里取（MOVE 才有）。"""
            act = p_units.get(u["uid"])
            if act and act[0] == "MOVE":
                return str(act[1].get("direction", ""))
            return ""

        migration = self.strategy._normal_migration or {}
        step_direction = migration.get("step_direction")
        heading = (self.strategy._manual_core_direction or
                   migration.get("direction"))
        movement_mode = ("migration" if self.strategy._migration_enabled
                         else "stationed")
        if not self.strategy._migration_enabled:
            phase = "stationed"
        elif migration:
            if self.strategy._normal_migration_paused:
                phase = "follower_wait"
            elif step_direction and heading and step_direction != heading:
                phase = "detouring"
            else:
                phase = "migrating"
        elif self.strategy._manual_core_direction is not None:
            phase = "pending"
        else:
            phase = "stationed"
        beacon = obs.beacon
        return {
            "kind": "live",
            "tick": tick,
            "uptime": time.time() - self.started_at,
            "player_status": str(player_status),
            "reconnects": self.reconnects, "missed_ticks": self.missed_ticks,
            "submit_ms": round(self.submit_ms, 1),
            "core": None if not core else {
                "pos": list(core["pos"]), "hp": core["hp"], "shield": core["shield"],
                "resources": core["resources"], "capacity": core["capacity"],
                "resource_space": core.get("resource_space"),
                "state": core.get("state"),
                "migration": core["migration"],
                "destination": core.get("destination"),
                "move_required_ticks": core.get("move_required_ticks"),
            },
            "population": obs.population,
            "composition": dict(Counter(u["utype"] for u in obs.units)),
            # 下一单位造价（动态涨价：>20 人口每 5 个 +30%）
            "unit_costs": _unit_costs(obs.population),
            "respawn_at_tick": obs.respawn_at_tick,
            "beacon": {
                "position": beacon["position"], "status": beacon.get("status"),
                "mine": any(u["carries_beacon"] for u in obs.units),
                "carrier_known": bool(carrier_known),
            },
            "units": [{"uid": str(u["uid"])[:8], "type": u["utype"],
                       "pos": list(u["pos"]), "hp": u["hp"], "cargo": u["cargo"],
                       "beacon": u["carries_beacon"],
                       "dir": _unit_dir(u),
                       "tag": (self.strategy._decisions.get(u["uid"]) or {}).get("tag"),
                       "goal": (self.strategy._decisions.get(u["uid"]) or {}).get("goal")}
                      for u in obs.units],
            "enemies": [{"uid": str(e["uid"])[:8], "type": e["utype"],
                         "pos": list(e["pos"]), "hp": e["hp"],
                         "owner": e.get("owner")}
                        for e in obs.enemies],
            "enemy_cores": [{"pos": list(c["pos"]), "hp": c["hp"],
                             "shield": c["shield"], "owner": str(c["owner"]),
                             "state": c.get("state"),
                             "destination": c.get("destination"),
                             "move_required_ticks": c.get("move_required_ticks")}
                            for c in obs.enemy_cores],
            "view": view,
            "plan": {"core": plan.get("core")[0] if plan and plan.get("core") else None,
                     # 单位级指令：uid8 -> {act, dir}（前端按单位列出已下发指令）
                     "units": {str(uid)[:8]: {"act": act,
                                              "dir": args.get("direction", "")
                                              if isinstance(args, dict) else "",
                                              "tag": (self.strategy._decisions.get(uid) or {}).get("tag"),
                                              "goal": (self.strategy._decisions.get(uid) or {}).get("goal")}
                               for uid, (act, args) in
                               (plan or {}).get("units", {}).items()},
                     "summary": dict(acts)},
            "memory": {"obstacles": len(mem.obstacles), "resources": len(mem.resources),
                       "area": len(mem.area_seen), "unreachable": len(mem.unreachable),
                       "enemies": len(mem.last_enemy_pos)},
            "core_control": {
                "movement_mode": movement_mode,
                "mode": "fixed" if self.strategy._manual_core_direction else "auto",
                "direction": heading,
                "heading": heading,
                "step_direction": step_direction,
                "phase": phase,
                "active": self.strategy._normal_migration is not None,
                "pending": (self.strategy._migration_enabled and
                            self.strategy._manual_core_direction is not None and
                            self.strategy._normal_migration is None),
                "paused": bool(self.strategy._normal_migration_paused),
                "pause_reason": ("follower_envelope" if
                                 self.strategy._normal_migration_paused else None),
            },
            # 地图分层：已探索区域 + 记忆地形（战争迷雾用）
            "explored": sorted(map(list, mem.visited)),
            "known_obstacles": sorted(map(list, mem.obstacles)),
            "known_resources": sorted(map(list, mem.resources)),
            "events": list(self.recent_events)[-80:],
        }


# ----------------------------------------------------------------------
# 主循环
# ----------------------------------------------------------------------
def run_online(api_key, genes, tick_limit=None, resume=None, save_memory=None,
               verbose=False, status_path=LIVE_STATUS, db_path=None,
               stop_event=None, client_holder=None, command_lock=None):
    """正式世界主循环。

    stop_event: threading.Event，被 set 时在下一个 tick 边界干净退出
        （保存记忆后 return）。CLI 调用传 None，行为不变。
    client_holder: 带 ``.game`` 属性的简单对象，主循环进入 ``with
        ArenaHeroClient`` 后会把当前 client 写进去，供外部线程在
        shutdown 时调用 ``client_holder.game.close()`` 让阻塞在
        ``turns()`` 的循环及时脱出。CLI 调用传 None，行为不变。
    """
    strat = HeuristicStrategy(genes=genes, bounds=None)
    if resume:
        try:
            import json as _json
            strat.load_memory(_json.load(open(resume)))
            print(f"[deploy] 已从 {resume} 恢复记忆 "
                  f"(障碍 {len(strat.mem.obstacles)} 格, 资源 {len(strat.mem.resources)} 点, "
                  f"区域 {len(strat.mem.area_seen)} 格)", flush=True)
        except Exception as exc:
            print(f"[deploy] 记忆恢复失败（忽略，从零开始）: {exc}", flush=True)

    def _save_memory(reason=""):
        if not save_memory:
            return
        try:
            atomic_write_json(save_memory, strat.mem.to_dict())
            print(f"[deploy] 记忆已保存 {save_memory} {reason}", flush=True)
        except Exception as exc:
            print(f"[deploy] 记忆保存失败: {exc}", flush=True)

    # SIGTERM/SIGINT 时保存记忆再退出（优雅停机）
    def _on_exit(signum, _frame):
        _save_memory(f"(signal {signum})")
        raise SystemExit(0)
    try:
        import signal as _sig
        _sig.signal(_sig.SIGTERM, _on_exit)
        _sig.signal(_sig.SIGINT, _on_exit)
    except (ValueError, OSError):
        pass  # 非主线程等无法注册信号

    adapter = LiveAdapter(strat)
    # 人工指令通道：commands.json（前端 monitor API 写入，agent 轮询应用）
    commands_path = Path(save_memory).with_name("commands.json") if save_memory else None
    applied_cmds = set()      # 已应用的指令 id（不重复应用）
    _cmd_mtime = 0
    command_lock = command_lock or threading.Lock()

    def _apply_commands(force=False, tick=999999999):
        """读取并应用未执行的指令。返回新应用数。

        tick: 当前对局 tick（写资源记忆用；启动时 force 无 turn，用大值
        表示"新鲜"，之后由视野/TTL 自然管理）。
        """
        with command_lock:
            return _apply_commands_locked(force=force, tick=tick)

    def _apply_commands_locked(force=False, tick=999999999):
        nonlocal _cmd_mtime
        if commands_path is None or not commands_path.exists():
            return 0
        try:
            mt = commands_path.stat().st_mtime
            if not force and mt == _cmd_mtime:
                return 0
            data = json.loads(commands_path.read_text(encoding="utf-8") or "{}")
            cmds = data.get("commands", [])
            n = 0
            for c in cmds:
                cid = c.get("id")
                if not cid or cid in applied_cmds:
                    continue
                ctype = c.get("type")
                pos = tuple(c.get("pos") or ())
                try:
                    if ctype == "mark_resource" and len(pos) == 2:
                        # 直接写入资源记忆：走正常 TTL/采集/视野反证流程
                        strat.mem.resources[pos] = tick
                        print(f"[deploy] 指令: 标记资源 {pos}", flush=True)
                    elif ctype == "mark_obstacle" and len(pos) == 2:
                        strat.mem.obstacles.add(pos)
                        strat.mem.new_obstacles = True
                        print(f"[deploy] 指令: 标记障碍 {pos}", flush=True)
                    elif ctype == "goto" and len(pos) == 2 and c.get("uid"):
                        strat._manual_goto[c["uid"]] = pos
                        print(f"[deploy] 指令: 单位 {c['uid'][:8]} → {pos}", flush=True)
                    elif ctype == "cancel_goto":
                        strat._manual_goto.pop(c.get("uid"), None)
                        print(f"[deploy] 指令: 取消单位 {str(c.get('uid'))[:8]} 目标", flush=True)
                    elif ctype == "core_migrate":
                        direction = c.get("direction")
                        strat.set_manual_core_migration(direction)
                        print(f"[deploy] 指令: Core 手动迁移 {direction}", flush=True)
                    elif ctype == "core_auto":
                        strat.resume_auto_core_migration()
                        print("[deploy] 指令: Core 恢复自动迁移", flush=True)
                    elif ctype == "core_station":
                        strat.station_core()
                        print("[deploy] 指令: Core 驻扎", flush=True)
                    elif ctype == "core_enable_migration":
                        strat.enable_core_migration()
                        print("[deploy] 指令: Core 恢复迁移模式", flush=True)
                    elif ctype == "clear_marks":
                        # 清空全部人工标记（等价于移除标记过的格）
                        strat.mem.resources.clear()
                        strat.mem.obstacles.clear()
                        strat.mem.unreachable.clear()
                        strat.mem.isolated.clear()
                        strat.mem.new_obstacles = True
                        print(f"[deploy] 指令: 清空标记", flush=True)
                    elif ctype == "remove_mark" and len(pos) == 2:
                        # 取消单个标记：从资源/障碍记忆移除该格（视野重新看到会
                        # 自然重记——那是事实，不压制）
                        strat.mem.resources.pop(pos, None)
                        if pos in strat.mem.obstacles:
                            strat.mem.obstacles.discard(pos)
                            strat.mem.new_obstacles = True
                        print(f"[deploy] 指令: 取消标记 {pos}", flush=True)
                    else:
                        print(f"[deploy] 指令 {cid} 类型无效: {ctype}", flush=True)
                except Exception as exc:
                    print(f"[deploy] 指令 {cid} 应用失败: {exc}", flush=True)
                # 无论成功失败都标记已处理：失败不重试（用户可重发），
                # 避免重启后 force 重放旧指令导致标记"复活"
                applied_cmds.add(cid)
                n += 1
            _cmd_mtime = mt
            # 已处理的指令从文件移除（goto 未完成保留，重启后可重新下发）
            done = getattr(strat, "_manual_done", None)
            if done:
                strat._manual_done = []
            todo = [c for c in cmds
                    if c.get("id") not in applied_cmds
                    or (c.get("type") == "goto"
                        and c.get("uid") not in (done or []))]
            if len(todo) != len(cmds):
                atomic_write_json(commands_path, {"commands": todo})
            return n
        except Exception as exc:
            print(f"[deploy] 指令应用失败: {exc}", flush=True)
            return 0

    _apply_commands(force=True)
    # 线上数据存储（SQLite 全量落盘；失败不干扰对局）
    db = None
    if db_path:
        try:
            from dbstore import DBStore
            db = DBStore(db_path)
        except Exception as exc:
            print(f"[deploy] DB 初始化失败（忽略，仅 JSONL）: {exc}", flush=True)
    # 外层保护：SDK 的 turns() 通常无限重连；若异常退出则重建 client 重试
    while True:
        if stop_event is not None and stop_event.is_set():
            _save_memory("(shutdown)")
            return
        try:
            with ArenaHeroClient(api_key=api_key) as game:
                if client_holder is not None:
                    client_holder.game = game
                for turn in game.turns():
                    if stop_event is not None and stop_event.is_set():
                        _save_memory("(shutdown)")
                        return
                    if tick_limit and turn.tick > tick_limit:
                        print(f"[deploy] 达到 tick 上限 {tick_limit}，退出", flush=True)
                        _save_memory()
                        return
                    adapter.detect_gap(turn)
                    obs = adapter.build_observation(turn)
                    # Core 重生：清临时目标（保留世界记忆——障碍永久有效）
                    if obs.core is not None and turn.state.events:
                        if any(getattr(ev, "event_type", "") == "CORE_DESTROYED"
                               for ev in turn.state.events):
                            strat.reset_transient()
                            print(f"[deploy] tick {turn.tick}: Core 重生，"
                                  f"已清临时目标（保留障碍/资源记忆 "
                                  f"{len(strat.mem.obstacles)} 格）", flush=True)
                    if verbose:
                        # 诊断：每 tick 打印核心状态
                        units = [(u["uid"], u["utype"], u["pos"], u["cargo"])
                                 for u in obs.units]
                        print(f"[dbg] tick {turn.tick}: core={obs.core['pos'] if obs.core else 'DEAD'}"
                              f" res={obs.core['resources'] if obs.core else 0}"
                              f" vis_res={len(obs.resources)} units={units}",
                              flush=True)
                        # 资源诊断：每 25 tick 输出视野内资源分布（含距 Core 距离）
                        if turn.tick % 25 == 0 and obs.core:
                            cp = obs.core["pos"]
                            rl = sorted(
                                (abs(r[0] - cp[0]) + abs(r[1] - cp[1]), r)
                                for r in obs.resources)
                            near = sum(1 for d, _ in rl if d <= 22)
                            memn = len(strat.mem.resources)
                            print(f"[resdbg] tick {turn.tick}: vis={len(rl)} near22={near}"
                                  f" mem={memn}"
                                  f" res={[(d, list(r)) for d, r in rl[:10]]}",
                                  flush=True)
                    # 人工指令轮询：每 tick 检查（线上 15s/tick，25 tick ≈ 6 分钟
                    # 感知延迟太长；每 tick = 最多 15 秒生效）
                    _apply_commands(tick=turn.tick)
                    plan = None
                    try:
                        plan = strat.decide(obs)
                        adapter.apply_plan(turn, plan)
                        # 出发进攻日志（tag 非战斗→战斗）
                        adapter._log_attack_launch(
                            turn.tick,
                            getattr(strat, "_decisions", {}) or {})
                        t_sub = time.time()
                        turn.submit()
                        dt_sub = time.time() - t_sub
                        adapter.submit_ms = dt_sub * 1000
                        if dt_sub > 2.0:
                            print(f"[deploy] tick {turn.tick} 提交慢: {dt_sub:.1f}s",
                                  flush=True)
                    except Exception as exc:
                        # 单个 tick 失败不退出：记录后继续（SDK 对提交有重试，
                        # 这里只兜底策略/提交的意外异常）
                        # 本 tick 没提交成功 → 方向记录作废，避免下一 tick 误判障碍
                        adapter._sent_dirs.clear()
                        print(f"[deploy] tick {turn.tick} 处理失败: {exc}", flush=True)
                    if status_path:
                        try:
                            write_status(status_path, adapter.build_status(
                                turn.tick, obs, plan,
                                player_status=turn.state.status,
                                carrier_known=turn.beacon.carrier_id is not None))
                        except Exception:
                            pass   # 监控是旁路，绝不影响对局
                        adapter.write_event(turn.tick, obs, plan,
                                            turn.events)
                    if db is not None:
                        # SQLite 全量落盘：ticks + units + events（单事务）
                        # 注意：官方 turn.events 是上一 tick 的解析结果，ev.tick
                        # 与 turn.tick 存在偏移 → 过滤条件需宽容（PK 幂等去重）
                        try:
                            evs = [
                                (e["type"], e.get("reason"), e.get("actor"),
                                 tuple(e["pos"]) if e.get("pos") else None,
                                 (e.get("values") or {}).get("amount"))
                                for e in adapter.recent_events
                                if turn.tick - (e.get("tick") or 0) <= 1]
                            db.write_tick_full(
                                turn.tick, time.time(), obs, plan,
                                adapter.submit_ms, evs,
                                strat.mem,
                                getattr(strat, "_decisions", {}) or {})
                            if turn.tick % 25 == 0:
                                db.write_mem_snap(turn.tick, strat.mem)
                        except Exception as exc:
                            print(f"[deploy] DB 写入失败（忽略）: {exc}",
                                  flush=True)
                    _save_memory()
                    if turn.tick % 25 == 0:
                        print(f"[deploy] tick {turn.tick}: pop={obs.population} "
                              f"res={obs.core['resources'] if obs.core else 0} "
                              f"reconnects={adapter.reconnects}", flush=True)
                        adapter.write_history(turn.tick, obs)
        except KeyboardInterrupt:
            _save_memory()
            raise
        except Exception as exc:
            _save_memory("(connection error)")
            # 打印底层原因（如 SDK 校验失败的具体字段），便于诊断连接层问题
            cause = getattr(exc, "__cause__", None)
            if cause is not None:
                print(f"[deploy] 连接层异常: {exc} | 底层: {cause}",
                      flush=True)
            else:
                print(f"[deploy] 连接层异常: {exc}，10s 后重试（策略记忆保留）",
                      flush=True)
            time.sleep(10)
            if stop_event is not None and stop_event.is_set():
                return
            continue
        return  # turns() 正常退出（仅当 client 关闭）


def run_local(genes, ticks=800, seed=42, players=8, status_path=None,
              tick_delay=0.0):
    """本地模拟器试运行同一基因，验证部署前行为。

    带 --status 时按正式世界同样的格式逐 tick 写快照，于是不用等上线，
    monitor.py 的「线上 Agent」页签就能先验证一遍。
    """
    from ahsim.game import Game
    from strategies.randombot import RandomBot
    from strategies.heuristic import HeuristicStrategy as SimStrategy

    class _Recording:
        """包一层记录最近一次决策，供状态快照的「本 tick 决策」面板展示。"""

        def __init__(self, inner):
            self.inner = inner
            self.last_plan = None

        def decide(self, obs):
            self.last_plan = self.inner.decide(obs)
            return self.last_plan

        def __getattr__(self, k):
            return getattr(self.inner, k)

    bounds = (-128, 127, -128, 127)
    me = _Recording(SimStrategy(genes=genes, bounds=bounds))
    strategies = {0: me}
    for i in range(1, players):
        strategies[i] = (RandomBot(seed=i * 7 + 1) if i % 3 == 0
                         else SimStrategy(bounds=bounds))
    g = Game(strategies=strategies, size=256, seed=seed, max_ticks=ticks)

    if not status_path:
        g.run()
    else:
        adapter = LiveAdapter(me.inner)
        for _ in range(ticks):
            g.tick += 1
            g.step()
            p = g.players[0]
            if p.core is None and not p.units:
                continue
            obs = g.build_observation(p)
            adapter.recent_events.extend(
                {"tick": g.tick, "type": e["type"], "reason": e.get("reason"),
                 "pos": list(e["at"]) if isinstance(e.get("at"), tuple) else None,
                 "values": None}
                for e in g.last_events if e.get("player", 0) == 0)
            write_status(status_path, adapter.build_status(
                g.tick, obs, me.last_plan))
            if tick_delay:
                time.sleep(tick_delay)

    st = g.results()[0]
    print(f"[local] {ticks} ticks: pop={st['final_population']} "
          f"harvest={st['harvested']} dmg={st['damage_dealt']} "
          f"alive={st['alive']}")


def main():
    ap = argparse.ArgumentParser(description="Arena Hero 正式世界部署")
    ap.add_argument("--genes", required=True, help="基因 JSON 路径")
    ap.add_argument("--local", action="store_true", help="本地模拟试运行")
    ap.add_argument("--ticks", type=int, default=800, help="本地运行的 tick 数")
    ap.add_argument("--players", type=int, default=8, help="本地试运行的玩家数")
    ap.add_argument("--tick-delay", type=float, default=0.0,
                    help="本地试运行每 tick 停顿秒数（配合监控面板慢放观察）")
    ap.add_argument("--resume", default=None, help="恢复记忆 JSON（重启后不迷路）")
    ap.add_argument("--save-memory", default=None,
                    help="定期把记忆写入此路径（配合 --resume 崩溃恢复）")
    ap.add_argument("--tick-limit", type=int, default=None,
                    help="运行到该 tick 后退出（观察期用）")
    ap.add_argument("--api-key", default=os.environ.get("ARENA_HERO_API_KEY", ""))
    ap.add_argument("--verbose", action="store_true", help="每 tick 打印诊断日志")
    ap.add_argument("--status", default=LIVE_STATUS,
                    help="每 tick 写运行状态快照供 monitor.py 读取；传空串关闭")
    ap.add_argument("--db", default=None,
                    help="SQLite 全量数据落盘路径（如 results/agent.db）；默认随 --status 目录")
    args = ap.parse_args()

    db_path = args.db
    if db_path is None and args.status:
        db_path = os.path.join(os.path.dirname(args.status) or ".", "agent.db")

    genes = load_genes(args.genes)
    if args.local:
        run_local(genes, ticks=args.ticks, players=args.players,
                  status_path=args.status or None, tick_delay=args.tick_delay)
        return
    if not args.api_key:
        print("缺少 API key：设置 ARENA_HERO_API_KEY 或 --api-key")
        sys.exit(1)
    run_online(args.api_key, genes, tick_limit=args.tick_limit,
               resume=args.resume, save_memory=args.save_memory,
               verbose=args.verbose, status_path=args.status or None,
               db_path=db_path)


if __name__ == "__main__":
    main()