"""LLM credential and grounding config resolution. Resolves the model name and litellm kwargs for OpenSpace's LLM client, and assembles grounding config from env-var overrides. """ from __future__ import annotations import json import logging import os import tempfile from typing import Any, Dict, Optional logger = logging.getLogger("openspace.host_detection") def build_llm_kwargs(model: str) -> tuple[str, Dict[str, Any]]: """Build litellm kwargs and resolve model for OpenSpace's LLM client. Resolution order (highest → lowest priority): Tier 1 — Explicit ``OPENSPACE_LLM_*`` env vars:: OPENSPACE_LLM_API_KEY → litellm ``api_key`` OPENSPACE_LLM_API_BASE → litellm ``api_base`` OPENSPACE_LLM_EXTRA_HEADERS → litellm ``extra_headers`` (JSON string) OPENSPACE_LLM_CONFIG → arbitrary litellm kwargs (JSON string) Tier 2 — Auto-detect from host agent config file:: ~/.nanobot/config.json → providers.{matched}.apiKey / apiBase Tier 3 — Provider-native env vars inherited from the parent process (e.g. ``OPENROUTER_API_KEY``). Read by litellm automatically. Returns: ``(resolved_model, llm_kwargs_dict)`` """ from openspace.host_detection.nanobot import try_read_nanobot_config kwargs: Dict[str, Any] = {} resolved_model = model source = "inherited env" # --- Tier 2: auto-detect from host config (filled first, may be overridden) --- host_config = try_read_nanobot_config(model) if host_config: host_model = host_config.pop("_model", None) forced_provider = host_config.pop("_forced_provider", None) if not resolved_model and host_model: resolved_model = host_model # If the host config forces a gateway provider (e.g. openrouter) # and the model name doesn't already carry that prefix, prepend # it so that litellm uses the correct request format (OpenAI- # compatible for gateways vs native for direct providers). _GATEWAY_PROVIDERS = {"openrouter", "aihubmix", "siliconflow"} if ( forced_provider and forced_provider in _GATEWAY_PROVIDERS and resolved_model and not resolved_model.lower().startswith(f"{forced_provider}/") ): resolved_model = f"{forced_provider}/{resolved_model}" logger.info( "Prepended gateway prefix: model=%r (forced_provider=%s)", resolved_model, forced_provider, ) kwargs.update(host_config) source = "nanobot config" # --- Tier 1: explicit env vars override everything --- api_key = os.environ.get("OPENSPACE_LLM_API_KEY") if api_key: kwargs["api_key"] = api_key source = "OPENSPACE_LLM_* env" api_base = os.environ.get("OPENSPACE_LLM_API_BASE") if api_base: kwargs["api_base"] = api_base extra_headers_raw = os.environ.get("OPENSPACE_LLM_EXTRA_HEADERS") if extra_headers_raw: try: headers = json.loads(extra_headers_raw) if isinstance(headers, dict): kwargs["extra_headers"] = headers except json.JSONDecodeError: logger.warning("Invalid JSON in OPENSPACE_LLM_EXTRA_HEADERS: %r", extra_headers_raw) llm_config_raw = os.environ.get("OPENSPACE_LLM_CONFIG") if llm_config_raw: try: llm_config = json.loads(llm_config_raw) if isinstance(llm_config, dict): kwargs.update(llm_config) source = "OPENSPACE_LLM_CONFIG env" except json.JSONDecodeError: logger.warning("Invalid JSON in OPENSPACE_LLM_CONFIG: %r", llm_config_raw) # Default model fallback if not resolved_model: resolved_model = "openrouter/anthropic/claude-sonnet-4.5" if kwargs: safe = { k: (v[:8] + "..." if k == "api_key" and isinstance(v, str) and len(v) > 8 else v) for k, v in kwargs.items() } logger.info("LLM kwargs resolved (source=%s): %s", source, safe) return resolved_model, kwargs def build_grounding_config_path() -> Optional[str]: """Resolve grounding config: inline JSON > file path > None. Supports: * ``OPENSPACE_CONFIG_JSON`` — inline JSON string (written to a temp file) * ``OPENSPACE_CONFIG_PATH`` — path to a JSON config file Granular env-var overrides (``OPENSPACE_SHELL_*``, ``OPENSPACE_SKILLS_*``, etc.) are merged before writing. Returns: Path to the resolved config file, or None. """ config_json_raw = os.environ.get("OPENSPACE_CONFIG_JSON", "").strip() overrides: Dict[str, Any] = {} if config_json_raw: try: overrides = json.loads(config_json_raw) if not isinstance(overrides, dict): logger.warning("OPENSPACE_CONFIG_JSON is not a dict, ignoring") overrides = {} else: logger.info("Loaded inline config from OPENSPACE_CONFIG_JSON") except json.JSONDecodeError as e: logger.warning("Invalid JSON in OPENSPACE_CONFIG_JSON: %s", e) # --- Granular env-var overrides --- conda_env = os.environ.get("OPENSPACE_SHELL_CONDA_ENV", "").strip() if conda_env: overrides.setdefault("shell", {})["conda_env"] = conda_env shell_wd = os.environ.get("OPENSPACE_SHELL_WORKING_DIR", "").strip() if shell_wd: overrides.setdefault("shell", {})["working_dir"] = shell_wd skills_dirs_raw = os.environ.get("OPENSPACE_SKILLS_DIRS", "").strip() if skills_dirs_raw: dirs = [d.strip() for d in skills_dirs_raw.split(",") if d.strip()] if dirs: overrides.setdefault("skills", {})["skill_dirs"] = dirs mcp_servers_raw = os.environ.get("OPENSPACE_MCP_SERVERS_JSON", "").strip() if mcp_servers_raw: try: servers = json.loads(mcp_servers_raw) if isinstance(servers, dict): overrides["mcpServers"] = servers except json.JSONDecodeError as e: logger.warning("Invalid JSON in OPENSPACE_MCP_SERVERS_JSON: %s", e) log_level = os.environ.get("OPENSPACE_LOG_LEVEL", "").strip().upper() if log_level and log_level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"): overrides["log_level"] = log_level if overrides: try: fd, tmp_path = tempfile.mkstemp(suffix=".json", prefix="openspace_cfg_") with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(overrides, f, ensure_ascii=False) logger.info( "Grounding config overrides written to %s (%d keys)", tmp_path, len(overrides), ) return tmp_path except Exception as e: logger.warning("Failed to write config overrides: %s", e) return os.environ.get("OPENSPACE_CONFIG_PATH")