server.py

import math
import sys
import io
import hashlib
import json
import os
import socket
import sqlite3
import time

# ── Early arg parse — must run before world_generation import ────────────────
# Sets TOKENTORCH_SEED so GenerationConstants.WORLD_SEED reads the right value
# when world_generation is imported below.
if __name__ == '__main__':
    import argparse as _ap
    _p = _ap.ArgumentParser(add_help=False)
    _p.add_argument('--world', default='default')
    _p.add_argument('--seed',  type=int, default=876567)
    _p.add_argument('--world-size', type=int, default=32768,
                    help='World size in blocks (32768=32km, 65536=64km)')
    _early, _ = _p.parse_known_args()
    os.environ['TOKENTORCH_SEED']  = str(_early.seed)
    os.environ['TOKENTORCH_WORLD'] = _early.world
    os.environ['TOKENTORCH_WORLD_SIZE'] = str(_early.world_size)
    os.environ['TOKENTORCH_HEADLESS'] = '1'

from node_router import NodeRouter
from tokengate.operations_coordinator import OperationsCoordinator
from tokengate.token_system import task_token_guard
from constants import GenerationConstants, ServerConstants
from physics import PhysicsEngine
from tokengate.unhashable_checker import HashPolicy
from world_generation import (
    GLOBAL_HEIGHTMAP, GLOBAL_BIOME_MAP,
    compute_chunk_hash, fetch_chunk_threaded,
    finished_chunks_queue, set_active_database
)
from monster_manager import MonsterManager

# Fix Windows console encoding for Unicode arrows/emoji in comments/tracebacks
if sys.stdout and hasattr(sys.stdout, 'buffer'):
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
if sys.stderr and hasattr(sys.stderr, 'buffer'):
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')

SAVES_DIR = "saves"
os.makedirs(SAVES_DIR, exist_ok=True)
os.makedirs(os.path.join(SAVES_DIR, "worlds"), exist_ok=True)

# Overridden in __main__ based on --world arg before start_server() is called
LEDGER_PATH = os.path.join(SAVES_DIR, "worlds", "hosted_default_ledger.db")
STATS_PATH = os.path.join(SAVES_DIR, "server_player_stats.jsonl")
SUSPECT_LOG = os.path.join(SAVES_DIR, "suspect_log.jsonl")

HOST = '0.0.0.0'
PORT = 5555
STATUS_PORT = PORT + 1   # 5556 — menu and AI agents poll this

# Mutable boot status — updated as each phase completes.
# AI agents report their kernel compilation back through this same socket.
_server_status: dict = {
    "phase":           "booting",   # booting → world_ready → outbound_ready
    "seed":            876567,
    "world_size":      32768,
    "villager_ready":  False,       # set True by villager_agent after kernels compile
    "nodes_ready":     False,       # set True by llm_node after startup
}

consts = GenerationConstants

const = ServerConstants

# --- GLOBAL STATE DICTIONARIES ---
player_profiles = {}
world_changes = {}
water_changes = {}
active_chunk_hosts = {}
active_monsters = {}  # NEW: Tracks all live AI monsters

water_chunk_last_saved: dict = {}
WATER_CHUNK_SAVE_COOLDOWN = 10.0

pending_block_writes = []
pending_water_writes = []

chunk_hash_registry: dict[tuple, str] = {}
dirty_server_chunks: set[tuple] = set()

RESPAWN_INTERVAL = 60.0

def _log_suspect(secure_id: str, cid: str, chunk: tuple, reason: str,
                 is_flying: bool, current_time: float):
    entry = {
        "timestamp": current_time,
        "secure_id": secure_id,
        "cid": cid,
        "chunk": list(chunk),
        "reason": reason,
        "is_flying": is_flying,
    }
    try:
        with open(SUSPECT_LOG, "a") as f:
            f.write(json.dumps(entry) + "\n")
        print(f"[SUSPECT] {cid} flagged: {reason} at chunk {chunk} | flying={is_flying}")
    except Exception as e:
        print(f"[SUSPECT LOG ERROR] {e}")


def _precompute_spawn_region_hashes():
    spawn_x = int(consts.WORLD_AREA / 2)
    spawn_z = int(consts.WORLD_AREA / 2)
    cx_center = spawn_x // consts.CHUNK_SIZE
    cz_center = spawn_z // consts.CHUNK_SIZE

    WARM_RADIUS = 2
    queued = 0
    for nx in range(cx_center - WARM_RADIUS, cx_center + WARM_RADIUS + 1):
        for ny in range(-2, 4):
            for nz in range(cz_center - WARM_RADIUS, cz_center + WARM_RADIUS + 1):
                fetch_chunk_threaded(nx, ny, nz, {}, {})
                queued += 1

    print(f"[SERVER] Queued {queued} spawn-region chunks for hash pre-computation.")
    return queued


def _drain_precomputed_hashes(timeout: float = 30.0):
    deadline = time.perf_counter() + timeout
    drained = 0
    while time.perf_counter() < deadline:
        if not finished_chunks_queue.empty():
            try:
                cx, cy, cz, _instances, volume, _water = finished_chunks_queue.get()
                chunk = (cx, cy, cz)
                chunk_hash_registry[chunk] = compute_chunk_hash(cx, cy, cz, volume)
                drained += 1
            except Exception as e:
                print(f"[SERVER] Hash pre-compute error: {e}")
        else:
            time.sleep(0.005)
    print(f"[SERVER] Pre-computed {drained} structural hashes for spawn region.")


def _compact_ledger() -> None:
    """Reclaim fragmented SQLite pages and rebuild the spatial index.

    The ledger schema uses PRIMARY KEY (x, y, z) with ON CONFLICT UPDATE,
    so rows are already unique per coordinate — there's no duplicate data
    to collapse.  What accumulates over time is page fragmentation from
    updates and WAL overhead.  VACUUM rewrites the file compactly;
    ANALYZE refreshes the query planner statistics so the spatial index
    stays fast on next boot.

    Called synchronously on shutdown — no guard needed since the server
    loop has already stopped at this point.
    """
    if not os.path.exists(LEDGER_PATH):
        return
    try:
        before = os.path.getsize(LEDGER_PATH) // 1024
        with sqlite3.connect(LEDGER_PATH, timeout=10.0) as conn:
            conn.execute('PRAGMA journal_mode=DELETE;')  # close WAL before VACUUM
            conn.execute('VACUUM;')
            conn.execute('ANALYZE;')
        after = os.path.getsize(LEDGER_PATH) // 1024
        print(f'[Ledger] Compacted {before} KB → {after} KB')
    except Exception as exc:
        print(f'[Ledger] Compaction failed: {exc}')


def _init_db():
    with sqlite3.connect(LEDGER_PATH) as conn:
        conn.execute("PRAGMA journal_mode=WAL;")
        conn.execute("""
            CREATE TABLE IF NOT EXISTS world_blocks (
                x INTEGER, y INTEGER, z INTEGER,
                block_id INTEGER, water_vol REAL, last_updated REAL,
                PRIMARY KEY (x, y, z)
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_spatial ON world_blocks(x, y, z);")


def _load_player_profiles():
    if os.path.exists(STATS_PATH):
        print("Restoring Player Profiles...")
        with open(STATS_PATH, "r") as f:
            for line in f:
                try:
                    data = json.loads(line)
                    for k, v in data.items():
                        player_profiles[k] = v
                except Exception:
                    pass


def _load_world_ledger():
    _init_db()
    if os.path.exists(LEDGER_PATH):
        print("Reconstructing world from SQLite ledger...")
        with sqlite3.connect(LEDGER_PATH) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT x, y, z, block_id, water_vol FROM world_blocks")
            for x, y, z, block_id, water_vol in cursor.fetchall():
                if block_id is not None:
                    world_changes[(x, y, z)] = block_id
                if water_vol is not None and water_vol > 0:
                    water_changes[(x, y, z)] = water_vol


@task_token_guard(
    operation_type='save_player',
    tags={'weight': 'medium',
          'storage_speed': 'SLOW',
          "hash_policy": HashPolicy.NONE}
)
def _save_player_profile(secure_id, cdata=None):
    if secure_id in player_profiles:
        if cdata:
            player_profiles[secure_id]['x'] = cdata['x']
            player_profiles[secure_id]['y'] = cdata['y']
            player_profiles[secure_id]['z'] = cdata['z']
            player_profiles[secure_id]['r'] = cdata['r']
        with  open(STATS_PATH, "a") as f:
            f.write(json.dumps({secure_id: player_profiles[secure_id]}) + "\n")


@task_token_guard(
    operation_type='ledger_write',
    tags={'weight': 'medium',
          'storage_speed': 'MODERATE',
          "hash_policy": HashPolicy.NONE}
)
def _write_ledger_entry(blocks_batch=None, water_batch=None, chunk_water_dict=None, chunk_coord=None):
    try:
        with sqlite3.connect(LEDGER_PATH, timeout=3.0) as conn:
            if blocks_batch:
                conn.executemany("""
                    INSERT INTO world_blocks (x, y, z, block_id, last_updated)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(x, y, z) DO UPDATE SET 
                        block_id=excluded.block_id, last_updated=excluded.last_updated;
                """, blocks_batch)
            if water_batch:
                conn.executemany("""
                    INSERT INTO world_blocks (x, y, z, water_vol, last_updated)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(x, y, z) DO UPDATE SET 
                        water_vol=excluded.water_vol, last_updated=excluded.last_updated;
                """, water_batch)
            if chunk_water_dict is not None and chunk_coord is not None:
                cx, cy, cz = chunk_coord
                min_x, max_x = cx * 16, (cx + 1) * 16
                min_y, max_y = cy * 16, (cy + 1) * 16
                min_z, max_z = cz * 16, (cz + 1) * 16
                conn.execute("""
                    UPDATE world_blocks SET water_vol = 0.0
                    WHERE x >= ? AND x < ? AND y >= ? AND y < ? AND z >= ? AND z < ?
                """, (min_x, max_x, min_y, max_y, min_z, max_z))
                ts = time.perf_counter()
                records = [
                    (int(coord.split(',')[0]), int(coord.split(',')[1]), int(coord.split(',')[2]), vol, ts)
                    for coord, vol in chunk_water_dict.items()
                ]
                conn.executemany("""
                    INSERT INTO world_blocks (x, y, z, water_vol, last_updated)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(x, y, z) DO UPDATE SET 
                        water_vol=excluded.water_vol, last_updated=excluded.last_updated;
                """, records)
        conn.commit()
    except Exception as e:
        print(f"Failed to batch-write to SQLite ledger: {e}")


def _generate_player_id(ip_address):
    salt = "TokenTorch_Secure_Salt_2026"
    return hashlib.sha256(f"{ip_address}_{salt}".encode()).hexdigest()[:16]


def _respawn_at_hash(secure_id, death_count):
    seed_string = f"{secure_id}_{death_count}"
    spawn_hash = int(hashlib.md5(seed_string.encode()).hexdigest(), 16)
    for offset in range(100):
        probe_x = (spawn_hash + offset * 73856) % consts.WORLD_AREA
        probe_z = (spawn_hash + offset * 19349) % consts.WORLD_AREA
        surface_y = float(GLOBAL_HEIGHTMAP[int(probe_z), int(probe_x)])
        biome_val = GLOBAL_BIOME_MAP[int(probe_z), int(probe_x)]
        if consts.SEA_LEVEL < surface_y <= 45 and biome_val < 0.8:
            return probe_x, surface_y + 2.0, probe_z
    center = consts.WORLD_AREA / 2
    return center, float(GLOBAL_HEIGHTMAP[int(center), int(center)]) + 2.0, center


def _run_status_server():
    """
    Lightweight UDP status endpoint on STATUS_PORT (5556).
    Runs in a daemon thread — menu and AI scripts poll this to know
    when each boot phase is complete before launching the next process.

    Query:  {"type": "status_query"}
    Update: {"type": "status_update", "villager_ready": true}
    """
    status_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    status_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    status_sock.bind(('0.0.0.0', STATUS_PORT))
    status_sock.settimeout(1.0)
    print(f'[Status] Endpoint live on port {STATUS_PORT}')

    while True:
        try:
            data, addr = status_sock.recvfrom(512)
            try:
                packet = json.loads(data.decode())
            except Exception:
                continue

            ptype = packet.get('type')

            if ptype == 'status_query':
                status_sock.sendto(json.dumps(_server_status).encode(), addr)

            elif ptype == 'status_update':
                if packet.get('villager_ready'):
                    _server_status['villager_ready'] = True
                    print('[Status] Villager kernels compiled — client clear to launch')
                if packet.get('nodes_ready'):
                    _server_status['nodes_ready'] = True
                    print('[Status] LLM nodes registered')

        except socket.timeout:
            continue
        except Exception as exc:
            print(f'[Status] Error: {exc}')


def start_server():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((HOST, PORT))
    sock.setblocking(False)

    coordinator = OperationsCoordinator()
    coordinator.start()

    # ── Declare shared state BEFORE anything that references it ──────
    clients = {}
    engine  = PhysicsEngine()


    # ── MonsterManager receives a live reference to clients ──────────
    monster_mgr = MonsterManager(
        sock            = sock,
        clients         = clients,
        heightmap       = GLOBAL_HEIGHTMAP,
        biome_map       = GLOBAL_BIOME_MAP,
        player_profiles = player_profiles,
        respawn_func    = _respawn_at_hash,
        world_changes   = world_changes,
    )
    router = NodeRouter(sock, clients, monster_mgr.active_monsters, GLOBAL_HEIGHTMAP)
    _load_player_profiles()
    _load_world_ledger()

    # Start status endpoint before precompute so menu can see "booting" phase
    import threading as _threading
    _status_thread = _threading.Thread(
        target=_run_status_server, daemon=True, name='status_server')
    _status_thread.start()

    print("[SERVER] Pre-warming spawn region...")
    _precompute_spawn_region_hashes()
    _drain_precomputed_hashes(timeout=30.0)

    # World is fully generated — AI agents and menu can now read seed/world_size
    _server_status['phase'] = 'world_ready'
    print(f'[Status] Phase → world_ready  (seed={_server_status["seed"]}  '
          f'world_size={_server_status["world_size"]})')
    print(f"UDP Game Server running on {HOST}:{PORT}")

    NETWORK_TICK_RATE = 1.0 / 30.0
    last_network_tick = 0.0

    def process_physics(dt):
        monster_mgr.process_ai(dt)

    def process_inbound():
        current_time = time.perf_counter()
        while True:
            try:
                data, addr = sock.recvfrom(2048)
                packet = json.loads(data.decode('utf-8'))
                cid = packet.get('id')
                ip_addr = addr[0]

                if cid and cid not in clients:
                    secure_id = _generate_player_id(ip_addr)
                    if secure_id not in player_profiles:
                        player_profiles[secure_id] = {'deaths': 0, 'kills': 0}

                    print(f"New player {cid} connected [SECURE_ID: {secure_id}]")
                    clients[cid] = {
                        'addr': addr, 'vy': 0.0, 'hp': 100.0,
                        'secure_id': secure_id, 'last_seen': current_time,
                        'x': 0, 'y': 0, 'z': 0, 'r': 0,
                        'last_x': 0, 'last_y': 0, 'last_z': 0, 'last_r': 0,
                        'invulnerable_until': 0.0
                    }

                    if 'x' in player_profiles[secure_id]:
                        sock.sendto(json.dumps({
                            "type": "respawn",
                            "x": player_profiles[secure_id]['x'],
                            "y": player_profiles[secure_id]['y'],
                            "z": player_profiles[secure_id]['z'],
                            "hp": 100.0
                        }).encode('utf-8'), addr)

                    changes_list = list(world_changes.items())
                    water_list = list(water_changes.items())

                    for i in range(0, len(changes_list), 200):
                        batch = changes_list[i:i + 200]
                        sync_packet = {
                            "type": "world_sync",
                            "changes": {f"{k[0]},{k[1]},{k[2]}": v for k, v in batch}
                        }
                        try:
                            sock.sendto(json.dumps(sync_packet).encode('utf-8'), addr)
                        except Exception as e:
                            pass

                    for i in range(0, len(water_list), 200):
                        batch = water_list[i:i + 200]
                        sync_packet = {
                            "type": "world_sync",
                            "water_changes": {f"{k[0]},{k[1]},{k[2]}": v for k, v in batch}
                        }
                        try:
                            sock.sendto(json.dumps(sync_packet).encode('utf-8'), addr)
                        except Exception as e:
                            pass

                    spawn_px = player_profiles[secure_id].get('x', consts.WORLD_AREA / 2)
                    spawn_pz = player_profiles[secure_id].get('z', consts.WORLD_AREA / 2)
                    monster_mgr.spawn_horde_around(spawn_px, spawn_pz)

                if cid in clients:
                    clients[cid]['last_seen'] = current_time

                # ── Top-level packet dispatch ──────────────────────────────────────────────
                ptype = packet.get("type")
                router.handle_packet(packet, addr, current_time)

                if ptype == "monster_spawn_request" and cid in clients:
                    req_x = float(packet.get("x", consts.WORLD_AREA / 2))
                    req_z = float(packet.get("z", consts.WORLD_AREA / 2))
                    monster_mgr.spawn_horde_around(req_x, req_z)
                    print(f"[AI Spawner] Client {cid} seeded monsters at ({req_x:.0f}, {req_z:.0f})")

                elif ptype == "player" or "x" in packet:
                    clients[cid].update({
                        'x': packet['x'], 'y': packet['y'], 'z': packet['z'], 'r': packet.get('r', 0),
                        'item': packet.get('item', 0),
                        'firing': packet.get('firing', False)
                    })

                elif packet.get("type") == "chunk_loaded":
                    cx, cy, cz = packet["chunk"]
                    chunk = (cx, cy, cz)

                    client_hash = packet.get("chunk_hash")
                    is_flying = clients[cid].get('is_flying', False)

                    if client_hash and chunk not in dirty_server_chunks:
                        expected = chunk_hash_registry.get(chunk)
                        if expected is None:
                            chunk_hash_registry[chunk] = client_hash
                        elif expected != client_hash:
                            secure_id = clients[cid].get('secure_id', 'unknown')
                            _log_suspect(secure_id, cid, chunk,
                                         f"chunk_hash_mismatch expected={expected} got={client_hash}", is_flying,
                                         current_time)

                    chunk_blocks = {}
                    chunk_water = {}

                    min_x, max_x = cx * 16, (cx + 1) * 16
                    min_y, max_y = cy * 16, (cy + 1) * 16
                    min_z, max_z = cz * 16, (cz + 1) * 16

                    for (x, y, z), b_id in list(world_changes.items()):
                        if min_x <= x < max_x and min_y <= y < max_y and min_z <= z < max_z:
                            chunk_blocks[f"{x},{y},{z}"] = b_id

                    for (x, y, z), vol in list(water_changes.items()):
                        if min_x <= x < max_x and min_y <= y < max_y and min_z <= z < max_z:
                            chunk_water[f"{x},{y},{z}"] = vol

                    if chunk_blocks or chunk_water:
                        sync_packet = {
                            "type": "world_sync",
                            "changes": chunk_blocks,
                            "water_changes": chunk_water
                        }
                        try:
                            sock.sendto(json.dumps(sync_packet).encode('utf-8'), clients[cid]['addr'])
                        except Exception:
                            pass

                    if chunk not in active_chunk_hosts:
                        active_chunk_hosts[chunk] = cid
                    else:
                        host_id = active_chunk_hosts[chunk]
                        if host_id in clients and host_id != cid:
                            req = {"type": "request_fluid_state", "chunk": [cx, cy, cz], "target": cid}
                            sock.sendto(json.dumps(req).encode('utf-8'), clients[host_id]['addr'])
                        else:
                            active_chunk_hosts[chunk] = cid

                elif packet.get("type") == "chat":
                    sender_id = packet.get("id")
                    sender_name = packet.get("name", "Unknown")
                    text = packet.get("text", "")
                    sx, sy, sz = 0, 0, 0
                    if sender_id in clients:
                        sx, sy, sz = clients[sender_id]['x'], clients[sender_id]['y'], clients[sender_id]['z']
                    chat_packet = json.dumps({"type": "chat", "name": sender_name, "text": text}).encode('utf-8')
                    for other_cid, other_cdata in clients.items():
                        dist = math.hypot(other_cdata['x'] - sx, other_cdata['z'] - sz)
                        if dist < 50.0:
                            try:
                                sock.sendto(chat_packet, other_cdata['addr'])
                            except Exception:
                                pass

                elif packet.get("type") == "relay_fluid_state":
                    target_id = packet.get("target_id")
                    if target_id in clients:
                        relay = {"type": "water_consensus_sync", "water_data": packet.get("water_data", {})}
                        sock.sendto(json.dumps(relay).encode('utf-8'), clients[target_id]['addr'])

                elif packet.get("type") == "block_update":
                    bx, by, bz = int(packet["bx"]), int(packet["by"]), int(packet["bz"])
                    block_id = packet["block_id"]
                    world_changes[(bx, by, bz)] = block_id
                    pending_block_writes.append((bx, by, bz, block_id, current_time))
                    encoded_update = json.dumps(packet).encode('utf-8')
                    for other_cid, other_cdata in clients.items():
                        if other_cid != cid:
                            try:
                                sock.sendto(encoded_update, other_cdata['addr'])
                            except Exception:
                                pass

                elif packet.get("type") == "water_update":
                    bx, by, bz = int(packet["bx"]), int(packet["by"]), int(packet["bz"])
                    vol = packet["volume"]
                    water_changes[(bx, by, bz)] = vol
                    pending_water_writes.append((bx, by, bz, vol, current_time))
                    encoded_update = json.dumps(packet).encode('utf-8')
                    for other_cid, other_cdata in clients.items():
                        if other_cid != cid:
                            try:
                                sock.sendto(encoded_update, other_cdata['addr'])
                            except Exception:
                                pass

                elif packet.get("type") == "batched_water_sync":
                    water_data = packet.get("water_data", {})
                    chunk_coord_list = packet.get("chunk")
                    if chunk_coord_list:
                        cx, cy, cz = chunk_coord_list
                        chunk_key = (cx, cy, cz)
                        min_x, max_x = cx * 16, (cx + 1) * 16
                        min_y, max_y = cy * 16, (cy + 1) * 16
                        min_z, max_z = cz * 16, (cz + 1) * 16

                        keys_to_delete = [
                            k for k in water_changes.keys()
                            if min_x <= k[0] < max_x and min_y <= k[1] < max_y and min_z <= k[2] < max_z
                        ]
                        for k in keys_to_delete:
                            del water_changes[k]

                        last_saved = water_chunk_last_saved.get(chunk_key, 0.0)
                        if current_time - last_saved >= WATER_CHUNK_SAVE_COOLDOWN:
                            water_chunk_last_saved[chunk_key] = current_time
                            _write_ledger_entry(chunk_water_dict=water_data, chunk_coord=chunk_coord_list)

                    for coord_str, vol in water_data.items():
                        x, y, z = map(int, coord_str.split(','))
                        water_changes[(x, y, z)] = vol

                elif packet.get("type") == "damage":
                    target_id = packet.get("target_id")
                    damage_amount = packet.get("amount", 0)
                    if target_id in clients and cid in clients:
                        if current_time < clients[target_id].get('invulnerable_until', 0.0):
                            continue
                        clients[target_id]['hp'] -= damage_amount
                        print(f"[PvP] {cid} blasted {target_id}! Target HP: {clients[target_id]['hp']}")
                        sock.sendto(
                            json.dumps({"type": "health_update", "hp": clients[target_id]['hp']}).encode('utf-8'),
                            clients[target_id]['addr'])
                        if clients[target_id]['hp'] <= 0:
                            clients[target_id]['hp'] = 100.0
                            clients[target_id]['invulnerable_until'] = current_time + 2.5
                            victim_secure_id = clients[target_id]['secure_id']
                            attacker_secure_id = clients[cid]['secure_id']
                            player_profiles[victim_secure_id]['deaths'] += 1
                            player_profiles[attacker_secure_id]['kills'] += 1
                            rx, ry, rz = _respawn_at_hash(victim_secure_id, player_profiles[victim_secure_id]['deaths'])
                            sock.sendto(
                                json.dumps({"type": "respawn", "x": rx, "y": ry, "z": rz, "hp": 100.0}).encode('utf-8'),
                                clients[target_id]['addr'])
                            print(f"[PvP] {target_id} Obliterated! Sent to Grassy Respawn.")

                elif ptype == "monster_damage":
                    target_mid = packet.get("target_id")
                    damage_amount = float(packet.get("amount", 0))
                    monster_mgr.handle_monster_damage(
                        target_mid, damage_amount, sock, clients, current_time
                    )

            except BlockingIOError:
                break
            except Exception:
                pass

    def process_outbound():
        nonlocal last_network_tick
        if not hasattr(process_outbound, 'last_heartbeat'):
            process_outbound.last_heartbeat = 0.0
            process_outbound.last_db_flush  = 0.0
            process_outbound.last_respawn   = 0.0

            # First tick — all subsystems live, clients may now connect
            _server_status['phase'] = 'outbound_ready'
            print('[Status] Phase → outbound_ready — accepting connections')

        current_time = time.perf_counter()
        if current_time - last_network_tick < NETWORK_TICK_RATE:
            return
        last_network_tick = current_time

        force_heartbeat = False
        if current_time - process_outbound.last_heartbeat > 1.0:
            force_heartbeat = True
            process_outbound.last_heartbeat = current_time

        if not hasattr(process_outbound, 'last_db_flush'):
            process_outbound.last_db_flush = current_time

        buffer_size = len(pending_block_writes) + len(pending_water_writes)
        time_since_last_flush = current_time - process_outbound.last_db_flush

        if buffer_size >= 100 or (buffer_size > 0 and time_since_last_flush > 5.0):
            process_outbound.last_db_flush = current_time
            b_batch = pending_block_writes[:]
            w_batch = pending_water_writes[:]
            pending_block_writes.clear()
            pending_water_writes.clear()
            _write_ledger_entry(blocks_batch=b_batch, water_batch=w_batch)

        timeout_disconnects = []
        active_players = []

        for cid, cdata in list(clients.items()):
            if current_time - cdata['last_seen'] > 15.0:
                print(f"[Disconnect] {cid} timed out. Dumping stats & pos...")
                _save_player_profile(cdata['secure_id'], cdata)
                timeout_disconnects.append(cid)
                leave_packet = json.dumps({"type": "player_leave", "id": cid}).encode('utf-8')
                for other_cid, other_cdata in clients.items():
                    if other_cid != cid:
                        try:
                            sock.sendto(leave_packet, other_cdata['addr'])
                        except Exception:
                            pass
                continue

            dx = abs(cdata['x'] - cdata['last_x'])
            dy = abs(cdata['y'] - cdata['last_y'])
            dz = abs(cdata['z'] - cdata['last_z'])
            dr = abs(cdata['r'] - cdata['last_r'])

            if dx > 0.01 or dy > 0.01 or dz > 0.01 or dr > 0.01 or cdata.get('firing', False) or force_heartbeat:
                cdata['last_x'], cdata['last_y'] = cdata['x'], cdata['y']
                cdata['last_z'], cdata['last_r'] = cdata['z'], cdata['r']
                active_players.append({
                    "id": cid, "x": cdata['x'], "y": cdata['y'],
                    "z": cdata['z'], "r": cdata['r'], "hp": cdata['hp'],
                    "item": cdata.get('item', 0),
                    "firing": cdata.get('firing', False)
                })

        router.tick(current_time)

        # Clean up dead connections
        for cid in timeout_disconnects:
            del clients[cid]

        if (current_time - process_outbound.last_respawn > RESPAWN_INTERVAL
                and len(monster_mgr.active_monsters) < const.MAX_TOTAL_MONSTERS):
            process_outbound.last_respawn = current_time
            if clients:
                # Spawn near each online player
                for _cid, cdata in list(clients.items()):
                    if 'x' in cdata:
                        monster_mgr.spawn_horde_around(cdata['x'], cdata['z'])

            print(f"[AI Spawner] Periodic re-seed. Active: {len(monster_mgr.active_monsters)}")

        # BATCH & SHIP PHASE
        if active_players:
            batch_packet = json.dumps({"type": "batch_players", "players": active_players}).encode('utf-8')
            for cdata in clients.values():
                try:
                    sock.sendto(batch_packet, cdata['addr'])
                except Exception:
                    pass

        # ── Monster batch (separate packet, separate dict on the client) ──
        monster_entries = monster_mgr.build_network_batch(force_heartbeat)
        if monster_entries:
            mbatch = json.dumps({
                "type":    "batch_monsters",
                "monsters": monster_entries
            }).encode('utf-8')
            for cdata in clients.values():
                try:
                    sock.sendto(mbatch, cdata['addr'])
                except Exception:
                    pass

        encoded_world = json.dumps({"type": "world", "time": time.perf_counter()}).encode('utf-8')
        for cdata in clients.values():
            try:
                sock.sendto(encoded_world, cdata['addr'])
            except Exception:
                pass
    try:
        while True:
            engine.process_loop(process_physics, process_inbound, process_outbound)
            time.sleep(0.001)

    except KeyboardInterrupt:
        print("\nShutting down server gracefully. Dumping ALL profiles...")
        for cdata in clients.values():
            _save_player_profile(cdata['secure_id'])
        sock.close()

        print("Stopping Operations Coordinator...")
        coordinator.stop()


if __name__ == "__main__":
    import argparse
    ap = argparse.ArgumentParser(description='TokenTorch Game Server')
    ap.add_argument('--world', default=os.environ.get('TOKENTORCH_WORLD', 'default'),
                    help='World name — determines save DB path')
    ap.add_argument('--seed',  type=int,
                    default=int(os.environ.get('TOKENTORCH_SEED', '876567')),
                    help='World generation seed')
    ap.add_argument('--world-size', type=int,
                    default=int(os.environ.get('TOKENTORCH_WORLD_SIZE', '32768')))
    args = ap.parse_args()

    world_db = os.path.join(SAVES_DIR, 'worlds', f'hosted_{args.world}.db')
    LEDGER_PATH = os.path.join(SAVES_DIR, 'worlds', f'hosted_{args.world}_ledger.db')

    set_active_database(os.path.join('worlds', f'hosted_{args.world}.db'))
    _server_status['seed']       = args.seed
    _server_status['world_size'] = args.world_size
    print(f'[Server] World: {args.world}  Seed: {args.seed}  '
          f'Size: {args.world_size}  DB: {world_db}')

    try:
        start_server()
    except Exception as e:
        import traceback
        print(f'\n[SERVER CRASHED]\n{traceback.format_exc()}')
    finally:
        print('\nServer stopped. Press Enter to close this window...')
        input()