EvoScientist-Multi/EvoScientist/config/settings.py

"""Configuration management for EvoScientist.

Handles loading, saving, and merging configuration from multiple sources.
See :func:`get_effective_config` for the authoritative priority chain —
``EVOSCIENTIST_*`` shell values and third-party keys are treated
asymmetrically with respect to workspace ``.env`` handling.
"""

from __future__ import annotations

import logging
import os
from dataclasses import asdict, dataclass, fields
from enum import StrEnum
from functools import lru_cache
from pathlib import Path
from typing import Any, Literal, get_type_hints

import yaml
from dotenv import dotenv_values, find_dotenv

# Tools that run shell commands and need manual HITL approval (subject to
# shell_allow_list). Single source of truth for every interrupt consumer
# (stream/display.py, channels/interaction.py) — keep aligned with the agent's
# `interrupt_on` set in EvoScientist.py.
HITL_SHELL_TOOLS = ("execute", "run_in_background")

# Armed non-shell destructive tools must always prompt — no allow-list carve-outs
# (their args carry paths, not commands). Keep aligned with HITL_INTERRUPT_ON.
HITL_ALWAYS_PROMPT_TOOLS = ("delete", "schedule_task")


class MemoryObservationTarget(StrEnum):
    """Runtime locations that can receive `record_observation`."""

    AGENT = "agent"
    TURN_WORKER = "turn_worker"
    SUBAGENT_WORKER = "subagent_worker"


class MemoryObservationWriter(StrEnum):
    """Configured observation-writing policy."""

    OFF = "off"
    AGENT = "agent"
    WORKER = "worker"
    ALL = "all"

    def enables(self, target: MemoryObservationTarget) -> bool:
        match self:
            case MemoryObservationWriter.OFF:
                return False
            case MemoryObservationWriter.AGENT:
                return target == MemoryObservationTarget.AGENT
            case MemoryObservationWriter.WORKER:
                return target in (
                    MemoryObservationTarget.TURN_WORKER,
                    MemoryObservationTarget.SUBAGENT_WORKER,
                )
            case MemoryObservationWriter.ALL:
                return target in (
                    MemoryObservationTarget.AGENT,
                    MemoryObservationTarget.TURN_WORKER,
                    MemoryObservationTarget.SUBAGENT_WORKER,
                )


class MemorySkillSynthesisMode(StrEnum):
    """Configured AutoSkills approval behavior."""

    REVIEW = "review"
    AUTO = "auto"


class MemorySkillSynthesisCadence(StrEnum):
    """Preset cadence for the built-in AutoSkills schedule."""

    NIGHTLY = "nightly"
    WEEKLY = "weekly"
    MONTHLY = "monthly"


DEFAULT_MEMORY_OBSERVATION_WRITER = MemoryObservationWriter.ALL
DEFAULT_MEMORY_SKILL_SYNTHESIS_MODE = MemorySkillSynthesisMode.REVIEW
DEFAULT_MEMORY_SKILL_SYNTHESIS_CADENCE = MemorySkillSynthesisCadence.WEEKLY
DEFAULT_MEMORY_SKILL_SYNTHESIS_TIME = "03:00"


def _normalize_hhmm(value: Any) -> str | None:
    parts = str(value).strip().split(":")
    if len(parts) != 2:
        return None
    hour, minute = parts
    if not (hour.isdecimal() and minute.isdecimal()):
        return None
    try:
        hour_int = int(hour)
        minute_int = int(minute)
    except ValueError:
        return None
    if not (0 <= hour_int <= 23 and 0 <= minute_int <= 59):
        return None
    return f"{hour_int:02d}:{minute_int:02d}"


# =============================================================================
# Configuration paths
# =============================================================================


def get_config_dir() -> Path:
    """Get the configuration directory path.

    Uses XDG_CONFIG_HOME if set, otherwise ~/.config/evoscientist/
    """
    xdg_config = os.environ.get("XDG_CONFIG_HOME")
    if xdg_config:
        return Path(xdg_config) / "evoscientist"
    return Path.home() / ".config" / "evoscientist"


def get_config_path() -> Path:
    """Get the path to the configuration file."""
    return get_config_dir() / "config.yaml"


# =============================================================================
# Configuration dataclass
# =============================================================================

# OpenRouter app-attribution defaults (issue #339). Single source of truth: the
# EvoScientistConfig fields below default to these, and llm/models.py imports
# them for its env-fallback, so the values never drift across the two layers.
OPENROUTER_DEFAULT_HTTP_REFERER = "https://github.com/EvoScientist/EvoScientist"
OPENROUTER_DEFAULT_APP_TITLE = "EvoScientist"
# OpenRouter honors only the first 2 categories per request (server-side limit)
# and silently ignores the rest, so keep the two most relevant ones. Chosen per
# maintainer review — creative-writing is a less competitive marketplace group.
OPENROUTER_DEFAULT_APP_CATEGORIES = "creative-writing,personal-agent"


@dataclass
class EvoScientistConfig:
    """EvoScientist configuration settings.

    Attributes:
        anthropic_api_key: Anthropic API key for Claude models.
        openai_api_key: OpenAI API key for GPT models.
        nvidia_api_key: NVIDIA API key for NVIDIA models.
        google_api_key: Google API key for Gemini models.
        tavily_api_key: Tavily API key for web search.
        provider: Default LLM provider ('anthropic', 'openai', 'google-genai', or 'nvidia').
        model: Default model name (short name or full ID).
        auxiliary_provider: Provider for auxiliary_model (empty = use main provider).
        auxiliary_model: Model for memory workers + tool selector + scheduler (empty = use main model).
        default_mode: Default workspace mode ('daemon' or 'run').
        default_workdir: Default workspace directory (empty = use current working directory).
        show_thinking: Whether to show thinking panels in CLI.
    """

    # API Keys
    anthropic_api_key: str = ""
    anthropic_base_url: str = ""
    anthropic_auth_mode: str = "api_key"  # "api_key" | "oauth"
    openai_api_key: str = ""
    openai_auth_mode: str = "api_key"  # "api_key" | "oauth"
    nvidia_api_key: str = ""
    google_api_key: str = ""
    minimax_api_key: str = ""
    minimax_base_url: str = ""
    siliconflow_api_key: str = ""
    openrouter_api_key: str = ""
    atlascloud_api_key: str = ""
    requesty_api_key: str = ""
    deepseek_api_key: str = ""
    zhipu_api_key: str = ""
    volcengine_api_key: str = ""
    dashscope_api_key: str = ""
    moonshot_api_key: str = ""
    kimi_api_key: str = ""
    custom_openai_api_key: str = ""
    custom_openai_base_url: str = ""
    custom_anthropic_api_key: str = ""
    custom_anthropic_base_url: str = ""
    ollama_base_url: str = ""
    tavily_api_key: str = ""

    # LLM Settings
    provider: str = "anthropic"
    model: str = "claude-sonnet-4-6"
    model_fallbacks: str = ""  # "model:provider,model:provider" fallback chain
    # Optional auxiliary model for background/helper LLM calls (memory workers +
    # tool selector). Empty = fall back to the main model/provider.
    auxiliary_provider: str = ""  # empty = use main provider
    auxiliary_model: str = ""  # empty = use main model

    # Async Sub-agent Settings
    # When True (default), the EvoSci CLI auto-starts a langgraph dev subprocess
    # so any sub-agent flagged ``async: true`` in subagents/<name>.yaml runs
    # non-blocking via AsyncSubAgent. Currently affects writing-agent and
    # data-analysis-agent. Adds ~10-15s to CLI startup (langgraph dev cold
    # start, mostly MCP server spawn time).
    #
    # Set False to run fully in-process — saves the startup cost in scenarios
    # where async isn't useful: short scripted EvoSci runs (CI / one-shot
    # ``-p "..."``), low-RAM environments, or workflows that only need the
    # synchronous sub-agents (planner / research / code / debug).
    enable_async_subagents: bool = True

    # Port for the auto-started langgraph dev subprocess. 6174 is Kaprekar's
    # constant — a memorable EvoScientist-themed default that avoids collisions
    # with common dev ports (3000/5000/8000/8080) and the langgraph CLI default
    # 2024. Override if it conflicts with another local service.
    langgraph_dev_port: int = 6174

    # Network interface the langgraph dev subprocess binds to. Loopback by
    # default: this server is the agent API — no authentication, and the
    # deployed agent can run shell commands — so it stays off the network until
    # asked. Set "0.0.0.0" to reach it from another machine (the WebUI talks to
    # it FROM THE BROWSER, and external SDK clients need it too); every launcher
    # then prints a red PUBLIC BIND banner while it is exposed.
    #
    # Callers that *connect* (health probes, async sub-agent self-dispatch) map
    # a wildcard bind back to loopback via manager._probe_host, so widening this
    # never redirects internal traffic off-box.
    langgraph_dev_host: str = "127.0.0.1"

    # Port for the WebUI front-end (Next.js server from @evoscientist/webui),
    # used only when ui_backend == "webui". 4716 is 6174 reversed — a memorable
    # pairing with the langgraph dev port that it connects to. The backend keeps
    # its own port (langgraph_dev_port); this is just the browser server.
    webui_port: int = 4716

    # Network interface the WebUI front-end binds to. Also all interfaces by
    # default; it serves the app shell only and holds no credentials (see
    # deploy/webui.py:_scrubbed_env). Set "127.0.0.1" to keep it local-only.
    webui_host: str = "0.0.0.0"

    # --- Scheduled tasks (cron) ---
    # Master switch for scheduled tasks (/schedule, NL tools, scheduler context). Defaults
    # True so the feature is available out-of-the-box; set False to disable.
    enable_scheduler: bool = True
    # Default IANA timezone for cron schedules created without an explicit tz.
    # Empty string => the host's local IANA zone (resolved via tzlocal), falling
    # back to UTC if it can't be determined; set e.g. "Europe/London" to pin one.
    scheduler_default_timezone: str = ""

    # Whether langgraph dev persists its runtime state to .langgraph_api/ next
    # to the subprocess cwd. True (default) keeps async-task, scheduler, and
    # Store API state across subprocess restarts — useful for future
    # cross-session async, cron, and Store features. Set False to suppress
    # writes (workspace stays cleaner; state is in-memory only and lost on
    # CLI exit). EvoScientist's main thread persistence uses sessions.db
    # regardless of this setting.
    langgraph_dev_file_persistence: bool = True

    # Concurrency: how many runs each langgraph dev worker processes in parallel.
    # 10 is the langgraph dev recommended default and works well on a typical
    # dev machine. Lower it (e.g., 4) on memory-constrained or low-core
    # machines if multiple async sub-agents in flight cause noticeable
    # slowdown.
    langgraph_dev_jobs_per_worker: int = 10

    # Max LangGraph super-steps (LLM call / tool call / sub-agent delegation
    # each count as 1) before raising GraphRecursionError. Resets on every
    # ``agent.invoke()`` — i.e., this is per-turn, NOT per-conversation. For
    # long conversations the relevant mechanisms are checkpointer persistence
    # (sessions.db), ContextEditingMiddleware (window management), and
    # EvoMemoryMiddleware (cross-turn memory).
    #
    # 1,000,000 is "effectively unlimited" — typical research turns use
    # 200-1000 steps; reaching 1M would cost ~$10K in tokens, by which point
    # rate limits, context overflow, or API quota errors would trip first.
    # Lower (e.g., 5000) if you want a tighter safety net against runaway loops.
    recursion_limit: int = 1_000_000

    # Memory Settings
    # Profile memory injects and maintains `/memories/profile/...` files.
    memory_profile_enabled: bool = True
    # Observation memory indexes `/memories/observations/...` and adds
    # observation-read guidance/context. Writes require this switch plus an
    # allowed `memory_observation_writer` role below.
    memory_observations_enabled: bool = True
    # Which observation-writing path receives the `record_observation` tool:
    # "off" disables writes; "agent" means live agents; "worker" means
    # post-run memory workers; "all" means live agents and post-run memory
    # workers.
    memory_observation_writer: MemoryObservationWriter = (
        DEFAULT_MEMORY_OBSERVATION_WRITER
    )
    # Post-turn and post-subagent memory workers. Disable for no-background-memory
    # controls while still allowing live agents to read configured memory.
    memory_workers_enabled: bool = True
    # Slow EvoMemory maintenance that periodically scans observation clusters
    # and drafts reusable skills.
    memory_skill_synthesis_enabled: bool = True
    memory_skill_synthesis_mode: MemorySkillSynthesisMode = (
        DEFAULT_MEMORY_SKILL_SYNTHESIS_MODE
    )
    memory_skill_synthesis_cadence: MemorySkillSynthesisCadence = (
        DEFAULT_MEMORY_SKILL_SYNTHESIS_CADENCE
    )
    memory_skill_synthesis_time: str = DEFAULT_MEMORY_SKILL_SYNTHESIS_TIME

    # Workspace Settings
    default_mode: Literal["daemon", "run"] = "daemon"
    default_workdir: str = ""

    # UI Settings
    show_thinking: bool = True
    # "webui" launches the browser front-end (@evoscientist/webui via npx) +
    # a deploy-style langgraph server instead of the in-terminal CLI/TUI.
    ui_backend: Literal["cli", "tui", "webui"] = "tui"
    log_level: str = "warning"
    # Empty means use the provider/model default. A non-empty value is an
    # explicit user override exported as EVOSCIENTIST_REASONING_EFFORT.
    reasoning_effort: str = ""
    # Anthropic prompt caching for OpenRouter anthropic/* models. Opt out if
    # cache-write costs outweigh the benefit for a workflow.
    openrouter_anthropic_prompt_cache: bool = True
    # OpenRouter app attribution (issue #339). Sent only for the openrouter
    # provider; identifies EvoScientist in OpenRouter's app rankings/analytics.
    # Override (e.g. a private fork) via these fields or their env vars.
    # Defaults live in the module constants above (also imported by llm/models.py).
    openrouter_http_referer: str = OPENROUTER_DEFAULT_HTTP_REFERER
    openrouter_app_title: str = OPENROUTER_DEFAULT_APP_TITLE
    # Comma-separated; split into a list before being passed to
    # langchain-openrouter (its app_categories kwarg expects list[str]).
    openrouter_app_categories: str = OPENROUTER_DEFAULT_APP_CATEGORIES

    # Channel Settings
    channel_enabled: str = ""  # "imessage" | "telegram" | "discord" | "slack" | "wechat" | "dingtalk" | "feishu" | "email" | "qq" | "signal" | "" (comma-separated for multiple)
    channel_send_thinking: bool = True  # forward thinking to any channel
    channel_debug_tracing: bool = False  # emit extra inbound diagnostics at DEBUG
    require_mention: str = "group"  # "always" | "group" | "off"
    text_chunk_limit: int = 0  # 0 = use capability default
    allowed_channels: str = ""  # comma-separated channel IDs, empty = allow all

    # iMessage Settings
    imessage_enabled: bool = False  # legacy compat
    imessage_allowed_senders: str = ""

    # Telegram Settings
    telegram_bot_token: str = ""
    telegram_allowed_senders: str = ""
    telegram_proxy: str = ""

    # Discord Settings
    discord_bot_token: str = ""
    discord_allowed_senders: str = ""
    discord_allowed_channels: str = ""
    discord_proxy: str = ""

    # Slack Settings
    slack_bot_token: str = ""
    slack_app_token: str = ""
    slack_allowed_senders: str = ""
    slack_allowed_channels: str = ""
    slack_proxy: str = ""

    # Feishu Settings
    feishu_app_id: str = ""
    feishu_app_secret: str = ""
    feishu_verification_token: str = ""
    feishu_encrypt_key: str = ""
    feishu_webhook_port: int = 9000
    feishu_allowed_senders: str = ""
    feishu_domain: str = "https://open.feishu.cn"
    feishu_proxy: str = ""
    feishu_subscription_mode: str = "webhook"  # "webhook" | "websocket"

    # WeChat Settings
    wechat_backend: str = "wecom"
    wechat_webhook_port: int = 9001
    wechat_allowed_senders: str = ""
    wechat_proxy: str = ""
    wechat_wecom_corp_id: str = ""
    wechat_wecom_agent_id: str = ""
    wechat_wecom_secret: str = ""
    wechat_wecom_token: str = ""
    wechat_wecom_encoding_aes_key: str = ""
    wechat_mp_app_id: str = ""
    wechat_mp_app_secret: str = ""
    wechat_mp_token: str = ""
    wechat_mp_encoding_aes_key: str = ""
    # Personal WeChat (iLink Bot) — credentials obtained via QR-code login.
    # Run: python -m EvoScientist.channels.wechat.serve --qr-login
    wechat_personal_account_id: str = ""
    wechat_personal_token: str = ""
    wechat_personal_base_url: str = ""
    wechat_personal_cdn_base_url: str = ""
    wechat_personal_dm_policy: str = "open"
    wechat_personal_group_policy: str = "disabled"
    wechat_personal_group_allowed: str = ""

    # DingTalk Settings
    dingtalk_client_id: str = ""
    dingtalk_client_secret: str = ""
    dingtalk_allowed_senders: str = ""
    dingtalk_proxy: str = ""

    # Email Settings
    email_imap_host: str = ""
    email_imap_port: int = 993
    email_imap_username: str = ""
    email_imap_password: str = ""
    email_imap_mailbox: str = "INBOX"
    email_imap_use_ssl: bool = True
    email_smtp_host: str = ""
    email_smtp_port: int = 587
    email_smtp_username: str = ""
    email_smtp_password: str = ""
    email_smtp_use_tls: bool = True
    email_from_address: str = ""
    email_poll_interval: int = 30
    email_mark_seen: bool = True
    email_max_body_chars: int = 12000
    email_subject_prefix: str = "Re: "
    email_allowed_senders: str = ""

    # QQ Settings
    qq_app_id: str = ""
    qq_app_secret: str = ""
    qq_allowed_senders: str = ""

    # Signal Settings
    signal_phone_number: str = ""
    signal_cli_path: str = "signal-cli"
    signal_config_dir: str = ""
    signal_allowed_senders: str = ""
    signal_rpc_port: int = 7583

    # Shared webhook port (0 = disabled)
    shared_webhook_port: int = 9000

    # HITL (Human-in-the-Loop) Settings
    auto_approve: bool = False  # Auto-approve all tool executions without prompting
    auto_mode: bool = False  # Run unattended: imply auto_approve and disable ask_user
    shell_allow_list: str = ""  # Comma-separated shell command prefixes to auto-approve

    # Dangerous mode: real-filesystem access (no workspace confinement). The agent
    # operates on real absolute paths anywhere on disk; the privileged-command
    # blocklist (sudo/chmod/dd/...) still applies. Implies auto_approve.
    dangerous_mode: bool = False

    # Agent features
    enable_ask_user: bool = True  # Enable ask_user tool for agent-initiated questions

    # CodeInterpreterMiddleware (PTC — Parallel Tool Calls) tuning
    # The PTC allowlist itself is hardcoded in
    # ``EvoScientist/middleware/code_interpreter.py`` as a load-bearing safety
    # decision (excludes ``execute`` so PTC can't bypass HITL approval,
    # excludes ``write_file``/``edit_file`` because batched writes have no
    # benefit). Only the resource budget knobs are user-tunable.
    code_interpreter_timeout: float = 60.0  # seconds per JS eval
    code_interpreter_max_result_chars: int = 10000  # truncate large JSON results

    # Default per-command timeout (seconds) for the sandbox `execute` tool.
    # Only the default — the agent can still override per command up to the
    # deepagents max_execute_timeout cap (3600s).
    sandbox_execute_timeout: int = 300

    # Checkpoint pruning (sessions.db retention per (thread_id, checkpoint_ns))
    # Safety net for runaway conversations. Under DeltaChannel (deepagents 0.6+)
    # normal usage produces linear growth, so this default is set well above
    # any realistic conversation length (~180-450 turns of dialogue) while
    # still capping legacy bloat at upgrade time. 0 disables ongoing pruning
    # entirely; the one-time legacy migration sweep still runs.
    checkpoint_keep_per_thread: int = 1000

    # DM access control policy
    dm_policy: str = "allowlist"

    # OpenAI API mode - "" = auto, "true" = force Responses, "false" = force Completions
    use_responses_api: str = ""

    # ccproxy
    ccproxy_port: int = 8000

    # STT (Speech-to-Text) Settings
    stt_enabled: bool = False
    stt_language: str = "auto"  # "auto" | "zh" | "en"
    stt_model: str = ""  # override model id; empty = auto-select by language
    stt_device: str = "cpu"  # "cpu" | "cuda"
    stt_compute_type: str = "int8"  # "int8" | "float16" | "float32"

    def __post_init__(self) -> None:
        # A non-positive or non-int sandbox_execute_timeout (e.g. a hand-edited
        # config file value — load_config does not coerce file values — or a
        # 0/negative env value) would raise inside CustomSandboxBackend.__init__
        # and crash agent/CLI startup. Fall back to the default instead, matching
        # how malformed env values already degrade to defaults.
        t = self.sandbox_execute_timeout
        if not isinstance(t, int) or isinstance(t, bool) or t <= 0:
            logging.getLogger(__name__).warning(
                "Invalid sandbox_execute_timeout %r; falling back to 300.", t
            )
            self.sandbox_execute_timeout = 300

        # auto_mode and dangerous_mode both imply auto_approve regardless of
        # source (CLI, env, config file, direct construction) — done here so the
        # "unattended → zero prompts" contract holds even when either is set via
        # `config set` or a config file rather than a CLI flag.
        if self.auto_mode or self.dangerous_mode:
            self.auto_approve = True

        _normalize_str_enum_fields(self)

        # Bind hosts reach socket.bind() / the langgraph CLI verbatim, where a
        # stray-whitespace or empty value surfaces as an opaque gaierror at
        # startup. Normalize to the field's own default instead.
        for _host_field, _host_default in (
            ("langgraph_dev_host", "127.0.0.1"),
            ("webui_host", "0.0.0.0"),
        ):
            _host = getattr(self, _host_field, _host_default)
            _host = _host.strip() if isinstance(_host, str) else ""
            setattr(self, _host_field, _host or _host_default)

        synthesis_time = _normalize_hhmm(self.memory_skill_synthesis_time)
        if synthesis_time is None:
            logging.getLogger(__name__).warning(
                "Invalid memory_skill_synthesis_time %r; falling back to %s.",
                self.memory_skill_synthesis_time,
                DEFAULT_MEMORY_SKILL_SYNTHESIS_TIME,
            )
            self.memory_skill_synthesis_time = DEFAULT_MEMORY_SKILL_SYNTHESIS_TIME
        else:
            self.memory_skill_synthesis_time = synthesis_time


@dataclass(frozen=True)
class MemoryControls:
    """Resolved memory feature switches used by agent and worker wiring."""

    profile_enabled: bool
    observations_enabled: bool
    observation_writer: MemoryObservationWriter
    workers_enabled: bool

    @classmethod
    def from_config(cls, config: EvoScientistConfig) -> MemoryControls:
        return cls(
            profile_enabled=config.memory_profile_enabled,
            observations_enabled=config.memory_observations_enabled,
            observation_writer=config.memory_observation_writer,
            workers_enabled=config.memory_workers_enabled,
        )

    @property
    def memory_enabled(self) -> bool:
        return self.profile_enabled or self.observations_enabled

    def observation_tool_enabled(self, target: MemoryObservationTarget) -> bool:
        return self.observations_enabled and self.observation_writer.enables(target)

    def worker_needed(self, target: MemoryObservationTarget) -> bool:
        if not self.workers_enabled:
            return False
        match target:
            case MemoryObservationTarget.TURN_WORKER:
                return self.profile_enabled or self.observation_tool_enabled(target)
            case MemoryObservationTarget.SUBAGENT_WORKER:
                return self.profile_enabled or self.observation_tool_enabled(target)
            case MemoryObservationTarget.AGENT:
                return False


# =============================================================================
# Config file operations
# =============================================================================


def load_config() -> EvoScientistConfig:
    """Load configuration from file.

    Returns:
        EvoScientistConfig instance with values from file, or defaults if
        file doesn't exist.
    """
    config_path = get_config_path()

    if not config_path.exists():
        return EvoScientistConfig()

    try:
        with open(config_path, encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}

        # Filter to only valid fields
        valid_fields = {f.name for f in fields(EvoScientistConfig)}
        filtered_data = {k: v for k, v in data.items() if k in valid_fields}

        return EvoScientistConfig(**filtered_data)
    except Exception:
        # On any error, return defaults
        return EvoScientistConfig()


def save_config(config: EvoScientistConfig) -> None:
    """Save configuration to file.

    Args:
        config: EvoScientistConfig instance to save.
    """
    config_path = get_config_path()
    config_path.parent.mkdir(parents=True, exist_ok=True)
    try:
        config_path.parent.chmod(0o700)
    except OSError:
        pass

    data = _config_to_dict(config)

    # Save all fields including empty API keys (users can set them via env vars instead)
    with open(config_path, "w", encoding="utf-8") as f:
        yaml.safe_dump(
            data,
            f,
            default_flow_style=False,
            sort_keys=False,
            allow_unicode=True,
        )
    try:
        config_path.chmod(0o600)
    except OSError:
        pass


def reset_config() -> None:
    """Reset configuration to defaults by deleting the config file."""
    config_path = get_config_path()
    if config_path.exists():
        config_path.unlink()


def _config_to_dict(config: EvoScientistConfig) -> dict[str, Any]:
    """Return a plain serializable config dict."""
    return {key: _plain_config_value(value) for key, value in asdict(config).items()}


# =============================================================================
# Config value operations
# =============================================================================


def _coerce_value(value: Any, field_type: Any) -> Any:
    """Coerce a value to the expected field type.

    Args:
        value: The value to coerce.
        field_type: The target type (from dataclass field).

    Returns:
        The coerced value.

    Raises:
        ValueError: If the value cannot be coerced.
        TypeError: If the value cannot be coerced.
    """
    if _is_str_enum_type(field_type):
        return field_type(str(value).strip().lower())
    if field_type == "bool" or field_type is bool:
        if isinstance(value, str):
            return value.lower() in ("true", "1", "yes", "on")
        return bool(value)
    if field_type == "int" or field_type is int:
        return int(value)
    if field_type == "float" or field_type is float:
        return float(value)
    return str(value)


def _plain_config_value(value: Any) -> Any:
    """Return the persisted/user-facing representation for a config value."""
    return value.value if isinstance(value, StrEnum) else value


@lru_cache(maxsize=1)
def _config_field_types() -> dict[str, Any]:
    """Return resolved dataclass annotations for config fields."""
    return get_type_hints(EvoScientistConfig)


def _config_field_type(key: str, fallback: Any) -> Any:
    """Return the resolved dataclass annotation for a config field."""
    return _config_field_types().get(key, fallback)


def _is_str_enum_type(field_type: Any) -> bool:
    return isinstance(field_type, type) and issubclass(field_type, StrEnum)


def _normalize_str_enum_fields(config: EvoScientistConfig) -> None:
    """Normalize all StrEnum config fields, falling back to field defaults."""
    for field in fields(config):
        field_type = _config_field_type(field.name, field.type)
        if not _is_str_enum_type(field_type):
            continue

        raw_value = getattr(config, field.name)
        try:
            value = _coerce_value(raw_value, field_type)
        except (ValueError, TypeError):
            default = field.default
            logging.getLogger(__name__).warning(
                "Invalid %s %r; falling back to %s.",
                field.name,
                raw_value,
                _plain_config_value(default),
            )
            value = default
        setattr(config, field.name, value)


def get_config_value(key: str) -> Any:
    """Get a single configuration value.

    Args:
        key: Configuration key name.

    Returns:
        The value, or None if key doesn't exist.
    """
    config = load_config()
    value = getattr(config, key, None)
    return _plain_config_value(value)


def set_config_value(key: str, value: Any) -> bool:
    """Set a single configuration value.

    Args:
        key: Configuration key name.
        value: New value.

    Returns:
        True if successful, False if key is invalid.
    """
    valid_fields = {f.name for f in fields(EvoScientistConfig)}
    if key not in valid_fields:
        return False

    config = load_config()

    # Type coercion based on field type
    field_info = next(f for f in fields(EvoScientistConfig) if f.name == key)
    field_type = _config_field_type(key, field_info.type)

    # __post_init__ only clamps on load, so validate here too. Reject bool before coercion
    # (_coerce_value(True, int) would turn it into 1 and slip past).
    if key == "sandbox_execute_timeout" and isinstance(value, bool):
        return False

    try:
        value = _coerce_value(value, field_type)
    except (ValueError, TypeError):
        return False

    if key == "sandbox_execute_timeout" and value <= 0:
        return False
    if key == "memory_skill_synthesis_time":
        value = _normalize_hhmm(value)
        if value is None:
            return False

    setattr(config, key, value)
    save_config(config)
    return True


def list_config() -> dict[str, Any]:
    """List all configuration values.

    Returns:
        Dictionary of all configuration key-value pairs.
    """
    return _config_to_dict(load_config())


# =============================================================================
# Effective configuration (merging sources)
# =============================================================================

# Environment variable mappings
_ENV_MAPPINGS = {
    "anthropic_api_key": "ANTHROPIC_API_KEY",
    "anthropic_base_url": "ANTHROPIC_BASE_URL",
    "anthropic_auth_mode": "EVOSCIENTIST_ANTHROPIC_AUTH_MODE",
    "openai_api_key": "OPENAI_API_KEY",
    "openai_auth_mode": "EVOSCIENTIST_OPENAI_AUTH_MODE",
    "nvidia_api_key": "NVIDIA_API_KEY",
    "google_api_key": "GOOGLE_API_KEY",
    "minimax_api_key": "MINIMAX_API_KEY",
    "minimax_base_url": "MINIMAX_BASE_URL",
    "siliconflow_api_key": "SILICONFLOW_API_KEY",
    "openrouter_api_key": "OPENROUTER_API_KEY",
    "atlascloud_api_key": "ATLASCLOUD_API_KEY",
    "requesty_api_key": "REQUESTY_API_KEY",
    "deepseek_api_key": "DEEPSEEK_API_KEY",
    "zhipu_api_key": "ZHIPU_API_KEY",
    "volcengine_api_key": "VOLCENGINE_API_KEY",
    "dashscope_api_key": "DASHSCOPE_API_KEY",
    "moonshot_api_key": "MOONSHOT_API_KEY",
    "kimi_api_key": "KIMI_API_KEY",
    "custom_openai_api_key": "CUSTOM_OPENAI_API_KEY",
    "custom_openai_base_url": "CUSTOM_OPENAI_BASE_URL",
    "custom_anthropic_api_key": "CUSTOM_ANTHROPIC_API_KEY",
    "custom_anthropic_base_url": "CUSTOM_ANTHROPIC_BASE_URL",
    "ollama_base_url": "OLLAMA_BASE_URL",
    "tavily_api_key": "TAVILY_API_KEY",
    "default_mode": "EVOSCIENTIST_DEFAULT_MODE",
    "default_workdir": "EVOSCIENTIST_WORKSPACE_DIR",
    "ui_backend": "EVOSCIENTIST_UI_BACKEND",
    "log_level": "EVOSCIENTIST_LOG_LEVEL",
    "model_fallbacks": "EVOSCIENTIST_MODEL_FALLBACKS",
    "auxiliary_provider": "EVOSCIENTIST_AUXILIARY_PROVIDER",
    "auxiliary_model": "EVOSCIENTIST_AUXILIARY_MODEL",
    "reasoning_effort": "EVOSCIENTIST_REASONING_EFFORT",
    "openrouter_anthropic_prompt_cache": (
        "EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE"
    ),
    "openrouter_http_referer": "EVOSCIENTIST_OPENROUTER_HTTP_REFERER",
    "openrouter_app_title": "EVOSCIENTIST_OPENROUTER_APP_TITLE",
    "openrouter_app_categories": "EVOSCIENTIST_OPENROUTER_APP_CATEGORIES",
    "dangerous_mode": "EVOSCIENTIST_DANGEROUS_MODE",
    "channel_debug_tracing": "EVOSCIENTIST_CHANNEL_DEBUG_TRACING",
    "ccproxy_port": "EVOSCIENTIST_CCPROXY_PORT",
    "use_responses_api": "EVOSCIENTIST_USE_RESPONSES_API",
    "checkpoint_keep_per_thread": "EVOSCIENTIST_CHECKPOINT_KEEP_PER_THREAD",
    "enable_async_subagents": "EVOSCIENTIST_ENABLE_ASYNC_SUBAGENTS",
    "langgraph_dev_port": "EVOSCIENTIST_LANGGRAPH_DEV_PORT",
    "langgraph_dev_host": "EVOSCIENTIST_LANGGRAPH_DEV_HOST",
    "webui_port": "EVOSCIENTIST_WEBUI_PORT",
    "webui_host": "EVOSCIENTIST_WEBUI_HOST",
    "enable_scheduler": "EVOSCIENTIST_ENABLE_SCHEDULER",
    "scheduler_default_timezone": "EVOSCIENTIST_SCHEDULER_DEFAULT_TIMEZONE",
    "code_interpreter_timeout": "EVOSCIENTIST_CODE_INTERPRETER_TIMEOUT",
    "code_interpreter_max_result_chars": "EVOSCIENTIST_CODE_INTERPRETER_MAX_RESULT_CHARS",
    "sandbox_execute_timeout": "EVOSCIENTIST_SANDBOX_EXECUTE_TIMEOUT",
    "langgraph_dev_file_persistence": "EVOSCIENTIST_LANGGRAPH_DEV_FILE_PERSISTENCE",
    "langgraph_dev_jobs_per_worker": "EVOSCIENTIST_LANGGRAPH_DEV_JOBS_PER_WORKER",
    "recursion_limit": "EVOSCIENTIST_RECURSION_LIMIT",
    "memory_profile_enabled": "EVOSCIENTIST_MEMORY_PROFILE_ENABLED",
    "memory_observations_enabled": "EVOSCIENTIST_MEMORY_OBSERVATIONS_ENABLED",
    "memory_observation_writer": "EVOSCIENTIST_MEMORY_OBSERVATION_WRITER",
    "memory_workers_enabled": "EVOSCIENTIST_MEMORY_WORKERS_ENABLED",
    "memory_skill_synthesis_enabled": "EVOSCIENTIST_MEMORY_SKILL_SYNTHESIS_ENABLED",
    "memory_skill_synthesis_mode": "EVOSCIENTIST_MEMORY_SKILL_SYNTHESIS_MODE",
    "memory_skill_synthesis_cadence": "EVOSCIENTIST_MEMORY_SKILL_SYNTHESIS_CADENCE",
    "memory_skill_synthesis_time": "EVOSCIENTIST_MEMORY_SKILL_SYNTHESIS_TIME",
}


def get_effective_config(
    cli_overrides: dict[str, Any] | None = None,
) -> EvoScientistConfig:
    """Get effective configuration by merging all sources.

    Priority (highest to lowest):
        1. CLI arguments (``cli_overrides``)
        2. Parent-process environment variables for any ``EVOSCIENTIST_*`` key
        3. ``.env`` file at (or above) the current working directory
        4. Parent-process environment variables for everything else
           (third-party API keys / base URLs, plus arbitrary unmapped keys)
        5. Config file (``~/.config/evoscientist/config.yaml``)
        6. Dataclass defaults

    Rows 2 and 4 differ because ``.env`` values need different treatment
    for our own namespaced config knobs vs third-party credentials.
    Third-party keys (``ANTHROPIC_API_KEY``, ``OPENAI_API_KEY``, ...)
    follow the industry convention that ``.env`` is the per-project
    credential store; extending shell-wins to them would silently flip
    a workspace key back to a global ``.bashrc`` key. Our own
    ``EVOSCIENTIST_*`` keys are the opposite: an explicit CLI/parent-
    process value (e.g. the bind port that ``EvoSci deploy --port X``
    hands to the langgraph dev subprocess) must not be shadowed by a
    workspace ``.env``. We implement this by reading ``.env`` into a
    dict via ``dotenv_values`` (no ``os.environ`` mutation), then
    writing third-party keys unconditionally and ``EVOSCIENTIST_*`` keys
    only when the shell doesn't already have a non-empty value.

    Tradeoff: ``OPENAI_API_KEY=xxx evoscientist ...`` inline overrides
    still lose to a workspace ``.env`` containing ``OPENAI_API_KEY``,
    because the merge writes third-party keys from ``.env``
    unconditionally. Users who need to override a ``.env``-defined
    credential inline must edit or unset the ``.env`` entry.

    Args:
        cli_overrides: Dictionary of CLI argument overrides.

    Returns:
        EvoScientistConfig with merged values.
    """
    # Merge workspace ``.env`` into ``os.environ`` without going through
    # ``load_dotenv``. The previous snapshot → ``load_dotenv`` → restore
    # sequence was a read-modify-write on ``os.environ`` that could race with
    # concurrent ``get_effective_config`` calls in the langgraph dev subprocess
    # (per-request threads in ``langgraph_dev/http.py``, ``sessions.py``
    # checkpoint writes, memory workers): one thread's mid-flight ``.env``
    # value could be re-captured by another as "parent env" and then restored
    # last, promoting the ``.env`` value into the snapshot permanently.
    #
    # ``dotenv_values`` returns a dict without touching ``os.environ``, so the
    # merge below is a pure write sequence and idempotent under interleaving.
    # Third-party keys keep ``.env``-wins (industry convention).
    # ``EVOSCIENTIST_*`` keys are our own namespaced config knobs where
    # CLI/parent-process intent should stay authoritative — write from ``.env``
    # only when the shell doesn't already have a non-empty value. Treating an
    # empty shell value as "unset" matches the ``if env_value:`` truthy check
    # in the ``_ENV_MAPPINGS`` loop below; without this, an empty parent export
    # would silently regress vs main by falling through to file/defaults.
    dotenv_path = find_dotenv(usecwd=True)
    dotenv_map = dotenv_values(dotenv_path) if dotenv_path else {}
    for env_key, env_value in dotenv_map.items():
        if env_value is None:
            continue  # bare ``FOO`` without ``=`` — nothing to write
        if env_key.startswith("EVOSCIENTIST_"):
            if not os.environ.get(env_key):
                os.environ[env_key] = env_value
        else:
            os.environ[env_key] = env_value

    # Start with file config (includes defaults for missing values)
    config = load_config()
    data = _config_to_dict(config)

    # Apply environment variable overrides
    for config_key, env_key in _ENV_MAPPINGS.items():
        env_value = os.environ.get(env_key)
        if env_value:
            field_info = next(
                f for f in fields(EvoScientistConfig) if f.name == config_key
            )
            try:
                data[config_key] = _coerce_value(
                    env_value,
                    _config_field_type(config_key, field_info.type),
                )
            except (ValueError, TypeError):
                pass

    # Apply CLI overrides (highest priority)
    if cli_overrides:
        for key, value in cli_overrides.items():
            if value is not None and key in data:
                data[key] = value

    return EvoScientistConfig(**data)


def apply_config_to_env(config: EvoScientistConfig) -> None:
    """Apply config API keys to environment variables if not already set.

    This allows the config file to provide API keys that downstream
    libraries (like langchain-anthropic) can pick up.

    Args:
        config: Configuration to apply.
    """
    if config.anthropic_api_key and not os.environ.get("ANTHROPIC_API_KEY"):
        os.environ["ANTHROPIC_API_KEY"] = config.anthropic_api_key
    if config.anthropic_base_url and not os.environ.get("ANTHROPIC_BASE_URL"):
        os.environ["ANTHROPIC_BASE_URL"] = config.anthropic_base_url
    if config.openai_api_key and not os.environ.get("OPENAI_API_KEY"):
        os.environ["OPENAI_API_KEY"] = config.openai_api_key
    if config.nvidia_api_key and not os.environ.get("NVIDIA_API_KEY"):
        os.environ["NVIDIA_API_KEY"] = config.nvidia_api_key
    if config.google_api_key and not os.environ.get("GOOGLE_API_KEY"):
        os.environ["GOOGLE_API_KEY"] = config.google_api_key
    if config.minimax_api_key and not os.environ.get("MINIMAX_API_KEY"):
        os.environ["MINIMAX_API_KEY"] = config.minimax_api_key
    if config.minimax_base_url and not os.environ.get("MINIMAX_BASE_URL"):
        os.environ["MINIMAX_BASE_URL"] = config.minimax_base_url
    if config.siliconflow_api_key and not os.environ.get("SILICONFLOW_API_KEY"):
        os.environ["SILICONFLOW_API_KEY"] = config.siliconflow_api_key
    if config.openrouter_api_key and not os.environ.get("OPENROUTER_API_KEY"):
        os.environ["OPENROUTER_API_KEY"] = config.openrouter_api_key
    if config.atlascloud_api_key and not os.environ.get("ATLASCLOUD_API_KEY"):
        os.environ["ATLASCLOUD_API_KEY"] = config.atlascloud_api_key
    if config.requesty_api_key and not os.environ.get("REQUESTY_API_KEY"):
        os.environ["REQUESTY_API_KEY"] = config.requesty_api_key
    if config.deepseek_api_key and not os.environ.get("DEEPSEEK_API_KEY"):
        os.environ["DEEPSEEK_API_KEY"] = config.deepseek_api_key
    if config.zhipu_api_key and not os.environ.get("ZHIPU_API_KEY"):
        os.environ["ZHIPU_API_KEY"] = config.zhipu_api_key
    if config.volcengine_api_key and not os.environ.get("VOLCENGINE_API_KEY"):
        os.environ["VOLCENGINE_API_KEY"] = config.volcengine_api_key
    if config.dashscope_api_key and not os.environ.get("DASHSCOPE_API_KEY"):
        os.environ["DASHSCOPE_API_KEY"] = config.dashscope_api_key
    if config.moonshot_api_key and not os.environ.get("MOONSHOT_API_KEY"):
        os.environ["MOONSHOT_API_KEY"] = config.moonshot_api_key
    if config.kimi_api_key and not os.environ.get("KIMI_API_KEY"):
        os.environ["KIMI_API_KEY"] = config.kimi_api_key
    if config.custom_openai_api_key and not os.environ.get("CUSTOM_OPENAI_API_KEY"):
        os.environ["CUSTOM_OPENAI_API_KEY"] = config.custom_openai_api_key
    if config.custom_openai_base_url and not os.environ.get("CUSTOM_OPENAI_BASE_URL"):
        os.environ["CUSTOM_OPENAI_BASE_URL"] = config.custom_openai_base_url
    if config.custom_anthropic_api_key and not os.environ.get(
        "CUSTOM_ANTHROPIC_API_KEY"
    ):
        os.environ["CUSTOM_ANTHROPIC_API_KEY"] = config.custom_anthropic_api_key
    if config.custom_anthropic_base_url and not os.environ.get(
        "CUSTOM_ANTHROPIC_BASE_URL"
    ):
        os.environ["CUSTOM_ANTHROPIC_BASE_URL"] = config.custom_anthropic_base_url
    if config.ollama_base_url and not os.environ.get("OLLAMA_BASE_URL"):
        os.environ["OLLAMA_BASE_URL"] = config.ollama_base_url
    if config.tavily_api_key and not os.environ.get("TAVILY_API_KEY"):
        os.environ["TAVILY_API_KEY"] = config.tavily_api_key
    if config.reasoning_effort and not os.environ.get("EVOSCIENTIST_REASONING_EFFORT"):
        os.environ["EVOSCIENTIST_REASONING_EFFORT"] = config.reasoning_effort
    if config.openrouter_http_referer and not os.environ.get(
        "EVOSCIENTIST_OPENROUTER_HTTP_REFERER"
    ):
        os.environ["EVOSCIENTIST_OPENROUTER_HTTP_REFERER"] = (
            config.openrouter_http_referer
        )
    if config.openrouter_app_title and not os.environ.get(
        "EVOSCIENTIST_OPENROUTER_APP_TITLE"
    ):
        os.environ["EVOSCIENTIST_OPENROUTER_APP_TITLE"] = config.openrouter_app_title
    if config.openrouter_app_categories and not os.environ.get(
        "EVOSCIENTIST_OPENROUTER_APP_CATEGORIES"
    ):
        os.environ["EVOSCIENTIST_OPENROUTER_APP_CATEGORIES"] = (
            config.openrouter_app_categories
        )
    if not config.openrouter_anthropic_prompt_cache and not os.environ.get(
        "EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE"
    ):
        os.environ["EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE"] = "false"
    # Round-trip dangerous_mode to env so it survives a fresh get_effective_config()
    # (warning banner, run_in_background) and is inherited by the langgraph dev
    # subprocess — otherwise a --dangerous CLI flag (not persisted to file/env)
    # is invisible to those consumers while the backend is already unconfined.
    # Bidirectional: clear it when off so a re-apply with a lower config (or a
    # stale value) can't leave the process stuck in dangerous mode.
    if config.dangerous_mode:
        os.environ["EVOSCIENTIST_DANGEROUS_MODE"] = "true"
    else:
        os.environ.pop("EVOSCIENTIST_DANGEROUS_MODE", None)
    if config.use_responses_api and not os.environ.get(
        "EVOSCIENTIST_USE_RESPONSES_API"
    ):
        os.environ["EVOSCIENTIST_USE_RESPONSES_API"] = config.use_responses_api