"""Configuration, constants, and model creation."""

from __future__ import annotations

import importlib
import json
import keyword
import logging
import os
import re
import shlex
import shutil
import sys
import threading
from dataclasses import dataclass
from enum import StrEnum
from importlib.metadata import PackageNotFoundError, distribution, version
from pathlib import Path
from typing import TYPE_CHECKING, Any, Protocol
from urllib.parse import unquote, urlparse

from deepagents_code._env_vars import HIDE_SPLASH_VERSION, is_env_truthy
from deepagents_code._git import resolve_git_branch
from deepagents_code._version import __version__
from deepagents_code.config_manifest import (
    INTERPRETER_ENABLE_DEFAULT,
    INTERPRETER_MAX_PTC_CALLS_DEFAULT,
    INTERPRETER_MAX_RESULT_CHARS_DEFAULT,
    INTERPRETER_MEMORY_LIMIT_MB_DEFAULT,
    INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE_DEFAULT,
    INTERPRETER_PTC_DEFAULT,
    INTERPRETER_TIMEOUT_SECONDS_DEFAULT,
)

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Lazy bootstrap: dotenv loading, LANGSMITH_PROJECT override, and start-path
# detection are deferred until first access of `settings` (via module
# `__getattr__`).  This avoids disk I/O and path traversal during import for
# callers that never touch `settings` (e.g. `deepagents --help`).
# ---------------------------------------------------------------------------

_bootstrap_done = False
"""Whether `_ensure_bootstrap()` has executed."""

_bootstrap_lock = threading.Lock()
"""Guards `_ensure_bootstrap()` against concurrent access from the main thread
and the prewarm worker thread."""

_singleton_lock = threading.Lock()
"""Guards lazy singleton construction in `_get_console` / `_get_settings`."""

_bootstrap_start_path: Path | None = None
"""Working directory captured at bootstrap time for dotenv and project discovery."""

_original_langsmith_project: str | None = None
"""Caller's `LANGSMITH_PROJECT` value before the app overrides it for agent traces.

Captured inside `_ensure_bootstrap()` after dotenv loading but before the
`LANGSMITH_PROJECT` override, so `.env`-only values are visible.
"""

_dotenv_loaded_values: dict[str, str] = {}
"""Environment values injected by our dotenv loader and safe to refresh later."""

_orphaned_tracing_disabled_notice: str | None = None
"""One-shot TUI notice populated when bootstrap disables orphaned tracing."""

_INHERITED_PYTHONPATH_ENV = "DEEPAGENTS_INHERITED_PYTHONPATH"
"""Carrier var that relays a launch-time `PYTHONPATH` to agent `execute` commands.

`PYTHONPATH` is stripped from the server interpreter's environment (see
`server._SERVER_ENV_DENYLIST`) to keep an untrusted import path off `sys.path`
during startup. The launch-time value is instead carried in this var and
re-applied only to the approval-gated shell backend's `execute` subprocesses by
`agent._apply_inherited_pythonpath`.
"""

_DOTENV_DENIED_ENV_KEYS = frozenset(
    {
        "DYLD_INSERT_LIBRARIES",
        "DYLD_LIBRARY_PATH",
        "GIT_ASKPASS",
        "LD_AUDIT",
        "LD_LIBRARY_PATH",
        "LD_PRELOAD",
        "NODE_OPTIONS",
        "PATH",
        "PYTHONEXECUTABLE",
        "PYTHONHOME",
        "PYTHONPATH",
        "PYTHONSTARTUP",
        "SSH_ASKPASS",
        _INHERITED_PYTHONPATH_ENV,
    }
)
"""Environment keys that project `.env` files must not inject.

`_INHERITED_PYTHONPATH_ENV` is denied so a project `.env` cannot smuggle a
`PYTHONPATH` into agent `execute` commands through the carrier var; the carrier
is only meant to relay a value the user set in their launch environment."""


def _find_dotenv_from_start_path(start_path: Path) -> Path | None:
    """Find the nearest `.env` file from an explicit start path upward.

    Args:
        start_path: Directory to start searching from.

    Returns:
        Path to the nearest `.env` file, or `None` if not found.
    """
    current = start_path.expanduser().resolve()
    for parent in [current, *list(current.parents)]:
        candidate = parent / ".env"
        try:
            if candidate.is_file():
                return candidate
        except OSError:
            logger.warning("Could not inspect .env candidate %s", candidate)
            continue
    return None


# Global user-level .env (~/.deepagents/.env); sentinel when Path.home() fails.
try:
    _GLOBAL_DOTENV_PATH = Path.home() / ".deepagents" / ".env"
except RuntimeError:
    _GLOBAL_DOTENV_PATH = Path("/nonexistent/.deepagents/.env")


def _preview_dotenv_environ(*, start_path: Path | None = None) -> dict[str, str]:
    """Return the environment after dotenv loading without mutating `os.environ`.

    Args:
        start_path: Directory to use for project `.env` discovery.

    Returns:
        Environment mapping with project and global dotenv values applied using
        the same first-write-wins precedence as `_load_dotenv`.
    """
    import dotenv

    env = dict(os.environ)
    for key, value in _dotenv_loaded_values.items():
        if env.get(key) == value:
            env.pop(key)

    def apply_dotenv(dotenv_path: Path | None) -> None:
        if dotenv_path is None:
            return
        try:
            values = dotenv.dotenv_values(dotenv_path=dotenv_path)
        except (OSError, ValueError):
            logger.warning(
                "Could not read dotenv at %s; previewed project env vars may be "
                "incomplete",
                dotenv_path,
                exc_info=True,
            )
            return
        for key, value in values.items():
            if value is not None and key not in env:
                env[key] = value

    project_dotenv: Path | None = None
    try:
        project_dotenv = (
            _find_dotenv_from_start_path(start_path)
            if start_path is not None
            else _find_dotenv_from_start_path(Path.cwd())
        )
    except OSError:
        logger.warning(
            "Could not inspect project dotenv at %s; previewed project env vars may "
            "be incomplete",
            start_path or "cwd",
            exc_info=True,
        )
    apply_dotenv(project_dotenv)

    try:
        global_dotenv = _GLOBAL_DOTENV_PATH if _GLOBAL_DOTENV_PATH.is_file() else None
    except OSError:
        logger.warning(
            "Could not inspect global dotenv at %s; previewed global defaults may "
            "be incomplete",
            _GLOBAL_DOTENV_PATH,
            exc_info=True,
        )
        global_dotenv = None
    apply_dotenv(global_dotenv)

    return env


def _resolve_env_var_from(env: dict[str, str], name: str) -> str | None:
    """Resolve an env var from a mapping using app prefix precedence.

    Returns:
        The resolved value, or `None` when absent or empty.
    """
    from deepagents_code.model_config import _ENV_PREFIX

    if not name.startswith(_ENV_PREFIX):
        prefixed = f"{_ENV_PREFIX}{name}"
        if prefixed in env:
            return env[prefixed] or None
    return env.get(name) or None


def _load_dotenv(
    *, start_path: Path | None = None, refresh_loaded: bool = False
) -> bool:
    """Load environment variables from project and global `.env` files.

    Loads in order (first write wins, `override=False`):

    1. Project/CWD `.env` — project-specific values
    2. `~/.deepagents/.env` — global user defaults

    Both layers use `override=False` (the python-dotenv default) so that
    shell-exported variables always take precedence over dotenv files.
    Because project loads first, the effective precedence is:

    ```text
    shell env (incl. inline `VAR=x`)  >  project `.env`  >  global `.env`
    ```

    !!! note

        To scope credentials to the app without colliding with
        identically-named shell exports, use the `DEEPAGENTS_CODE_` env-var
        prefix (see `resolve_env_var` in `deepagents_code.model_config`).

    Args:
        start_path: Directory to use for project `.env` discovery.
        refresh_loaded: Remove values previously injected by this loader before
            applying the current project/global dotenv stack. Values modified
            after loading are preserved.

    Returns:
        `True` when at least one dotenv file was loaded, `False` otherwise.
    """
    import dotenv

    loaded = False

    if refresh_loaded:
        for key, value in list(_dotenv_loaded_values.items()):
            if os.environ.get(key) == value:
                os.environ.pop(key)
        _dotenv_loaded_values.clear()

    def apply_dotenv(dotenv_path: Path) -> bool:
        values = dotenv.dotenv_values(dotenv_path=dotenv_path)
        applied = False
        for key, value in values.items():
            if value is None or key in os.environ or key in _DOTENV_DENIED_ENV_KEYS:
                continue
            os.environ[key] = value
            _dotenv_loaded_values[key] = value
            applied = True
        return applied

    # 1. Project/CWD .env — loads first so project values are set before the
    # global file, which can only fill in vars not already present.
    dotenv_path: Path | str | None = None
    try:
        if start_path is None:
            found = dotenv.find_dotenv(usecwd=True)
            if found:
                dotenv_path = found
                loaded = apply_dotenv(Path(found)) or loaded
        else:
            dotenv_path = _find_dotenv_from_start_path(start_path)
            if dotenv_path is not None:
                loaded = apply_dotenv(dotenv_path) or loaded
    except (OSError, ValueError):
        logger.warning(
            "Could not read project dotenv at %s; project env vars will not be loaded",
            dotenv_path or start_path or "cwd",
            exc_info=True,
        )

    # 2. Global (~/.deepagents/.env) — fills in any vars not already set by
    # the shell or the project dotenv.
    # try/except wraps both is_file() and load_dotenv() to cover the TOCTOU
    # window where the file can vanish between stat and open.
    try:
        if _GLOBAL_DOTENV_PATH.is_file() and apply_dotenv(_GLOBAL_DOTENV_PATH):
            loaded = True
            logger.debug("Loaded global dotenv: %s", _GLOBAL_DOTENV_PATH)
    except (OSError, ValueError):
        logger.warning(
            "Could not read global dotenv at %s; global defaults will not be applied",
            _GLOBAL_DOTENV_PATH,
            exc_info=True,
        )

    return loaded


_TRACING_ENABLE_ENV_VARS = (
    "LANGSMITH_TRACING_V2",
    "LANGCHAIN_TRACING_V2",
    "LANGSMITH_TRACING",
    "LANGCHAIN_TRACING",
)
"""Env vars LangChain/LangSmith read to decide whether tracing is enabled."""

_TRACING_API_KEY_ENV_VARS = ("LANGSMITH_API_KEY", "LANGCHAIN_API_KEY")
"""Env vars that hold the LangSmith API key used for trace ingestion."""

_TRACING_ENDPOINT_ENV_VARS = ("LANGSMITH_ENDPOINT", "LANGCHAIN_ENDPOINT")
"""Env vars that point tracing at a non-default (self-hosted/proxied) endpoint."""


class _LangSmithProfileConfig(Protocol):
    """Subset of LangSmith profile client config fields used at bootstrap."""

    api_url: str | None
    """Base URL for a custom self-hosted or proxied LangSmith endpoint."""

    api_key: str | None
    """API key from the active LangSmith profile."""

    oauth_access_token: str | None
    """OAuth access token from the active LangSmith profile."""

    oauth_refresh_token: str | None
    """OAuth refresh token from the active LangSmith profile."""


def _quiet_sdk_tracing_logging() -> None:
    """Keep LangSmith/LangChain SDK logging from corrupting the TUI.

    These SDK loggers emit ingestion/auth errors (e.g. repeated 401s) on their
    own loggers. With no handler attached they reach Python's last-resort stderr
    handler and bleed onto the alternate-screen TUI. Route them to the debug log
    when `DEEPAGENTS_CODE_DEBUG` is set, otherwise attach a `NullHandler` so they
    stay off the terminal.
    """
    from deepagents_code._debug import configure_debug_logging

    for name in ("langsmith", "langchain"):
        sdk_logger = logging.getLogger(name)
        configure_debug_logging(sdk_logger)
        if not sdk_logger.handlers:
            sdk_logger.addHandler(logging.NullHandler())


def _load_langsmith_profile_config() -> _LangSmithProfileConfig | None:
    """Return the active LangSmith profile client config, if available."""
    try:
        client_module = importlib.import_module("langsmith.client")
    except ImportError:
        return None

    profiles = getattr(client_module, "_profiles", None)
    if profiles is None:
        return None

    return profiles.load_profile_client_config()


def _has_langsmith_profile_credentials() -> bool:
    """Return whether the LangSmith profile config has usable auth material."""
    config = _load_langsmith_profile_config()
    if config is None:
        return False

    return bool(
        config.api_key or config.oauth_access_token or config.oauth_refresh_token
    )


def _has_langsmith_profile_custom_endpoint() -> bool:
    """Return whether the LangSmith profile points at a custom endpoint."""
    config = _load_langsmith_profile_config()
    if config is None:
        return False

    return bool((config.api_url or "").strip())


def _build_orphaned_tracing_disabled_notice() -> str:
    """Return the user-facing notice for disabled orphaned tracing."""
    base = (
        "LangSmith tracing was disabled because tracing is enabled but no "
        "credentials were found."
    )
    if shutil.which("langsmith"):
        return (
            f"{base} Set LANGSMITH_API_KEY or run `langsmith auth login`, "
            "then restart Deep Agents Code."
        )
    return f"{base} Set LANGSMITH_API_KEY, then restart Deep Agents Code."


def consume_orphaned_tracing_disabled_notice() -> str | None:
    """Return and clear the pending orphaned-tracing notice, if any."""
    global _orphaned_tracing_disabled_notice  # noqa: PLW0603

    notice = _orphaned_tracing_disabled_notice
    _orphaned_tracing_disabled_notice = None
    return notice


def _disable_orphaned_tracing() -> None:
    """Disable LangSmith tracing when enabled without a usable API key.

    LangChain enables tracing whenever a tracing flag is truthy, regardless of
    credentials. With no env or profile key the background tracer retries
    ingestion and floods `langsmith.client` 401 errors into the TUI (most visibly
    at the atexit flush). When a tracing flag is set but no credentials are
    resolvable, unset the flags so tracing never starts.

    A custom endpoint (`LANGSMITH_ENDPOINT`/`LANGCHAIN_ENDPOINT`, or a profile
    `api_url`) signals a self-hosted or proxied LangSmith that may ingest without
    an API key, so an explicitly configured endpoint is trusted and left alone
    rather than risk disabling a working keyless setup. The SDK loggers are
    quieted separately by `_quiet_sdk_tracing_logging`, so any residual ingest
    errors stay off the TUI.
    """
    global _orphaned_tracing_disabled_notice  # noqa: PLW0603

    from deepagents_code._env_vars import classify_env_bool

    tracing_on = any(
        classify_env_bool(os.environ[var])
        for var in _TRACING_ENABLE_ENV_VARS
        if var in os.environ
    )
    if not tracing_on:
        return

    has_custom_endpoint = any(
        (os.environ.get(var) or "").strip() for var in _TRACING_ENDPOINT_ENV_VARS
    )
    if has_custom_endpoint or _has_langsmith_profile_custom_endpoint():
        return

    has_key = any(
        (os.environ.get(var) or "").strip() for var in _TRACING_API_KEY_ENV_VARS
    )
    if has_key or _has_langsmith_profile_credentials():
        return

    disabled = [var for var in _TRACING_ENABLE_ENV_VARS if var in os.environ]
    for var in disabled:
        os.environ[var] = "false"
    _orphaned_tracing_disabled_notice = _build_orphaned_tracing_disabled_notice()
    logger.warning(
        "LangSmith tracing is enabled (%s) but no API key is set; disabling "
        "tracing to avoid repeated authentication failures. Set LANGSMITH_API_KEY "
        "to enable tracing, or unset the tracing flag to silence this warning.",
        ", ".join(disabled),
    )


def _ensure_bootstrap() -> None:
    """Run one-time bootstrap: dotenv loading and `LANGSMITH_PROJECT` override.

    Idempotent and thread-safe — subsequent calls are no-ops. Called
    automatically by `_get_settings()` when `settings` is first accessed.

    The flag is set in `finally` so that partial failures (e.g. a
    malformed `.env`) still mark bootstrap as done — preventing infinite retry
    loops. Exceptions are caught and logged at ERROR level; the app proceeds
    with the environment as-is.
    """
    global _bootstrap_done, _bootstrap_start_path, _original_langsmith_project  # noqa: PLW0603

    if _bootstrap_done:
        return

    with _bootstrap_lock:
        if _bootstrap_done:  # double-check after acquiring lock
            return

        try:
            from deepagents_code.project_utils import (
                get_server_project_context as _get_server_project_context,
            )

            ctx = _get_server_project_context()
            _bootstrap_start_path = ctx.user_cwd if ctx else None
            _load_dotenv(start_path=_bootstrap_start_path)

            # `configure_debug_logging` already ran at import, before the `.env`
            # above was loaded. Re-run it so a `DEEPAGENTS_CODE_DEBUG` set only in
            # `.env` installs the file handler now (idempotent for the same path),
            # ensuring later failures are actually written to the debug log.
            from deepagents_code._debug import configure_debug_logging

            configure_debug_logging(logging.getLogger("deepagents_code"))

            # Keep LangSmith/LangChain SDK logging off the TUI (route to the
            # debug log when enabled, else swallow via NullHandler).
            _quiet_sdk_tracing_logging()

            # Capture AFTER dotenv loading so .env-only values are visible,
            # but BEFORE the override below replaces it.
            _original_langsmith_project = os.environ.get("LANGSMITH_PROJECT")

            # CRITICAL: Override LANGSMITH_PROJECT to route agent traces to a
            # separate project. LangSmith reads LANGSMITH_PROJECT at invocation
            # time, so we override it here and preserve the user's original
            # value for shell commands.
            from deepagents_code._env_vars import LANGSMITH_PROJECT

            deepagents_project = os.environ.get(LANGSMITH_PROJECT)
            if deepagents_project:
                os.environ["LANGSMITH_PROJECT"] = deepagents_project

            # Propagate prefixed LangSmith env vars to canonical names.
            # The app resolves prefixed vars via resolve_env_var(), but the
            # LangSmith SDK reads os.environ directly and has no knowledge
            # of the DEEPAGENTS_CODE_ prefix. Setting canonical vars here
            # bridges that gap.
            from deepagents_code.model_config import _ENV_PREFIX

            for canonical in (
                "LANGSMITH_API_KEY",
                "LANGCHAIN_API_KEY",
                "LANGSMITH_TRACING",
                "LANGCHAIN_TRACING_V2",
            ):
                prefixed = f"{_ENV_PREFIX}{canonical}"
                if prefixed not in os.environ:
                    continue
                prefixed_val = os.environ[prefixed]
                if canonical not in os.environ:
                    # Propagate (including empty string for explicit disable).
                    os.environ[canonical] = prefixed_val
                elif os.environ[canonical] != prefixed_val:
                    os.environ[canonical] = prefixed_val
                    logger.warning(
                        "Both %s and %s are set with different values; "
                        "using %s. Unset %s to silence this warning.",
                        canonical,
                        prefixed,
                        prefixed,
                        canonical,
                    )

            # Tracing enabled without a key floods the TUI with 401 ingest
            # errors; disable it before any traced run starts.
            _disable_orphaned_tracing()

            # Bridge stored service keys (e.g. Tavily web search, entered via
            # `/auth`) onto their canonical env vars before settings detection,
            # so a stored key activates the feature without exporting the var.
            from deepagents_code.model_config import apply_stored_service_credentials

            apply_stored_service_credentials()
        except Exception:
            logger.exception(
                "Bootstrap failed; .env values and LANGSMITH_PROJECT override "
                "may be missing. The app will proceed with environment as-is.",
            )
        finally:
            _bootstrap_done = True


if TYPE_CHECKING:
    from langchain_core.language_models import BaseChatModel
    from langchain_core.runnables import RunnableConfig
    from rich.console import Console

    # Static type stubs for lazy module attributes resolved by __getattr__.
    # At runtime these are created on first access by _get_settings() /
    # _get_console() and cached in globals().
    settings: Settings
    console: Console

MODE_PREFIXES: dict[str, str] = {
    "shell_incognito": "!!",
    "shell": "!",
    "command": "/",
}
"""Maps each non-normal mode to its trigger character."""

MODE_DISPLAY_GLYPHS: dict[str, str] = {
    "shell_incognito": "$",
    "shell": "$",
    "command": "/",
}
"""Maps each non-normal mode to its display glyph shown in the prompt/UI."""

if MODE_PREFIXES.keys() != MODE_DISPLAY_GLYPHS.keys():
    _only_prefixes = MODE_PREFIXES.keys() - MODE_DISPLAY_GLYPHS.keys()
    _only_glyphs = MODE_DISPLAY_GLYPHS.keys() - MODE_PREFIXES.keys()
    msg = (
        "MODE_PREFIXES and MODE_DISPLAY_GLYPHS have mismatched keys: "
        f"only in PREFIXES={_only_prefixes}, only in GLYPHS={_only_glyphs}"
    )
    raise ValueError(msg)

_MODE_PREFIXES_BY_LENGTH: tuple[tuple[str, str], ...] = tuple(
    sorted(MODE_PREFIXES.items(), key=lambda item: len(item[1]), reverse=True)
)
"""Mode entries ordered longest-prefix-first.

Pre-sorted at import so `detect_mode_prefix` runs in constant time per
keystroke without re-sorting.
"""


def detect_mode_prefix(text: str) -> tuple[str, str] | None:
    """Return the longest mode prefix and mode for `text`, if any.

    Longer prefixes win so multi-character triggers like `!!` are matched
    before their single-character prefixes (`!`).

    Args:
        text: Input text that may start with a mode trigger.

    Returns:
        Tuple of `(prefix, mode)` for the longest matching trigger, otherwise
        `None`.
    """
    for mode, prefix in _MODE_PREFIXES_BY_LENGTH:
        if text.startswith(prefix):
            return prefix, mode
    return None


class CharsetMode(StrEnum):
    """Character set mode for TUI display."""

    UNICODE = "unicode"
    """Always use Unicode glyphs (e.g. `⏺`, `✓`, `…`)."""

    ASCII = "ascii"
    """Always use ASCII-safe fallbacks (e.g. `(*)`, `[OK]`, `...`)."""

    AUTO = "auto"
    """Detect charset support at runtime and pick Unicode or ASCII."""


@dataclass(frozen=True)
class Glyphs:
    """Character glyphs for TUI display."""

    tool_prefix: str  # ⏺ vs (*)
    ellipsis: str  # … vs ...
    checkmark: str  # ✓ vs [OK]
    error: str  # ✗ vs [X]
    circle_empty: str  # ○ vs [ ]
    circle_filled: str  # ● vs [*]
    output_prefix: str  # ⎿ vs L
    spinner_frames: tuple[str, ...]  # Braille vs ASCII spinner
    pause: str  # ⏸ vs ||
    newline: str  # ⏎ vs \\n
    warning: str  # ⚠ vs [!]
    question: str  # ? vs [?]
    arrow_up: str  # up arrow vs ^
    arrow_down: str  # down arrow vs v
    bullet: str  # bullet vs -
    cursor: str  # cursor vs >
    disclosure_collapsed: str  # ▸ vs >
    disclosure_expanded: str  # ▾ vs v

    # Box-drawing characters
    box_vertical: str  # │ vs |
    box_horizontal: str  # ─ vs -
    box_double_horizontal: str  # ═ vs =

    # Diff-specific
    gutter_bar: str  # ▌ vs |

    # Status bar
    git_branch: str  # "↗" vs "git:"


UNICODE_GLYPHS = Glyphs(
    tool_prefix="⏺",
    ellipsis="…",
    checkmark="✓",
    error="✗",
    circle_empty="○",
    circle_filled="●",
    output_prefix="⎿",
    spinner_frames=("⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"),
    pause="⏸",
    newline="⏎",
    warning="⚠",
    question="?",
    arrow_up="↑",
    arrow_down="↓",
    bullet="•",
    cursor="›",  # noqa: RUF001  # Intentional Unicode glyph
    disclosure_collapsed="▸",
    disclosure_expanded="▾",
    # Box-drawing characters
    box_vertical="│",
    box_horizontal="─",
    box_double_horizontal="═",
    gutter_bar="▌",
    git_branch="↗",
)
"""Glyph set for terminals with full Unicode support."""

ASCII_GLYPHS = Glyphs(
    tool_prefix="(*)",
    ellipsis="...",
    checkmark="[OK]",
    error="[X]",
    circle_empty="[ ]",
    circle_filled="[*]",
    output_prefix="L",
    spinner_frames=("(-)", "(\\)", "(|)", "(/)"),
    pause="||",
    newline="\\n",
    warning="[!]",
    question="[?]",
    arrow_up="^",
    arrow_down="v",
    bullet="-",
    cursor=">",
    disclosure_collapsed=">",
    disclosure_expanded="v",
    # Box-drawing characters
    box_vertical="|",
    box_horizontal="-",
    box_double_horizontal="=",
    gutter_bar="|",
    git_branch="git:",
)
"""Glyph set for terminals limited to 7-bit ASCII."""

_glyphs_cache: Glyphs | None = None
"""Module-level cache for detected glyphs."""

_charset_mode_cache: CharsetMode | None = None
"""Module-level cache for the detected charset mode."""

_editable_cache: tuple[bool, str | None] | None = None
"""Module-level cache for editable install info: (is_editable, source_path)."""

_langsmith_url_cache: tuple[str, str] | None = None
"""Module-level cache for successful LangSmith project URL lookups."""

_LANGSMITH_URL_LOOKUP_TIMEOUT_SECONDS = 2.0
"""Max seconds to wait for LangSmith project URL lookup.

Kept short so tracing metadata can never stall app flows.
"""


def _get_deepagents_version() -> str | None:
    """Read the installed Deep Agents SDK version from package metadata.

    Returns:
        The installed Deep Agents SDK version, or `None` when package metadata
        is unavailable.
    """
    try:
        return version("deepagents")
    except PackageNotFoundError:
        logger.debug(
            "Failed to read deepagents version from package metadata",
            exc_info=True,
        )
        return None


def _resolve_editable_info() -> tuple[bool, str | None]:
    """Parse PEP 610 `direct_url.json` once and cache both results.

    Returns:
        Tuple of (is_editable, contracted_source_path). The path is
        `~`-contracted when it falls under the user's home directory, or
        `None` when the install is non-editable or the path is unavailable.
    """
    global _editable_cache  # noqa: PLW0603  # Module-level cache requires global statement
    if _editable_cache is not None:
        return _editable_cache

    editable = False
    path: str | None = None

    try:
        dist = distribution("deepagents-code")
        raw = dist.read_text("direct_url.json")
        if raw:
            data = json.loads(raw)
            editable = data.get("dir_info", {}).get("editable", False)
            if editable:
                url = data.get("url", "")
                if url.startswith("file://"):
                    path = unquote(urlparse(url).path)
                    home = str(Path.home())
                    if path.startswith(home):
                        path = "~" + path[len(home) :]
    except (PackageNotFoundError, FileNotFoundError, json.JSONDecodeError, TypeError):
        logger.debug(
            "Failed to read editable install info from PEP 610 metadata",
            exc_info=True,
        )

    _editable_cache = (editable, path)
    return _editable_cache


def _is_editable_install() -> bool:
    """Check if deepagents-code is installed in editable mode.

    Uses PEP 610 `direct_url.json` metadata to detect editable installs.

    Returns:
        `True` if installed in editable mode, `False` otherwise.
    """
    return _resolve_editable_info()[0]


def _get_editable_install_path() -> str | None:
    """Return the `~`-contracted source directory for an editable install.

    Returns `None` for non-editable installs or when the path cannot be
    determined.
    """
    return _resolve_editable_info()[1]


def _detect_charset_mode() -> CharsetMode:
    """Auto-detect terminal charset capabilities (cached for the process).

    Returns:
        The detected CharsetMode based on environment and terminal encoding.
    """
    global _charset_mode_cache  # noqa: PLW0603  # Module-level cache requires global statement
    if _charset_mode_cache is not None:
        return _charset_mode_cache
    _charset_mode_cache = _compute_charset_mode()
    return _charset_mode_cache


def _compute_charset_mode() -> CharsetMode:
    """Compute terminal charset capabilities from environment and encoding.

    Returns:
        The detected CharsetMode based on environment and terminal encoding.
    """
    from deepagents_code.model_config import resolve_env_var

    env_mode = (resolve_env_var("UI_CHARSET_MODE") or "auto").lower()
    if env_mode == "unicode":
        return CharsetMode.UNICODE
    if env_mode == "ascii":
        return CharsetMode.ASCII

    # Auto: check stdout encoding and LANG
    encoding = getattr(sys.stdout, "encoding", "") or ""
    if "utf" in encoding.lower():
        return CharsetMode.UNICODE
    lang = os.environ.get("LANG", "") or os.environ.get("LC_ALL", "")
    if "utf" in lang.lower():
        return CharsetMode.UNICODE
    return CharsetMode.ASCII


def get_glyphs() -> Glyphs:
    """Get the glyph set for the current charset mode.

    Returns:
        The appropriate Glyphs instance based on charset mode detection.
    """
    global _glyphs_cache  # noqa: PLW0603  # Module-level cache requires global statement
    if _glyphs_cache is not None:
        return _glyphs_cache

    mode = _detect_charset_mode()
    _glyphs_cache = ASCII_GLYPHS if mode == CharsetMode.ASCII else UNICODE_GLYPHS
    return _glyphs_cache


def reset_glyphs_cache() -> None:
    """Reset the glyphs and charset-mode caches (for testing)."""
    global _glyphs_cache, _charset_mode_cache  # noqa: PLW0603  # Module-level caches require global statement
    _glyphs_cache = None
    _charset_mode_cache = None


def is_ascii_mode() -> bool:
    """Check whether the terminal is in ASCII charset mode.

    Convenience wrapper so widgets can branch on charset without importing
    both `_detect_charset_mode` and `CharsetMode`.

    Returns:
        `True` when the detected charset mode is ASCII.
    """
    return _detect_charset_mode() == CharsetMode.ASCII


def newline_shortcut() -> str:
    """Return the terminal-appropriate label for the newline keyboard shortcut.

    Prefers `Shift+Enter` when the terminal is known to support the kitty
    keyboard protocol, either via conservative terminal-identity heuristics
    or the `DEEPAGENTS_CODE_KITTY_KEYBOARD` override. Falls back to
    `Option+Enter` on macOS and `Ctrl+J` elsewhere — both survive legacy
    terminals that strip the shift modifier from `Enter`.

    Returns:
        A human-readable shortcut string,
            e.g. `'Shift+Enter'`, `'Option+Enter'`, or `'Ctrl+J'`.
    """
    from deepagents_code.terminal_capabilities import supports_kitty_keyboard_protocol

    if supports_kitty_keyboard_protocol():
        return "Shift+Enter"
    return "Option+Enter" if sys.platform == "darwin" else "Ctrl+J"


_UNICODE_BANNER = f"""
██████╗  ███████╗ ███████╗ ██████╗    ▄▓▓▄
██╔══██╗ ██╔════╝ ██╔════╝ ██╔══██╗  ▓•███▙
██║  ██║ █████╗   █████╗   ██████╔╝  ░▀▀████▙▖
██║  ██║ ██╔══╝   ██╔══╝   ██╔═══╝      █▓████▙▖
██████╔╝ ███████╗ ███████╗ ██║          ▝█▓█████▙
╚═════╝  ╚══════╝ ╚══════╝ ╚═╝           ░▜█▓████▙
                                          ░█▀█▛▀▀▜▙▄
                                        ░▀░▀▒▛░░  ▝▀▘

 █████╗   ██████╗  ███████╗ ███╗   ██╗ ████████╗ ███████╗
██╔══██╗ ██╔════╝  ██╔════╝ ████╗  ██║ ╚══██╔══╝ ██╔════╝
███████║ ██║  ███╗ █████╗   ██╔██╗ ██║    ██║    ███████╗
██╔══██║ ██║   ██║ ██╔══╝   ██║╚██╗██║    ██║    ╚════██║
██║  ██║ ╚██████╔╝ ███████╗ ██║ ╚████║    ██║    ███████║
╚═╝  ╚═╝  ╚═════╝  ╚══════╝ ╚═╝  ╚═══╝    ╚═╝    ╚══════╝
                                                  v{__version__}
"""
_ASCII_BANNER = f"""
 ____  ____  ____  ____
|  _ \\| ___|| ___||  _ \\
| | | | |_  | |_  | |_) |
| |_| |  _| |  _| |  __/
|____/|____||____||_|

    _    ____  ____  _   _  _____  ____
   / \\  / ___|| ___|| \\ | ||_   _|/ ___|
  / _ \\| |  _ | |_  |  \\| |  | |  \\___ \\
 / ___ \\ |_| ||  _| | |\\  |  | |   ___) |
/_/   \\_\\____||____||_| \\_|  |_|  |____/
                                  v{__version__}
"""


def get_banner() -> str:
    """Get the appropriate banner for the current charset mode.

    Returns:
        The text art banner string (Unicode or ASCII based on charset mode).

            Includes "(local)" suffix when installed in editable mode.
    """
    if _detect_charset_mode() == CharsetMode.ASCII:
        banner = _ASCII_BANNER
    else:
        banner = _UNICODE_BANNER

    if is_env_truthy(HIDE_SPLASH_VERSION):
        return banner.replace(f"v{__version__}", "")

    if _is_editable_install():
        banner = banner.replace(f"v{__version__}", f"v{__version__} (local)")

    return banner


MAX_ARG_LENGTH = 150
"""Character limit for tool argument values in the UI.

Longer values are truncated with an ellipsis by `truncate_value`
in `tool_display`.
"""

config: RunnableConfig = {
    "recursion_limit": 1000,
}
"""Default LangGraph runnable config.

Sets `recursion_limit` to 1000 to accommodate deeply nested agent graphs without
hitting the default LangGraph ceiling.
"""

_git_branch_cache: dict[str, str | None] = {}
"""Per-cwd cache of resolved git branch names.

Avoids repeated git branch resolution within the same session. Keyed by
`str(Path.cwd())`; `None` values indicate the directory is not inside a git
repository or that resolution failed.
"""


def _get_git_branch() -> str | None:
    """Return the current git branch name, or `None` if not in a repo."""
    try:
        cwd = str(Path.cwd())
    except OSError:
        logger.debug("Could not determine cwd for git branch lookup", exc_info=True)
        return None
    if cwd in _git_branch_cache:
        return _git_branch_cache[cwd]

    try:
        branch = resolve_git_branch(cwd) or None
    except OSError:
        logger.debug("Could not determine git branch", exc_info=True)
        branch = None

    _git_branch_cache[cwd] = branch
    return branch


def build_stream_config(
    thread_id: str,
    assistant_id: str | None,
    *,
    sandbox_type: str | None = None,
) -> RunnableConfig:
    """Build the LangGraph stream config dict.

    Injects the dcode version into `metadata["lc_versions"]` so LangSmith traces
    can be correlated with specific releases. `create_deep_agent` supplies the
    SDK version through the compiled graph config, and LangChain merges nested
    metadata dictionaries so both versions survive at stream time.

    Also records `dcode_client_deepagents_version` as a dcode-client diagnostic.
    This describes the Deep Agents package installed alongside the TUI, which
    can differ from a remote graph's Deep Agents runtime version.

    Includes `ls_integration` metadata so LangSmith traces originating from
    the app are distinguishable from bare SDK usage.

    Args:
        thread_id: The app session thread identifier.
        assistant_id: The dcode agent identifier, if any. When set, it is
            surfaced in trace metadata under `dcode_agent_name` and
            `agent_name`.
        sandbox_type: Sandbox provider name for trace metadata, or `None` if no
            sandbox is active.

    Returns:
        Config dict with `configurable` and `metadata` keys.
    """
    from datetime import UTC, datetime

    try:
        cwd = str(Path.cwd())
    except OSError:
        logger.warning("Could not determine working directory", exc_info=True)
        cwd = ""

    metadata: dict[str, Any] = {
        "lc_versions": {"deepagents-code": __version__},
        "ls_integration": "deepagents-code",
    }
    deepagents_version = _get_deepagents_version()
    if deepagents_version is not None:
        metadata["dcode_client_deepagents_version"] = deepagents_version

    from deepagents_code._env_vars import USER_ID

    user_id = os.environ.get(USER_ID)
    if user_id:
        metadata["user_id"] = user_id
    if cwd:
        metadata["cwd"] = cwd
    if assistant_id:
        metadata.update(
            {
                "dcode_agent_name": assistant_id,
                "agent_name": assistant_id,
                "updated_at": datetime.now(UTC).isoformat(),
            }
        )
    branch = _get_git_branch()
    if branch:
        metadata["git_branch"] = branch
    if sandbox_type and sandbox_type != "none":
        metadata["sandbox_type"] = sandbox_type
    return {
        "configurable": {"thread_id": thread_id},
        "metadata": metadata,
    }


class _ShellAllowAll(list):  # noqa: FURB189  # sentinel type, not a general-purpose list subclass
    """Sentinel subclass for unrestricted shell access.

    Using a dedicated type instead of a plain list lets consumers use
    `isinstance` checks, which survive serialization/copy unlike identity
    checks (`is`).
    """


SHELL_ALLOW_ALL: list[str] = _ShellAllowAll(["__ALL__"])
"""Sentinel value returned by `parse_shell_allow_list` for `--shell-allow-list=all`."""


def parse_shell_allow_list(allow_list_str: str | None) -> list[str] | None:
    """Parse shell allow-list from string.

    Args:
        allow_list_str: Comma-separated list of commands, `'recommended'` for
            safe defaults, or `'all'` to allow any command.

            `'all'` must be the sole value — it is not recognized inside a
            comma-separated list (unlike `'recommended'`).

            Can also include `'recommended'` in the list to merge with custom
            commands.

    Returns:
        List of allowed commands, `SHELL_ALLOW_ALL` if `'all'` was specified,
            or `None` if no allow-list configured.

    Raises:
        ValueError: If `'all'` is combined with other commands.
    """
    if not allow_list_str:
        return None

    # Special value 'all' allows any shell command
    if allow_list_str.strip().lower() == "all":
        return SHELL_ALLOW_ALL

    # Special value 'recommended' uses our curated safe list
    if allow_list_str.strip().lower() == "recommended":
        return list(RECOMMENDED_SAFE_SHELL_COMMANDS)

    # Split by comma and strip whitespace
    commands = [cmd.strip() for cmd in allow_list_str.split(",") if cmd.strip()]

    # Reject ambiguous input: 'all' mixed with other commands
    if any(cmd.lower() == "all" for cmd in commands):
        msg = (
            "Cannot combine 'all' with other commands in --shell-allow-list. "
            "Use '--shell-allow-list all' alone to allow any command."
        )
        raise ValueError(msg)

    # If "recommended" is in the list, merge with recommended commands
    result = []
    for cmd in commands:
        if cmd.lower() == "recommended":
            result.extend(RECOMMENDED_SAFE_SHELL_COMMANDS)
        else:
            result.append(cmd)

    # Remove duplicates while preserving order
    seen: set[str] = set()
    unique: list[str] = []
    for cmd in result:
        if cmd not in seen:
            seen.add(cmd)
            unique.append(cmd)
    return unique


INTERPRETER_PTC_SAFE_PRESET: frozenset[str] = frozenset({"read_file", "glob", "grep"})
"""Strictly read-only PTC allowlist for `interpreter_ptc="safe"`.

Limited to tools that are **not** in `_add_interrupt_on()` to begin with, so
exposing them through PTC does not introduce a new HITL bypass. Network
tools (`web_search`, `fetch_url`), subagent dispatch (`task`), shell
execution (`execute`), and file writes (`write_file`, `edit_file`, MCP
write tools) are deliberately excluded — they are HITL-gated outside the
REPL, and PTC bypasses `interrupt_on`, so including them would silently
escalate privileges. Users who need network or subagent access from inside
the REPL must list those tools explicitly (which signals intent at config
time) or use `interpreter_ptc="all"` with the unsafe acknowledgement.
"""

INTERPRETER_PTC_ALL_SENTINEL = "all"
"""Sentinel string for `interpreter_ptc="all"` — resolved at agent-build time
from the live tool list. Requires `interpreter_ptc_acknowledge_unsafe=True`
when `auto_approve` is `False`."""

INTERPRETER_PTC_SAFE_SENTINEL = "safe"
"""Sentinel string for `interpreter_ptc="safe"` — expanded from
`INTERPRETER_PTC_SAFE_PRESET`."""


def _parse_interpreter_ptc(
    raw: Any,  # noqa: ANN401  # accepts TOML-shaped value
) -> str | bool | list[str]:
    """Coerce a raw `interpreter_ptc` value into the canonical shape.

    Args:
        raw: Value loaded from TOML or supplied by the CLI.

    Returns:
        `False` for `False`/`None`/`[]`, the string `"safe"`/`"all"` when
        either sentinel is given, otherwise a validated list of tool names.
        A list may include the `"safe"` preset (expanded at agent-build time)
        but never `"all"`.

    Raises:
        ValueError: If `raw` is a list with empty or non-string entries, a
            list containing `"all"`, or a string other than `"safe"`/`"all"`.
    """
    if raw is None or raw is False:
        return False
    if raw is True:
        msg = (
            "`interpreter_ptc` cannot be set to True; use 'safe', 'all', or "
            "an explicit list of tool names."
        )
        raise ValueError(msg)
    if isinstance(raw, str):
        normalized = raw.strip().lower()
        if normalized in {INTERPRETER_PTC_SAFE_SENTINEL, INTERPRETER_PTC_ALL_SENTINEL}:
            return normalized
        msg = (
            f"Invalid `interpreter_ptc` string {raw!r}; expected 'safe', 'all', "
            "or a list of tool names."
        )
        raise ValueError(msg)
    if isinstance(raw, list):
        if not raw:
            return False
        names: list[str] = []
        for entry in raw:
            if not isinstance(entry, str) or not entry.strip():
                msg = (
                    "`interpreter_ptc` list entries must be non-empty strings; "
                    f"got {entry!r}."
                )
                raise ValueError(msg)
            cleaned = entry.strip()
            if cleaned.lower() == INTERPRETER_PTC_ALL_SENTINEL:
                msg = (
                    "`interpreter_ptc` list entries cannot include 'all'; use "
                    "'all' as a standalone value or list explicit tool names "
                    "(optionally with the 'safe' preset)."
                )
                raise ValueError(msg)
            names.append(cleaned)
        return names
    msg = (
        f"`interpreter_ptc` must be False, 'safe', 'all', or a list of tool "
        f"names; got {type(raw).__name__}."
    )
    raise ValueError(msg)


def _read_config_toml_retries() -> dict[str, Any] | None:
    """Read and lightly validate `[retries]` from `~/.deepagents/config.toml`.

    Provider sub-table names are checked against the set of providers the app
    knows how to authenticate so a mistyped provider (e.g. `[retries.fireorks]`)
    surfaces a warning rather than being silently dropped. Value validation is
    deferred to `_resolve_retry_kwargs`, which runs per active provider.

    Returns:
        The raw `[retries]` mapping, or `None` when the section is absent or the
            file cannot be read.
    """
    import tomllib

    from deepagents_code.model_config import (
        DEFAULT_CONFIG_PATH,
        IMPLICIT_AUTH_PROVIDERS,
        NO_AUTH_REQUIRED_PROVIDERS,
        PROVIDER_API_KEY_ENV,
        RETRY_PARAM_BY_PROVIDER,
    )

    try:
        with DEFAULT_CONFIG_PATH.open("rb") as f:
            data = tomllib.load(f)
    except FileNotFoundError:
        return None
    except (PermissionError, OSError, tomllib.TOMLDecodeError):
        logger.warning(
            "Could not read retries config from %s",
            DEFAULT_CONFIG_PATH,
            exc_info=True,
        )
        return None

    section = data.get("retries")
    if not isinstance(section, dict):
        return None

    known_providers = (
        set(PROVIDER_API_KEY_ENV)
        | set(NO_AUTH_REQUIRED_PROVIDERS)
        | set(IMPLICIT_AUTH_PROVIDERS)
        | set(RETRY_PARAM_BY_PROVIDER)
    )
    for key, value in section.items():
        if (
            isinstance(value, dict)
            and key not in known_providers
            and "param" not in value
        ):
            logger.warning(
                "Ignoring [retries.%s] in config.toml; %r is not a known provider",
                key,
                key,
            )
    return section


def _coerce_max_retries(raw: Any, *, source: str) -> int | None:  # noqa: ANN401
    """Validate a TOML retry count.

    Args:
        raw: Value loaded from TOML.
        source: Human-readable config path for warnings.

    Returns:
        The retry count, or `None` when invalid.
    """
    if isinstance(raw, int) and not isinstance(raw, bool) and raw >= 0:
        return raw
    logger.warning("Ignoring %s=%r in config.toml (expected int >= 0)", source, raw)
    return None


def _coerce_retry_param(raw: Any, *, source: str) -> str | None:  # noqa: ANN401
    """Validate a constructor kwarg name for retry configuration.

    Args:
        raw: Value loaded from TOML.
        source: Human-readable config path for warnings.

    Returns:
        The retry parameter name, or `None` when invalid.
    """
    if isinstance(raw, str) and raw.isidentifier() and not keyword.iskeyword(raw):
        return raw
    logger.warning(
        "Ignoring %s=%r in config.toml (expected Python identifier string)",
        source,
        raw,
    )
    return None


def _resolve_retry_kwargs(
    section: dict[str, Any] | None,
    provider: str,
) -> dict[str, int]:
    """Resolve the retry-count kwarg for `provider` from a `[retries]` section.

    A per-provider `[retries.<provider>].max_retries` overrides the global
    `[retries].max_retries`. Known providers use `RETRY_PARAM_BY_PROVIDER`;
    arbitrary providers can opt in with `[retries.<provider>].param`.
    Unknown providers without a configured parameter receive nothing, and
    unknown or malformed keys are dropped with a warning.

    Args:
        section: Raw `[retries]` mapping from `config.toml`, or `None`.
        provider: Provider the kwargs are being resolved for.

    Returns:
        `{retry_param_name: count}` when a valid retry count resolves, else an
            empty dict.
    """
    if not section:
        return {}

    from deepagents_code.model_config import RETRY_PARAM_BY_PROVIDER

    for key, value in section.items():
        if key == "max_retries" or isinstance(value, dict):
            continue
        logger.warning("Ignoring [retries].%s=%r in config.toml", key, value)

    retry_param = RETRY_PARAM_BY_PROVIDER.get(provider)
    resolved: int | None = None
    if "max_retries" in section:
        resolved = _coerce_max_retries(
            section["max_retries"], source="[retries].max_retries"
        )

    provider_section = section.get(provider)
    if provider_section is not None and not isinstance(provider_section, dict):
        logger.warning(
            "Ignoring [retries].%s=%r in config.toml (expected table)",
            provider,
            provider_section,
        )
    elif provider_section:
        for key, value in provider_section.items():
            if key not in {"max_retries", "param"}:
                logger.warning(
                    "Ignoring [retries.%s].%s=%r in config.toml",
                    provider,
                    key,
                    value,
                )
        if "max_retries" in provider_section:
            provider_value = _coerce_max_retries(
                provider_section["max_retries"],
                source=f"[retries.{provider}].max_retries",
            )
            if provider_value is not None:
                resolved = provider_value
        if "param" in provider_section:
            provider_param = _coerce_retry_param(
                provider_section["param"],
                source=f"[retries.{provider}].param",
            )
            if provider_param is not None:
                retry_param = provider_param

    if retry_param is None:
        logger.warning(
            "Ignoring [retries] config for provider %r; provider does not support "
            "a registered or configured retry parameter",
            provider,
        )
        return {}

    if resolved is None:
        return {}
    return {retry_param: resolved}


CLI_MAX_RETRIES_KEY = "__deepagents_cli_max_retries__"
"""Internal carrier key for the `--max-retries` CLI flag.

`cli_main` stashes the flag value under this key in the `model_params` dict it
forwards to the run, and `create_model` pops it before constructing the model.
This lets the CLI value ride the existing `model_params`/`extra_kwargs` carrier
to the one place that authoritatively resolves the provider, where it can be
folded under the provider's *resolved* retry-param name (see
`_resolve_retry_param_name`) rather than a hardcoded `max_retries`.

The key is internal-only: it is popped before reaching any model constructor and
is never serialized or surfaced to users. It is deliberately unlikely to collide
with a real constructor kwarg name.
"""


def _resolve_retry_param_name(provider: str) -> str:
    """Resolve the constructor kwarg name that sets `provider`'s retry count.

    Honors a `[retries.<provider>].param` override in `config.toml`, then the
    registered `RETRY_PARAM_BY_PROVIDER` mapping, and finally falls back to
    `max_retries` -- the near-universal LangChain chat-model kwarg -- for
    providers that are neither registered nor configured.

    Args:
        provider: Provider the retry kwarg name is being resolved for.

    Returns:
        The constructor kwarg name to use for the retry count.
    """
    from deepagents_code.model_config import RETRY_PARAM_BY_PROVIDER

    section = _read_config_toml_retries()
    if section:
        provider_section = section.get(provider)
        if isinstance(provider_section, dict) and "param" in provider_section:
            configured = _coerce_retry_param(
                provider_section["param"],
                source=f"[retries.{provider}].param",
            )
            if configured is not None:
                return configured

    return RETRY_PARAM_BY_PROVIDER.get(provider, "max_retries")


def _read_config_toml_skills_dirs() -> list[str] | None:
    """Read `[skills].extra_allowed_dirs` from `~/.deepagents/config.toml`.

    Returns:
        List of path strings, or `None` if the key is absent or the file
            cannot be read.
    """
    import tomllib

    from deepagents_code.model_config import DEFAULT_CONFIG_PATH

    try:
        with DEFAULT_CONFIG_PATH.open("rb") as f:
            data = tomllib.load(f)
    except FileNotFoundError:
        return None
    except (PermissionError, OSError, tomllib.TOMLDecodeError):
        logger.warning(
            "Could not read skills config from %s",
            DEFAULT_CONFIG_PATH,
            exc_info=True,
        )
        return None

    skills_section = data.get("skills", {})
    dirs = skills_section.get("extra_allowed_dirs")
    if isinstance(dirs, list):
        return dirs
    return None


def _parse_extra_skills_dirs(
    env_raw: str | None,
    config_toml_dirs: list[str] | None = None,
) -> list[Path] | None:
    """Merge extra skill directories from env var and config.toml.

    Extra skills directories extend the containment allowlist used by
    `load_skill_content` to validate that a resolved skill path lives inside a
    trusted root. They do **not** add new skill discovery locations — skills are
    still discovered only from the standard directories. This exists so that
    symlinks inside standard skill directories can legitimately point to targets
    in user-specified locations without being rejected by the path
    containment check.

    The env var (`DEEPAGENTS_CODE_EXTRA_SKILLS_DIRS`, colon-separated) takes
    precedence: when set, `config.toml` values are ignored.

    Args:
        env_raw: Value of `DEEPAGENTS_CODE_EXTRA_SKILLS_DIRS` (colon-separated), or
            `None` if unset.
        config_toml_dirs: List of path strings from
            `[skills].extra_allowed_dirs` in `~/.deepagents/config.toml`.

    Returns:
        List of resolved `Path` objects, or `None` if not configured.
    """
    # Env var takes precedence when set
    if env_raw:
        dirs = [
            Path(p.strip()).expanduser().resolve()
            for p in env_raw.split(":")
            if p.strip()
        ]
        return dirs or None

    if config_toml_dirs:
        dirs = [
            Path(p).expanduser().resolve()
            for p in config_toml_dirs
            if isinstance(p, str) and p.strip()
        ]
        return dirs or None

    return None


_RELOADABLE_FIELDS = (
    "openai_api_key",
    "anthropic_api_key",
    "google_api_key",
    "nvidia_api_key",
    "tavily_api_key",
    "google_cloud_project",
    "deepagents_langchain_project",
    "project_root",
    "shell_allow_list",
    "extra_skills_dirs",
)
"""Fields refreshed on `/reload` and cwd switches.

Runtime model state (`model_name`, `model_provider`, `model_context_limit`) and
the original user LangSmith project are intentionally excluded -- they are set
once and should not change across reloads.
"""

_API_KEY_FIELDS = frozenset(
    field for field in _RELOADABLE_FIELDS if field.endswith("_api_key")
)
"""Reloadable fields that hold API keys and must be masked in change reports.

Derived from `_RELOADABLE_FIELDS` so new `*_api_key` fields are picked up
automatically.
"""


@dataclass
class Settings:
    """Global settings and environment detection for deepagents-code.

    This class is initialized once at startup and provides access to:
    - Available models and API keys
    - Current project information
    - Tool availability (e.g., Tavily)
    - File system paths
    """

    openai_api_key: str | None
    """OpenAI API key if available."""

    anthropic_api_key: str | None
    """Anthropic API key if available."""

    google_api_key: str | None
    """Google API key if available."""

    nvidia_api_key: str | None
    """NVIDIA API key if available."""

    tavily_api_key: str | None
    """Tavily API key if available."""

    google_cloud_project: str | None
    """Google Cloud project ID for VertexAI authentication."""

    deepagents_langchain_project: str | None
    """LangSmith project name for deepagents agent tracing."""

    user_langchain_project: str | None
    """Original `LANGSMITH_PROJECT` from environment (for user code)."""

    model_name: str | None = None
    """Currently active model name, set after model creation."""

    model_provider: str | None = None
    """Provider identifier (e.g., `openai`, `anthropic`, `google_genai`)."""

    model_context_limit: int | None = None
    """Maximum input token count from the model profile."""

    model_unsupported_modalities: frozenset[str] = frozenset()
    """Input modalities not indicated as supported by the model profile."""

    project_root: Path | None = None
    """Current project root directory, or `None` if not in a git project."""

    shell_allow_list: list[str] | None = None
    """Shell commands that don't require user approval."""

    extra_skills_dirs: list[Path] | None = None
    """Extra directories added to the skill path containment allowlist.

    These do NOT add new skill discovery locations — skills are still only
    discovered from the standard directories. They exist so that symlinks inside
    standard skill directories can point to targets in these additional
    locations without being rejected by the containment check
    in `load_skill_content`.

    Set via `DEEPAGENTS_CODE_EXTRA_SKILLS_DIRS` env var (colon-separated) or
    `[skills].extra_allowed_dirs` in `~/.deepagents/config.toml`.
    """

    enable_interpreter: bool = INTERPRETER_ENABLE_DEFAULT
    """Wire `CodeInterpreterMiddleware` from `langchain-quickjs` into the main
    agent. Local-mode only; raises `ValueError` at agent-build time when a
    remote sandbox is active. Subagents never receive the interpreter in v1.

    The `quickjs` optional extra must be installed when this flag is `True`.

    Defaults are owned by `config_manifest` (the canonical config surface) so
    they are defined in exactly one place.
    """

    interpreter_timeout_seconds: float = INTERPRETER_TIMEOUT_SECONDS_DEFAULT
    """Per-`js_eval`-call wall-clock timeout (seconds) for the QuickJS REPL."""

    interpreter_memory_limit_mb: int = INTERPRETER_MEMORY_LIMIT_MB_DEFAULT
    """QuickJS heap memory cap (MB), shared across all calls within a session."""

    interpreter_max_ptc_calls: int = INTERPRETER_MAX_PTC_CALLS_DEFAULT
    """Maximum `tools.*` host-bridge invocations allowed per `js_eval` call.

    PTC calls bypass `interrupt_on`/HITL approval — this budget is the only
    runtime limiter on bursty tool fan-out from inside the REPL.
    """

    interpreter_max_result_chars: int = INTERPRETER_MAX_RESULT_CHARS_DEFAULT
    """Independent cap (chars) on `js_eval` result and stdout blocks before
    truncation."""

    interpreter_ptc: str | bool | list[str] = INTERPRETER_PTC_DEFAULT
    """Programmatic tool calling allowlist for `js_eval`.

    Accepted values:

    - `False` or `[]`: pure REPL, no `tools.*` bridge.
    - `"safe"`: expand to `INTERPRETER_PTC_SAFE_PRESET`.
    - `"all"`: every tool passed to `create_cli_agent` is exposed. Requires
        `interpreter_ptc_acknowledge_unsafe=True` when `auto_approve` is `False`.
    - `list[str]`: explicit tool names. The list may also include the `"safe"`
        preset (expanded to `INTERPRETER_PTC_SAFE_PRESET`); `"all"` is rejected
        inside a list. Names are matched against the live tool registry at
        runtime, so names not present are simply not exposed.
    """

    interpreter_ptc_acknowledge_unsafe: bool = (
        INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE_DEFAULT
    )
    """Explicit acknowledgement required when `interpreter_ptc="all"` is set
    without `auto_approve`.

    `"all"` exposes every host tool to `tools.*` calls from inside the REPL,
    bypassing HITL approval — this flag is a deliberate sanity gate, not a
    feature toggle.
    """

    @classmethod
    def from_environment(cls, *, start_path: Path | None = None) -> Settings:
        """Create settings by detecting the current environment.

        Args:
            start_path: Directory to start project detection from (defaults to cwd)

        Returns:
            Settings instance with detected configuration
        """
        # Detect API keys (normalize empty strings to None).
        from deepagents_code.model_config import resolve_env_var

        openai_key = resolve_env_var("OPENAI_API_KEY")
        anthropic_key = resolve_env_var("ANTHROPIC_API_KEY")
        google_key = resolve_env_var("GOOGLE_API_KEY")
        nvidia_key = resolve_env_var("NVIDIA_API_KEY")
        tavily_key = resolve_env_var("TAVILY_API_KEY")
        google_cloud_project = resolve_env_var("GOOGLE_CLOUD_PROJECT")

        # Detect LangSmith configuration
        # DEEPAGENTS_CODE_LANGSMITH_PROJECT: Project for deepagents agent tracing
        # user_langchain_project: User's ORIGINAL LANGSMITH_PROJECT (before override)
        # When accessed via the module-level `settings` singleton,
        # _ensure_bootstrap() has already run and may have overridden
        # LANGSMITH_PROJECT. We use the saved original value, not the
        # current os.environ value. Direct callers should ensure
        # bootstrap has run if they depend on the override.
        from deepagents_code._env_vars import (
            EXTRA_SKILLS_DIRS,
            LANGSMITH_PROJECT,
            SHELL_ALLOW_LIST,
        )

        deepagents_langchain_project = resolve_env_var(LANGSMITH_PROJECT)
        user_langchain_project = _original_langsmith_project  # Use saved original!

        # Detect project
        from deepagents_code.project_utils import find_project_root

        project_root = find_project_root(start_path)

        # Parse shell command allow-list from environment
        # Format: comma-separated list of commands (e.g., "ls,cat,grep,pwd")

        shell_allow_list_str = os.environ.get(SHELL_ALLOW_LIST)
        shell_allow_list = parse_shell_allow_list(shell_allow_list_str)

        # Parse extra skill containment roots from env var or config.toml.
        # These extend the path allowlist for load_skill_content but do not
        # add new skill discovery locations.
        extra_skills_dirs = _parse_extra_skills_dirs(
            os.environ.get(EXTRA_SKILLS_DIRS),
            _read_config_toml_skills_dirs(),
        )

        from deepagents_code.config_manifest import resolve_interpreter_kwargs

        interpreter_kwargs = resolve_interpreter_kwargs()

        return cls(
            openai_api_key=openai_key,
            anthropic_api_key=anthropic_key,
            google_api_key=google_key,
            nvidia_api_key=nvidia_key,
            tavily_api_key=tavily_key,
            google_cloud_project=google_cloud_project,
            deepagents_langchain_project=deepagents_langchain_project,
            user_langchain_project=user_langchain_project,
            project_root=project_root,
            shell_allow_list=shell_allow_list,
            extra_skills_dirs=extra_skills_dirs,
            **interpreter_kwargs,
        )

    @staticmethod
    def _reload_values(
        *,
        start_path: Path | None,
        env: dict[str, str],
        previous: dict[str, object],
    ) -> dict[str, object]:
        """Resolve reloadable settings from an environment mapping.

        Returns:
            Reloadable setting values keyed by field name.
        """
        from deepagents_code._env_vars import (
            EXTRA_SKILLS_DIRS,
            LANGSMITH_PROJECT,
            SHELL_ALLOW_LIST,
        )

        try:
            shell_allow_list = parse_shell_allow_list(env.get(SHELL_ALLOW_LIST))
        except ValueError:
            logger.warning(
                "Invalid %s during reload; keeping previous value",
                SHELL_ALLOW_LIST,
            )
            shell_allow_list = previous["shell_allow_list"]

        try:
            from deepagents_code.project_utils import find_project_root

            project_root = find_project_root(start_path)
        except OSError:
            logger.warning(
                "Could not detect project root during reload; keeping previous value"
            )
            project_root = previous["project_root"]

        try:
            extra_skills_dirs = _parse_extra_skills_dirs(
                env.get(EXTRA_SKILLS_DIRS),
                _read_config_toml_skills_dirs(),
            )
        except (OSError, ValueError):
            # Path resolution can fail (e.g. broken symlink loop). Keep the
            # previous value rather than letting the failure escape reload --
            # callers such as the cwd switch run this after `os.chdir`, where an
            # uncaught error would strand the process in a half-applied cwd.
            logger.warning(
                "Could not resolve %s during reload; keeping previous value",
                EXTRA_SKILLS_DIRS,
                exc_info=True,
            )
            extra_skills_dirs = previous["extra_skills_dirs"]

        return {
            "openai_api_key": _resolve_env_var_from(env, "OPENAI_API_KEY"),
            "anthropic_api_key": _resolve_env_var_from(env, "ANTHROPIC_API_KEY"),
            "google_api_key": _resolve_env_var_from(env, "GOOGLE_API_KEY"),
            "nvidia_api_key": _resolve_env_var_from(env, "NVIDIA_API_KEY"),
            "tavily_api_key": _resolve_env_var_from(env, "TAVILY_API_KEY"),
            "google_cloud_project": _resolve_env_var_from(env, "GOOGLE_CLOUD_PROJECT"),
            "deepagents_langchain_project": _resolve_env_var_from(
                env,
                LANGSMITH_PROJECT,
            ),
            "project_root": project_root,
            "shell_allow_list": shell_allow_list,
            "extra_skills_dirs": extra_skills_dirs,
        }

    @staticmethod
    def _format_reload_changes(
        previous: dict[str, object], refreshed: dict[str, object]
    ) -> list[str]:
        """Format changed reloadable settings for logs and messages.

        Returns:
            Human-readable change descriptions.
        """

        def display(field: str, value: object) -> str:
            if field in _API_KEY_FIELDS:
                return "set" if value else "unset"
            return str(value)

        changes: list[str] = []
        for field in _RELOADABLE_FIELDS:
            old_value = previous[field]
            new_value = refreshed[field]
            if old_value != new_value:
                changes.append(
                    f"{field}: {display(field, old_value)} -> "
                    f"{display(field, new_value)}"
                )
        return changes

    def preview_reload_from_environment(
        self, *, start_path: Path | None = None
    ) -> list[str]:
        """Preview runtime settings changes without applying them.

        Args:
            start_path: Directory to start project detection from (defaults to cwd).

        Returns:
            A list of human-readable change descriptions that would be produced by
            `reload_from_environment`.
        """
        previous = {field: getattr(self, field) for field in _RELOADABLE_FIELDS}
        env = _preview_dotenv_environ(start_path=start_path)
        refreshed = self._reload_values(
            start_path=start_path,
            env=env,
            previous=previous,
        )
        return self._format_reload_changes(previous, refreshed)

    def reload_from_environment(self, *, start_path: Path | None = None) -> list[str]:
        """Reload selected settings from environment variables and project files.

        This refreshes only fields that are expected to change at runtime
        (API keys, Google Cloud project, project root, shell allow-list, and
        LangSmith tracing project).

        Runtime model state (`model_name`, `model_provider`,
        `model_context_limit`) and the original user LangSmith project
        (`user_langchain_project`) are intentionally preserved -- they are
        not in `_RELOADABLE_FIELDS` and are never touched by this method.

        !!! note

            Shell-exported variables always take precedence. Values previously
            injected from `.env` files are refreshed so an accepted cwd switch
            can pick up the resumed project's `.env`.

        Args:
            start_path: Directory to start project detection from (defaults to cwd).

        Returns:
            A list of human-readable change descriptions.
        """
        _load_dotenv(start_path=start_path, refresh_loaded=True)

        previous = {field: getattr(self, field) for field in _RELOADABLE_FIELDS}
        refreshed = self._reload_values(
            start_path=start_path,
            env=dict(os.environ),
            previous=previous,
        )

        for field, value in refreshed.items():
            setattr(self, field, value)

        # Sync the LANGSMITH_PROJECT env var so LangSmith tracing picks up
        # the change
        new_project = refreshed["deepagents_langchain_project"]
        if new_project:
            os.environ["LANGSMITH_PROJECT"] = str(new_project)
        elif previous["deepagents_langchain_project"]:
            # Override was previously active but new value is unset; restore.
            if _original_langsmith_project:
                os.environ["LANGSMITH_PROJECT"] = _original_langsmith_project
            else:
                os.environ.pop("LANGSMITH_PROJECT", None)

        return self._format_reload_changes(previous, refreshed)

    @property
    def has_anthropic(self) -> bool:
        """Check if Anthropic API key is configured."""
        return self.anthropic_api_key is not None

    @property
    def has_google(self) -> bool:
        """Check if Google API key is configured."""
        return self.google_api_key is not None

    @property
    def has_vertex_ai(self) -> bool:
        """Check if VertexAI is available (Google Cloud project set, no API key).

        VertexAI uses Application Default Credentials (ADC) for authentication,
        so if GOOGLE_CLOUD_PROJECT is set and GOOGLE_API_KEY is not, we assume
        VertexAI.
        """
        return self.google_cloud_project is not None and self.google_api_key is None

    @property
    def has_tavily(self) -> bool:
        """Check if Tavily API key is configured."""
        return self.tavily_api_key is not None

    @property
    def user_deepagents_dir(self) -> Path:
        """Get the base user-level .deepagents directory.

        Returns:
            Path to ~/.deepagents
        """
        return Path.home() / ".deepagents"

    @staticmethod
    def get_user_agent_md_path(agent_name: str) -> Path:
        """Get user-level AGENTS.md path for a specific agent.

        Returns path regardless of whether the file exists.

        Args:
            agent_name: Name of the agent

        Returns:
            Path to ~/.deepagents/{agent_name}/AGENTS.md
        """
        return Path.home() / ".deepagents" / agent_name / "AGENTS.md"

    def get_project_agent_md_path(self) -> list[Path]:
        """Get project-level AGENTS.md paths.

        Checks both `{project_root}/.deepagents/AGENTS.md` and
        `{project_root}/AGENTS.md`, returning all that exist. If both are
        present, both are loaded and their instructions are combined, with
        `.deepagents/AGENTS.md` first.

        Returns:
            Existing AGENTS.md paths.

                Empty if neither file exists or not in a project, one entry if
                only one is present, or two entries if both locations have the
                file.
        """
        if not self.project_root:
            return []
        from deepagents_code.project_utils import find_project_agent_md

        return find_project_agent_md(self.project_root)

    @staticmethod
    def _is_valid_agent_name(agent_name: str) -> bool:
        """Validate to prevent invalid filesystem paths and security issues.

        Returns:
            True if the agent name is valid, False otherwise.
        """
        if not agent_name or not agent_name.strip():
            return False
        # Allow only alphanumeric, hyphens, underscores, and whitespace
        return bool(re.match(r"^[a-zA-Z0-9_\-\s]+$", agent_name))

    def get_agent_dir(self, agent_name: str) -> Path:
        """Get the global agent directory path.

        Args:
            agent_name: Name of the agent

        Returns:
            Path to ~/.deepagents/{agent_name}

        Raises:
            ValueError: If the agent name contains invalid characters.
        """
        if not self._is_valid_agent_name(agent_name):
            msg = (
                f"Invalid agent name: {agent_name!r}. Agent names can only "
                "contain letters, numbers, hyphens, underscores, and spaces."
            )
            raise ValueError(msg)
        return Path.home() / ".deepagents" / agent_name

    def ensure_agent_dir(self, agent_name: str) -> Path:
        """Ensure the global agent directory exists and return its path.

        Args:
            agent_name: Name of the agent

        Returns:
            Path to ~/.deepagents/{agent_name}

        Raises:
            ValueError: If the agent name contains invalid characters.
        """
        if not self._is_valid_agent_name(agent_name):
            msg = (
                f"Invalid agent name: {agent_name!r}. Agent names can only "
                "contain letters, numbers, hyphens, underscores, and spaces."
            )
            raise ValueError(msg)
        agent_dir = self.get_agent_dir(agent_name)
        agent_dir.mkdir(parents=True, exist_ok=True)
        return agent_dir

    def get_user_skills_dir(self, agent_name: str) -> Path:
        """Get user-level skills directory path for a specific agent.

        Args:
            agent_name: Name of the agent

        Returns:
            Path to ~/.deepagents/{agent_name}/skills/
        """
        return self.get_agent_dir(agent_name) / "skills"

    def ensure_user_skills_dir(self, agent_name: str) -> Path:
        """Ensure user-level skills directory exists and return its path.

        Args:
            agent_name: Name of the agent

        Returns:
            Path to ~/.deepagents/{agent_name}/skills/
        """
        skills_dir = self.get_user_skills_dir(agent_name)
        skills_dir.mkdir(parents=True, exist_ok=True)
        return skills_dir

    def get_project_skills_dir(self) -> Path | None:
        """Get project-level skills directory path.

        Returns:
            Path to {project_root}/.deepagents/skills/, or None if not in a project
        """
        if not self.project_root:
            return None
        return self.project_root / ".deepagents" / "skills"

    def ensure_project_skills_dir(self) -> Path | None:
        """Ensure project-level skills directory exists and return its path.

        Returns:
            Path to {project_root}/.deepagents/skills/, or None if not in a project
        """
        if not self.project_root:
            return None
        skills_dir = self.get_project_skills_dir()
        if skills_dir is None:
            return None
        skills_dir.mkdir(parents=True, exist_ok=True)
        return skills_dir

    def get_user_agents_dir(self, agent_name: str) -> Path:
        """Get user-level agents directory path for custom subagent definitions.

        Args:
            agent_name: Name of the agent (e.g., "deepagents")

        Returns:
            Path to ~/.deepagents/{agent_name}/agents/
        """
        return self.get_agent_dir(agent_name) / "agents"

    def get_project_agents_dir(self) -> Path | None:
        """Get project-level agents directory path for custom subagent definitions.

        Returns:
            Path to {project_root}/.deepagents/agents/, or None if not in a project
        """
        if not self.project_root:
            return None
        return self.project_root / ".deepagents" / "agents"

    @property
    def user_agents_dir(self) -> Path:
        """Get the base user-level `.agents` directory (`~/.agents`).

        Returns:
            Path to `~/.agents`
        """
        return Path.home() / ".agents"

    def get_user_agent_skills_dir(self) -> Path:
        """Get user-level `~/.agents/skills/` directory.

        This is a generic alias path for skills that is tool-agnostic.

        Returns:
            Path to `~/.agents/skills/`
        """
        return self.user_agents_dir / "skills"

    def get_project_agent_skills_dir(self) -> Path | None:
        """Get project-level `.agents/skills/` directory.

        This is a generic alias path for skills that is tool-agnostic.

        Returns:
            Path to `{project_root}/.agents/skills/`, or `None` if not in a project
        """
        if not self.project_root:
            return None
        return self.project_root / ".agents" / "skills"

    @staticmethod
    def get_user_claude_skills_dir() -> Path:
        """Get user-level `~/.claude/skills/` directory (experimental).

        Convenience bridge for cross-tool skill sharing with Claude Code.
        This is experimental and may be removed.

        Returns:
            Path to `~/.claude/skills/`
        """
        return Path.home() / ".claude" / "skills"

    def get_project_claude_skills_dir(self) -> Path | None:
        """Get project-level `.claude/skills/` directory (experimental).

        Convenience bridge for cross-tool skill sharing with Claude Code.
        This is experimental and may be removed.

        Returns:
            Path to `{project_root}/.claude/skills/`, or `None` if not in a project.
        """
        if not self.project_root:
            return None
        return self.project_root / ".claude" / "skills"

    @staticmethod
    def get_built_in_skills_dir() -> Path:
        """Get the directory containing built-in skills that ship with the app.

        Returns:
            Path to the `built_in_skills/` directory within the package.
        """
        return Path(__file__).parent / "built_in_skills"

    def get_extra_skills_dirs(self) -> list[Path]:
        """Get user-configured extra skill directories.

        Set via `DEEPAGENTS_CODE_EXTRA_SKILLS_DIRS` (colon-separated paths) or
        `[skills].extra_allowed_dirs` in `~/.deepagents/config.toml`.

        Returns:
            List of extra skill directory paths, or empty list if not configured.
        """
        return self.extra_skills_dirs or []


DANGEROUS_SHELL_PATTERNS = (
    "$(",  # Command substitution
    "`",  # Backtick command substitution
    "$'",  # ANSI-C quoting (can encode dangerous chars via escape sequences)
    "\n",  # Newline (command injection)
    "\r",  # Carriage return (command injection)
    "\t",  # Tab (can be used for injection in some shells)
    "<(",  # Process substitution (input)
    ">(",  # Process substitution (output)
    "<<<",  # Here-string
    "<<",  # Here-doc (can embed commands)
    ">>",  # Append redirect
    ">",  # Output redirect
    "<",  # Input redirect
    "${",  # Variable expansion with braces (can run commands via ${var:-$(cmd)})
)
"""Literal substrings that indicate shell injection risk.

Used by `contains_dangerous_patterns` to reject commands that embed arbitrary
execution via redirects, substitution operators, or control characters — even
when the base command is on the allow-list.
"""

RECOMMENDED_SAFE_SHELL_COMMANDS = (
    # Directory listing
    "ls",
    "dir",
    # File content viewing (read-only)
    "cat",
    "head",
    "tail",
    # Text searching (read-only)
    "grep",
    "wc",
    "strings",
    # Text processing (read-only, no shell execution)
    "cut",
    "tr",
    "diff",
    "md5sum",
    "sha256sum",
    # Path utilities
    "pwd",
    "which",
    # System info (read-only)
    "uname",
    "hostname",
    "whoami",
    "id",
    "groups",
    "uptime",
    "nproc",
    "lscpu",
    "lsmem",
    # Process viewing (read-only)
    "ps",
)
"""Read-only commands auto-approved in non-interactive mode.

Only includes readers and formatters — shells, editors, interpreters, package
managers, network tools, archivers, and anything on GTFOBins/LOOBins is
intentionally excluded. File-write and injection vectors are blocked separately
by `DANGEROUS_SHELL_PATTERNS`.
"""


def contains_dangerous_patterns(command: str) -> bool:
    """Check if a command contains dangerous shell patterns.

    These patterns can be used to bypass allow-list validation by embedding
    arbitrary commands within seemingly safe commands. The check includes
    both literal substring patterns (redirects, substitution operators, etc.)
    and regex patterns for bare variable expansion (`$VAR`) and the background
    operator (`&`).

    Args:
        command: The shell command to check.

    Returns:
        True if dangerous patterns are found, False otherwise.
    """
    if any(pattern in command for pattern in DANGEROUS_SHELL_PATTERNS):
        return True

    # Bare variable expansion ($VAR without braces) can leak sensitive paths.
    # We already block ${ and $( above; this catches plain $HOME, $IFS, etc.
    if re.search(r"\$[A-Za-z_]", command):
        return True

    # Standalone & (background execution) changes the execution model and
    # should not be allowed.  We check for & that is NOT part of &&.
    return bool(re.search(r"(?<![&])&(?![&])", command))


def is_shell_command_allowed(command: str, allow_list: list[str] | None) -> bool:
    """Check if a shell command is in the allow-list.

    The allow-list matches against the first token of the command (the executable
    name). This allows read-only commands like ls, cat, grep, etc. to be
    auto-approved.

    When `allow_list` is the `SHELL_ALLOW_ALL` sentinel, all non-empty commands
    are approved unconditionally — dangerous pattern checks are skipped.

    SECURITY: For regular allow-lists, this function rejects commands containing
    dangerous shell patterns (command substitution, redirects, process
    substitution, etc.) BEFORE parsing, to prevent injection attacks that could
    bypass the allow-list.

    Args:
        command: The full shell command to check.
        allow_list: List of allowed command names (e.g., `["ls", "cat", "grep"]`),
            the `SHELL_ALLOW_ALL` sentinel to allow any command, or `None`.

    Returns:
        `True` if the command is allowed, `False` otherwise.
    """
    if not allow_list or not command or not command.strip():
        return False

    # SHELL_ALLOW_ALL sentinel — skip pattern and token checks
    if isinstance(allow_list, _ShellAllowAll):
        return True

    # SECURITY: Check for dangerous patterns BEFORE any parsing
    # This prevents injection attacks like: ls "$(rm -rf /)"
    if contains_dangerous_patterns(command):
        return False

    allow_set = set(allow_list)

    # Extract the first command token
    # Handle pipes and other shell operators by checking each command in the pipeline
    # Split by compound operators first (&&, ||), then single-char operators (|, ;).
    # Note: standalone & (background) is blocked by contains_dangerous_patterns above.
    segments = re.split(r"&&|\|\||[|;]", command)

    # Track if we found at least one valid command
    found_command = False

    for raw_segment in segments:
        segment = raw_segment.strip()
        if not segment:
            continue

        try:
            # Try to parse as shell command to extract the executable name
            tokens = shlex.split(segment)
            if tokens:
                found_command = True
                cmd_name = tokens[0]
                # Check if this command is in the allow set
                if cmd_name not in allow_set:
                    return False
        except ValueError:
            # If we can't parse it, be conservative and require approval
            return False

    # All segments are allowed (and we found at least one command)
    return found_command


def get_langsmith_project_name() -> str | None:
    """Resolve the LangSmith project name if tracing is configured.

    Checks for the required API key and tracing environment variables.
    When both are present, resolves the project name with priority:
    `settings.deepagents_langchain_project` (from
    `DEEPAGENTS_CODE_LANGSMITH_PROJECT`), then `LANGSMITH_PROJECT` from the
    environment (note: this may already have been overridden at bootstrap time
    to match `DEEPAGENTS_CODE_LANGSMITH_PROJECT`), then `'deepagents-code'`.

    Returns:
        Project name string when LangSmith tracing is active, None otherwise.
    """
    from deepagents_code.config_manifest import LANGSMITH_PROJECT_DEFAULT
    from deepagents_code.model_config import resolve_env_var

    langsmith_key = resolve_env_var("LANGSMITH_API_KEY") or resolve_env_var(
        "LANGCHAIN_API_KEY"
    )
    langsmith_tracing = resolve_env_var("LANGSMITH_TRACING") or resolve_env_var(
        "LANGCHAIN_TRACING_V2"
    )
    if not (langsmith_key and langsmith_tracing):
        return None

    return (
        _get_settings().deepagents_langchain_project
        or os.environ.get("LANGSMITH_PROJECT")
        or LANGSMITH_PROJECT_DEFAULT
    )


def get_langsmith_replica_projects() -> list[str]:
    """Extra LangSmith project names to dual-write agent traces to.

    Parses `DEEPAGENTS_CODE_LANGSMITH_REPLICA_PROJECTS` (comma-separated) into a
    de-duplicated, order-preserving list.

    Returns:
        Project names, or `[]` when the env var is unset or empty.
    """
    from deepagents_code._env_vars import LANGSMITH_REPLICA_PROJECTS

    raw = os.environ.get(LANGSMITH_REPLICA_PROJECTS)
    if not raw:
        return []
    return list(dict.fromkeys(p.strip() for p in raw.split(",") if p.strip()))


def get_langsmith_replica_project() -> str | None:
    """The single extra LangSmith project to mirror agent runs to, if configured.

    dcode agent runs execute inside the LangGraph server subprocess, so the only
    way to mirror them to another project is the server's own replica path: the
    SDK forwards a `langsmith_tracing` project in the run-create request, and the
    server wraps the run in a `tracing_context` whose write replicas are that
    project plus the server's primary project. Client-side callbacks and
    `tracing_context(replicas=...)` cannot reach the run because it is created
    server-side, not in the app process.

    Implementation detail (subject to change): as of `langgraph-api` 0.10.0 this
    happens in `langgraph_api.stream` and `langgraph_api.models.run`.

    The server mirrors to exactly one extra project, so when
    `DEEPAGENTS_CODE_LANGSMITH_REPLICA_PROJECTS` lists several, only the first is
    used and the rest are dropped with a warning.

    Returns:
        The first configured replica project name, or `None` when none are set.
    """
    extras = get_langsmith_replica_projects()
    if not extras:
        return None
    if len(extras) > 1:
        logger.warning(
            "DEEPAGENTS_CODE_LANGSMITH_REPLICA_PROJECTS lists %d projects, but the "
            "LangGraph server mirrors runs to only one extra project; tracing to "
            "%r and ignoring %s.",
            len(extras),
            extras[0],
            extras[1:],
        )
    return extras[0]


class LangSmithLookupError(Exception):
    """Base class for typed LangSmith project URL lookup failures.

    Concrete subclasses (`LangSmithImportError`, `LangSmithLookupTimeoutError`,
    `LangSmithApiError`) let interactive callers like `/trace` show the user
    the actual cause instead of collapsing every failure into a generic
    "could not reach LangSmith" message.
    """


class LangSmithImportError(LangSmithLookupError):
    """The `langsmith` package is not installed."""


class LangSmithLookupTimeoutError(LangSmithLookupError):
    """The LangSmith project URL lookup exceeded its hard timeout."""


class LangSmithApiError(LangSmithLookupError):
    """The LangSmith SDK call raised — auth, 404, network, etc.

    Wraps the underlying SDK exception in `__cause__`.
    """


def _assemble_langsmith_thread_url(project_url: str, thread_id: str) -> str:
    """Format a LangSmith thread URL from a project URL prefix.

    Args:
        project_url: Project URL prefix from `fetch_langsmith_project_url`
            (e.g. `https://smith.langchain.com/o/<org>/projects/p/<proj>`).
        thread_id: Thread identifier to append.

    Returns:
        Full thread URL with the `deepagents-code` utm tag.
    """
    return f"{project_url.rstrip('/')}/t/{thread_id}?utm_source=deepagents-code"


def fetch_langsmith_project_url_or_raise(project_name: str) -> str:
    """Fetch the LangSmith project URL, raising on any failure.

    Successful results are cached at module level so repeated calls do not
    make additional network requests.

    The network call runs in a daemon thread with a hard timeout of
    `_LANGSMITH_URL_LOOKUP_TIMEOUT_SECONDS`, so this function blocks the
    calling thread for at most that duration even if LangSmith is unreachable.

    Args:
        project_name: LangSmith project name to look up.

    Returns:
        Project URL string.

    Raises:
        LangSmithImportError: `langsmith` is not installed.
        LangSmithLookupTimeoutError: lookup exceeded the hard timeout.
        LangSmithApiError: the SDK call raised (auth, 404, network, etc.);
            wraps the original exception in `__cause__`.
    """
    global _langsmith_url_cache  # noqa: PLW0603  # Module-level cache requires global statement

    if _langsmith_url_cache is not None:
        cached_name, cached_url = _langsmith_url_cache
        if cached_name == project_name:
            return cached_url
        # Different project name — fall through to fetch.

    try:
        from langsmith import Client
    except ImportError as exc:
        logger.debug(
            "langsmith package not installed; cannot fetch project URL for '%s'",
            project_name,
            exc_info=True,
        )
        msg = "langsmith package is not installed"
        raise LangSmithImportError(msg) from exc

    result: str | None = None
    lookup_error: Exception | None = None
    done = threading.Event()

    def _lookup_url() -> None:
        nonlocal result, lookup_error
        try:
            from deepagents_code.model_config import resolve_env_var

            # Explicit api_key because Client() reads os.environ directly
            # and doesn't know about the DEEPAGENTS_CODE_ prefix.
            api_key = resolve_env_var("LANGSMITH_API_KEY") or resolve_env_var(
                "LANGCHAIN_API_KEY"
            )
            project = Client(api_key=api_key).read_project(project_name=project_name)
            result = project.url or None
        except Exception as exc:  # noqa: BLE001  # LangSmith SDK error types are not stable
            lookup_error = exc
        finally:
            done.set()

    thread = threading.Thread(target=_lookup_url, daemon=True)
    thread.start()

    if not done.wait(_LANGSMITH_URL_LOOKUP_TIMEOUT_SECONDS):
        logger.debug(
            "Timed out fetching LangSmith project URL for '%s' after %.1fs",
            project_name,
            _LANGSMITH_URL_LOOKUP_TIMEOUT_SECONDS,
        )
        msg = (
            f"LangSmith project URL lookup timed out after "
            f"{_LANGSMITH_URL_LOOKUP_TIMEOUT_SECONDS:.1f}s"
        )
        raise LangSmithLookupTimeoutError(msg)

    if lookup_error is not None:
        logger.debug(
            "Could not fetch LangSmith project URL for '%s'",
            project_name,
            exc_info=(
                type(lookup_error),
                lookup_error,
                lookup_error.__traceback__,
            ),
        )
        msg = str(lookup_error) or repr(lookup_error)
        raise LangSmithApiError(msg) from lookup_error

    if not result:
        # SDK returned a project with an empty URL — treat as an API anomaly.
        msg = f"LangSmith returned no URL for project '{project_name}'"
        raise LangSmithApiError(msg)

    _langsmith_url_cache = (project_name, result)
    return result


def fetch_langsmith_project_url(project_name: str) -> str | None:
    """Fetch the LangSmith project URL, returning None on any failure.

    Thin back-compat wrapper around `fetch_langsmith_project_url_or_raise`
    for passive callers (status banners, non-interactive output) that just
    want a URL-or-nothing answer. Interactive callers that need to tell the
    user *why* the lookup failed should use the raising variant directly.

    Args:
        project_name: LangSmith project name to look up.

    Returns:
        Project URL string if found, None otherwise.
    """
    try:
        return fetch_langsmith_project_url_or_raise(project_name)
    except LangSmithLookupError:
        return None


def build_langsmith_thread_url(thread_id: str) -> str | None:
    """Build a full LangSmith thread URL if tracing is configured.

    Combines `get_langsmith_project_name` and `fetch_langsmith_project_url`
    into a single convenience helper.

    Args:
        thread_id: Thread identifier to build the URL for.

    Returns:
        Full thread URL string, or `None` if unavailable (LangSmith is not
            configured or the project URL cannot be resolved.)
    """
    project_name = get_langsmith_project_name()
    if not project_name:
        return None

    project_url = fetch_langsmith_project_url(project_name)
    if not project_url:
        return None

    return _assemble_langsmith_thread_url(project_url, thread_id)


def reset_langsmith_url_cache() -> None:
    """Reset the LangSmith URL cache (for testing)."""
    global _langsmith_url_cache  # noqa: PLW0603  # Module-level cache requires global statement
    _langsmith_url_cache = None


def get_default_coding_instructions() -> str:
    """Get the default coding agent instructions.

    These are the immutable base instructions that cannot be modified by the agent.
    Long-term memory (AGENTS.md) is handled separately by the middleware.

    Returns:
        The default agent instructions as a string.
    """
    default_prompt_path = Path(__file__).parent / "default_agent_prompt.md"
    return default_prompt_path.read_text()


def detect_provider(model_name: str) -> str | None:
    """Auto-detect provider from model name.

    Intentionally duplicates a subset of LangChain's
    `_attempt_infer_model_provider` because we need to resolve the provider
    **before** calling `init_chat_model` in order to:

    1. Build provider-specific kwargs (API base URLs, headers, etc.) that are
       passed *into* `init_chat_model`.
    2. Validate credentials early to surface user-friendly errors.

    Args:
        model_name: Model name to detect provider from.

    Returns:
        Provider name (openai, anthropic, google_genai, google_vertexai,
            nvidia) or `None` if the provider cannot be determined from the
            name alone.
    """
    model_lower = model_name.lower()

    if model_lower.startswith(("gpt-", "o1", "o3", "o4", "chatgpt")):
        return "openai"

    if model_lower.startswith("claude"):
        s = _get_settings()
        if not s.has_anthropic and s.has_vertex_ai:
            return "google_vertexai"
        return "anthropic"

    if model_lower.startswith("gemini"):
        s = _get_settings()
        if s.has_vertex_ai and not s.has_google:
            return "google_vertexai"
        return "google_genai"

    if model_lower.startswith(("nemotron", "nvidia/")):
        return "nvidia"

    return None


def _get_default_model_spec() -> str:
    """Get default model specification based on available credentials.

    Checks in order:

    1. `[models].default` in config file (user's intentional preference).
    2. `[models].recent` in config file (last `/model` switch).
    3. Auto-detection based on available API credentials.

    Returns:
        Model specification in `provider:model` format.

    Raises:
        NoCredentialsConfiguredError: If no credentials are configured for any
            of the auto-detectable providers. Callers may catch this to defer
            startup and prompt for credentials interactively.
    """
    from deepagents_code.model_config import (
        ModelConfig,
        NoCredentialsConfiguredError,
        get_provider_auth_status,
    )

    config = ModelConfig.load()
    if config.default_model:
        return config.default_model

    if config.recent_model:
        return config.recent_model

    # `is True` deliberately excludes `ProviderAuthState.UNKNOWN` (which maps
    # to `as_legacy_bool() -> None`). For the three explicit-credential
    # providers below, an UNKNOWN result means we cannot prove auth works, so
    # we fall through rather than pick an unverifiable default. If an
    # implicit-auth provider (e.g., Vertex ADC) is added to this fallback
    # list, switch to checking `state` against the relevant
    # `ProviderAuthState` members directly.
    if get_provider_auth_status("openai").as_legacy_bool() is True:
        return "openai:gpt-5.5"
    if get_provider_auth_status("anthropic").as_legacy_bool() is True:
        return "anthropic:claude-opus-4-7"
    if get_provider_auth_status("google_genai").as_legacy_bool() is True:
        return "google_genai:gemini-3.1-pro-preview"

    msg = (
        "No credentials configured. Please set one of: "
        "ANTHROPIC_API_KEY, OPENAI_API_KEY, or GOOGLE_API_KEY"
    )
    raise NoCredentialsConfiguredError(msg)


_OPENROUTER_APP_URL = "https://pypi.org/project/deepagents-code/"
"""Default `app_url` (maps to `HTTP-Referer`) for OpenRouter attribution.

See https://openrouter.ai/docs/app-attribution for details.
"""

_OPENROUTER_APP_TITLE = "Deep Agents Code"
"""Default `app_title` (maps to `X-Title`) for OpenRouter attribution."""

_OPENROUTER_APP_CATEGORIES: list[str] = ["cli-agent"]
"""Default `app_categories` (maps to `X-OpenRouter-Categories`) for OpenRouter."""

_cli_openrouter_profile_registered = False
"""Process-wide guard so the app's OpenRouter profile is registered exactly once."""


def _cli_openrouter_attribution_kwargs() -> dict[str, Any]:
    """App-specific OpenRouter attribution kwargs.

    Layered on top of the SDK's built-in factory via profile stacking; these
    values override the SDK defaults but still sit beneath any caller-supplied
    `kwargs` (i.e. `config.toml`-resolved values), preserving the precedence
    documented on `apply_provider_profile`.

    Returns:
        Mapping of `app_url` and `app_title` to spread into `init_chat_model`.
    """
    return {
        "app_url": _OPENROUTER_APP_URL,
        "app_title": _OPENROUTER_APP_TITLE,
    }


def _ensure_cli_openrouter_profile_registered() -> None:
    """Stack the app's OpenRouter attribution onto the SDK's built-in profile.

    Stacking (vs. duplicating the inline `_get_provider_kwargs` path) means the
    SDK's `pre_init` version check fires exactly once and the app's app-
    attribution defaults are composed via the same `apply_provider_profile`
    path used for every other provider. `register_provider_profile` merges on
    top of the existing built-in registration: the app's `init_kwargs` and
    factory output win on shared keys, while the built-in's `pre_init` and
    factory still chain.
    """
    global _cli_openrouter_profile_registered  # noqa: PLW0603
    if _cli_openrouter_profile_registered:
        return

    from deepagents.profiles.provider import ProviderProfile, register_provider_profile

    register_provider_profile(
        "openrouter",
        ProviderProfile(
            init_kwargs={"app_categories": _OPENROUTER_APP_CATEGORIES},
            init_kwargs_factory=_cli_openrouter_attribution_kwargs,
        ),
    )
    _cli_openrouter_profile_registered = True


def _get_provider_kwargs(
    provider: str, *, model_name: str | None = None
) -> dict[str, Any]:
    """Get provider-specific kwargs from the config file.

    Reads `base_url`, `api_key_env`, and the `params` table from the user's
    `config.toml` for the given provider.

    When `model_name` is provided, per-model overrides from the `params`
    sub-table are shallow-merged on top.

    Args:
        provider: Provider name (e.g., openai, anthropic, fireworks, ollama).
        model_name: Optional model name for per-model overrides.

    Returns:
        Dictionary of provider-specific kwargs.
    """
    from deepagents_code.model_config import ModelConfig

    config = ModelConfig.load()
    result: dict[str, Any] = config.get_kwargs(provider, model_name=model_name)
    base_url = config.get_base_url(provider)
    if base_url:
        result["base_url"] = base_url
    from deepagents_code.model_config import (
        OPTIONAL_AUTH_ENV,
        PROVIDER_API_KEY_ENV,
        resolve_env_var,
    )

    api_key_env = config.get_api_key_env(provider)
    if not api_key_env:
        api_key_env = PROVIDER_API_KEY_ENV.get(provider)
        if api_key_env:
            logger.debug(
                "No api_key_env in config.toml for '%s';"
                " using hardcoded provider env var",
                provider,
            )
    if api_key_env:
        api_key = resolve_env_var(api_key_env)
        if api_key:
            result["api_key"] = api_key

    # `langchain-ollama` has no `api_key` kwarg; hosted Ollama (Cloud or
    # gateway) needs the bearer token threaded through `client_kwargs.headers`.
    if provider == "ollama":
        optional_env = OPTIONAL_AUTH_ENV.get(provider)
        optional_key = resolve_env_var(optional_env) if optional_env else None
        if optional_key:
            client_kwargs = result.get("client_kwargs")
            if client_kwargs is not None and not isinstance(client_kwargs, dict):
                logger.warning(
                    "Provider 'ollama' has non-mapping client_kwargs (%s);"
                    " skipping Authorization header injection",
                    type(client_kwargs).__name__,
                )
            else:
                client_kwargs = dict(client_kwargs) if client_kwargs else {}
                headers = client_kwargs.get("headers")
                if headers is not None and not isinstance(headers, dict):
                    logger.warning(
                        "Provider 'ollama' has non-mapping client_kwargs.headers"
                        " (%s); skipping Authorization header injection",
                        type(headers).__name__,
                    )
                else:
                    headers = dict(headers) if headers else {}
                    has_auth_header = any(
                        isinstance(k, str) and k.lower() == "authorization"
                        for k in headers
                    )
                    if not has_auth_header:
                        headers["Authorization"] = f"Bearer {optional_key}"
                        client_kwargs["headers"] = headers
                        result["client_kwargs"] = client_kwargs

    retry_section = _read_config_toml_retries()
    retry_kwargs = _resolve_retry_kwargs(retry_section, provider)
    for key, value in retry_kwargs.items():
        result.setdefault(key, value)

    return result


def _create_model_from_class(
    class_path: str,
    model_name: str,
    provider: str,
    kwargs: dict[str, Any],
) -> BaseChatModel:
    """Import and instantiate a custom `BaseChatModel` class.

    Args:
        class_path: Fully-qualified class in `module.path:ClassName` format.
        model_name: Model identifier to pass as `model` kwarg.
        provider: Provider name (for error messages).
        kwargs: Additional keyword arguments for the constructor.

    Returns:
        Instantiated `BaseChatModel`.

    Raises:
        ModelConfigError: If the class cannot be imported, is not a
            `BaseChatModel` subclass, or fails to instantiate.
    """
    from langchain_core.language_models import (
        BaseChatModel as _BaseChatModel,  # Runtime import; module level is typing only
    )

    from deepagents_code.model_config import ModelConfigError

    if ":" not in class_path:
        msg = (
            f"Invalid class_path '{class_path}' for provider '{provider}': "
            "must be in module.path:ClassName format"
        )
        raise ModelConfigError(msg)

    module_path, class_name = class_path.rsplit(":", 1)

    try:
        module = importlib.import_module(module_path)
    except ImportError as e:
        msg = f"Could not import module '{module_path}' for provider '{provider}': {e}"
        raise ModelConfigError(msg) from e

    cls = getattr(module, class_name, None)
    if cls is None:
        msg = (
            f"Class '{class_name}' not found in module '{module_path}' "
            f"for provider '{provider}'"
        )
        raise ModelConfigError(msg)

    if not (isinstance(cls, type) and issubclass(cls, _BaseChatModel)):
        msg = (
            f"'{class_path}' is not a BaseChatModel subclass (got {type(cls).__name__})"
        )
        raise ModelConfigError(msg)

    try:
        return cls(model=model_name, **kwargs)
    except Exception as e:
        msg = f"Failed to instantiate '{class_path}' for '{provider}:{model_name}': {e}"
        raise ModelConfigError(msg) from e


def _create_model_via_init(
    model_name: str,
    provider: str,
    kwargs: dict[str, Any],
) -> BaseChatModel:
    """Create a model using langchain's `init_chat_model`.

    Args:
        model_name: Model identifier.
        provider: Provider name (may be empty for auto-detection).
        kwargs: Additional keyword arguments.

    Returns:
        Instantiated `BaseChatModel`.

    Raises:
        UnknownProviderError: When `provider` is empty and
            `init_chat_model` also fails to infer one. Carries the
            model spec and docs URL as attributes so the UI can render
            a clickable link.
        MissingProviderPackageError: When the provider's LangChain package
            is not installed. Carries the `provider` and `package` to install
            so the UI can render a targeted recovery hint.
        ModelConfigError: On other import, value, or runtime errors.
    """
    from langchain.chat_models import init_chat_model

    from deepagents_code.model_config import (
        MissingProviderPackageError,
        ModelConfigError,
        UnknownProviderError,
    )

    try:
        if provider:
            return init_chat_model(model_name, model_provider=provider, **kwargs)
        return init_chat_model(model_name, **kwargs)
    except ImportError as e:
        import importlib.util

        package_map = {
            "anthropic": "langchain-anthropic",
            "openai": "langchain-openai",
            "google_genai": "langchain-google-genai",
            "google_vertexai": "langchain-google-vertexai",
            "nvidia": "langchain-nvidia-ai-endpoints",
        }
        package = package_map.get(provider, f"langchain-{provider}")
        # Convert pip package name to Python module name for import check.
        module_name = package.replace("-", "_")
        try:
            spec_found = importlib.util.find_spec(module_name) is not None
        except (ImportError, ValueError) as spec_exc:
            # A broken finder is indistinguishable from "not installed" here;
            # log so a real corruption doesn't masquerade as the missing-package
            # hint without leaving a trail.
            logger.debug(
                "find_spec failed for %s; treating provider package as missing: %s",
                module_name,
                spec_exc,
            )
            spec_found = False
        if spec_found:
            # Package is installed but an internal import failed — surface
            # the real error instead of the misleading "missing package" hint.
            msg = (
                f"Provider package '{package}' is installed but failed to "
                f"import for provider '{provider}': {e}"
            )
        else:
            from deepagents_code.extras_info import extra_for_package

            extra = extra_for_package(package)
            if extra is not None:
                msg = (
                    f"Missing package for provider '{provider}'. "
                    f"Install: /install {extra}"
                )
            else:
                from deepagents_code.extras_info import ExtrasIntrospectionError
                from deepagents_code.update_check import install_package_command

                try:
                    install_cmd = install_package_command(package)
                except (ValueError, ExtrasIntrospectionError) as exc:
                    logger.debug(
                        "install_package_command failed; falling back to "
                        "manual hint: %s",
                        exc,
                    )
                    install_hint = f"Install the '{package}' package manually"
                else:
                    install_hint = f"Install with: {install_cmd}"
                msg = (
                    f"Missing package for provider '{provider}'. "
                    f"{install_hint}, then retry with `/model`."
                )
            raise MissingProviderPackageError(
                msg, provider=provider, package=package
            ) from e
        raise ModelConfigError(msg) from e
    except (ValueError, TypeError) as e:
        if not provider:
            # Both app auto-detection and `init_chat_model`'s own inference
            # failed; surface a structured error so the UI can render the
            # docs URL as a clickable link.
            raise UnknownProviderError(model_spec=model_name) from e
        spec = f"{provider}:{model_name}"
        msg = f"Invalid model configuration for '{spec}': {e}"
        raise ModelConfigError(msg) from e
    except Exception as e:  # provider SDK auth/network errors
        spec = f"{provider}:{model_name}" if provider else model_name
        msg = f"Failed to initialize model '{spec}': {e}"
        raise ModelConfigError(msg) from e


@dataclass(frozen=True)
class ModelResult:
    """Result of creating a chat model, bundling the model with its metadata.

    This separates model creation from settings mutation so callers can decide
    when to commit the metadata to global settings.

    Attributes:
        model: The instantiated chat model.
        model_name: Resolved model name.
        provider: Resolved provider name.
        context_limit: Max input tokens from the model profile, or `None`.
        unsupported_modalities: Input modalities not indicated as supported by
            the model profile (e.g. `{"audio", "video"}`).
    """

    model: BaseChatModel
    model_name: str
    provider: str
    context_limit: int | None = None
    unsupported_modalities: frozenset[str] = frozenset()

    def apply_to_settings(self) -> None:
        """Commit this result's metadata to global `settings`."""
        s = _get_settings()
        s.model_name = self.model_name
        s.model_provider = self.provider
        s.model_context_limit = self.context_limit
        s.model_unsupported_modalities = self.unsupported_modalities


def _apply_profile_overrides(
    model: BaseChatModel,
    overrides: dict[str, Any],
    model_name: str,
    *,
    label: str,
    raise_on_failure: bool = False,
) -> None:
    """Merge `overrides` into `model.profile`.

    If the model already has a dict profile, overrides are layered on top
    so existing keys (e.g., `tool_calling`) are preserved unchanged.

    Args:
        model: The chat model whose profile will be updated.
        overrides: Key/value pairs to merge into the profile.
        model_name: Model name used in log/error messages.
        label: Human-readable source label for messages
            (e.g., `"config.toml"`, `"CLI --profile-override"`).
        raise_on_failure: When `True`, raise `ModelConfigError` instead
            of logging a warning if assignment fails.

    Raises:
        ModelConfigError: If `raise_on_failure` is `True` and the model
            rejects profile assignment.
    """
    from deepagents_code.model_config import ModelConfigError

    logger.debug("Applying %s profile overrides: %s", label, overrides)
    profile = getattr(model, "profile", None)
    merged = {**profile, **overrides} if isinstance(profile, dict) else overrides
    try:
        model.profile = merged  # ty: ignore[invalid-assignment]
    except (AttributeError, TypeError, ValueError) as exc:
        if raise_on_failure:
            msg = (
                f"Could not apply {label} to model '{model_name}': {exc}. "
                f"The model may not support profile assignment."
            )
            raise ModelConfigError(msg) from exc
        logger.warning(
            "Could not apply %s profile overrides to model '%s': %s. "
            "Overrides will be ignored.",
            label,
            model_name,
            exc,
        )


def create_model(
    model_spec: str | None = None,
    *,
    extra_kwargs: dict[str, Any] | None = None,
    profile_overrides: dict[str, Any] | None = None,
) -> ModelResult:
    """Create a chat model.

    Uses `init_chat_model` for standard providers, or imports a custom
    `BaseChatModel` subclass when the provider has a `class_path` in config.

    Supports `provider:model` format (e.g., `'openai:gpt-5.5'`)
    for explicit provider selection, or bare model names for auto-detection.

    Args:
        model_spec: Model specification in `provider:model` format (e.g.,
            `'anthropic:claude-sonnet-4-5'`, `'openai:gpt-5.5'`) or just the model
            name for auto-detection (e.g., `'claude-sonnet-4-5'`).

                If not provided, uses environment-based defaults.
        extra_kwargs: Additional kwargs to pass to the model constructor.

            These take highest priority, overriding values from the config file.

            A `CLI_MAX_RETRIES_KEY` entry (set by the `--max-retries` flag) is
            treated specially: it is popped here and re-applied under the
            provider's resolved retry-param name with top precedence, rather than
            being forwarded verbatim to the constructor.
        profile_overrides: Extra profile fields from `--profile-override`.

            Merged on top of config file profile overrides (dcode wins).

    Returns:
        A `ModelResult` containing the model and its metadata.

    Raises:
        ModelConfigError: If provider cannot be determined from the model name
            or required provider package is not installed.
        MissingCredentialsError: If no credentials are configured for the
            resolved provider.

    Examples:
        >>> model = create_model("anthropic:claude-sonnet-4-5")
        >>> model = create_model("openai:gpt-5.5")
        >>> model = create_model("gpt-5.5")  # Auto-detects openai
        >>> model = create_model()  # Uses environment defaults
    """
    from deepagents_code.model_config import (
        IMPLICIT_AUTH_PROVIDERS,
        ModelConfig,
        ModelConfigError,
        ModelSpec,
        apply_stored_credentials,
        get_credential_env_var,
        has_provider_credentials,
        warn_on_split_credential_source,
    )

    if not model_spec:
        model_spec = _get_default_model_spec()

    # Parse provider:model syntax
    provider: str
    model_name: str
    parsed = ModelSpec.try_parse(model_spec)
    if parsed:
        # Explicit provider:model (e.g., "anthropic:claude-sonnet-4-5")
        provider, model_name = parsed.provider, parsed.model
    elif ":" in model_spec:
        # Contains colon but ModelSpec rejected it (empty provider or model)
        _, _, after = model_spec.partition(":")
        if after:
            # Leading colon (e.g., ":claude-opus-4-6") — treat as bare model name
            model_name = after
            provider = detect_provider(model_name) or ""
        else:
            msg = (
                f"Invalid model spec '{model_spec}': model name is required "
                "(e.g., 'anthropic:claude-sonnet-4-5' or 'claude-sonnet-4-5')"
            )
            raise ModelConfigError(msg)
    else:
        # Bare model name — auto-detect provider or let init_chat_model infer
        model_name = model_spec
        provider = detect_provider(model_spec) or ""

    # Stored API keys (added via `/auth`) take effect by being copied onto
    # the env var name LangChain reads. Apply before the credential check so
    # `has_provider_credentials` and the downstream SDK see the same value.
    if provider:
        # Flag a key/endpoint resolved from different env tiers *before*
        # `apply_stored_credentials` bridges stored values onto plain env vars,
        # so the check sees the user's raw env intent rather than post-bridge
        # state. Diagnostic only -- never alters resolution.
        warn_on_split_credential_source(provider)
        apply_stored_credentials(provider)

    from deepagents_code.model_config import CODEX_PROVIDER

    # Early credential check — fail fast with an actionable message instead of
    # letting the provider SDK raise an opaque auth error on first invocation.
    # Providers that support implicit auth (e.g., Vertex AI ADC) are excluded
    # because their env-var mapping is not a reliable indicator.
    if provider and provider not in IMPLICIT_AUTH_PROVIDERS:
        cred_status = has_provider_credentials(provider)
        if cred_status is False:
            from deepagents_code.model_config import MissingCredentialsError

            if provider == CODEX_PROVIDER:
                # No env var to set; point the user at `/auth` instead.
                msg = (
                    "Not signed in to ChatGPT. Run `/auth` and select "
                    "openai_codex to sign in with your ChatGPT account."
                )
                raise MissingCredentialsError(msg, provider=provider, env_var=None)
            env_var = get_credential_env_var(provider)
            display_env = env_var or f"<{provider} API key>"
            msg = (
                f"No credentials found for provider '{provider}'. "
                f"Please set the {display_env} environment variable."
            )
            raise MissingCredentialsError(msg, provider=provider, env_var=env_var)

    # Provider-specific kwargs (with per-model overrides)
    kwargs = _get_provider_kwargs(provider, model_name=model_name)

    # Compose under existing kwargs: profile < config.toml < --model-params
    # (applied below). The app's OpenRouter profile is stacked on top of the
    # built-in SDK profile so its `pre_init` (version check) and factory
    # (app attribution) compose into a single `apply_provider_profile` call.
    if provider:
        from deepagents.profiles.provider import apply_provider_profile

        if provider == "openrouter":
            _ensure_cli_openrouter_profile_registered()

        spec = f"{provider}:{model_name}" if model_name else provider
        try:
            kwargs = apply_provider_profile(spec, kwargs)
        except ModelConfigError:
            raise
        except Exception as exc:
            # `pre_init` and `init_kwargs_factory` callables registered on a
            # `ProviderProfile` may raise arbitrary exceptions (e.g. an
            # `ImportError` from the OpenRouter min-version check). Surface
            # them as `ModelConfigError` so the app's error path renders an
            # actionable message instead of a raw stack trace.
            logger.debug(
                "ProviderProfile resolution for %r failed.", spec, exc_info=True
            )
            msg = (
                f"Failed to apply provider profile for '{spec}': {exc}. "
                f"Check that the provider package is installed and up to date, "
                f"or set explicit kwargs via `--model-params`."
            )
            raise ModelConfigError(msg) from exc

    # App --model-params take highest priority. Copy defensively before popping
    # the CLI sentinel so a caller that retains and reuses this dict (e.g. the
    # app re-creating the model on a runtime `/model` switch) keeps the sentinel
    # for the next provider's resolution.
    cli_max_retries: int | None = None
    if extra_kwargs:
        extra_kwargs = dict(extra_kwargs)
        cli_max_retries = extra_kwargs.pop(CLI_MAX_RETRIES_KEY, None)
        kwargs.update(extra_kwargs)

    # `--max-retries` outranks everything: fold it under the provider's resolved
    # retry-param name (honoring `[retries.<provider>].param`) so a custom
    # provider whose kwarg is not `max_retries` is still served. Applied after
    # the `extra_kwargs` merge so it wins over a `max_retries` in `--model-params`.
    if cli_max_retries is not None:
        kwargs[_resolve_retry_param_name(provider)] = cli_max_retries

    # Check if this provider uses a custom BaseChatModel class
    config = ModelConfig.load()
    class_path = config.get_class_path(provider) if provider else None

    if provider == CODEX_PROVIDER:
        # Codex models are constructed directly via `_ChatOpenAICodex` so the
        # `token_provider=` kwarg is wired to the on-disk OAuth token store
        # before any request goes out. `init_chat_model` does not know about
        # this class and would route through API-key `ChatOpenAI` instead.
        from deepagents_code.integrations import openai_codex as _codex
        from deepagents_code.model_config import (
            MissingCredentialsError,
            ModelConfigError,
        )

        # Drop any `api_key` left in kwargs (e.g. from a config-level
        # `api_key_env` set on the codex provider, or a `--model-params
        # api_key=...`) so the bearer always comes from the OAuth
        # `token_provider` rather than a static key.
        kwargs.pop("api_key", None)
        try:
            model = _codex.build_chat_model(model_name, **kwargs)
        except FileNotFoundError as exc:
            msg = (
                "Not signed in to ChatGPT. Run `/auth` and select "
                "openai_codex to sign in with your ChatGPT account."
            )
            raise MissingCredentialsError(msg, provider=provider, env_var=None) from exc
        except _codex.CodexAuthExpiredError as exc:
            # A token exists but its refresh token is dead. Route through the
            # same `MissingCredentialsError` recovery path as a missing token
            # (which the retry flow re-attempts after `/auth`) instead of the
            # generic `ModelConfigError` below, which would not offer sign-in.
            msg = (
                "ChatGPT session expired. Run `/auth` and select openai_codex "
                "to sign in again."
            )
            raise MissingCredentialsError(msg, provider=provider, env_var=None) from exc
        except Exception as exc:
            spec = f"{provider}:{model_name}"
            msg = f"Failed to initialize Codex model '{spec}': {exc}"
            raise ModelConfigError(msg) from exc
    elif class_path:
        model = _create_model_from_class(class_path, model_name, provider, kwargs)
    else:
        model = _create_model_via_init(model_name, provider, kwargs)

    resolved_provider = provider or getattr(model, "_model_provider", provider)

    # Apply profile overrides from config.toml (e.g., max_input_tokens)
    if provider:
        config_profile_overrides = config.get_profile_overrides(
            provider, model_name=model_name
        )
        if config_profile_overrides:
            _apply_profile_overrides(
                model,
                config_profile_overrides,
                model_name,
                label=f"config.toml (provider '{provider}')",
            )

    # App --profile-override takes highest priority (on top of config.toml)
    if profile_overrides:
        _apply_profile_overrides(
            model,
            profile_overrides,
            model_name,
            label="CLI --profile-override",
            raise_on_failure=True,
        )

    # Extract context limit and modality support from model profile
    context_limit: int | None = None
    unsupported_modalities: frozenset[str] = frozenset()
    profile = getattr(model, "profile", None)
    if isinstance(profile, dict):
        if isinstance(profile.get("max_input_tokens"), int):
            context_limit = profile["max_input_tokens"]

        modality_keys = {
            "image_inputs": "image",
            "audio_inputs": "audio",
            "video_inputs": "video",
            "pdf_inputs": "pdf",
        }
        unsupported_modalities = frozenset(
            label for key, label in modality_keys.items() if profile.get(key) is False
        )

    return ModelResult(
        model=model,
        model_name=model_name,
        provider=resolved_provider,
        context_limit=context_limit,
        unsupported_modalities=unsupported_modalities,
    )


def validate_model_capabilities(model: BaseChatModel, model_name: str) -> None:
    """Validate that the model has required capabilities for `deepagents`.

    Checks the model's profile (if available) to ensure it supports tool calling, which
    is required for agent functionality. Issues warnings for models without profiles or
    with limited context windows.

    Args:
        model: The instantiated model to validate.
        model_name: Model name for error/warning messages.

    Note:
        This validation is best-effort. Models without profiles will pass with
        a warning. Calls `sys.exit(1)` if the model's profile explicitly
        indicates `tool_calling=False`.
    """
    console = _get_console()
    profile = getattr(model, "profile", None)

    if profile is None:
        # Model doesn't have profile data - warn but allow
        console.print(
            f"[dim][yellow]Note:[/yellow] No capability profile for "
            f"'{model_name}'. Cannot verify tool calling support.[/dim]"
        )
        return

    if not isinstance(profile, dict):
        return

    # Check required capability: tool_calling
    tool_calling = profile.get("tool_calling")
    if tool_calling is False:
        console.print(
            f"[bold red]Error:[/bold red] Model '{model_name}' "
            "does not support tool calling."
        )
        console.print(
            "\nDeep Agents requires tool calling for agent functionality. "
            "Please choose a model that supports tool calling."
        )
        console.print("\nSee MODELS.md for supported models.")
        sys.exit(1)

    # Warn about potentially limited context (< 8k tokens)
    max_input_tokens = profile.get("max_input_tokens")
    if max_input_tokens and max_input_tokens < 8000:  # noqa: PLR2004  # Model context window default
        console.print(
            f"[dim][yellow]Warning:[/yellow] Model '{model_name}' has limited context "
            f"({max_input_tokens:,} tokens). Agent performance may be affected.[/dim]"
        )


def _get_console() -> Console:
    """Return the lazily-initialized global `Console` instance.

    Defers the `rich.console` import until console output is actually
    needed. The result is cached in `globals()["console"]`.

    Returns:
        The global Rich `Console` singleton.
    """
    cached = globals().get("console")
    if cached is not None:
        return cached
    with _singleton_lock:
        cached = globals().get("console")
        if cached is not None:
            return cached
        from rich.console import Console

        inst = Console(highlight=False)
        globals()["console"] = inst
        return inst


def _get_settings() -> Settings:
    """Return the lazily-initialized global `Settings` instance.

    Ensures bootstrap has run before constructing settings. The result is cached
    in `globals()["settings"]` so subsequent access — including
    `from config import settings` in other modules — resolves instantly.

    Returns:
        The global `Settings` singleton.
    """
    cached = globals().get("settings")
    if cached is not None:
        return cached
    with _singleton_lock:
        cached = globals().get("settings")
        if cached is not None:
            return cached
        _ensure_bootstrap()
        try:
            inst = Settings.from_environment(start_path=_bootstrap_start_path)
        except Exception:
            logger.exception(
                "Failed to initialize settings from environment (start_path=%s)",
                _bootstrap_start_path,
            )
            raise
        globals()["settings"] = inst
        return inst


def __getattr__(name: str) -> Settings | Console:
    """Lazy module attributes for `settings` and `console`.

    Defers heavy initialization until first access. Subsequent accesses hit
    the module-level attribute directly (no `__getattr__` overhead).

    Returns:
        The requested lazy singleton.

    Raises:
        AttributeError: If *name* is not a lazily-provided attribute.
    """
    if name == "settings":
        return _get_settings()
    if name == "console":
        return _get_console()
    msg = f"module {__name__!r} has no attribute {name!r}"
    raise AttributeError(msg)