agentic-mobile-control/apps/device-host-agent/host_agent/config.py

from __future__ import annotations

import os
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
from urllib.parse import urlparse


class HostAgentConfigurationError(ValueError):
    """Raised when Host Agent process configuration is invalid."""


_LOOPBACK_BIND_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})
_AI_PLANNER_TRANSPORTS = frozenset({"direct", "cloud"})


@dataclass(frozen=True)
class HostAgentConfig:
    control_plane_url: str
    host_id: str = ""
    token: str = field(default="", repr=False)
    identity_path: Path = Path("tasks/host_identity.json")
    local_account_path: Path = Path("tasks/host_local_account.json")
    enrollment_managed: bool = False
    display_name: str | None = None
    heartbeat_interval_seconds: float = 30.0
    poll_timeout_seconds: float = 20.0
    retry_backoff_seconds: float = 1.0
    max_retry_backoff_seconds: float = 30.0
    max_retry_attempts: int = 5
    console_bind_host: str = "127.0.0.1"
    console_port: int = 8765
    console_allow_non_loopback: bool = False
    console_session_ttl_seconds: float = 43200.0
    console_history_limit: int = 200
    ai_planner_transport: str = "direct"


def load_host_agent_config(
    env: Mapping[str, str] | None = None,
) -> HostAgentConfig:
    values = os.environ if env is None else env
    control_plane_url = (
        values.get(
            "HOST_AGENT_CONTROL_PLANE_URL",
            "https://amcp.home.jerryyan.top",
        )
        .strip()
        .rstrip("/")
    )
    parsed_url = urlparse(control_plane_url)
    if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
        raise HostAgentConfigurationError(
            "HOST_AGENT_CONTROL_PLANE_URL must be an HTTP(S) URL"
        )

    identity_path = Path(
        values.get("HOST_AGENT_IDENTITY_PATH", "tasks/host_identity.json").strip()
    )
    local_account_path = Path(
        values.get(
            "HOST_AGENT_LOCAL_ACCOUNT_PATH", "tasks/host_local_account.json"
        ).strip()
    )

    config = HostAgentConfig(
        control_plane_url=control_plane_url,
        identity_path=identity_path,
        local_account_path=local_account_path,
        enrollment_managed=True,
        display_name=values.get("HOST_AGENT_DISPLAY_NAME") or None,
        heartbeat_interval_seconds=_positive_float(
            values,
            "HOST_AGENT_HEARTBEAT_INTERVAL_SECONDS",
            30.0,
        ),
        poll_timeout_seconds=_positive_float(
            values,
            "HOST_AGENT_POLL_TIMEOUT_SECONDS",
            20.0,
        ),
        retry_backoff_seconds=_positive_float(
            values,
            "HOST_AGENT_RETRY_BACKOFF_SECONDS",
            1.0,
        ),
        max_retry_backoff_seconds=_positive_float(
            values,
            "HOST_AGENT_MAX_RETRY_BACKOFF_SECONDS",
            30.0,
        ),
        max_retry_attempts=_positive_int(
            values,
            "HOST_AGENT_MAX_RETRY_ATTEMPTS",
            5,
        ),
        console_bind_host=values.get(
            "HOST_AGENT_CONSOLE_BIND_HOST", "127.0.0.1"
        ).strip(),
        console_port=_positive_int(values, "HOST_AGENT_CONSOLE_PORT", 8765),
        console_allow_non_loopback=_truthy(
            values, "HOST_AGENT_CONSOLE_ALLOW_NON_LOOPBACK", False
        ),
        console_session_ttl_seconds=_positive_float(
            values,
            "HOST_AGENT_CONSOLE_SESSION_TTL_SECONDS",
            43200.0,
        ),
        console_history_limit=_positive_int(
            values,
            "HOST_AGENT_CONSOLE_HISTORY_LIMIT",
            200,
        ),
        ai_planner_transport=_parse_ai_planner_transport(
            values.get("AI_PLANNER_TRANSPORT")
        ),
    )
    if config.max_retry_backoff_seconds < config.retry_backoff_seconds:
        raise HostAgentConfigurationError(
            "maximum retry backoff must not be less than initial backoff"
        )
    if (
        config.console_bind_host not in _LOOPBACK_BIND_HOSTS
        and not config.console_allow_non_loopback
    ):
        raise HostAgentConfigurationError(
            "HOST_AGENT_CONSOLE_BIND_HOST must be a loopback address "
            f"({', '.join(sorted(_LOOPBACK_BIND_HOSTS))}) unless "
            "HOST_AGENT_CONSOLE_ALLOW_NON_LOOPBACK is set"
        )
    return config


def _parse_ai_planner_transport(value: str | None) -> str:
    if value is None:
        return "direct"
    transport = value.strip().lower()
    if transport not in _AI_PLANNER_TRANSPORTS:
        raise HostAgentConfigurationError(
            "AI_PLANNER_TRANSPORT must be one of "
            f"{', '.join(sorted(_AI_PLANNER_TRANSPORTS))}"
        )
    return transport


def _positive_float(
    values: Mapping[str, str],
    name: str,
    default: float,
) -> float:
    raw_value = values.get(name)
    if raw_value is None:
        return default
    try:
        value = float(raw_value)
    except ValueError as exc:
        raise HostAgentConfigurationError(f"{name} must be a number") from exc
    if value <= 0:
        raise HostAgentConfigurationError(f"{name} must be greater than zero")
    return value


def _positive_int(
    values: Mapping[str, str],
    name: str,
    default: int,
) -> int:
    raw_value = values.get(name)
    if raw_value is None:
        return default
    try:
        value = int(raw_value)
    except ValueError as exc:
        raise HostAgentConfigurationError(f"{name} must be an integer") from exc
    if value <= 0:
        raise HostAgentConfigurationError(f"{name} must be greater than zero")
    return value


def _truthy(
    values: Mapping[str, str],
    name: str,
    default: bool,
) -> bool:
    raw_value = values.get(name)
    if raw_value is None:
        return default
    return raw_value.strip().lower() in {"true", "1"}