thoxmesh_node/config.py

"""Typed configuration for the node agent, resolved from the environment.



Every field is settable by environment variable so the same module runs

unmodified in a Colab cell, a Docker Space and a local shell. ``.env.example``

documents the same names with placeholder values; no real secret is ever

committed.

"""

from __future__ import annotations

import os
from dataclasses import dataclass, field
from typing import Any, Mapping

from .errors import ConfigError

#: Supabase project hosting the MeshStack v2 control plane.
DEFAULT_SUPABASE_PROJECT = "rtaktyshnzbjktdyeemg"

#: Transport used to publish this node into the mesh.
TRANSPORT_DEVICE_V2 = "device_v2"
TRANSPORT_CONTROLLER = "controller"
TRANSPORT_AUTO = "auto"
VALID_TRANSPORTS = (TRANSPORT_AUTO, TRANSPORT_DEVICE_V2, TRANSPORT_CONTROLLER)

#: Protocol values allowed by the v2 schema CHECK constraint.
VALID_PROTOCOLS = ("openai", "ollama", "llama_cpp", "litert", "custom")


def _env_int(source: Mapping[str, str], key: str, default: int, *, minimum: int = 1) -> int:
    raw = (source.get(key) or "").strip()
    if not raw:
        return default
    try:
        value = int(raw)
    except ValueError as exc:
        raise ConfigError(f"{key} must be an integer, got {raw!r}") from exc
    if value < minimum:
        raise ConfigError(f"{key} must be >= {minimum}, got {value}")
    return value


def _env_bool(source: Mapping[str, str], key: str, default: bool) -> bool:
    raw = (source.get(key) or "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on")


def _env_flags(source: Mapping[str, str], key: str, default: tuple[str, ...]) -> dict[str, bool]:
    """Parse a comma-separated capability list into the jsonb flag map.



    ThoxRoute and ``resolve_route`` both test capabilities by truthy key lookup,

    so the list is expanded into ``{"chat": True, ...}`` rather than kept as an

    array.

    """
    raw = (source.get(key) or "").strip()
    names = [part.strip() for part in raw.split(",") if part.strip()] if raw else list(default)
    return {name: True for name in names}


@dataclass(frozen=True)
class NodeConfig:
    """Fully resolved node configuration."""

    # Identity of what we serve
    model_id: str
    alias: str
    protocol: str
    context_window: int
    capabilities: dict[str, bool]

    # Where we serve it
    host: str
    port: int
    public_url: str

    # How we join the mesh
    transport: str
    supabase_project: str
    controller_url: str
    node_kind: str
    ephemeral: bool
    heartbeat_seconds: int
    ttl_seconds: int
    register_retry_seconds: int

    # Local backend
    model_repo: str
    model_file: str
    n_threads: int
    n_gpu_layers: str
    max_tokens_cap: int
    request_timeout_seconds: int

    extra: dict[str, Any] = field(default_factory=dict)

    @property
    def device_function_url(self) -> str:
        """Absolute URL of the ``meshstack-device-v2`` edge function."""
        return f"https://{self.supabase_project}.supabase.co/functions/v1/meshstack-device-v2"

    def validate(self) -> None:
        """Fail fast on contradictions the agent cannot resolve at runtime."""
        if self.protocol not in VALID_PROTOCOLS:
            raise ConfigError(
                f"THOX_PROTOCOL must be one of {VALID_PROTOCOLS}, got {self.protocol!r}; "
                "the v2 schema rejects anything else"
            )
        if self.transport not in VALID_TRANSPORTS:
            raise ConfigError(f"THOX_MESH_TRANSPORT must be one of {VALID_TRANSPORTS}")
        if self.ttl_seconds <= self.heartbeat_seconds:
            raise ConfigError(
                f"TTL ({self.ttl_seconds}s) must exceed the heartbeat interval "
                f"({self.heartbeat_seconds}s), otherwise the lease expires between beats"
            )
        if self.transport == TRANSPORT_CONTROLLER and not self.controller_url:
            raise ConfigError("controller transport selected but THOX_CONTROLLER_URL is empty")
        if not self.model_id:
            raise ConfigError("THOX_MODEL_ID must not be empty")
        if not self.alias:
            raise ConfigError("THOX_ALIAS must not be empty")


def load_config(env: Mapping[str, str] | None = None) -> NodeConfig:
    """Resolve configuration from ``env`` (defaults to ``os.environ``).



    Defaults target the free-tier case deliberately: a CPU-sized GGUF, a 30s

    heartbeat and a 120s TTL. The TTL is four heartbeats so a node survives a

    couple of transient failures but a dead Colab session drops out of routing

    inside two minutes.

    """
    source = env if env is not None else os.environ

    heartbeat = _env_int(source, "THOX_HEARTBEAT_SECONDS", 30)
    ttl = _env_int(source, "THOX_TTL_SECONDS", 120)

    config = NodeConfig(
        # Defaults name a model that exists. An earlier default was
        # "thox-micro-125m", which is not published on the Hub at all, so any
        # caller relying on defaults was pointed at nothing.
        model_id=(source.get("THOX_MODEL_ID") or "thoxmini-3b").strip(),
        alias=(source.get("THOX_ALIAS") or source.get("THOX_MODEL_ID") or "thoxmini-3b").strip(),
        protocol=(source.get("THOX_PROTOCOL") or "openai").strip(),
        context_window=_env_int(source, "THOX_N_CTX", 2048),
        capabilities=_env_flags(source, "THOX_CAPABILITIES", ("chat", "completion")),
        host=(source.get("THOX_HOST") or "0.0.0.0").strip(),
        port=_env_int(source, "THOX_PORT", 7860),
        public_url=(source.get("THOX_PUBLIC_URL") or "").strip().rstrip("/"),
        transport=(source.get("THOX_MESH_TRANSPORT") or TRANSPORT_AUTO).strip().lower(),
        supabase_project=(source.get("THOX_SUPABASE_PROJECT") or DEFAULT_SUPABASE_PROJECT).strip(),
        controller_url=(source.get("THOX_CONTROLLER_URL") or "").strip().rstrip("/"),
        node_kind=(source.get("THOX_NODE_KIND") or "unknown").strip(),
        ephemeral=_env_bool(source, "THOX_EPHEMERAL", True),
        heartbeat_seconds=heartbeat,
        ttl_seconds=ttl,
        register_retry_seconds=_env_int(source, "THOX_REGISTER_RETRY_SECONDS", 10),
        model_repo=(source.get("THOX_MODEL_REPO") or "Thox-ai/ThoxMini-3B").strip(),
        model_file=(source.get("THOX_MODEL_FILE") or "thoxmini-3b-Q4_K_M.gguf").strip(),
        n_threads=_env_int(source, "THOX_N_THREADS", 0, minimum=0),
        # "auto" offloads everything when a GPU is detected and nothing when it
        # is not, so one value is correct on both Colab runtime types.
        n_gpu_layers=(source.get("THOX_N_GPU_LAYERS") or "auto").strip(),
        max_tokens_cap=_env_int(source, "THOX_MAX_TOKENS_CAP", 512),
        request_timeout_seconds=_env_int(source, "THOX_REQUEST_TIMEOUT_SECONDS", 120),
    )
    config.validate()
    return config


def detect_node_kind(env: Mapping[str, str] | None = None) -> str:
    """Best-effort guess at where we are running.



    Used for telemetry labelling and to pick sensible tunnel behaviour. A Space

    is already publicly addressable and needs no tunnel; Colab always does.

    """
    source = env if env is not None else os.environ
    if source.get("SPACE_ID") or source.get("SPACE_HOST"):
        return "hf-space"
    if source.get("COLAB_RELEASE_TAG") or source.get("COLAB_GPU") is not None:
        return "colab"
    if "google.colab" in (source.get("PYTHONPATH") or ""):
        return "colab"
    return "local"


def space_public_url(env: Mapping[str, str] | None = None) -> str:
    """Public origin of the current Space, or ``""`` when not on a Space.



    Hugging Face injects ``SPACE_HOST`` (``owner-name.hf.space``) but not

    ``SPACE_URL``; this repo previously registered ``http://0.0.0.0:7860``

    because it read the variable that does not exist. Returning ``""`` when the

    origin is unknown is deliberate — advertising an unreachable address is

    worse than not registering.

    """
    source = env if env is not None else os.environ
    host = (source.get("SPACE_HOST") or "").strip()
    if host:
        return host if "://" in host else f"https://{host}"
    space_id = (source.get("SPACE_ID") or "").strip()
    if space_id and "/" in space_id:
        owner, name = space_id.split("/", 1)
        slug = f"{owner}-{name}".replace("_", "-").replace(".", "-").lower()
        return f"https://{slug}.hf.space"
    return ""