# Copyright 2026 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Application settings and environment variable configuration management.""" import os from pathlib import Path from typing import Literal from dotenv import load_dotenv from filelock import Timeout from pydantic import Field, SecretStr, model_validator from pydantic_settings import BaseSettings from artemis.config.constants import ( DEFAULT_ADB_HOST, DEFAULT_ADB_PORT, DEFAULT_MODEL, DEFAULT_PROFILE, ENV_DATA_ENGINE_DB_PATH, ENV_OCR_API_KEY, ENV_OPENAI_API_KEY, ENV_VISION_API_KEY, ) from artemis.config.environment import migrate_transport_default from artemis.config.paths import ( GLOBAL_APP_DIR, get_data_engine_db_path, get_default_traces_path, get_env_file, get_temp_dir, ) from artemis.utils.logger import get_logger # Installed wheels load .env from the user directory, outside site-packages. _canonical_env = get_env_file() logger = get_logger(__name__) try: if migrate_transport_default(_canonical_env): logger.info("已将全局模型传输协议迁移为 WebSocket；之后可在模型配置中切回 HTTP。") except (OSError, Timeout): os.environ["OPENAI_TRANSPORT"] = "websocket" logger.warning("无法保存 WebSocket 默认迁移；本次使用 WebSocket，下次启动重试。") load_dotenv(dotenv_path=_canonical_env, verbose=True) _global_env = GLOBAL_APP_DIR / ".env" if _global_env.exists() and _global_env.resolve() != _canonical_env.resolve(): load_dotenv(dotenv_path=_global_env, verbose=True) def is_placeholder_key(val: str | SecretStr | None) -> bool: """Check if an API credential is empty or an unconfigured placeholder value.""" if val is None: return True raw_str = val.get_secret_value() if isinstance(val, SecretStr) else str(val) raw_str = raw_str.strip().lower() if not raw_str: return True if raw_str in ( "api_key", "your_api_key", "your_api_key_here", "your_gemini_api_key_here", "your_google_cloud_vision_api_key_here", "your_openai_api_key_here", "your_anthropic_api_key_here", "your_openrouter_api_key_here", "your_xai_api_key_here", "none", "empty", "null", "undefined", ): return True if ( (raw_str.startswith("your_") and raw_str.endswith("_here")) or (raw_str.startswith("<") and raw_str.endswith(">")) or (raw_str.startswith("[") and raw_str.endswith("]")) ): return True return False class Settings(BaseSettings): """Centralized ARTEMIS runtime settings loaded from environment and .env files.""" ARTEMIS_PLUGIN_MARKETPLACE_URL: str = ( "https://git.playones.com/chenmingxuan/Artemis-Plugins/raw/branch/main/catalog-v4.json" ) ARTEMIS_PLUGIN_START_TIMEOUT_SECONDS: float = Field(default=30, gt=0, le=300) ARTEMIS_PLUGIN_TOOL_TIMEOUT_SECONDS: float = Field(default=60, gt=0, le=3600) ARTEMIS_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS: float = Field(default=15, gt=0, le=60) ARTEMIS_PLUGIN_DOWNLOAD_MAX_BYTES: int = Field(default=134217728, gt=0) ARTEMIS_PLUGIN_DOWNLOAD_TIMEOUT_SECONDS: float = Field(default=300, gt=0, le=1800) ARTEMIS_PLUGIN_EXTRACT_MAX_BYTES: int = Field(default=536870912, gt=0) ARTEMIS_PLUGIN_RESULT_MAX_BYTES: int = Field(default=16777216, gt=0) # LLM Provider Authentication OPENAI_API_KEY: SecretStr | None = None # Google Cloud Vision OCR Authentication OCR_API_KEY: SecretStr | None = None VISION_API_KEY: SecretStr | None = None API_KEY: SecretStr | None = None # Custom Provider Endpoints OPENAI_BASE_URL: str | None = "https://api-n-cd.playones.com/v1" OPENAI_MODEL: str = Field(default=DEFAULT_MODEL, min_length=1) OPENAI_TIMEOUT_SECONDS: float = Field(default=300.0, gt=0) OPENAI_TRANSPORT: Literal["http", "websocket"] = "websocket" # Environment only; deliberately absent from the saved UI/task configuration. OPENAI_USE_SYSTEM_PROXY: bool = False OPENAI_MAX_RETRIES: int = Field(default=5, ge=0, le=10) OPENAI_RETRY_INTERVAL_SECONDS: float = Field(default=10.0, ge=0, le=300) OPENAI_RATE_LIMIT_RETRY_SECONDS: float = Field(default=60.0, ge=0, le=3600) OPENAI_TOOL_FALLBACK_MODEL: str = Field(default="gpt-6-luna", min_length=1) OPENAI_TOOL_FALLBACK_REASONING_EFFORT: str = "high" OPENAI_TOOL_FALLBACK_MAX_RETRIES: int = Field(default=3, ge=0, le=10) ARTEMIS_TOOL_FALLBACK_MAX_NO_PROGRESS_ATTEMPTS: int = Field(default=3, ge=1, le=10) OPENAI_REASONING_EFFORT: str = "high" ARTEMIS_CREDENTIAL_TIMEOUT_SECONDS: float = Field(default=12.0, gt=0) ARTEMIS_MODEL_CACHE_TTL_SECONDS: float = Field(default=10.0, ge=0) ARTEMIS_DAEMON_PROBE_TIMEOUT_SECONDS: float = Field(default=2.0, gt=0) ARTEMIS_VIDEO_TIMEOUT_SECONDS: float = Field(default=120.0, gt=0) ARTEMIS_VIDEO_MAX_RETRIES: int = Field(default=1, ge=0) ARTEMIS_VIDEO_CHUNK_SECONDS: float = Field(default=60.0, gt=0) ARTEMIS_CHECKPOINT_TIMEOUT_SECONDS: float = Field(default=180.0, gt=0) ARTEMIS_SETTLEMENT_TIMEOUT_SECONDS: float = Field(default=120.0, gt=0) ARTEMIS_VIDEO_ENABLED: bool = False ARTEMIS_RECORDING_ENABLED: bool = False ARTEMIS_RECORDING_CONFIG: dict = Field(default_factory=dict) # Worker-only override, applied after the user's JSON recording defaults. ARTEMIS_TASK_RECORDING_CONFIG: dict = Field(default_factory=dict) ARTEMIS_OUTPUTTER_ENABLED: bool = True ARTEMIS_FLASH_FINAL_CHECK_ENABLED: bool = False ARTEMIS_STEP_SUMMARIZER_ENABLED: bool = True ARTEMIS_SCREENSHOT_LIMIT_ENABLED: bool = True # Consecutive Pro responses without a valid tool call before failing the task. ARTEMIS_OPERATOR_MAX_NO_PROGRESS_ATTEMPTS: int = Field(default=3, ge=1, le=20) # Consecutive Flash responses without any tool call before failing the task. ARTEMIS_FLASH_MAX_NO_PROGRESS_ATTEMPTS: int = Field(default=3, ge=1, le=20) ARTEMIS_TASK_TIMEOUT_SECONDS: int = Field(default=1800, ge=30, le=86400) ARTEMIS_BATCH_MAX_EXECUTIONS: int = Field(default=10000, ge=1) ARTEMIS_OBSERVATION_TIMEOUT_SECONDS: float = Field(default=60, gt=0, le=300) ARTEMIS_OBSERVATION_MAX_ATTEMPTS: int = Field(default=3, ge=1, le=5) ARTEMIS_DEVICE_COMMAND_TIMEOUT_SECONDS: float = Field(default=10, gt=0, le=60) ARTEMIS_ACTION_FAILURE_ATTEMPTS: int = Field(default=3, ge=1, le=10) ARTEMIS_WORKER_CLEANUP_TIMEOUT_SECONDS: float = Field(default=30, gt=0, le=300) # Android ADB Connectivity ADB_HOST: str | None = Field(default=DEFAULT_ADB_HOST) ADB_PORT: int | None = Field(default=DEFAULT_ADB_PORT) ADB_DEVICE_SERIAL: str | None = None ARTEMIS_ADB_AUTO_CONNECT: bool = True ARTEMIS_ADB_DISCOVERY_INTERVAL_SECONDS: float = Field(default=10, ge=1, le=300) ARTEMIS_ADB_DISCOVERY_TIMEOUT_SECONDS: float = Field(default=5, gt=0, le=30) # UI hierarchy backend: "auto" = Accessibility Helper with UIAutomator2 # fallback, "helper" = helper only, "uiautomator" = UIAutomator2 only. ARTEMIS_HIERARCHY_BACKEND: str = Field(default="auto") # Whether a task may install / upgrade the Accessibility Helper APK on a device # it holds. False = only attach to a helper installed by `artemis helper install`. ARTEMIS_HELPER_AUTO_INSTALL: bool = Field(default=True) # Execution Defaults PROJECT_NAME: str | None = None ARTEMIS_DEFAULT_PROFILE: str = Field(default=DEFAULT_PROFILE) ARTEMIS_DEFAULT_MODEL: str = Field(default=DEFAULT_MODEL) # LLM Reliability LLM_PAUSE_TIMEOUT_SECONDS: float = Field( default=900.0, description=( "How long a task waits in the paused state after exhausting LLM" " retries before failing with LLMExhaustedError. <= 0 waits" " forever (legacy interactive behavior)." ), ) # Paths & Storage TRACES_PATH: Path = Field(default_factory=get_default_traces_path) DATA_ENGINE_DB_PATH: Path = Field(default_factory=get_data_engine_db_path) TEMP_PATH: Path = Field(default_factory=get_temp_dir) model_config = {"env_file": ".env", "extra": "ignore"} @model_validator(mode="after") def fallback_api_keys(self) -> "Settings": """Normalize OpenAI and OCR API keys and filter placeholders.""" # Sanitize any placeholder values loaded from environment or .env for attr in ( "OPENAI_API_KEY", "OCR_API_KEY", "VISION_API_KEY", "API_KEY", ): val = getattr(self, attr, None) if val and is_placeholder_key(val): setattr(self, attr, None) # Fallback for OCR & Vision API keys if not self.OCR_API_KEY and self.VISION_API_KEY: self.OCR_API_KEY = self.VISION_API_KEY return self def get_api_key(self, provider: str) -> SecretStr | None: """Get API key for a specified provider. Args: provider: Provider name ('openai', 'ocr', or 'vision'). Returns: SecretStr containing the API key or None if not configured. """ provider_lower = provider.lower() key: SecretStr | None = None if provider_lower in ("ocr", "vision", "google_vision"): key = self.OCR_API_KEY or self.VISION_API_KEY elif provider_lower == "openai": key = self.OPENAI_API_KEY if key and not is_placeholder_key(key): return key return None def set_api_key(self, provider: str, key: str, persist_to_env: bool = False) -> None: """Dynamically set an API key at runtime, optionally persisting to .env in the app dir. Args: provider: Target provider name. key: Secret API key string. persist_to_env: Whether to save the key to the app directory's .env file. """ secret = SecretStr(key) provider_lower = provider.lower() env_key_name = None if provider_lower == "openai": self.OPENAI_API_KEY = secret env_key_name = ENV_OPENAI_API_KEY os.environ[ENV_OPENAI_API_KEY] = key elif provider_lower in ("ocr", "vision", "google_vision"): self.OCR_API_KEY = secret self.VISION_API_KEY = secret env_key_name = ENV_OCR_API_KEY os.environ[ENV_OCR_API_KEY] = key os.environ[ENV_VISION_API_KEY] = key else: raise ValueError(f"Unsupported credential provider: {provider}") if persist_to_env and env_key_name: target_env_files = [get_env_file()] seen_paths = set() for env_file in target_env_files: try: env_file.parent.mkdir(parents=True, exist_ok=True) resolved = env_file.resolve() if resolved in seen_paths: continue seen_paths.add(resolved) lines = [] if env_file.exists(): lines = env_file.read_text(encoding="utf-8").splitlines() # Determine keys to update keys_to_update = [env_key_name] if provider_lower in ("ocr", "vision", "google_vision"): keys_to_update = [ENV_OCR_API_KEY, ENV_VISION_API_KEY] new_lines = [] updated_set = set() for line in lines: replaced = False for k in keys_to_update: if ( line.startswith(f"{k}=") or line.startswith(f"#{k}=") or line.startswith(f"# {k}=") ): new_lines.append(f"{k}={key}") updated_set.add(k) replaced = True break if not replaced: new_lines.append(line) # Ensure the primary key is present if not replaced primary_key = keys_to_update[0] if primary_key not in updated_set: new_lines.append(f"{primary_key}={key}") env_file.write_text("\n".join(new_lines) + "\n", encoding="utf-8") except Exception as e: logger.warning(f"Could not persist {env_key_name} to {env_file}: {e}") # Singleton instance settings = Settings() # Synchronize DATA_ENGINE_DB_PATH in environment for external sub-processes/tools if settings.DATA_ENGINE_DB_PATH: os.environ[ENV_DATA_ENGINE_DB_PATH] = str(settings.DATA_ENGINE_DB_PATH)