from __future__ import annotations import os from dataclasses import dataclass, field from pathlib import Path from urllib.parse import urlparse OMNI_ROOT = Path(__file__).resolve().parent.parent def _env_bool(name: str, default: bool) -> bool: value = os.getenv(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def _loopback_url(value: str, *, schemes: set[str]) -> str: parsed = urlparse(value) if parsed.scheme not in schemes: raise ValueError(f"Unsupported URL scheme for {value!r}") if parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: raise ValueError( f"Non-loopback endpoint {value!r} requires OMNI_ALLOW_REMOTE_ENDPOINTS=1" ) return value.rstrip("/") @dataclass(slots=True) class OrchestratorSettings: """Runtime settings. Defaults never start a model or contact the internet.""" host: str = "127.0.0.1" port: int = 8100 conductor_id: str = "omni-conductor" api_token: str | None = field(default=None, repr=False) hermes_api_key: str | None = field(default=None, repr=False) asr_url: str = "ws://127.0.0.1:8101/v1/stream" router_url: str = "http://127.0.0.1:8102/v1" qwen_url: str = "http://127.0.0.1:8103/v1" hermes_url: str = "http://127.0.0.1:8642" manager_url: str = "http://127.0.0.1:8110" media_url: str = "http://127.0.0.1:8111" event_store_path: Path = field( default_factory=lambda: OMNI_ROOT / ".state" / "orchestrator" / "events.v1.jsonl" ) node_trust_store_path: Path = field( default_factory=lambda: OMNI_ROOT / ".state" / "orchestrator" / "trusted-nodes.v1.json" ) router_enabled: bool = False router_model: str = "local-tool-router" connect_timeout_s: float = 0.75 request_timeout_s: float = 20.0 stream_timeout_s: float = 300.0 max_replay_events: int = 10_000 provider_health_ttl_s: float = 5.0 mdns_discovery_timeout_s: float = 0.75 allow_remote_endpoints: bool = False remote_nodes_enabled: bool = False mdns_discovery_enabled: bool = False @classmethod def from_env(cls) -> "OrchestratorSettings": return cls( host=os.getenv("OMNI_CONDUCTOR_HOST", "127.0.0.1"), port=int(os.getenv("OMNI_CONDUCTOR_PORT", "8100")), conductor_id=os.getenv("OMNI_CONDUCTOR_ID", "omni-conductor"), api_token=(os.getenv("OMNI_CONDUCTOR_TOKEN") or None), hermes_api_key=( os.getenv("HERMES_API_KEY") or os.getenv("API_SERVER_KEY") or os.getenv("OMNI_HERMES_API_KEY") or None ), asr_url=os.getenv("OMNI_ASR_URL", "ws://127.0.0.1:8101/v1/stream"), router_url=os.getenv("OMNI_ROUTER_URL", "http://127.0.0.1:8102/v1"), qwen_url=os.getenv("OMNI_QWEN_URL", "http://127.0.0.1:8103/v1"), hermes_url=os.getenv("OMNI_HERMES_URL", "http://127.0.0.1:8642"), manager_url=os.getenv("OMNI_MANAGER_URL", "http://127.0.0.1:8110"), media_url=os.getenv("OMNI_MEDIA_URL", "http://127.0.0.1:8111"), event_store_path=Path( os.getenv( "OMNI_EVENT_STORE_PATH", str(OMNI_ROOT / ".state" / "orchestrator" / "events.v1.jsonl"), ) ).resolve(), node_trust_store_path=Path( os.getenv( "OMNI_NODE_TRUST_STORE_PATH", str(OMNI_ROOT / ".state" / "orchestrator" / "trusted-nodes.v1.json"), ) ).resolve(), router_enabled=_env_bool("OMNI_ROUTER_ENABLED", False), router_model=os.getenv("OMNI_ROUTER_MODEL", "local-tool-router"), connect_timeout_s=float(os.getenv("OMNI_CONNECT_TIMEOUT_S", "0.75")), request_timeout_s=float(os.getenv("OMNI_REQUEST_TIMEOUT_S", "20")), stream_timeout_s=float(os.getenv("OMNI_STREAM_TIMEOUT_S", "300")), max_replay_events=int(os.getenv("OMNI_MAX_REPLAY_EVENTS", "10000")), provider_health_ttl_s=float(os.getenv("OMNI_PROVIDER_HEALTH_TTL_S", "5")), mdns_discovery_timeout_s=float(os.getenv("OMNI_MDNS_TIMEOUT_S", "0.75")), allow_remote_endpoints=_env_bool("OMNI_ALLOW_REMOTE_ENDPOINTS", False), remote_nodes_enabled=_env_bool("OMNI_REMOTE_NODES_ENABLED", False), mdns_discovery_enabled=_env_bool("OMNI_MDNS_DISCOVERY_ENABLED", False), ).validated() def validated(self) -> "OrchestratorSettings": if ( self.host not in {"127.0.0.1", "localhost", "::1"} and not self.allow_remote_endpoints ): raise ValueError( "A non-loopback conductor bind requires OMNI_ALLOW_REMOTE_ENDPOINTS=1" ) if ( self.host not in {"127.0.0.1", "localhost", "::1"} and not self.api_token ): raise ValueError( "A non-loopback conductor bind requires OMNI_CONDUCTOR_TOKEN" ) if not 1 <= self.port <= 65_535: raise ValueError("OMNI_CONDUCTOR_PORT must be a valid TCP port") if min( self.connect_timeout_s, self.request_timeout_s, self.stream_timeout_s, self.provider_health_ttl_s, self.mdns_discovery_timeout_s, ) <= 0: raise ValueError("Timeouts must be positive") if self.mdns_discovery_enabled and not self.remote_nodes_enabled: raise ValueError( "OMNI_MDNS_DISCOVERY_ENABLED requires OMNI_REMOTE_NODES_ENABLED=1" ) if not self.allow_remote_endpoints: self.asr_url = _loopback_url(self.asr_url, schemes={"ws", "wss"}) for name in ("router_url", "qwen_url", "hermes_url", "manager_url", "media_url"): setattr(self, name, _loopback_url(getattr(self, name), schemes={"http", "https"})) else: self.asr_url = self.asr_url.rstrip("/") for name in ("router_url", "qwen_url", "hermes_url", "manager_url", "media_url"): setattr(self, name, getattr(self, name).rstrip("/")) return self def topology(self) -> dict[str, dict[str, object]]: return { "conductor": {"url": f"http://{self.host}:{self.port}", "port": self.port}, "asr": {"url": self.asr_url, "port": 8101, "streaming": True}, "router": { "url": self.router_url, "port": 8102, "optional": True, "enabled": self.router_enabled, }, "qwen": {"url": self.qwen_url, "port": 8103, "owner": "hermes"}, "hermes": {"url": self.hermes_url, "port": 8642, "sse": True}, "model_manager": {"url": self.manager_url, "port": 8110, "on_demand": True}, "media": {"url": self.media_url, "port": 8111, "on_demand": True}, }