Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 89 additions & 29 deletions libs/code/deepagents_code/doctor.py
Original file line number Diff line number Diff line change
Expand Up @@ -202,62 +202,122 @@ def _collect_diagnostics() -> DiagnosticSection:
def _collect_updates() -> DiagnosticSection:
"""Collect update-channel status from local config and the offline cache.

The same four rows always render, so two `doctor` outputs stay
line-comparable and the `--json` item labels are stable; `Update checks` and
`Auto-updates` therefore report configuration as configured, even when this
install never acts on it. The `Latest version` row carries the reason no
cached answer is available, which is what makes it consistent with the
`Last checked` stamp below it.

Returns:
The `Updates` section.
"""
from deepagents_code.config import _is_editable_install
from deepagents_code.update_check import (
get_cached_update_available,
get_last_update_check_time,
is_auto_update_enabled,
is_update_check_enabled,
)

items = [
DiagnosticItem(
"Update checks",
"enabled" if is_update_check_enabled() else "disabled",
),
]
if _is_editable_install():
items.append(DiagnosticItem("Auto-updates", "disabled (editable install)"))
editable = _is_editable_install()
checks_enabled = is_update_check_enabled()
# Read once and share with both rows below so they cannot straddle a
# concurrent cache refresh and disagree.
checked_at = get_last_update_check_time()
if editable:
auto_updates = "disabled (editable install)"
else:
items.append(
DiagnosticItem(
"Auto-updates",
"enabled" if is_auto_update_enabled() else "disabled",
)
)
auto_updates = "enabled" if is_auto_update_enabled() else "disabled"

available, latest = get_cached_update_available()
if latest is None:
update_status = "unknown (no recent check)"
elif available:
update_status = f"v{latest} available"
else:
update_status = "up to date"
items.extend(
(
DiagnosticItem("Latest version", update_status),
DiagnosticItem("Last checked", _format_last_checked()),
)

return DiagnosticSection(
title="Updates",
items=[
DiagnosticItem(
"Update checks",
"enabled" if checks_enabled else "disabled",
),
DiagnosticItem("Auto-updates", auto_updates),
DiagnosticItem(
"Latest version",
_format_latest_version(
available,
latest,
editable=editable,
checks_enabled=checks_enabled,
checked_at=checked_at,
),
),
DiagnosticItem("Last checked", _format_last_checked(checked_at)),
],
)

return DiagnosticSection(title="Updates", items=items)

def _format_latest_version(
available: bool,
latest: str | None,
*,
editable: bool,
checks_enabled: bool,
checked_at: float | None,
) -> str:
"""Describe the cached latest version, or why no answer is available.

A cached answer is reported whenever one exists, since it is a true
statement about the installed version. Otherwise the cause is named:
editable installs and disabled checks never contact PyPI at all, so any
stamp on disk was written by another install sharing the state directory.
Failing that, the stamp separates an expired cache (repeated fetch failures
leave it untouched) from one never written at all, and a current cache that
still yields no answer is reported as incomplete rather than stale: it holds
no entry this install can use, as when only pre-release pins were recorded
or a pre-release install meets a stable-only payload.

Args:
available: Whether the cached answer is newer than the running version.
latest: Cached latest version, or `None` when the cache holds no usable
answer.
editable: Whether this is an editable install.
checks_enabled: Whether update checks are enabled by config and env.
checked_at: Epoch time of the last recorded check, or `None`.

Returns:
The `Latest version` value.
"""
from deepagents_code.update_check import is_update_cache_fresh

if latest is not None:
return f"v{latest} available" if available else "up to date"
if editable:
return "not checked (editable install)"
if not checks_enabled:
return "not checked (checks disabled)"
if checked_at is None:
return "unknown (never checked)"
if not is_update_cache_fresh(checked_at):
return "unknown (cache stale)"
return "unknown (cache incomplete)"


def _format_last_checked() -> str:
def _format_last_checked(checked_at: float | None) -> str:
"""Return a relative description of the last update check, or `never`.

`never` covers both the no-check-recorded case and, defensively, a stamp
that cannot be formatted. `get_last_update_check_time` only returns finite,
in-range epochs, so the formatting path does not raise here.

Args:
checked_at: Epoch time of the last recorded check, or `None`.

Returns:
The `Last checked` value.
"""
from datetime import UTC, datetime

from deepagents_code.sessions import format_relative_timestamp
from deepagents_code.update_check import get_last_update_check_time

checked_at = get_last_update_check_time()
if checked_at is None:
return "never"
iso = datetime.fromtimestamp(checked_at, tz=UTC).isoformat()
Expand Down
20 changes: 20 additions & 0 deletions libs/code/deepagents_code/update_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -349,6 +349,26 @@ def get_last_update_check_time() -> float | None:
return _coerce_checked_at(checked_at)


def is_update_cache_fresh(checked_at: float | None) -> bool:
"""Return whether a recorded check stamp is still within `CACHE_TTL`.

Lets status surfaces tell "the cache expired" apart from "the cache is
current but holds no usable answer for this install" — states that
`get_cached_update_available` collapses into the same `(False, None)`
result. Takes the stamp instead of reading it so a caller that already
called `get_last_update_check_time` does not read `CACHE_FILE` twice and
risk straddling a concurrent refresh.

Args:
checked_at: Epoch time of the last recorded check, as returned by
`get_last_update_check_time`.

Returns:
`True` when `checked_at` is set and younger than `CACHE_TTL`.
"""
return checked_at is not None and time.time() - checked_at < CACHE_TTL


def _canonical_prerelease_pin(raw: object) -> str | None:
"""Return the canonical targeted pre-release pin for `raw`, or `None`.

Expand Down
98 changes: 94 additions & 4 deletions libs/code/tests/unit_tests/test_doctor.py
Original file line number Diff line number Diff line change
Expand Up @@ -472,7 +472,14 @@ def test_bracket_malformed_ipv6_is_unknown(self) -> None:
class TestCollectUpdates:
"""Tests for the Updates diagnostic section."""

def _labels(self, cache_file: Path) -> dict[str, str]:
def _labels(
self,
cache_file: Path,
*,
editable: bool = False,
checks_enabled: bool = True,
cached: tuple[bool, str | None] = (False, "1.0.0"),
) -> dict[str, str]:
"""Collect the Updates labels, reading `checked_at` from `cache_file`.

Patches `CACHE_FILE` rather than `get_last_update_check_time` so the
Expand All @@ -482,24 +489,43 @@ def _labels(self, cache_file: Path) -> dict[str, str]:
from deepagents_code.doctor import _collect_updates

with (
patch("deepagents_code.config._is_editable_install", return_value=False),
patch(
"deepagents_code.config._is_editable_install",
return_value=editable,
),
patch(
"deepagents_code.update_check.is_update_check_enabled",
return_value=True,
return_value=checks_enabled,
),
patch(
"deepagents_code.update_check.is_auto_update_enabled",
return_value=True,
),
patch(
"deepagents_code.update_check.get_cached_update_available",
return_value=(False, "1.0.0"),
return_value=cached,
),
patch("deepagents_code.update_check.CACHE_FILE", cache_file),
):
section = _collect_updates()
return {item.label: item.value for item in section.items}

def _stale_cache(self, tmp_path: Path) -> Path:
"""Write a cache stamped three days ago, well past `CACHE_TTL`."""
cache = tmp_path / "latest_version.json"
cache.write_text(
json.dumps({"checked_at": time.time() - 3 * 86_400}), encoding="utf-8"
)
return cache

def _fresh_cache(self, tmp_path: Path) -> Path:
"""Write a cache stamped five minutes ago, well inside `CACHE_TTL`."""
cache = tmp_path / "latest_version.json"
cache.write_text(
json.dumps({"checked_at": time.time() - 300}), encoding="utf-8"
)
return cache

def test_last_checked_shows_relative_time(self, tmp_path: Path) -> None:
"""A check stamped an hour ago renders as `1h ago` via the real read."""
cache = tmp_path / "latest_version.json"
Expand All @@ -526,6 +552,70 @@ def test_last_checked_never_on_corrupt_stamp(self, tmp_path: Path) -> None:
cache.write_text(json.dumps({"checked_at": float("nan")}), encoding="utf-8")
assert self._labels(cache)["Last checked"] == "never"

def test_latest_version_reports_cached_answer(self, tmp_path: Path) -> None:
"""A cached answer is reported even though it is older than the TTL."""
cache = self._stale_cache(tmp_path)
assert self._labels(cache)["Latest version"] == "up to date"
available = self._labels(cache, cached=(True, "9.9.9"))
assert available["Latest version"] == "v9.9.9 available"

def test_latest_version_blames_editable_install(self, tmp_path: Path) -> None:
"""Editable installs never check, so the row names that as the cause."""
labels = self._labels(
self._stale_cache(tmp_path), editable=True, cached=(False, None)
)
assert labels["Latest version"] == "not checked (editable install)"
assert labels["Auto-updates"] == "disabled (editable install)"
assert labels["Last checked"] == "3d ago"

def test_latest_version_blames_disabled_checks(self, tmp_path: Path) -> None:
"""Disabled checks freeze the cache, so the row names that as the cause."""
labels = self._labels(
self._stale_cache(tmp_path), checks_enabled=False, cached=(False, None)
)
assert labels["Latest version"] == "not checked (checks disabled)"
assert labels["Update checks"] == "disabled"

def test_latest_version_reports_stale_cache(self, tmp_path: Path) -> None:
"""An enabled checker with a rejected cache reads as stale, not unknown."""
labels = self._labels(self._stale_cache(tmp_path), cached=(False, None))
assert labels["Latest version"] == "unknown (cache stale)"
assert labels["Last checked"] == "3d ago"

def test_latest_version_reports_incomplete_cache(self, tmp_path: Path) -> None:
"""A current cache with no usable entry is incomplete, not stale.

Reachable when only pre-release pins were written or a pre-release
install meets a stable-only payload, so the row must not claim the cache
expired.
"""
labels = self._labels(self._fresh_cache(tmp_path), cached=(False, None))
assert labels["Latest version"] == "unknown (cache incomplete)"
assert labels["Last checked"] == "5m ago"

def test_latest_version_reports_never_checked(self, tmp_path: Path) -> None:
"""With no stamp on disk, no check has ever completed."""
labels = self._labels(tmp_path / "latest_version.json", cached=(False, None))
assert labels["Latest version"] == "unknown (never checked)"
assert labels["Last checked"] == "never"

def test_row_set_is_fixed(self, tmp_path: Path) -> None:
"""Every state renders the same labels so outputs stay comparable."""
expected = ["Update checks", "Auto-updates", "Latest version", "Last checked"]
stale = self._stale_cache(tmp_path)
assert list(self._labels(stale)) == expected
assert (
list(self._labels(stale, editable=True, cached=(False, None))) == expected
)
assert (
list(self._labels(stale, checks_enabled=False, cached=(False, None)))
== expected
)
assert (
list(self._labels(tmp_path / "missing.json", cached=(False, None)))
== expected
)


class TestCommitHash:
"""Tests for git commit hash detection."""
Expand Down
32 changes: 32 additions & 0 deletions libs/code/tests/unit_tests/test_update_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
_run_install_subprocess,
_terminate_install_process,
_uv_tool_bin_dir,
_write_release_prerelease_pins,
cleanup_update_logs,
clear_resume_auto_update_deferral,
clear_startup_auto_update_failure,
Expand Down Expand Up @@ -78,6 +79,7 @@
is_installation_stale,
is_installed_version_at_least,
is_update_available,
is_update_cache_fresh,
is_valid_extra_name,
is_valid_package_name,
mark_auto_update_default_acknowledged,
Expand Down Expand Up @@ -357,6 +359,36 @@ def test_invalid_numeric_checked_at_returns_none(
assert get_last_update_check_time() is None


class TestIsUpdateCacheFresh:
"""Unit tests for `is_update_cache_fresh`."""

def test_recent_stamp_is_fresh(self) -> None:
"""A stamp inside the TTL window is fresh."""
assert is_update_cache_fresh(time.time() - 60) is True

def test_expired_stamp_is_not_fresh(self) -> None:
"""A stamp at or past the TTL boundary is not fresh."""
assert is_update_cache_fresh(time.time() - CACHE_TTL) is False
assert is_update_cache_fresh(time.time() - CACHE_TTL - 1) is False

def test_missing_stamp_is_not_fresh(self) -> None:
"""No recorded check cannot be fresh."""
assert is_update_cache_fresh(None) is False

def test_separates_fresh_cache_from_missing_answer(self, cache_file) -> None:
"""A pins-only cache is fresh yet yields no cached version answer.

`_write_release_prerelease_pins` seeds `checked_at` without any version
keys, so freshness and "has an answer" genuinely differ and callers
cannot infer staleness from a `None` answer.
"""
_write_release_prerelease_pins("1.1.0", ["deepagents==0.7.0a2"])

assert get_cached_update_available() == (False, None)
assert is_update_cache_fresh(get_last_update_check_time()) is True
assert cache_file.exists()


class TestGetLatestVersion:
def test_fresh_fetch(self, cache_file) -> None:
"""Successful PyPI fetch writes cache and returns version."""
Expand Down