"""Update lifecycle for `deepagents-code`.

Handles version checking against PyPI (with caching), install-method detection,
auto-upgrade execution, config-driven opt-in/out, notification throttling, and
"what's new" tracking.

Most public entry points absorb errors and return sentinel values.
`set_auto_update` raises on write failures so callers can surface
actionable feedback.
"""

from __future__ import annotations

import asyncio
import json
import logging
import operator
import os
import re
import shlex
import shutil
import sys
import time
import tomllib
from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
from contextlib import suppress
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any, Literal, TextIO

from packaging.version import InvalidVersion, Version

from deepagents_code._version import PYPI_URL, SDK_PYPI_URL, USER_AGENT, __version__
from deepagents_code.model_config import DEFAULT_CONFIG_PATH, DEFAULT_STATE_DIR

if TYPE_CHECKING:
    from pathlib import Path

logger = logging.getLogger(__name__)

CACHE_FILE: Path = DEFAULT_STATE_DIR / "latest_version.json"
"""On-disk cache of the latest published dcode/SDK versions and SDK release times.

Populated by `get_latest_version`; reads short-circuit on the cached payload
when it is younger than `CACHE_TTL`. SDK upload timestamps are stored under
`_SDK_RELEASE_TIMES_KEY`.
"""

UPDATE_STATE_FILE: Path = DEFAULT_STATE_DIR / "update_state.json"
"""Persistent flags for the update-notification UX.

Tracks which version the user has been notified about (`notified_version`,
`notified_at`) and the most recent version they've seen the splash for
(`seen_version`, `seen_at`). Read by `should_notify_update` and friends
to suppress repeat notifications across invocations. Auto-update opt-outs
live in `config.toml`, not here.
"""

CACHE_TTL = 86_400  # 24 hours
"""Maximum age in seconds before `CACHE_FILE` entries are considered stale.

A cached `latest_version.json` younger than this is reused without an HTTP
call to PyPI; older payloads trigger a fresh fetch. Set conservatively at
24h since release cadence is on the order of days, not minutes.
"""

INSTALLED_AGE_NOTICE_DAYS = 7
"""Minimum installed-version age before update notices call it out explicitly."""

_SDK_RELEASE_TIMES_KEY = "sdk_release_times"
"""`CACHE_FILE` key for cached SDK upload timestamps, keyed by version string."""

InstallMethod = Literal["uv", "brew", "other", "unknown"]

FALLBACK_UPGRADE_COMMAND = "uv tool upgrade deepagents-code"
"""Generic upgrade hint used when install-method detection fails.

Callers that surface an upgrade command in user-facing text should prefer
`upgrade_command()`; this constant exists so those callers have something
to render when detection raises unexpectedly. The documented install path
is `uv tool install` (see `scripts/install.sh`), so the uv command is the
right display fallback. Execution paths still refuse unrecognized installs
instead of updating a separate environment.
"""

_UPGRADE_COMMANDS: dict[InstallMethod, str] = {
    "uv": "uv tool upgrade deepagents-code",
    "brew": "brew upgrade deepagents-code",
}
"""Upgrade commands keyed by install method.

`perform_upgrade` runs only the command matching the detected install method;
no fallback chain. Unknown non-editable installs are refused rather than
upgraded with a different package manager, because that can update a separate
environment from the one currently providing `dcode`.
"""

_UV_PRERELEASE_UPGRADE_COMMAND = "uv tool upgrade deepagents-code --prerelease allow"
"""uv upgrade command that opts into alpha/beta/rc release resolution."""

_PRERELEASE_UNSUPPORTED_MESSAGE = (
    "Pre-release updates aren't supported for this install. Reinstall with "
    "pre-releases enabled:\n"
    '  curl -LsSf https://langch.in/dcode | DEEPAGENTS_CODE_PRERELEASE="allow" bash'
)
"""User-facing reason a pre-release upgrade is refused on non-uv installs.

Points at the install script (uv under the hood) rather than raw uv commands,
since that one-liner is the path we promote.
"""

_UPGRADE_TIMEOUT = 120  # seconds
"""Wall-clock cap for `perform_upgrade` and `perform_install_extra`."""

UPDATE_LOG_DIR: Path = DEFAULT_STATE_DIR / "update_logs"
"""Directory for persisted update command logs."""

UPDATE_LOG_RETENTION_DAYS = 14
"""Delete update logs older than this many days."""

UPDATE_LOG_MAX_FILES = 10
"""Keep at most this many newest update logs."""

UpgradeProgressCallback = Callable[[str], Awaitable[None] | None]


def _parse_version(v: str) -> Version:
    """Parse a PEP 440 version string into a comparable `Version` object.

    Supports stable (`1.2.3`) and pre-release (`1.2.3a1`, `1.2.3rc2`) versions.

    Args:
        v: Version string like `'1.2.3'` or `'1.2.3a1'`.

    Returns:
        A `packaging.version.Version` instance.
    """
    return Version(v.strip())  # raises InvalidVersion for non-PEP 440 strings


def is_installed_version_at_least(version: str) -> bool:
    """Return whether installed package metadata is at least `version`."""
    try:
        from importlib.metadata import PackageNotFoundError, version as pkg_version

        installed = _parse_version(pkg_version("deepagents-code"))
        target = _parse_version(version)
    except (InvalidVersion, PackageNotFoundError):
        return False
    return installed >= target


def _latest_from_releases(
    releases: Mapping[str, Sequence[object]],
    *,
    include_prereleases: bool,
) -> str | None:
    """Pick the newest version from a PyPI `releases` mapping.

    Skips versions with no uploaded files (empty entries) and, when
    *include_prereleases* is `False`, skips pre-release versions.

    Args:
        releases: The `releases` dict from the PyPI JSON API.
        include_prereleases: Whether to consider pre-release versions.

    Returns:
        The highest matching version string, or `None` if none qualify.
    """
    best: Version | None = None
    best_str: str | None = None
    for ver_str, files in releases.items():
        if not files:
            continue
        try:
            ver = Version(ver_str)
        except InvalidVersion:
            logger.debug("Skipping unparseable release key: %s", ver_str)
            continue
        if not include_prereleases and ver.is_prerelease:
            continue
        if best is None or ver > best:
            best = ver
            best_str = ver_str
    return best_str


def get_cached_update_available() -> tuple[bool, str | None]:
    """Check for updates using only a fresh local cache entry.

    This is the startup fast path: it never contacts PyPI. Stale, missing,
    corrupt, or unparsable cache data is treated as "no cached update answer" so
    callers can launch immediately and let a background update check refresh the
    cache later.

    Returns:
        A `(available, latest)` tuple. `latest` is `None` when the cache cannot
            provide a fresh answer.
    """
    try:
        installed = _parse_version(__version__)
    except InvalidVersion:
        logger.warning(
            "Installed version %r is not PEP 440 compliant; "
            "cache-only update checks disabled for this install",
            __version__,
        )
        return False, None

    cache_key = "version_prerelease" if installed.is_prerelease else "version"
    try:
        if not CACHE_FILE.exists():
            return False, None
        data = json.loads(CACHE_FILE.read_text(encoding="utf-8"))
        if not isinstance(data, dict):
            return False, None
        checked_at = data.get("checked_at")
        if not isinstance(checked_at, (int, float)):
            return False, None
        if time.time() - checked_at >= CACHE_TTL:
            return False, None
        value = data.get(cache_key)
        if not isinstance(value, str):
            return False, None
        return _parse_version(value) > installed, value
    except (OSError, json.JSONDecodeError, TypeError, InvalidVersion):
        logger.debug("Failed to read cache-only update answer", exc_info=True)
        return False, None


def get_latest_version(
    *,
    bypass_cache: bool = False,
    include_prereleases: bool = False,
) -> str | None:
    """Fetch the latest deepagents-code version from PyPI, with caching.

    Results are cached to `CACHE_FILE` to avoid repeated network calls.
    The cache stores both the latest stable and pre-release versions so a
    single PyPI request serves both code paths.

    Args:
        bypass_cache: Skip the cache and always hit PyPI.
        include_prereleases: When `True`, consider pre-release versions
            (alpha, beta, rc). Stable users should leave this `False`.

    Returns:
        The latest version string, or `None` on any failure.
    """
    cache_key = "version_prerelease" if include_prereleases else "version"
    cached_version: str | None = None

    try:
        if not bypass_cache and CACHE_FILE.exists():
            data = json.loads(CACHE_FILE.read_text(encoding="utf-8"))
            fresh = time.time() - data.get("checked_at", 0) < CACHE_TTL
            if fresh and cache_key in data:
                value = data[cache_key]
                cached_version = value if isinstance(value, str) else None
            release_times = data.get("release_times")
            has_installed_release_time = (
                isinstance(release_times, dict) and __version__ in release_times
            )
            if fresh and cache_key in data and has_installed_release_time:
                return cached_version
    except (OSError, json.JSONDecodeError, TypeError):
        logger.debug("Failed to read update-check cache", exc_info=True)

    try:
        import requests
    except ImportError:
        logger.warning(
            "requests package not installed — update checks disabled. "
            "Install with: uv tool install -U deepagents-code --with requests"
        )
        return cached_version

    try:
        resp = requests.get(
            PYPI_URL,
            headers={"User-Agent": USER_AGENT},
            timeout=3,
        )
        resp.raise_for_status()
        payload = resp.json()
        stable: str = payload["info"]["version"]
        releases: dict[str, list[object]] = payload.get("releases", {})
        if not releases:
            logger.debug("PyPI response missing or empty 'releases' key")
        prerelease = _latest_from_releases(releases, include_prereleases=True)
    except (requests.RequestException, OSError, KeyError, json.JSONDecodeError):
        logger.debug("Failed to fetch latest version from PyPI", exc_info=True)
        return cached_version

    release_times = _extract_release_times(
        payload, stable=stable, prerelease=prerelease, installed=__version__
    )

    try:
        CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
        CACHE_FILE.write_text(
            json.dumps(
                {
                    "version": stable,
                    "version_prerelease": prerelease,
                    "release_times": release_times,
                    "checked_at": time.time(),
                }
            ),
            encoding="utf-8",
        )
    except OSError:
        logger.debug("Failed to write update-check cache", exc_info=True)

    return prerelease if include_prereleases else stable


def _extract_release_times(
    payload: dict[str, Any],
    *,
    stable: str,
    prerelease: str | None,
    installed: str | None = None,
) -> dict[str, str]:
    """Pull `upload_time_iso_8601` for the given versions out of a PyPI payload.

    PyPI lists per-file uploads; the first file's timestamp is used as a
    stand-in for the release's publish time (files typically land within
    seconds of each other). Looks up both versions under `releases[ver]`
    rather than `payload["urls"]`, which reflects the project's
    `info.version` and may not match `stable` when the latest on PyPI is
    a pre-release.

    Args:
        payload: Parsed PyPI JSON response.
        stable: Latest stable version string.
        prerelease: Latest pre-release version string, if any.
        installed: Currently installed version string, if it should be cached.

    Returns:
        Mapping of version string to ISO-8601 upload time. Silently drops
        versions whose timestamp is missing or malformed.
    """
    times: dict[str, str] = {}
    releases = payload.get("releases")
    if not isinstance(releases, dict):
        return times
    for ver in (stable, prerelease, installed):
        if not ver:
            continue
        files = releases.get(ver)
        if not isinstance(files, list) or not files:
            continue
        ts = _upload_time(files[0])
        if ts:
            times[ver] = ts
    return times


def _upload_time(file_entry: object) -> str | None:
    """Return `upload_time_iso_8601` from a PyPI file entry, or `None`."""
    if not isinstance(file_entry, dict):
        return None
    # `isinstance(..., dict)` narrows to `dict[Unknown, Unknown]`, so `.get()`
    # overload resolution is ambiguous. PyPI payloads are str-keyed in practice
    # and the `isinstance(value, str)` check below validates the result anyway.
    value = file_entry.get("upload_time_iso_8601")  # ty: ignore[invalid-argument-type]
    return value if isinstance(value, str) else None


def get_release_time(version: str | None) -> str | None:
    """Return the cached ISO-8601 upload time for `version`, or `None`.

    Only versions captured during a prior `get_latest_version` call are
    available; unknown versions, or a `None` input, return `None`.
    """
    if not version:
        return None
    try:
        if CACHE_FILE.exists():
            data = json.loads(CACHE_FILE.read_text(encoding="utf-8"))
            if isinstance(data, dict):
                times = data.get("release_times")
                if isinstance(times, dict):
                    value = times.get(version)
                    if isinstance(value, str):
                        return value
    except (OSError, json.JSONDecodeError):
        logger.debug("Failed to read release_times from cache", exc_info=True)
    return None


def _format_age_from_iso(iso: str | None) -> str:
    """Return `'released Nd ago'` for an ISO-8601 timestamp, or `""` on failure."""
    if not iso:
        return ""
    from deepagents_code.sessions import format_relative_timestamp

    age = format_relative_timestamp(iso)
    return f"released {age}" if age else ""


def format_release_age(version: str | None) -> str:
    """Return a human-readable age for `version` (e.g., `'released 3d ago'`).

    Returns an empty string when the upload time is unknown (cache entry
    lacks `release_times` for this version, or a `None` version) so callers
    can concatenate unconditionally.
    """
    return _format_age_from_iso(get_release_time(version))


def format_age_suffix(version: str | None) -> str:
    """Return `", released Nd ago"` for `version`, or `""` when unknown.

    The `", "` separator is included so callers can splice the age into a
    parenthetical unconditionally — if the age is unknown, the empty
    string collapses cleanly into the surrounding text.
    """
    age = format_release_age(version)
    return f", {age}" if age else ""


def format_release_age_parenthetical(version: str | None) -> str:
    """Return `" (released Nd ago)"` for `version`, or `""` when unknown."""
    age = format_release_age(version)
    return f" ({age})" if age else ""


def _days_old_from_iso(iso: str | None) -> int | None:
    """Return whole elapsed days for an ISO-8601 timestamp, or `None` on failure."""
    if not iso:
        return None
    try:
        dt = datetime.fromisoformat(iso).astimezone()
    except (ValueError, TypeError):
        logger.debug(
            "Failed to parse release timestamp %r for installed age",
            iso,
            exc_info=True,
        )
        return None

    days = (datetime.now(tz=dt.tzinfo) - dt).days
    return max(days, 0)


def format_installed_age_suffix(version: str | None) -> str:
    """Return `" (N days old)"` for installed versions at least a week old."""
    days = _days_old_from_iso(get_release_time(version))
    if days is None or days < INSTALLED_AGE_NOTICE_DAYS:
        return ""
    unit = "day" if days == 1 else "days"
    return f" ({days} {unit} old)"


def get_sdk_release_time(
    version: str | None, *, bypass_cache: bool = False
) -> str | None:
    """Return the ISO-8601 upload time for `deepagents` SDK `version`.

    Reads from `CACHE_FILE` under `sdk_release_times`, falling back to a
    single PyPI fetch on cache miss and writing the result back so
    subsequent calls stay local.

    Args:
        version: Installed SDK version string.
        bypass_cache: Skip the cache read and always hit PyPI.

            The result is still written back to the cache.

    Returns:
        The ISO-8601 upload timestamp, or `None` on any failure (missing
            version, unresolvable on PyPI, `requests` unavailable, or
            network error).
    """
    if not version:
        return None

    try:
        if not bypass_cache and CACHE_FILE.exists():
            data = json.loads(CACHE_FILE.read_text(encoding="utf-8"))
            if isinstance(data, dict):
                times = data.get(_SDK_RELEASE_TIMES_KEY)
                if isinstance(times, dict):
                    cached = times.get(version)
                    if isinstance(cached, str):
                        return cached
    except (OSError, json.JSONDecodeError):
        logger.debug("Failed to read sdk release_times from cache", exc_info=True)

    try:
        import requests
    except ImportError:
        logger.debug("requests unavailable — SDK release time lookup disabled")
        return None

    try:
        resp = requests.get(
            SDK_PYPI_URL,
            headers={"User-Agent": USER_AGENT},
            timeout=3,
        )
        resp.raise_for_status()
        payload = resp.json()
        releases = payload.get("releases")
        if not isinstance(releases, dict):
            return None
        files = releases.get(version)
        if not isinstance(files, list) or not files:
            return None
        iso = _upload_time(files[0])
    except (requests.RequestException, OSError, json.JSONDecodeError):
        logger.debug("Failed to fetch SDK release time from PyPI", exc_info=True)
        return None

    if iso:
        _write_sdk_release_time(version, iso)
    return iso


def _write_sdk_release_time(version: str, iso: str) -> None:
    """Merge a single SDK release timestamp into `CACHE_FILE`.

    A corrupt existing cache is overwritten rather than propagating the
    decode error — otherwise every caller would keep paying the PyPI
    round-trip because the write never succeeds.
    """
    data: dict[str, object] = {}
    if CACHE_FILE.exists():
        try:
            raw = json.loads(CACHE_FILE.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            logger.warning(
                "SDK release-time cache is corrupt; overwriting", exc_info=True
            )
        except OSError:
            logger.debug("Failed to read SDK release-time cache", exc_info=True)
            return
        else:
            if isinstance(raw, dict):
                data = raw

    times: dict[str, str] = {}
    existing = data.get(_SDK_RELEASE_TIMES_KEY)
    if isinstance(existing, dict):
        times.update(
            {
                k: v
                for k, v in existing.items()
                if isinstance(k, str) and isinstance(v, str)
            }
        )
    times[version] = iso
    data[_SDK_RELEASE_TIMES_KEY] = times
    try:
        CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
        CACHE_FILE.write_text(json.dumps(data), encoding="utf-8")
    except OSError:
        logger.debug("Failed to write SDK release time to cache", exc_info=True)


def format_sdk_release_age(version: str | None) -> str:
    """Return a human-readable age for SDK `version` (e.g., `'released 3d ago'`).

    May trigger a single PyPI fetch on cache miss (3s timeout). Returns an
    empty string on any failure so callers can concatenate unconditionally.
    """
    return _format_age_from_iso(get_sdk_release_time(version))


def format_sdk_age_suffix(version: str | None) -> str:
    """Return `", released Nd ago"` for SDK `version`, or `""` when unknown.

    The `", "` separator is included so callers can splice the age into a
    line unconditionally — if the age is unknown, the empty string
    collapses cleanly into the surrounding text. May trigger a single
    PyPI fetch on cache miss.
    """
    age = format_sdk_release_age(version)
    return f", {age}" if age else ""


def _read_update_state() -> dict[str, object]:
    """Read the shared update state file.

    Returns:
        Parsed dict, or empty dict on missing/corrupt file.
    """
    try:
        if UPDATE_STATE_FILE.exists():
            raw = json.loads(UPDATE_STATE_FILE.read_text(encoding="utf-8"))
            if isinstance(raw, dict):
                return raw
    except (OSError, json.JSONDecodeError):
        logger.debug("Failed to read update state file", exc_info=True)
    return {}


def _write_update_state(
    patch: dict[str, object], *, remove_keys: tuple[str, ...] = ()
) -> bool:
    """Merge *patch* into the shared update state file and drop *remove_keys*.

    Args:
        patch: Keys to merge into the existing state.
        remove_keys: Keys to drop from the existing state before writing.

    Returns:
        `True` if the state was persisted, `False` if the write failed (the
            error is logged, not raised, so callers stay fail-soft but can surface
            the miss when a stale state has user-visible consequences).
    """
    data = _read_update_state()
    for key in remove_keys:
        data.pop(key, None)
    data.update(patch)
    try:
        UPDATE_STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
        UPDATE_STATE_FILE.write_text(json.dumps(data), encoding="utf-8")
    except OSError:
        logger.warning(
            "Failed to write update state to %s",
            UPDATE_STATE_FILE,
            exc_info=True,
        )
        return False
    return True


def should_notify_update(latest: str) -> bool:
    """Return whether the user should be notified about version *latest*.

    Throttles notifications to at most once per `CACHE_TTL` period for a
    given version, preventing repeated banners every session.

    Args:
        latest: The version string to check against.

    Returns:
        `True` if the user should see the update banner, `False` if the
            notification was already shown within the `CACHE_TTL` window.
    """
    data = _read_update_state()
    notified_at = data.get("notified_at", 0)
    notified_version = data.get("notified_version")
    return not (
        isinstance(notified_at, (int, float))
        and notified_version == latest
        and time.time() - notified_at < CACHE_TTL
    )


def mark_update_notified(latest: str) -> None:
    """Record that the user was notified about version *latest*.

    Writes into the shared update state file so a subsequent
    `should_notify_update` call can suppress duplicate banners.

    Args:
        latest: The version string that was shown.
    """
    _write_update_state({"notified_at": time.time(), "notified_version": latest})


def clear_update_notified() -> None:
    """Clear the "already notified" marker so the update modal re-opens next launch.

    Removes both `notified_at` and `notified_version` from the shared
    update state file.
    """
    _write_update_state({}, remove_keys=("notified_at", "notified_version"))


def is_update_available(
    *,
    bypass_cache: bool = False,
    include_prereleases: bool | None = None,
) -> tuple[bool, str | None]:
    """Check whether a newer version of deepagents-code is available.

    When the installed version is a pre-release (e.g. `0.0.35a1`),
    pre-release versions on PyPI are included in the comparison so alpha
    testers are notified of newer alphas and the eventual stable release.
    Stable installs only compare against stable PyPI releases unless
    `include_prereleases` is explicitly set.

    Args:
        bypass_cache: Skip the cache and always hit PyPI.
        include_prereleases: Override whether alpha/beta/rc releases are
            considered. When `None`, this follows the installed version.

    Returns:
        A `(available, latest)` tuple.

            `latest` is the PyPI version string when it was fetched and parsed
            successfully, or `None` when the PyPI check itself fails (network
            error, unparseable response, or non-PEP 440 installed version).
            `available` is `True` only when `latest` is strictly newer than
            the installed version. Callers can therefore distinguish "already
            up to date" (`(False, "1.2.3")`) from "could not reach PyPI"
            (`(False, None)`).
    """
    try:
        installed = _parse_version(__version__)
    except InvalidVersion:
        logger.warning(
            "Installed version %r is not PEP 440 compliant; "
            "update checks disabled for this install",
            __version__,
        )
        return False, None

    include_prereleases = _resolve_include_prereleases(
        include_prereleases,
        installed=installed,
    )
    latest = get_latest_version(
        bypass_cache=bypass_cache,
        include_prereleases=include_prereleases,
    )
    if latest is None:
        return False, None

    try:
        return _parse_version(latest) > installed, latest
    except InvalidVersion:
        logger.debug("Failed to compare versions", exc_info=True)
        return False, None


# ---------------------------------------------------------------------------
# Install method detection
# ---------------------------------------------------------------------------


def _resolve_include_prereleases(
    include_prereleases: bool | None,
    *,
    installed: Version | None = None,
) -> bool:
    """Resolve update channel preference from the requested or installed channel.

    Args:
        include_prereleases: Explicit channel preference, or `None` to infer
            from the installed version.
        installed: Parsed installed version to reuse when the caller already
            has one.

    Returns:
        `True` when pre-release versions should be considered.
    """
    if include_prereleases is not None:
        return include_prereleases
    if installed is None:
        try:
            installed = _parse_version(__version__)
        except InvalidVersion:
            logger.warning(
                "Installed version %r is not PEP 440 compliant; "
                "defaulting to stable-only upgrades",
                __version__,
            )
            return False
    return installed.is_prerelease


def detect_install_method() -> InstallMethod:
    """Detect how `deepagents-code` was installed.

    Checks `sys.prefix` against known paths for uv and Homebrew.

    Returns:
        The detected install method: `'uv'`, `'brew'`, `'other'`, or `'unknown'`
            (editable/dev installs).
    """
    from deepagents_code.config import _is_editable_install

    prefix = sys.prefix
    # uv tool installs live under ~/.local/share/uv/tools/
    if "/uv/tools/" in prefix or "\\uv\\tools\\" in prefix:
        return "uv"
    # Homebrew prefixes
    if any(
        prefix.startswith(p)
        for p in ("/opt/homebrew", "/usr/local/Cellar", "/home/linuxbrew")
    ):
        return "brew"
    # Editable / dev installs — don't auto-upgrade
    if _is_editable_install():
        return "unknown"
    return "other"


def upgrade_command(
    method: InstallMethod | None = None,
    *,
    include_prereleases: bool | None = None,
) -> str:
    """Return the shell command to upgrade `deepagents-code`.

    Falls back to the documented uv command for display-only guidance.

    Args:
        method: Install method override.

            Auto-detected if `None`.
        include_prereleases: Whether to include alpha/beta/rc releases. When
            `None`, follows the installed version's channel. When `True`,
            returns the uv pre-release command regardless of `method`, since
            only uv can be steered onto the pre-release channel.
    """
    include_prereleases = _resolve_include_prereleases(include_prereleases)
    if include_prereleases:
        return _UV_PRERELEASE_UPGRADE_COMMAND
    if method is None:
        method = detect_install_method()
    return _UPGRADE_COMMANDS.get(method, FALLBACK_UPGRADE_COMMAND)


def prerelease_upgrade_supported(
    method: InstallMethod | None = None,
) -> tuple[bool, str | None]:
    """Return whether pre-release upgrades are supported for the install method.

    Pre-release channel switching is only safe for `uv tool` installs, where
    `uv tool upgrade --prerelease allow` re-resolves against the pre-release
    feed. Other package managers can't be steered onto that channel, so callers
    should refuse before promising an upgrade.

    Args:
        method: Install method override.

            Auto-detected if `None`.

    Returns:
        A `(supported, reason)` tuple. `reason` is `None` when supported, else a
        user-facing explanation of why the pre-release upgrade is refused.
    """
    if method is None:
        method = detect_install_method()
    if method != "uv":
        return False, _PRERELEASE_UNSUPPORTED_MESSAGE
    return True, None


def cleanup_update_logs(
    *,
    retention_days: int = UPDATE_LOG_RETENTION_DAYS,
    max_files: int = UPDATE_LOG_MAX_FILES,
) -> None:
    """Remove old update logs while preserving the newest recent logs.

    Args:
        retention_days: Maximum age in days to keep.
        max_files: Maximum number of newest log files to keep.
    """
    try:
        if not UPDATE_LOG_DIR.exists():
            return
        logs = sorted(
            (
                (p, p.stat().st_mtime)
                for p in UPDATE_LOG_DIR.glob("*.log")
                if p.is_file()
            ),
            key=operator.itemgetter(1),
            reverse=True,
        )
        cutoff = time.time() - (retention_days * 86_400)
        for idx, (path, mtime) in enumerate(logs):
            if idx >= max_files or mtime < cutoff:
                path.unlink(missing_ok=True)
    except OSError:
        logger.debug("Failed to clean up update logs", exc_info=True)


def create_update_log_path() -> Path:
    """Return a new timestamped update log path and clean stale logs."""
    cleanup_update_logs()
    stamp = datetime.now(tz=UTC).strftime("%Y%m%d-%H%M%S")
    return UPDATE_LOG_DIR / f"{stamp}-update.log"


async def _emit_progress(callback: UpgradeProgressCallback | None, line: str) -> None:
    """Send a progress line to *callback*, supporting sync or async callbacks."""
    if callback is None:
        return
    result = callback(line)
    if isinstance(result, Awaitable):
        await result


async def _read_stream(
    stream: asyncio.StreamReader,
    *,
    lines: list[str],
    log_file: TextIO | None,
    progress: UpgradeProgressCallback | None,
) -> None:
    """Read subprocess output, append it to the log file, and emit progress."""
    while True:
        raw = await stream.readline()
        if not raw:
            return
        line = raw.decode(errors="replace").rstrip("\n")
        lines.append(line)
        if log_file is not None:
            with suppress(OSError):
                log_file.write(f"{line}\n")
                log_file.flush()
        await _emit_progress(progress, line)


async def _run_install_subprocess(
    cmd: str,
    *,
    progress: UpgradeProgressCallback | None,
    log_path: Path | None,
) -> tuple[bool, str]:
    """Run a shell command, streaming stdout/stderr to *progress* and a log file.

    Shared subprocess plumbing for `perform_upgrade` and
    `perform_install_extra`. Returns `(success, combined_output)` where
    *combined_output* is the concatenated stdout+stderr, stripped.

    On timeout or `OSError`, the process is killed and a synthetic error
    line is emitted both to the log and via *progress*. The wall-clock cap
    is `_UPGRADE_TIMEOUT`.

    Args:
        cmd: Shell command to execute.
        progress: Optional callback invoked for each output line.
        log_path: Optional path to persist command output. Falls back to a
            fresh `create_update_log_path()` when `None`.

    Returns:
        `(success, output)` — *success* is `True` iff the subprocess exited 0.
    """
    timeout = _UPGRADE_TIMEOUT
    if log_path is None:
        log_path = create_update_log_path()

    output_lines: list[str] = []
    proc: asyncio.subprocess.Process | None = None
    log_file: TextIO | None = None
    try:
        log_path.parent.mkdir(parents=True, exist_ok=True)
        log_file = log_path.open("w", encoding="utf-8")
        log_file.write(f"$ {cmd}\n")
        log_file.flush()
    except OSError:
        logger.warning(
            "Could not create install log at %s; subprocess output will not be "
            "persisted to disk",
            log_path,
            exc_info=True,
        )
        log_file = None

    try:
        proc = await asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            stdin=asyncio.subprocess.DEVNULL,
        )
        await asyncio.wait_for(
            asyncio.gather(
                _read_stream(
                    proc.stdout,  # ty: ignore[invalid-argument-type]
                    lines=output_lines,
                    log_file=log_file,
                    progress=progress,
                ),
                _read_stream(
                    proc.stderr,  # ty: ignore[invalid-argument-type]
                    lines=output_lines,
                    log_file=log_file,
                    progress=progress,
                ),
                proc.wait(),
            ),
            timeout=timeout,
        )
    except TimeoutError:
        if proc is not None:
            proc.kill()
            await proc.wait()
        msg = f"Command timed out after {timeout}s: {cmd}"
        if log_file is not None:
            with suppress(OSError):
                log_file.write(f"{msg}\n")
                log_file.close()
        await _emit_progress(progress, msg)
        logger.warning(msg)
        return False, msg
    except OSError as exc:
        if log_file is not None:
            with suppress(OSError):
                log_file.close()
        logger.warning("Failed to execute command: %s", cmd, exc_info=True)
        return False, f"Failed to execute: {cmd}\n{type(exc).__name__}: {exc}"

    if log_file is not None:
        with suppress(OSError):
            log_file.close()
    output = "\n".join(output_lines).strip()
    if proc.returncode == 0:
        return True, output
    logger.warning(
        "Command exited with code %d: %s\n%s",
        proc.returncode,
        cmd,
        output,
    )
    return False, output


async def perform_upgrade(
    *,
    progress: UpgradeProgressCallback | None = None,
    log_path: Path | None = None,
    include_prereleases: bool | None = None,
) -> tuple[bool, str]:
    """Attempt to upgrade `deepagents-code` using the detected install method.

    Only tries the detected method — does not fall back to other package
    managers to avoid cross-environment contamination.

    Args:
        progress: Optional callback invoked for each output line.
        log_path: Optional path to persist command output.
        include_prereleases: Whether to include alpha/beta/rc releases. When
            `None`, follows the installed version's channel. Pre-release
            upgrades require the uv install method; returns failure otherwise.

    Returns:
        `(success, output)` — *output* is the combined stdout/stderr.
    """
    method = detect_install_method()
    if method == "unknown":
        return False, "Editable install detected — skipping auto-update."
    if method == "other":
        return False, (
            "Unsupported install method detected — cannot auto-update without "
            "knowing which environment provides `dcode`. Reinstall with "
            "`uv tool install -U deepagents-code` or upgrade with the package "
            "manager originally used for this install."
        )
    resolved_include_prereleases = _resolve_include_prereleases(include_prereleases)
    if resolved_include_prereleases:
        supported, reason = prerelease_upgrade_supported(method)
        if not supported:
            return False, reason or _PRERELEASE_UNSUPPORTED_MESSAGE

    cmd = upgrade_command(method, include_prereleases=resolved_include_prereleases)

    # Skip brew if binary not on PATH
    if method == "brew" and not shutil.which("brew"):
        return False, "brew not found on PATH."

    return await _run_install_subprocess(cmd, progress=progress, log_path=log_path)


_EXTRA_NAME_RE = re.compile(r"^[A-Za-z0-9](?:[-_.A-Za-z0-9]*[A-Za-z0-9])?$")
"""Conservative package-extra name pattern used before shell command display."""


_PACKAGE_NAME_RE = re.compile(r"^[A-Za-z0-9](?:[-_.A-Za-z0-9]*[A-Za-z0-9])?$")
"""Conservative package name pattern used before shell command display."""


def is_valid_extra_name(extra: str) -> bool:
    """Return whether `extra` is safe to embed in package-extra syntax.

    Args:
        extra: Candidate extra name from CLI or slash-command input.

    Returns:
        `True` when the value is a conservative PEP 508-style extra name.
    """
    return bool(_EXTRA_NAME_RE.fullmatch(extra))


def is_valid_package_name(package: str) -> bool:
    """Return whether `package` is safe to embed in a `--with` install command.

    Args:
        package: Candidate package name from CLI or slash-command input.

    Returns:
        `True` when the value is a conservative PEP 508-style package name.
    """
    return bool(_PACKAGE_NAME_RE.fullmatch(package))


def _dcode_extras_requirement(extras: Iterable[str]) -> str:
    """Return the validated `deepagents-code[...]` requirement for a uv install.

    Shared by the extra- and package-install commands so already-installed
    extras survive a `uv tool install` reinstall — a bare `deepagents-code`
    request would replace the tool and drop them. Returns plain
    `deepagents-code` when no extras are selected; otherwise the single-quoted
    bracket form, which keeps zsh from globbing the brackets.

    Args:
        extras: Extra names to encode. Each is validated against PEP 508
            grammar before interpolation. This is the authoritative gate for
            caller-supplied extras (`install_extras_command`) and a
            redundant re-check for extras read from distribution metadata
            (`install_package_command`).

    Returns:
        Shell-safe requirement token, e.g. `deepagents-code` or
            `'deepagents-code[baseten,nvidia]'`.

    Raises:
        ValueError: If any extra fails PEP 508 validation.
    """
    names = sorted(set(extras))
    for name in names:
        if not is_valid_extra_name(name):
            msg = (
                f"Invalid extra name {name!r}: must match PEP 508 "
                f"({_EXTRA_NAME_RE.pattern})"
            )
            raise ValueError(msg)
    if not names:
        return "deepagents-code"
    extras_part = ",".join(names)
    return f"'deepagents-code[{extras_part}]'"


def install_package_command(
    package: str,
    *,
    distribution_name: str = "deepagents-code",
) -> str:
    """Return the shell command that adds a package to the dcode tool env.

    The result is built for *execution* (via `perform_install_package`), not for
    display — surfacing raw `uv tool` invocations to the user is intentionally
    avoided. `package` is validated and then `shlex.quote`-d: the validation
    already blocks shell metacharacters, so the quoting is defense in depth that
    keeps the command safe even if the pattern is later loosened.

    Already-installed extras are folded into the `deepagents-code[...]`
    requirement via the shared `_dcode_extras_requirement` helper, the same way
    `install_extras_command` builds its requirement. Without this the reinstall
    would replace the tool with a plain `deepagents-code`, silently dropping any
    extras the user added through `/install <extra>`.

    Args:
        package: Package name to install into the existing tool environment.
        distribution_name: Name of the installed distribution to inspect for
            already-installed extras.

    Returns:
        Shell command string suitable for execution via the shell.

    Raises:
        ExtrasIntrospectionError: If installed extras cannot be determined
            safely from distribution metadata (refused rather than risk
            dropping them).
        ValueError: If `package` or any already-installed extra fails PEP 508
            validation.
    """
    if not _PACKAGE_NAME_RE.fullmatch(package):
        msg = (
            f"Invalid package name {package!r}: must match PEP 508 "
            f"({_PACKAGE_NAME_RE.pattern})"
        )
        raise ValueError(msg)
    from deepagents_code.extras_info import (
        ExtrasIntrospectionError,
        installed_extra_names,
    )

    try:
        extras = installed_extra_names(distribution_name, strict=True)
    except ExtrasIntrospectionError as exc:
        msg = str(exc)
        raise ExtrasIntrospectionError(msg) from exc
    requirement = _dcode_extras_requirement(extras)
    return f"uv tool install -U {requirement} --with {shlex.quote(package)}"


def install_extras_command(extras: Iterable[str]) -> str:
    """Return the uv command that installs the exact set of dcode extras.

    Args:
        extras: Extra names to include in the tool reinstall. Validated by
            `_dcode_extras_requirement`, which raises `ValueError` on any name
            that fails PEP 508 validation.

    Returns:
        Shell command string suitable for display in error messages and
            execution via `perform_install_extra`.
    """
    return f"uv tool install -U {_dcode_extras_requirement(extras)}"


def install_extra_command(
    extra: str,
    *,
    distribution_name: str = "deepagents-code",
) -> str:
    """Return the shell command that adds `extra` to the installed dcode tool.

    The documented install path is `uv tool install` (see
    `scripts/install.sh`), so extras must be preserved across reinstalls.
    Single-quoting the bracket form keeps zsh from globbing it.

    Args:
        extra: The extra name (e.g. `'quickjs'`, `'daytona'`, `'fireworks'`).
            Validated internally against PEP 508 grammar before interpolation
            into the shell command.
        distribution_name: Name of the installed distribution to inspect for
            already-installed extras.

    Returns:
        Shell command string suitable for display in error messages and
            for execution via `perform_install_extra`.

    Raises:
        ExtrasIntrospectionError: If installed extras cannot be determined
            safely from distribution metadata.
        ValueError: If `extra` or any already-installed extra fails PEP 508
            validation.
    """
    from deepagents_code.extras_info import (
        ExtrasIntrospectionError,
        installed_extra_names,
    )

    if not is_valid_extra_name(extra):
        msg = (
            f"Invalid extra name {extra!r}: must match PEP 508 "
            f"({_EXTRA_NAME_RE.pattern})"
        )
        raise ValueError(msg)
    try:
        extras = installed_extra_names(distribution_name, strict=True)
    except ExtrasIntrospectionError as exc:
        msg = str(exc)
        raise ExtrasIntrospectionError(msg) from exc
    extras.add(extra)
    return install_extras_command(extras)


def editable_extra_hint(extra: str) -> str:
    """Return the canonical action hint for editable installs missing an extra.

    Shared by every site that detects an editable install and points the user
    at the correct `uv tool install --editable` invocation, so wording stays
    consistent and the literal `[<extra>]` bracket fragment is centrally
    defined (callers that print through Rich markup must still escape it).
    """
    return (
        "Rerun your `uv tool install --editable` command with "
        f"`--with 'deepagents-code[{extra}]'` added so the extra is "
        "resolved against the editable source."
    )


def editable_package_hint(package: str) -> str:
    """Return the canonical action hint for editable installs needing a package.

    Editable installs can't have packages added automatically, so this points
    the user at adding it to their own development environment. Phrased without
    a raw install command, since surfacing `uv tool` invocations to the user is
    intentionally avoided.
    """
    return (
        f"Add '{package}' to your editable checkout's environment (the one your "
        "editable install of Deep Agents Code runs from), then relaunch."
    )


async def perform_install_extra(
    extra: str,
    *,
    progress: UpgradeProgressCallback | None = None,
    log_path: Path | None = None,
) -> tuple[bool, str]:
    """Add `extra` to the installed dcode tool environment.

    Runs `uv tool install -U 'deepagents-code[<extras>]'`, preserving any
    extras that are already installed. Editable installs are refused — the
    caller should rerun their `uv tool install --editable` command with `--with
    'deepagents-code[<extra>]'` added so the extra is resolved against the
    editable source.

    Args:
        extra: The extra name to install. Must satisfy `is_valid_extra_name`;
            invalid names are rejected without invoking uv (defense in depth
            against shell injection via the `--force`/`--yes` bypass paths).
        progress: Optional callback invoked for each output line.
        log_path: Optional path to persist command output.

    Returns:
        `(success, output)` — *output* is the combined stdout/stderr, or an
            explanatory error message when the install method is unsupported
            or `extra` is malformed.
    """
    if not is_valid_extra_name(extra):
        return False, (
            f"Invalid extra name {extra!r}: must match {_EXTRA_NAME_RE.pattern}"
        )
    method = detect_install_method()
    if method == "unknown":
        return False, (
            "Editable install detected — cannot add extras automatically.\n"
            + editable_extra_hint(extra)
        )
    if method == "brew":
        # Homebrew formula doesn't expose extras; uv tool install is the
        # right escape hatch but would conflict with the brew-managed binary.
        return False, (
            "Homebrew install detected — extras are not supported via brew. "
            "Reinstall with `uv tool install -U 'deepagents-code["
            f"{extra}]'` to switch to a uv-managed tool install with extras."
        )
    if method == "other":
        return False, (
            "Unsupported install method detected — cannot add extras without "
            "knowing which environment provides `dcode`. Reinstall with "
            f"`uv tool install -U 'deepagents-code[{extra}]'` to switch to a "
            "uv-managed tool install with extras."
        )

    if not shutil.which("uv"):
        return False, (
            "`uv` not found on PATH. Reinstall dcode following the docs, or "
            "install uv (https://docs.astral.sh/uv/) so extras can be added."
        )

    from deepagents_code.extras_info import ExtrasIntrospectionError

    try:
        cmd = install_extra_command(extra)
    except (ExtrasIntrospectionError, ValueError) as exc:
        return False, f"{type(exc).__name__}: {exc}"
    return await _run_install_subprocess(cmd, progress=progress, log_path=log_path)


async def perform_install_package(
    package: str,
    *,
    progress: UpgradeProgressCallback | None = None,
    log_path: Path | None = None,
) -> tuple[bool, str]:
    """Add an arbitrary `package` to the installed dcode tool environment.

    Runs `uv tool install -U 'deepagents-code[<extras>]' --with <package>`, the
    escape hatch for a provider whose package is not a `deepagents-code` extra
    (e.g. a custom or in-house `class_path` model). Already-installed extras are
    preserved so the reinstall does not drop them. Editable installs are refused
    — the caller should rerun their `uv tool install --editable` command with
    `--with <package>` added so it resolves against the editable source.

    Args:
        package: The package name to install. Must satisfy
            `is_valid_package_name`; invalid names are rejected without invoking
            uv (defense in depth against shell injection via the
            `--force`/`--yes` bypass paths).
        progress: Optional callback invoked for each output line.
        log_path: Optional path to persist command output.

    Returns:
        `(success, output)` — on success, *output* is the combined
            stdout/stderr from the install. On failure it is an explanatory
            message: when the install method is unsupported, `package` is
            malformed, `uv` is unavailable, or the install subprocess fails or
            times out.
    """
    if not is_valid_package_name(package):
        return False, (
            f"Invalid package name {package!r}: must match {_PACKAGE_NAME_RE.pattern}"
        )
    method = detect_install_method()
    if method == "unknown":
        return False, (
            "Editable install detected — cannot add packages automatically.\n"
            + editable_package_hint(package)
        )
    if method == "brew":
        return False, (
            "Homebrew install detected — packages can't be added to a brew "
            "install. Reinstall Deep Agents Code as a uv-managed tool (see the "
            "installation docs) to enable adding packages."
        )
    if method == "other":
        return False, (
            "Unsupported install method detected — cannot add packages without "
            "knowing which environment provides `dcode`. Reinstall Deep Agents "
            "Code as a uv-managed tool (see the installation docs) to enable "
            "adding packages."
        )

    if not shutil.which("uv"):
        return False, (
            "Package installs require uv, which was not found. Reinstall Deep "
            "Agents Code following the installation docs so packages can be "
            "added."
        )

    from deepagents_code.extras_info import ExtrasIntrospectionError

    try:
        cmd = install_package_command(package)
    except ValueError as exc:
        return False, f"{type(exc).__name__}: {exc}"
    except ExtrasIntrospectionError as exc:
        # Distinct from a malformed package name: the running distribution's own
        # metadata could not be read or parsed. Leave a breadcrumb so the cause
        # is recoverable from logs, even though the user message is unchanged.
        logger.warning(
            "Could not introspect installed extras for package install of %r",
            package,
            exc_info=True,
        )
        return False, f"{type(exc).__name__}: {exc}"
    return await _run_install_subprocess(cmd, progress=progress, log_path=log_path)


# ---------------------------------------------------------------------------
# Config helpers
# ---------------------------------------------------------------------------


def is_update_check_enabled() -> bool:
    """Return whether update checks are enabled.

    Checks `DEEPAGENTS_CODE_NO_UPDATE_CHECK` env var and the `[update].check` key
    in `config.toml`.

    Defaults to enabled.
    """
    from deepagents_code._env_vars import NO_UPDATE_CHECK

    if os.environ.get(NO_UPDATE_CHECK):
        return False
    return _read_update_config().get("check", True)


def is_auto_update_enabled() -> bool:
    """Return whether auto-update is enabled.

    Opt-out via `DEEPAGENTS_CODE_AUTO_UPDATE=0` env var or
    `[update].auto_update = false` in `config.toml`.

    Defaults to `True`.

    Unrecognized env values (neither truthy nor falsy) are ignored with a
    warning and fall through to the config read below.

    If `config.toml` exists but cannot be parsed, returns `False` (fail-closed):
    a corrupt file may hold an explicit opt-out, so it is not treated as the
    permissive default. A genuinely absent config falls through to `True`.

    Always disabled for editable installs.
    """
    from deepagents_code._env_vars import AUTO_UPDATE, classify_env_bool
    from deepagents_code.config import _is_editable_install

    if _is_editable_install():
        return False
    if AUTO_UPDATE in os.environ:
        raw = os.environ[AUTO_UPDATE]
        classified = classify_env_bool(raw)
        if classified is not None:
            return classified
        # Unrecognized boolean token: warn and fall through to the config read
        # below (which itself fails closed on a corrupt config), mirroring
        # `config_manifest._coerce_env`. With the opt-out default an absent or
        # default config leaves auto-update on, so an ignored disable attempt
        # (e.g. a typo like `ture`) must be surfaced rather than swallowed.
        logger.warning("Ignoring %s=%r (expected bool)", AUTO_UPDATE, raw)
    try:
        config = _read_update_config_strict()
    except _ConfigReadError:
        # The config exists but cannot be parsed. Fail *closed* here even though
        # the default is opt-out: a corrupt file may hold an explicit
        # `auto_update = false`, and silently re-enabling auto-update (which
        # upgrades and re-execs the process) against an unreadable opt-out is
        # worse than skipping the upgrade. A genuinely absent config still
        # falls through to the opt-out default below.
        logger.warning(
            "Could not read [update] config; disabling auto-update until it is "
            "readable",
            exc_info=True,
        )
        return False
    return config.get("auto_update", True)


def set_auto_update(enabled: bool) -> None:
    """Persist the auto-update preference to `config.toml`.

    Writes `[update].auto_update` so the setting survives across sessions.

    Args:
        enabled: Whether auto-update should be enabled.
    """
    import contextlib
    import tempfile
    from pathlib import Path

    import tomli_w

    DEFAULT_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
    if DEFAULT_CONFIG_PATH.exists():
        with DEFAULT_CONFIG_PATH.open("rb") as f:
            data = tomllib.load(f)
    else:
        data = {}

    if "update" not in data:
        data["update"] = {}
    data["update"]["auto_update"] = enabled

    fd, tmp_path = tempfile.mkstemp(dir=DEFAULT_CONFIG_PATH.parent, suffix=".tmp")
    try:
        with os.fdopen(fd, "wb") as f:
            tomli_w.dump(data, f)
        Path(tmp_path).replace(DEFAULT_CONFIG_PATH)
    except BaseException:
        with contextlib.suppress(OSError):
            Path(tmp_path).unlink()
        raise


class _ConfigReadError(Exception):
    """Internal: `config.toml` exists but could not be read or parsed.

    Lets callers that care about the difference (e.g. `is_auto_update_enabled`,
    which fails closed) distinguish a corrupt config from a genuinely absent
    one. A missing file is *not* an error and returns an empty config.
    """


def _read_update_config_strict() -> dict[str, bool]:
    """Read `[update]` section from `config.toml`, surfacing read errors.

    Returns:
        A dict of boolean config values; empty when the file is absent.

    Raises:
        _ConfigReadError: When the file exists but cannot be opened or parsed.
    """
    if not DEFAULT_CONFIG_PATH.exists():
        return {}
    try:
        with DEFAULT_CONFIG_PATH.open("rb") as f:
            data = tomllib.load(f)
    except (OSError, tomllib.TOMLDecodeError) as exc:
        raise _ConfigReadError from exc
    section = data.get("update", {})
    return {k: v for k, v in section.items() if isinstance(v, bool)}


def _read_update_config() -> dict[str, bool]:
    """Read `[update]` section from `config.toml`.

    Returns:
        A dict of boolean config values, empty on missing/unreadable file.
    """
    try:
        return _read_update_config_strict()
    except _ConfigReadError:
        logger.warning("Could not read [update] config — using defaults", exc_info=True)
        return {}


def is_auto_update_explicitly_set() -> bool:
    """Return whether the user explicitly chose an auto-update preference.

    `True` when `DEEPAGENTS_CODE_AUTO_UPDATE` holds a recognized boolean or
    `[update].auto_update` is present in `config.toml`. Distinguishes a
    deliberate opt-in/out from the implicit opt-out default.
    """
    from deepagents_code._env_vars import AUTO_UPDATE, classify_env_bool

    if (
        AUTO_UPDATE in os.environ
        and classify_env_bool(os.environ[AUTO_UPDATE]) is not None
    ):
        return True
    return "auto_update" in _read_update_config()


def should_announce_auto_update_default() -> bool:
    """Return whether to show the one-time auto-update default migration notice.

    `True` when no explicit env/config preference is set (so auto-update is on
    only *implicitly*, via the opt-out default) and the notice has not been
    acknowledged yet. This does not itself verify that auto-update is enabled;
    callers must gate on `is_auto_update_enabled` first (e.g. an editable
    install has no explicit preference but never auto-updates).
    """
    if is_auto_update_explicitly_set():
        return False
    return not _read_update_state().get("auto_update_default_acknowledged", False)


def mark_auto_update_default_acknowledged() -> bool:
    """Record that the one-time auto-update default migration notice was shown.

    Returns:
        `True` if the acknowledgement was persisted. `False` means the state
            write failed, so the notice will fire again on the next launch;
            callers should surface that rather than letting the repeat
            look like a bug.
    """
    return _write_update_state({"auto_update_default_acknowledged": True})


# ---------------------------------------------------------------------------
# "What's new" tracking
# ---------------------------------------------------------------------------


def get_seen_version() -> str | None:
    """Return the last version the user saw the "what's new" banner for."""
    value = _read_update_state().get("seen_version")
    return value if isinstance(value, str) else None


def mark_version_seen(version: str) -> None:
    """Record that the user has seen the "what's new" banner for *version*."""
    _write_update_state({"seen_version": version, "seen_at": time.time()})


def should_show_whats_new() -> bool:
    """Return `True` if this is the first launch on a newer version."""
    seen = get_seen_version()
    if seen is None:
        # First run ever — mark current as seen, don't show banner.
        mark_version_seen(__version__)
        return False
    try:
        return _parse_version(__version__) > _parse_version(seen)
    except InvalidVersion:
        logger.debug("Failed to compare versions for what's-new check", exc_info=True)
        return False