src/world_simulator/config.py

from __future__ import annotations

import json
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Literal, cast

TerrainKind = Literal["plain_green"]
ConnectorKind = Literal["deterministic", "openai_compatible"]
OverseerMode = Literal["off", "advisor", "autopilot"]
FORCE_DETERMINISTIC_ENV = "WORLD_SIMULATOR_FORCE_DETERMINISTIC"


@dataclass(frozen=True, slots=True)
class WorldConfig:
    width: int
    depth: int
    terrain: TerrainKind
    seed: int
    survival: bool = False


@dataclass(frozen=True, slots=True)
class NpcConfig:
    count: int


@dataclass(frozen=True, slots=True)
class SimulationConfig:
    tick_ms: int


@dataclass(frozen=True, slots=True)
class ServerConfig:
    host: str
    port: int


@dataclass(frozen=True, slots=True)
class ConnectorConfig:
    type: ConnectorKind
    base_url: str | None = None
    api_url: str | None = None
    model: str | None = None
    api_key_env: str | None = None
    timeout_seconds: int = 60
    max_tokens: int = 1400
    temperature: float = 0.6
    top_p: float = 0.95
    tool_choice: str | dict[str, Any] | None = None
    max_parallel_npc_calls: int = 5
    fallback_to_deterministic: bool = True
    extra_body: dict[str, Any] | None = None


@dataclass(frozen=True, slots=True)
class OverseerConfig:
    mode: OverseerMode = "off"
    cycle_ticks: int = 8
    max_directives: int = 3
    connector: ConnectorConfig = field(
        default_factory=lambda: ConnectorConfig(type="deterministic")
    )


@dataclass(frozen=True, slots=True)
class GameConfig:
    world: WorldConfig
    npcs: NpcConfig
    simulation: SimulationConfig
    server: ServerConfig
    connector: ConnectorConfig = field(
        default_factory=lambda: ConnectorConfig(type="deterministic")
    )
    god_console: ConnectorConfig | None = None
    overseer: OverseerConfig = field(default_factory=OverseerConfig)
    secondary_connectors: dict[str, ConnectorConfig] = field(default_factory=dict)


def load_game_config(path: Path) -> GameConfig:
    _load_dotenv_for_config(path)
    with path.open("r", encoding="utf-8") as config_file:
        raw = json.load(config_file)
    return parse_game_config(raw)


def apply_runtime_env_overrides(config: GameConfig) -> GameConfig:
    if not _truthy_env(os.getenv(FORCE_DETERMINISTIC_ENV)):
        return config

    deterministic_connector = ConnectorConfig(type="deterministic")
    return GameConfig(
        world=config.world,
        npcs=config.npcs,
        simulation=config.simulation,
        server=config.server,
        connector=deterministic_connector,
        god_console=None,
        overseer=OverseerConfig(
            mode=config.overseer.mode,
            cycle_ticks=config.overseer.cycle_ticks,
            max_directives=config.overseer.max_directives,
            connector=deterministic_connector,
        ),
    )


def parse_game_config(raw: dict[str, Any]) -> GameConfig:
    world = _required_mapping(raw, "world")
    npcs = _required_mapping(raw, "npcs")
    simulation = _required_mapping(raw, "simulation")
    server = _required_mapping(raw, "server")
    connector = _optional_mapping(raw, "connector") or {"type": "deterministic"}
    god_console = _optional_mapping(raw, "god_console")
    overseer = _optional_mapping(raw, "overseer")

    terrain_raw = _required_str(world, "terrain")
    if terrain_raw != "plain_green":
        raise ValueError(f"Unsupported terrain kind: {terrain_raw}")
    terrain: TerrainKind = "plain_green"

    secondary_raw = raw.get("secondary_connectors", {})
    if not isinstance(secondary_raw, dict):
        raise ValueError("'secondary_connectors' must be an object.")
    secondary_connectors = {
        cid: _parse_connector_config(cfg)
        for cid, cfg in secondary_raw.items()
        if isinstance(cfg, dict)
    }

    return GameConfig(
        world=WorldConfig(
            width=_required_positive_int(world, "width"),
            depth=_required_positive_int(world, "depth"),
            terrain=terrain,
            seed=_required_int(world, "seed"),
            survival=_optional_bool(world, "survival", default=False),
        ),
        npcs=NpcConfig(count=_required_non_negative_int(npcs, "count")),
        simulation=SimulationConfig(tick_ms=_required_positive_int(simulation, "tick_ms")),
        server=ServerConfig(
            host=_required_str(server, "host"),
            port=_required_positive_int(server, "port"),
        ),
        connector=_parse_connector_config(connector),
        god_console=_parse_connector_config(god_console) if god_console else None,
        overseer=_parse_overseer_config(overseer),
        secondary_connectors=secondary_connectors,
    )


def _parse_overseer_config(raw: dict[str, Any] | None) -> OverseerConfig:
    if raw is None:
        return OverseerConfig()
    mode_raw = _optional_str(raw, "mode") or "off"
    if mode_raw not in ("off", "advisor", "autopilot"):
        raise ValueError(f"Unsupported overseer mode: {mode_raw}")
    connector_raw = _optional_mapping(raw, "connector")
    if connector_raw is None:
        connector_keys = {
            "type",
            "base_url",
            "api_url",
            "model",
            "api_key_env",
            "timeout_seconds",
            "max_tokens",
            "temperature",
            "top_p",
            "tool_choice",
            "max_parallel_npc_calls",
            "fallback_to_deterministic",
            "extra_body",
            "base_url_env",
            "api_url_env",
            "model_env",
        }
        connector_raw = {
            key: value
            for key, value in raw.items()
            if key in connector_keys
        } or {"type": "deterministic"}
    return OverseerConfig(
        mode=cast(OverseerMode, mode_raw),
        cycle_ticks=_optional_positive_int(raw, "cycle_ticks") or 8,
        max_directives=_optional_positive_int(raw, "max_directives") or 3,
        connector=_parse_connector_config(connector_raw),
    )


def _parse_connector_config(raw: dict[str, Any]) -> ConnectorConfig:
    connector_type_raw = _optional_str(raw, "type") or "deterministic"
    if connector_type_raw not in ("deterministic", "openai_compatible"):
        raise ValueError(f"Unsupported connector type: {connector_type_raw}")

    connector_type = cast(ConnectorKind, connector_type_raw)

    base_url = _optional_str(raw, "base_url")
    api_url = _optional_str(raw, "api_url")
    model = _optional_str(raw, "model")
    base_url_env = _optional_str(raw, "base_url_env")
    api_url_env = _optional_str(raw, "api_url_env")
    model_env = _optional_str(raw, "model_env")

    env_base_url = os.getenv(base_url_env) if base_url_env else None
    env_api_url = os.getenv(api_url_env) if api_url_env else None

    if env_base_url:
        base_url = env_base_url
    elif env_api_url:
        api_url = env_api_url
        base_url = None
    if model_env:
        model = os.getenv(model_env) or model

    temperature = _optional_non_negative_float(raw, "temperature")
    top_p = _optional_probability(raw, "top_p")

    return ConnectorConfig(
        type=connector_type,
        base_url=base_url,
        api_url=api_url,
        model=model,
        api_key_env=_optional_str(raw, "api_key_env"),
        timeout_seconds=_optional_positive_int(raw, "timeout_seconds") or 60,
        max_tokens=_optional_positive_int(raw, "max_tokens") or 1400,
        temperature=0.6 if temperature is None else temperature,
        top_p=0.95 if top_p is None else top_p,
        tool_choice=_optional_tool_choice(raw, "tool_choice"),
        max_parallel_npc_calls=_optional_positive_int(raw, "max_parallel_npc_calls") or 5,
        fallback_to_deterministic=_optional_bool(raw, "fallback_to_deterministic", default=True),
        extra_body=_optional_mapping(raw, "extra_body"),
    )


def _required_mapping(raw: dict[str, Any], key: str) -> dict[str, Any]:
    value = raw.get(key)
    if not isinstance(value, dict):
        raise ValueError(f"Expected '{key}' to be an object.")
    return value


def _optional_mapping(raw: dict[str, Any], key: str) -> dict[str, Any] | None:
    value = raw.get(key)
    if value is None:
        return None
    if not isinstance(value, dict):
        raise ValueError(f"Expected '{key}' to be an object.")
    return value


def _required_int(raw: dict[str, Any], key: str) -> int:
    value = raw.get(key)
    if not isinstance(value, int) or isinstance(value, bool):
        raise ValueError(f"Expected '{key}' to be an integer.")
    return value


def _required_positive_int(raw: dict[str, Any], key: str) -> int:
    value = _required_int(raw, key)
    if value <= 0:
        raise ValueError(f"Expected '{key}' to be positive.")
    return value


def _required_non_negative_int(raw: dict[str, Any], key: str) -> int:
    value = _required_int(raw, key)
    if value < 0:
        raise ValueError(f"Expected '{key}' to be non-negative.")
    return value


def _optional_positive_int(raw: dict[str, Any], key: str) -> int | None:
    value = raw.get(key)
    if value is None:
        return None
    if not isinstance(value, int) or isinstance(value, bool):
        raise ValueError(f"Expected '{key}' to be an integer.")
    if value <= 0:
        raise ValueError(f"Expected '{key}' to be positive.")
    return value


def _optional_non_negative_float(raw: dict[str, Any], key: str) -> float | None:
    value = raw.get(key)
    if value is None:
        return None
    if not isinstance(value, int | float) or isinstance(value, bool):
        raise ValueError(f"Expected '{key}' to be a number.")
    if value < 0:
        raise ValueError(f"Expected '{key}' to be non-negative.")
    return float(value)


def _optional_probability(raw: dict[str, Any], key: str) -> float | None:
    value = raw.get(key)
    if value is None:
        return None
    if not isinstance(value, int | float) or isinstance(value, bool):
        raise ValueError(f"Expected '{key}' to be a number.")
    if value <= 0 or value > 1:
        raise ValueError(f"Expected '{key}' to be between 0 and 1.")
    return float(value)


def _required_str(raw: dict[str, Any], key: str) -> str:
    value = raw.get(key)
    if not isinstance(value, str) or not value:
        raise ValueError(f"Expected '{key}' to be a non-empty string.")
    return value


def _optional_str(raw: dict[str, Any], key: str) -> str | None:
    value = raw.get(key)
    if value is None:
        return None
    if not isinstance(value, str) or not value:
        raise ValueError(f"Expected '{key}' to be a non-empty string.")
    return value


def _optional_bool(raw: dict[str, Any], key: str, *, default: bool) -> bool:
    value = raw.get(key)
    if value is None:
        return default
    if not isinstance(value, bool):
        raise ValueError(f"Expected '{key}' to be a boolean.")
    return value


def _optional_tool_choice(raw: dict[str, Any], key: str) -> str | dict[str, Any] | None:
    value = raw.get(key)
    if value is None:
        return None
    if isinstance(value, str):
        if not value:
            raise ValueError(f"Expected '{key}' to be a non-empty string or object.")
        return value
    if isinstance(value, dict):
        return value
    raise ValueError(f"Expected '{key}' to be a string or object.")


def _load_dotenv_for_config(config_path: Path) -> None:
    candidates = [
        Path.cwd() / ".env",
        config_path.resolve().parent / ".env",
        config_path.resolve().parent.parent / ".env",
    ]
    seen: set[Path] = set()
    for candidate in candidates:
        if candidate in seen or not candidate.is_file():
            continue
        seen.add(candidate)
        _load_dotenv(candidate)


def _load_dotenv(path: Path) -> None:
    with path.open("r", encoding="utf-8") as dotenv_file:
        for raw_line in dotenv_file:
            line = raw_line.strip()
            if not line or line.startswith("#"):
                continue
            if line.startswith("export "):
                line = line.removeprefix("export ").strip()
            key, separator, value = line.partition("=")
            if not separator:
                continue
            key = key.strip()
            if not key:
                continue
            os.environ.setdefault(key, _dotenv_value(value))


def _dotenv_value(raw: str) -> str:
    value = raw.strip()
    if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
        return value[1:-1]
    return value


def _truthy_env(value: str | None) -> bool:
    return value is not None and value.strip().lower() in ("1", "true", "yes", "on")