mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-21 17:25:26 -04:00
feat(code): dcode doctor diagnostics command (#4148)
Add a `dcode doctor` command that prints install diagnostics (versions, platform, update status, config locations), with `--json` support. --- Adds a `dcode doctor` subcommand to `deepagents-code`, inspired by `claude doctor`. It prints a grouped, tree-style summary of the install (versions, platform, install method, path), update status, and on-disk config locations, with a `--json` mode for machine-readable output. It deliberately runs offline — the update section reads only the local cache via `get_cached_update_available` and never contacts PyPI — and reuses existing helpers (`detect_install_method`, `is_auto_update_enabled`, editable-install detection, config paths) so the output stays consistent with `dcode --version` and `dcode update`. All rendered values go through `rich.markup.escape` with `highlight=False`. The new command is a leaf subcommand (like `update`), so it is intentionally not registered in `_HELP_SPECS`. `ui.show_doctor_help` backs `dcode doctor -h`, and the command is listed in the top-level help usage. Made by [Open SWE](https://openswe.vercel.app/agents/74c251d7-548f-ee57-6e48-c76e1bd66a1a) --------- Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
@@ -0,0 +1,69 @@
|
||||
"""Filesystem-path classification shared by the diagnostic CLI commands.
|
||||
|
||||
`dcode doctor` and `dcode config path` both probe whether config locations
|
||||
exist. `Path.exists()` can report `False` for `OSError` cases such as EACCES
|
||||
when a parent directory denies traversal, so a bare `.exists()` can hide the
|
||||
very permissions problem these commands should diagnose. `classify_path`
|
||||
centralizes the guard and reports an unreadable path as a distinct state instead
|
||||
of conflating it with "missing".
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import logging
|
||||
from enum import StrEnum
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_MISSING_ERRNOS = {errno.ENOENT, errno.ENOTDIR}
|
||||
|
||||
|
||||
class PathState(StrEnum):
|
||||
"""Whether a probed path exists, is absent, or could not be read.
|
||||
|
||||
A `StrEnum` so the value serializes directly to JSON without a custom
|
||||
encoder.
|
||||
"""
|
||||
|
||||
EXISTS = "exists"
|
||||
"""The path is present on disk."""
|
||||
|
||||
MISSING = "missing"
|
||||
"""The path is absent (and its parents are readable)."""
|
||||
|
||||
UNREADABLE = "unreadable"
|
||||
"""Existence could not be determined because `Path.stat()` raised.
|
||||
|
||||
Typically EACCES when a parent directory denies traversal. Kept distinct
|
||||
from `MISSING` so diagnostics can flag it as a genuine problem rather than
|
||||
a not-yet-created path.
|
||||
"""
|
||||
|
||||
|
||||
def classify_path(path: Path) -> PathState:
|
||||
"""Classify a path as existing, missing, or unreadable.
|
||||
|
||||
Args:
|
||||
path: Filesystem path to probe.
|
||||
|
||||
Returns:
|
||||
`PathState.EXISTS` for a present path, `PathState.MISSING` for expected
|
||||
absent-path errors, and `PathState.UNREADABLE` when `Path.stat()`
|
||||
raises another `OSError` (e.g. a parent directory denies traversal).
|
||||
The error is logged at debug level so an unreadable path is never
|
||||
silently indistinguishable from a missing one.
|
||||
"""
|
||||
try:
|
||||
path.stat()
|
||||
except OSError as exc:
|
||||
if exc.errno in _MISSING_ERRNOS:
|
||||
return PathState.MISSING
|
||||
logger.debug("Could not stat %s", path, exc_info=True)
|
||||
return PathState.UNREADABLE
|
||||
else:
|
||||
return PathState.EXISTS
|
||||
@@ -4378,11 +4378,6 @@ class DeepAgentsApp(App):
|
||||
resolved versions of the core LangChain-ecosystem dependencies, which
|
||||
helps diagnose local checkouts.
|
||||
"""
|
||||
from importlib.metadata import (
|
||||
PackageNotFoundError,
|
||||
version as _pkg_version,
|
||||
)
|
||||
|
||||
lines: list[str] = []
|
||||
try:
|
||||
from deepagents_code._version import __version__ as cli_version
|
||||
@@ -4397,17 +4392,15 @@ class DeepAgentsApp(App):
|
||||
logger.warning("Unexpected error looking up app version", exc_info=True)
|
||||
lines.append("deepagents-code version: unknown")
|
||||
|
||||
try:
|
||||
from deepagents_code.extras_info import resolve_sdk_version
|
||||
|
||||
sdk_version, sdk_status = resolve_sdk_version()
|
||||
if sdk_status == "resolved":
|
||||
from deepagents_code.update_check import format_sdk_age_suffix
|
||||
|
||||
sdk_version = _pkg_version("deepagents")
|
||||
sdk_age_suffix = await asyncio.to_thread(format_sdk_age_suffix, sdk_version)
|
||||
lines.append(f"deepagents (SDK) version: {sdk_version}{sdk_age_suffix}")
|
||||
except PackageNotFoundError:
|
||||
logger.debug("deepagents SDK package not found in environment")
|
||||
lines.append("deepagents (SDK) version: unknown")
|
||||
except Exception:
|
||||
logger.warning("Unexpected error looking up SDK version", exc_info=True)
|
||||
else:
|
||||
lines.append("deepagents (SDK) version: unknown")
|
||||
|
||||
editable = False
|
||||
|
||||
@@ -754,17 +754,17 @@ 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.
|
||||
|
||||
This intentionally calls `importlib.metadata.version` directly instead of
|
||||
`resolve_sdk_version`: `config` is on the startup hot path, while
|
||||
`resolve_sdk_version` lives in `extras_info` and imports `packaging`.
|
||||
|
||||
Returns:
|
||||
The installed Deep Agents SDK version, or `None` when package metadata
|
||||
is unavailable.
|
||||
is unavailable.
|
||||
"""
|
||||
try:
|
||||
return version("deepagents")
|
||||
except PackageNotFoundError:
|
||||
logger.debug(
|
||||
"Failed to read deepagents version from package metadata",
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@@ -471,17 +471,15 @@ def _config_paths() -> list[tuple[str, Any, bool]]:
|
||||
("recent models", DEFAULT_STATE_DIR / RECENT_MODELS_FILENAME),
|
||||
]
|
||||
|
||||
from deepagents_code._paths import PathState, classify_path
|
||||
|
||||
rows: list[tuple[str, Any, bool]] = []
|
||||
for label, path in candidates:
|
||||
if path is None:
|
||||
continue
|
||||
try:
|
||||
exists = path.exists()
|
||||
except OSError:
|
||||
# A permission/transient FS error is not the same as "missing"; log
|
||||
# it (at debug, to keep normal output clean) so a developer can tell
|
||||
# the two apart when triaging a `config path` report.
|
||||
logger.debug("Could not stat %s", path, exc_info=True)
|
||||
exists = False
|
||||
# `classify_path` logs unreadable paths at debug level. `config path`
|
||||
# reports a plain exists/missing bool, so unreadable still collapses to
|
||||
# missing here while `doctor` can surface it as a problem.
|
||||
exists = classify_path(path) is PathState.EXISTS
|
||||
rows.append((label, path, exists))
|
||||
return rows
|
||||
|
||||
@@ -0,0 +1,328 @@
|
||||
"""The `dcode doctor` command: report install health and diagnostics.
|
||||
|
||||
Inspired by `claude doctor`, this prints a grouped, tree-style summary of the
|
||||
running install, update status, and configuration locations so the output is
|
||||
safe to paste into a bug report. It stays offline: the update section reads
|
||||
only the local cache and never contacts PyPI.
|
||||
|
||||
Help rendering for `dcode doctor -h` is served by `ui.show_doctor_help`, which
|
||||
does not import this module, so the help path stays light.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import platform
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from deepagents_code.output import write_json
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import argparse
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DiagnosticItem:
|
||||
"""A single labeled diagnostic fact.
|
||||
|
||||
`ok` is `False` only for genuine problems (e.g. a missing dependency), not
|
||||
for informational states such as an available update.
|
||||
"""
|
||||
|
||||
label: str
|
||||
value: str
|
||||
ok: bool = True
|
||||
|
||||
|
||||
@dataclass
|
||||
class DiagnosticSection:
|
||||
"""A named group of related diagnostic items."""
|
||||
|
||||
title: str
|
||||
items: list[DiagnosticItem] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
"""Whether every item in the section is healthy."""
|
||||
return all(item.ok for item in self.items)
|
||||
|
||||
|
||||
def _platform_tag() -> str:
|
||||
"""Return a compact `<os>-<arch>` platform tag (e.g. `darwin-arm64`)."""
|
||||
return f"{platform.system()}-{platform.machine()}".lower()
|
||||
|
||||
|
||||
def _sdk_version() -> tuple[str, bool]:
|
||||
"""Return the installed `deepagents` SDK version and whether it resolved.
|
||||
|
||||
Maps the shared `resolve_sdk_version` outcome onto doctor's display: a
|
||||
genuinely missing package reads as `not installed`, an unexpected lookup
|
||||
failure as `unknown`, and only a resolved version is marked healthy.
|
||||
"""
|
||||
from deepagents_code.extras_info import resolve_sdk_version
|
||||
|
||||
sdk_version, status = resolve_sdk_version()
|
||||
if status == "resolved":
|
||||
# `sdk_version` is a real string when the status is "resolved".
|
||||
return sdk_version or "unknown", True
|
||||
if status == "not_installed":
|
||||
return "not installed", False
|
||||
return "unknown", False
|
||||
|
||||
|
||||
def _commit_hash(path: str) -> str:
|
||||
"""Return the short git commit hash for a source path, if available.
|
||||
|
||||
Args:
|
||||
path: Directory used as the git command working directory.
|
||||
|
||||
Returns:
|
||||
The short commit hash, or `unknown` when git metadata cannot be read.
|
||||
"""
|
||||
import shutil
|
||||
import subprocess # noqa: S404 # fixed-argv git metadata probe
|
||||
from pathlib import Path
|
||||
|
||||
cwd = Path(path).expanduser()
|
||||
git = shutil.which("git")
|
||||
if git is None:
|
||||
return "unknown"
|
||||
try:
|
||||
git_path = Path(git).expanduser().resolve(strict=True)
|
||||
result = subprocess.run( # noqa: S603 # fixed argv with absolute Git path
|
||||
[str(git_path), "rev-parse", "--short", "HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=2,
|
||||
check=False,
|
||||
cwd=cwd,
|
||||
)
|
||||
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
|
||||
logger.debug("Git commit hash detection failed", exc_info=True)
|
||||
return "unknown"
|
||||
if result.returncode != 0:
|
||||
return "unknown"
|
||||
return result.stdout.strip() or "unknown"
|
||||
|
||||
|
||||
def _collect_diagnostics() -> DiagnosticSection:
|
||||
"""Collect core version, platform, and install-location facts.
|
||||
|
||||
Returns:
|
||||
The `Diagnostics` section.
|
||||
"""
|
||||
from deepagents_code._version import __version__
|
||||
from deepagents_code.config import (
|
||||
_get_editable_install_path,
|
||||
_is_editable_install,
|
||||
)
|
||||
from deepagents_code.update_check import detect_install_method
|
||||
|
||||
sdk_version, sdk_ok = _sdk_version()
|
||||
editable = _is_editable_install()
|
||||
if editable:
|
||||
method = "editable"
|
||||
path = _get_editable_install_path() or sys.prefix
|
||||
else:
|
||||
method = detect_install_method()
|
||||
path = sys.prefix
|
||||
|
||||
return DiagnosticSection(
|
||||
title="Diagnostics",
|
||||
items=[
|
||||
DiagnosticItem("deepagents-code", __version__),
|
||||
DiagnosticItem("deepagents (SDK)", sdk_version, ok=sdk_ok),
|
||||
DiagnosticItem("Commit hash", _commit_hash(path)),
|
||||
DiagnosticItem("Python", platform.python_version()),
|
||||
DiagnosticItem("Platform", _platform_tag()),
|
||||
DiagnosticItem("Install method", method),
|
||||
DiagnosticItem("Path", path),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def _collect_updates() -> DiagnosticSection:
|
||||
"""Collect update-channel status from local config and the offline cache.
|
||||
|
||||
Returns:
|
||||
The `Updates` section.
|
||||
"""
|
||||
from deepagents_code.config import _is_editable_install
|
||||
from deepagents_code.update_check import (
|
||||
get_cached_update_available,
|
||||
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)"))
|
||||
else:
|
||||
items.append(
|
||||
DiagnosticItem(
|
||||
"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.append(DiagnosticItem("Latest version", update_status))
|
||||
|
||||
return DiagnosticSection(title="Updates", items=items)
|
||||
|
||||
|
||||
def _path_status(label: str, path: object) -> DiagnosticItem:
|
||||
"""Build an item reporting a path and whether it exists on disk.
|
||||
|
||||
An unreadable path (e.g. a parent directory that denies traversal) is
|
||||
flagged as a genuine problem (`ok=False`) so it surfaces in the section
|
||||
health and exit code, rather than being mistaken for a not-yet-created one.
|
||||
|
||||
Args:
|
||||
label: Human-readable name for the path.
|
||||
path: Filesystem path to probe.
|
||||
|
||||
Returns:
|
||||
A diagnostic item describing the path and its existence.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
from deepagents_code._paths import PathState, classify_path
|
||||
|
||||
resolved = Path(str(path))
|
||||
state = classify_path(resolved)
|
||||
suffix = {
|
||||
PathState.EXISTS: "exists",
|
||||
PathState.MISSING: "not created",
|
||||
PathState.UNREADABLE: "unreadable",
|
||||
}[state]
|
||||
return DiagnosticItem(
|
||||
label, f"{resolved} ({suffix})", ok=state is not PathState.UNREADABLE
|
||||
)
|
||||
|
||||
|
||||
def _collect_configuration() -> DiagnosticSection:
|
||||
"""Collect on-disk configuration and data locations.
|
||||
|
||||
Returns:
|
||||
The `Configuration` section.
|
||||
"""
|
||||
from deepagents_code.model_config import (
|
||||
DEFAULT_CONFIG_DIR,
|
||||
DEFAULT_CONFIG_PATH,
|
||||
)
|
||||
|
||||
return DiagnosticSection(
|
||||
title="Configuration",
|
||||
items=[
|
||||
_path_status("Data directory", DEFAULT_CONFIG_DIR),
|
||||
_path_status("Config file", DEFAULT_CONFIG_PATH),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def collect_sections() -> list[DiagnosticSection]:
|
||||
"""Gather every diagnostic section in display order.
|
||||
|
||||
Returns:
|
||||
The diagnostic sections, in render order.
|
||||
"""
|
||||
return [
|
||||
_collect_diagnostics(),
|
||||
_collect_updates(),
|
||||
_collect_configuration(),
|
||||
]
|
||||
|
||||
|
||||
def _tree_connectors() -> tuple[str, str]:
|
||||
"""Return the `(tee, corner)` tree connectors for the active charset."""
|
||||
from deepagents_code.config import is_ascii_mode
|
||||
|
||||
if is_ascii_mode():
|
||||
return "|-", "`-"
|
||||
return "\u251c", "\u2514" # ├ └
|
||||
|
||||
|
||||
def _render_text(sections: list[DiagnosticSection]) -> None:
|
||||
"""Print the diagnostic sections as a styled tree to the console."""
|
||||
from rich.markup import escape
|
||||
|
||||
from deepagents_code import theme
|
||||
from deepagents_code.config import console, get_glyphs
|
||||
|
||||
glyphs = get_glyphs()
|
||||
tee, corner = _tree_connectors()
|
||||
|
||||
console.print()
|
||||
for section in sections:
|
||||
status_glyph = glyphs.checkmark if section.ok else glyphs.warning
|
||||
status_color = theme.SUCCESS if section.ok else theme.WARNING
|
||||
console.print(
|
||||
f" [bold]{escape(section.title)}[/bold] "
|
||||
f"[{status_color}]{status_glyph}[/{status_color}]"
|
||||
)
|
||||
for index, item in enumerate(section.items):
|
||||
connector = corner if index == len(section.items) - 1 else tee
|
||||
value_color = theme.MUTED if item.ok else "red"
|
||||
console.print(
|
||||
f" {connector} {escape(item.label)}: "
|
||||
f"[{value_color}]{escape(item.value)}[/{value_color}]",
|
||||
highlight=False,
|
||||
)
|
||||
console.print()
|
||||
|
||||
|
||||
def run_doctor_command(args: argparse.Namespace) -> int:
|
||||
"""Run `dcode doctor`, printing diagnostics as text or JSON.
|
||||
|
||||
Args:
|
||||
args: Parsed CLI namespace. Only `output_format` is read.
|
||||
|
||||
Returns:
|
||||
Process exit code: `0` when all sections are healthy, `1` otherwise.
|
||||
"""
|
||||
sections = collect_sections()
|
||||
healthy = all(section.ok for section in sections)
|
||||
output_format = getattr(args, "output_format", "text")
|
||||
|
||||
if output_format == "json":
|
||||
write_json(
|
||||
"doctor",
|
||||
{
|
||||
"healthy": healthy,
|
||||
"sections": [
|
||||
{
|
||||
"title": section.title,
|
||||
"ok": section.ok,
|
||||
"items": [
|
||||
{
|
||||
"label": item.label,
|
||||
"value": item.value,
|
||||
"ok": item.ok,
|
||||
}
|
||||
for item in section.items
|
||||
],
|
||||
}
|
||||
for section in sections
|
||||
],
|
||||
},
|
||||
)
|
||||
else:
|
||||
_render_text(sections)
|
||||
|
||||
return 0 if healthy else 1
|
||||
@@ -16,12 +16,46 @@ from importlib.metadata import (
|
||||
distribution,
|
||||
version as pkg_version,
|
||||
)
|
||||
from typing import Literal
|
||||
|
||||
from packaging.requirements import InvalidRequirement, Requirement
|
||||
from packaging.utils import canonicalize_name
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SdkVersionStatus = Literal["resolved", "not_installed", "error"]
|
||||
"""Outcome of an SDK version lookup.
|
||||
|
||||
`"not_installed"` means the package metadata is genuinely absent;
|
||||
`"error"` means an unexpected failure occurred while reading it. Callers
|
||||
that don't care which kind of failure happened can treat both the same.
|
||||
"""
|
||||
|
||||
|
||||
def resolve_sdk_version() -> tuple[str | None, SdkVersionStatus]:
|
||||
"""Resolve the installed `deepagents` SDK version from package metadata.
|
||||
|
||||
Single source of truth for the lookup that `--version`, `/version`, and
|
||||
`doctor` each used to reimplement. Distinguishes a genuinely missing
|
||||
package from an unexpected metadata error so diagnostic callers can report
|
||||
the two differently, while collapse-friendly callers can ignore the split.
|
||||
|
||||
Returns:
|
||||
`(version, status)`. `version` is the resolved version string when
|
||||
`status` is `"resolved"`, otherwise `None`.
|
||||
"""
|
||||
try:
|
||||
return pkg_version("deepagents"), "resolved"
|
||||
except PackageNotFoundError:
|
||||
logger.debug("deepagents SDK package not found in environment")
|
||||
return None, "not_installed"
|
||||
except Exception: # Best-effort lookup; never propagate to the caller
|
||||
logger.warning(
|
||||
"Unexpected error looking up deepagents SDK version", exc_info=True
|
||||
)
|
||||
return None, "error"
|
||||
|
||||
|
||||
_EXTRA_MARKER_RE = re.compile(r"""extra\s*==\s*["']([^"']+)["']""")
|
||||
|
||||
|
||||
|
||||
@@ -55,19 +55,10 @@ def build_version_text() -> str:
|
||||
Returns:
|
||||
Multi-line version string suitable for stdout.
|
||||
"""
|
||||
from importlib.metadata import (
|
||||
PackageNotFoundError,
|
||||
version as _pkg_version,
|
||||
)
|
||||
from deepagents_code.extras_info import resolve_sdk_version
|
||||
|
||||
try:
|
||||
sdk_version = _pkg_version("deepagents")
|
||||
except PackageNotFoundError:
|
||||
logger.debug("deepagents SDK package not found in environment")
|
||||
sdk_version = "unknown"
|
||||
except Exception: # Best-effort SDK version lookup
|
||||
logger.warning("Unexpected error looking up SDK version", exc_info=True)
|
||||
sdk_version = "unknown"
|
||||
sdk_version_value, status = resolve_sdk_version()
|
||||
sdk_version = sdk_version_value if status == "resolved" else "unknown"
|
||||
|
||||
text = f"deepagents-code {__version__}\ndeepagents (SDK) {sdk_version}"
|
||||
|
||||
@@ -1186,6 +1177,14 @@ def parse_args() -> argparse.Namespace:
|
||||
)
|
||||
add_json_output_arg(update_parser)
|
||||
|
||||
doctor_parser = subparsers.add_parser(
|
||||
"doctor",
|
||||
help="Print install health and diagnostics",
|
||||
add_help=False,
|
||||
parents=help_parent(_lazy_help("show_doctor_help")),
|
||||
)
|
||||
add_json_output_arg(doctor_parser)
|
||||
|
||||
# Default interactive mode — argument order here determines the
|
||||
# usage line printed by argparse; keep in sync with ui.show_help().
|
||||
parser.add_argument(
|
||||
@@ -2244,6 +2243,29 @@ def cli_main() -> None:
|
||||
if _show_bare_command_group_help(args):
|
||||
return
|
||||
|
||||
# Keep self-contained commands that do not need global settings here, before
|
||||
# state migration and settings bootstrap. If a future command only reads
|
||||
# local files or delegates bootstrap to specific subcommands, dispatch it here
|
||||
# so lightweight diagnostic paths stay fast.
|
||||
# Use `getattr` because this fast-path block is for optional top-level
|
||||
# subcommands only. ACP/root-mode invocations may not define `command`,
|
||||
# and should fall through to the later handlers instead of raising here.
|
||||
command = getattr(args, "command", None)
|
||||
if command == "config":
|
||||
from deepagents_code.config_commands import run_config_command
|
||||
|
||||
sys.exit(run_config_command(args))
|
||||
|
||||
if command == "auth" and getattr(args, "auth_command", None) == "path":
|
||||
from deepagents_code.auth_commands import run_auth_command
|
||||
|
||||
sys.exit(run_auth_command(args))
|
||||
|
||||
if command == "doctor":
|
||||
from deepagents_code.doctor import run_doctor_command
|
||||
|
||||
sys.exit(run_doctor_command(args))
|
||||
|
||||
# Best-effort, idempotent migration. Placed after parse_args and the
|
||||
# bare-help fast path so --help / --version / `deepagents <group>`
|
||||
# exit before any I/O. Wrapped broadly so an unexpected non-OSError
|
||||
@@ -2265,6 +2287,11 @@ def cli_main() -> None:
|
||||
# `deepagents <group>` pays the settings bootstrap cost.
|
||||
from deepagents_code.config import console, settings
|
||||
|
||||
if command == "auth":
|
||||
from deepagents_code.auth_commands import run_auth_command
|
||||
|
||||
sys.exit(run_auth_command(args))
|
||||
|
||||
model_params: dict[str, Any] | None = None
|
||||
raw_kwargs = getattr(args, "model_params", None)
|
||||
if raw_kwargs:
|
||||
@@ -2351,16 +2378,6 @@ def cli_main() -> None:
|
||||
)
|
||||
sys.exit(exit_code)
|
||||
|
||||
if args.command == "config":
|
||||
from deepagents_code.config_commands import run_config_command
|
||||
|
||||
sys.exit(run_config_command(args))
|
||||
|
||||
if args.command == "auth":
|
||||
from deepagents_code.auth_commands import run_auth_command
|
||||
|
||||
sys.exit(run_auth_command(args))
|
||||
|
||||
# Apply shell-allow-list from command line if provided (overrides env var)
|
||||
if args.shell_allow_list:
|
||||
from deepagents_code.config import parse_shell_allow_list
|
||||
|
||||
@@ -113,6 +113,9 @@ def show_help() -> None:
|
||||
console.print(
|
||||
" dcode update Check for and install updates"
|
||||
)
|
||||
console.print(
|
||||
" dcode doctor Print install diagnostics"
|
||||
)
|
||||
console.print()
|
||||
|
||||
console.print("[bold]Options:[/bold]", style=theme.PRIMARY)
|
||||
@@ -452,6 +455,30 @@ def show_update_help() -> None:
|
||||
console.print()
|
||||
|
||||
|
||||
def show_doctor_help() -> None:
|
||||
"""Show help information for the `doctor` subcommand."""
|
||||
console.print()
|
||||
console.print("[bold]Usage:[/bold]", style=theme.PRIMARY)
|
||||
console.print(" dcode doctor [options]", markup=False)
|
||||
console.print()
|
||||
console.print(
|
||||
"Print install health and diagnostics (versions, platform, install",
|
||||
)
|
||||
console.print(
|
||||
"method, update status, and config locations). Runs offline and is",
|
||||
)
|
||||
console.print(
|
||||
"safe to paste into a bug report.",
|
||||
)
|
||||
console.print()
|
||||
_print_option_section()
|
||||
console.print()
|
||||
console.print("[bold]Examples:[/bold]", style=theme.PRIMARY)
|
||||
console.print(" dcode doctor")
|
||||
console.print(" dcode doctor --json")
|
||||
console.print()
|
||||
|
||||
|
||||
def _print_mcp_discovery_paths() -> None:
|
||||
"""Print the auto-discovered MCP config paths in precedence order."""
|
||||
from deepagents_code.mcp_tools import MCP_CONFIG_DISCOVERY_PATHS
|
||||
|
||||
@@ -1030,7 +1030,7 @@ def test_resolve_theme_unknown_saved_warns(monkeypatch, caplog) -> None:
|
||||
|
||||
|
||||
def test_config_paths_logs_and_reports_missing_on_oserror(monkeypatch, caplog) -> None:
|
||||
"""An `OSError` from `path.exists()` is logged and reported as missing."""
|
||||
"""An `OSError` from `path.stat()` is logged and reported as missing."""
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
@@ -1038,16 +1038,18 @@ def test_config_paths_logs_and_reports_missing_on_oserror(monkeypatch, caplog) -
|
||||
from deepagents_code.config_commands import _config_paths
|
||||
|
||||
target = model_config.DEFAULT_CONFIG_PATH
|
||||
real_exists = Path.exists
|
||||
real_stat = Path.stat
|
||||
|
||||
def fake_exists(self, *args: object, **kwargs: object) -> bool:
|
||||
def fake_stat(self, *, follow_symlinks: bool = True) -> object:
|
||||
if self == target:
|
||||
msg = "boom"
|
||||
raise OSError(msg)
|
||||
return real_exists(self, *args, **kwargs)
|
||||
return real_stat(self, follow_symlinks=follow_symlinks)
|
||||
|
||||
monkeypatch.setattr(Path, "exists", fake_exists)
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code.config_commands"):
|
||||
monkeypatch.setattr(Path, "stat", fake_stat)
|
||||
# The OSError guard and its debug log now live in the shared `_paths`
|
||||
# classifier that `_config_paths` delegates to.
|
||||
with caplog.at_level(logging.DEBUG, logger="deepagents_code._paths"):
|
||||
rows = _config_paths()
|
||||
config_row = next(row for row in rows if row[0] == "config.toml")
|
||||
assert config_row[2] is False
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
"""Unit tests for the `dcode doctor` command."""
|
||||
|
||||
import argparse
|
||||
import io
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
from deepagents_code.doctor import (
|
||||
DiagnosticItem,
|
||||
DiagnosticSection,
|
||||
_commit_hash,
|
||||
collect_sections,
|
||||
run_doctor_command,
|
||||
)
|
||||
from deepagents_code.main import parse_args
|
||||
|
||||
|
||||
class TestDoctorArgs:
|
||||
"""Tests for `doctor` argument parsing."""
|
||||
|
||||
def test_command_parsed(self) -> None:
|
||||
"""`dcode doctor` selects the doctor command."""
|
||||
with patch.object(sys, "argv", ["deepagents", "doctor"]):
|
||||
args = parse_args()
|
||||
assert args.command == "doctor"
|
||||
|
||||
def test_json_flag(self) -> None:
|
||||
"""`dcode doctor --json` selects JSON output."""
|
||||
with patch.object(sys, "argv", ["deepagents", "doctor", "--json"]):
|
||||
args = parse_args()
|
||||
assert args.command == "doctor"
|
||||
assert args.output_format == "json"
|
||||
|
||||
|
||||
class TestDiagnosticSection:
|
||||
"""Tests for the section dataclass health aggregation."""
|
||||
|
||||
def test_ok_when_all_items_ok(self) -> None:
|
||||
"""A section is healthy when every item is healthy."""
|
||||
section = DiagnosticSection(
|
||||
title="X",
|
||||
items=[DiagnosticItem("a", "1"), DiagnosticItem("b", "2")],
|
||||
)
|
||||
assert section.ok is True
|
||||
|
||||
def test_not_ok_when_any_item_fails(self) -> None:
|
||||
"""A single failing item makes the section unhealthy."""
|
||||
section = DiagnosticSection(
|
||||
title="X",
|
||||
items=[DiagnosticItem("a", "1"), DiagnosticItem("b", "2", ok=False)],
|
||||
)
|
||||
assert section.ok is False
|
||||
|
||||
|
||||
class TestCollectSections:
|
||||
"""Tests for the diagnostic data collection."""
|
||||
|
||||
def test_section_titles(self) -> None:
|
||||
"""All three sections are collected in display order."""
|
||||
sections = collect_sections()
|
||||
assert [s.title for s in sections] == [
|
||||
"Diagnostics",
|
||||
"Updates",
|
||||
"Configuration",
|
||||
]
|
||||
|
||||
def test_diagnostics_reports_version(self) -> None:
|
||||
"""The Diagnostics section reports the running CLI version."""
|
||||
from deepagents_code._version import __version__
|
||||
|
||||
diagnostics = collect_sections()[0]
|
||||
labels = {item.label: item.value for item in diagnostics.items}
|
||||
assert labels["deepagents-code"] == __version__
|
||||
assert "Commit hash" in labels
|
||||
assert labels["Commit hash"]
|
||||
assert "Platform" in labels
|
||||
assert "Install method" in labels
|
||||
|
||||
|
||||
class TestCommitHash:
|
||||
"""Tests for git commit hash detection."""
|
||||
|
||||
def test_uses_absolute_git_path(self, tmp_path) -> None:
|
||||
"""Git metadata probing must not rely on subprocess PATH lookup."""
|
||||
git = tmp_path / "git"
|
||||
git.write_text("", encoding="utf-8")
|
||||
git.chmod(0o755)
|
||||
|
||||
with (
|
||||
patch("shutil.which", return_value=str(git)),
|
||||
patch(
|
||||
"subprocess.run",
|
||||
return_value=SimpleNamespace(returncode=0, stdout="abc123\n"),
|
||||
) as run,
|
||||
):
|
||||
assert _commit_hash(str(tmp_path)) == "abc123"
|
||||
|
||||
argv = run.call_args.args[0]
|
||||
assert Path(argv[0]).is_absolute()
|
||||
assert argv[1:] == ["rev-parse", "--short", "HEAD"]
|
||||
|
||||
def test_missing_git_returns_unknown(self) -> None:
|
||||
"""Missing Git should degrade to `unknown` without spawning a process."""
|
||||
with (
|
||||
patch("shutil.which", return_value=None),
|
||||
patch("subprocess.run") as run,
|
||||
):
|
||||
assert _commit_hash("/tmp") == "unknown"
|
||||
|
||||
run.assert_not_called()
|
||||
|
||||
|
||||
class TestRunDoctorCommand:
|
||||
"""Tests for the text and JSON rendering paths."""
|
||||
|
||||
def _run_text(self) -> tuple[int, str]:
|
||||
buf = io.StringIO()
|
||||
test_console = Console(file=buf, highlight=False, width=200)
|
||||
args = argparse.Namespace(output_format="text")
|
||||
with patch("deepagents_code.config.console", test_console):
|
||||
code = run_doctor_command(args)
|
||||
return code, buf.getvalue()
|
||||
|
||||
def test_text_output_contains_sections(self) -> None:
|
||||
"""Text output renders each section title and key facts."""
|
||||
code, output = self._run_text()
|
||||
assert code == 0
|
||||
assert "Diagnostics" in output
|
||||
assert "Updates" in output
|
||||
assert "Configuration" in output
|
||||
assert "deepagents-code" in output
|
||||
|
||||
def test_json_output_envelope(self, capsys) -> None:
|
||||
"""JSON output is a stable envelope with section data."""
|
||||
args = argparse.Namespace(output_format="json")
|
||||
code = run_doctor_command(args)
|
||||
assert code == 0
|
||||
|
||||
captured = capsys.readouterr()
|
||||
envelope = json.loads(captured.out)
|
||||
assert envelope["command"] == "doctor"
|
||||
assert envelope["schema_version"] == 1
|
||||
data = envelope["data"]
|
||||
assert data["healthy"] is True
|
||||
titles = [section["title"] for section in data["sections"]]
|
||||
assert titles == ["Diagnostics", "Updates", "Configuration"]
|
||||
|
||||
def test_unhealthy_returns_nonzero(self) -> None:
|
||||
"""An unhealthy section yields a non-zero exit code."""
|
||||
unhealthy = [
|
||||
DiagnosticSection(
|
||||
title="Diagnostics",
|
||||
items=[DiagnosticItem("deepagents (SDK)", "not installed", ok=False)],
|
||||
)
|
||||
]
|
||||
args = argparse.Namespace(output_format="text")
|
||||
buf = io.StringIO()
|
||||
with (
|
||||
patch("deepagents_code.doctor.collect_sections", return_value=unhealthy),
|
||||
patch(
|
||||
"deepagents_code.config.console",
|
||||
Console(file=buf, highlight=False, width=200),
|
||||
),
|
||||
):
|
||||
code = run_doctor_command(args)
|
||||
assert code == 1
|
||||
|
||||
|
||||
class TestPathStatus:
|
||||
"""Tests for the path-existence diagnostic item."""
|
||||
|
||||
def test_existing_path_is_healthy(self, tmp_path) -> None:
|
||||
"""An existing path reports `exists` and stays healthy."""
|
||||
from deepagents_code.doctor import _path_status
|
||||
|
||||
item = _path_status("Data directory", tmp_path)
|
||||
assert item.ok is True
|
||||
assert "exists" in item.value
|
||||
|
||||
def test_missing_path_is_healthy(self, tmp_path) -> None:
|
||||
"""A not-yet-created path is informational, not a failure."""
|
||||
from deepagents_code.doctor import _path_status
|
||||
|
||||
item = _path_status("Data directory", tmp_path / "absent")
|
||||
assert item.ok is True
|
||||
assert "not created" in item.value
|
||||
|
||||
def test_unreadable_path_is_unhealthy(self, monkeypatch) -> None:
|
||||
"""An unreadable path is flagged as a genuine problem (`ok=False`)."""
|
||||
from pathlib import Path
|
||||
|
||||
from deepagents_code.doctor import _path_status
|
||||
|
||||
def _raise(self: Path) -> object: # noqa: ARG001 # must match Path.stat signature
|
||||
msg = "permission denied"
|
||||
raise PermissionError(msg)
|
||||
|
||||
monkeypatch.setattr(Path, "stat", _raise)
|
||||
item = _path_status("Config file", "/some/protected/path")
|
||||
assert item.ok is False
|
||||
assert "unreadable" in item.value
|
||||
|
||||
|
||||
class TestDoctorHelp:
|
||||
"""Tests for the doctor help screen."""
|
||||
|
||||
def test_help_renders(self) -> None:
|
||||
"""`show_doctor_help` prints usage and examples."""
|
||||
from deepagents_code.ui import show_doctor_help
|
||||
|
||||
buf = io.StringIO()
|
||||
test_console = Console(file=buf, highlight=False, width=200)
|
||||
with patch("deepagents_code.ui.console", test_console):
|
||||
show_doctor_help()
|
||||
output = buf.getvalue()
|
||||
assert "dcode doctor [options]" in output
|
||||
assert "Usage:" in output
|
||||
@@ -19,6 +19,7 @@ from deepagents_code.extras_info import (
|
||||
format_known_extras,
|
||||
get_extras_status,
|
||||
get_optional_dependency_status,
|
||||
resolve_sdk_version,
|
||||
verify_interpreter_deps,
|
||||
)
|
||||
|
||||
@@ -355,3 +356,34 @@ def test_format_extras_status_renders_markdown_table() -> None:
|
||||
assert lines[3] == "| --- | --- | --- |"
|
||||
assert lines[4] == "| anthropic | langchain-anthropic | 1.4.0 |"
|
||||
assert lines[5] == "| daytona | langchain-daytona | 0.0.4 |"
|
||||
|
||||
|
||||
class TestResolveSdkVersion:
|
||||
"""Tests for the shared `deepagents` SDK version resolver."""
|
||||
|
||||
def test_resolved_returns_version(self) -> None:
|
||||
"""A successful lookup reports the version and `resolved` status."""
|
||||
with patch(
|
||||
"deepagents_code.extras_info.pkg_version", return_value="1.2.3"
|
||||
) as mock:
|
||||
version, status = resolve_sdk_version()
|
||||
mock.assert_called_once_with("deepagents")
|
||||
assert (version, status) == ("1.2.3", "resolved")
|
||||
|
||||
def test_not_installed_distinguished_from_error(self) -> None:
|
||||
"""A missing package reports `not_installed`, never `error`."""
|
||||
with patch(
|
||||
"deepagents_code.extras_info.pkg_version",
|
||||
side_effect=PackageNotFoundError("deepagents"),
|
||||
):
|
||||
version, status = resolve_sdk_version()
|
||||
assert (version, status) == (None, "not_installed")
|
||||
|
||||
def test_unexpected_error_reports_error_status(self) -> None:
|
||||
"""Any non-`PackageNotFoundError` failure reports `error`, not a crash."""
|
||||
with patch(
|
||||
"deepagents_code.extras_info.pkg_version",
|
||||
side_effect=RuntimeError("corrupt metadata"),
|
||||
):
|
||||
version, status = resolve_sdk_version()
|
||||
assert (version, status) == (None, "error")
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Unit tests for `deepagents_code._paths`."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from deepagents_code._paths import PathState, classify_path
|
||||
|
||||
|
||||
class TestClassifyPath:
|
||||
"""Tests for the shared path classifier."""
|
||||
|
||||
def test_existing_path(self, tmp_path: Path) -> None:
|
||||
"""A path that exists classifies as EXISTS."""
|
||||
target = tmp_path / "present"
|
||||
target.write_text("x")
|
||||
assert classify_path(target) is PathState.EXISTS
|
||||
|
||||
def test_existing_directory(self, tmp_path: Path) -> None:
|
||||
"""A directory that exists classifies as EXISTS."""
|
||||
assert classify_path(tmp_path) is PathState.EXISTS
|
||||
|
||||
def test_missing_path(self, tmp_path: Path) -> None:
|
||||
"""A path that does not exist classifies as MISSING."""
|
||||
assert classify_path(tmp_path / "absent") is PathState.MISSING
|
||||
|
||||
def test_unreadable_path(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""An OSError from `Path.stat()` classifies as UNREADABLE.
|
||||
|
||||
Simulates EACCES on a parent directory rather than relying on chmod,
|
||||
which is ignored when running as root and varies by platform.
|
||||
"""
|
||||
|
||||
def _raise(self: Path) -> object: # noqa: ARG001 # must match Path.stat signature
|
||||
msg = "permission denied"
|
||||
raise PermissionError(msg)
|
||||
|
||||
monkeypatch.setattr(Path, "stat", _raise)
|
||||
assert classify_path(Path("/anything")) is PathState.UNREADABLE
|
||||
|
||||
def test_state_value_is_json_friendly(self) -> None:
|
||||
"""`PathState` is a str enum, so its value serializes directly."""
|
||||
assert PathState.UNREADABLE == "unreadable"
|
||||
assert PathState.EXISTS.value == "exists"
|
||||
@@ -46,24 +46,25 @@ def _run_cli_main(argv: list[str]) -> subprocess.CompletedProcess[str]:
|
||||
from deepagents_code.main import cli_main
|
||||
|
||||
argv = ["deepagents", *json.loads(sys.argv[1])]
|
||||
with (
|
||||
patch.object(sys, "argv", argv),
|
||||
patch("deepagents_code.main.check_cli_dependencies"),
|
||||
):
|
||||
cli_main()
|
||||
|
||||
prefixes = tuple(json.loads(sys.argv[2]))
|
||||
loaded = sorted(
|
||||
name for name in sys.modules if name.startswith(prefixes)
|
||||
)
|
||||
config_module = sys.modules.get("deepagents_code.config")
|
||||
bootstrap_done = (
|
||||
getattr(config_module, "_bootstrap_done", None)
|
||||
if config_module is not None
|
||||
else None
|
||||
)
|
||||
print("LOADED_MODULES=" + json.dumps(loaded), file=sys.stderr)
|
||||
print("BOOTSTRAP_DONE=" + json.dumps(bootstrap_done), file=sys.stderr)
|
||||
try:
|
||||
with (
|
||||
patch.object(sys, "argv", argv),
|
||||
patch("deepagents_code.main.check_cli_dependencies"),
|
||||
):
|
||||
cli_main()
|
||||
finally:
|
||||
prefixes = tuple(json.loads(sys.argv[2]))
|
||||
loaded = sorted(
|
||||
name for name in sys.modules if name.startswith(prefixes)
|
||||
)
|
||||
config_module = sys.modules.get("deepagents_code.config")
|
||||
bootstrap_done = (
|
||||
getattr(config_module, "_bootstrap_done", None)
|
||||
if config_module is not None
|
||||
else None
|
||||
)
|
||||
print("LOADED_MODULES=" + json.dumps(loaded), file=sys.stderr)
|
||||
print("BOOTSTRAP_DONE=" + json.dumps(bootstrap_done), file=sys.stderr)
|
||||
"""
|
||||
return subprocess.run(
|
||||
[
|
||||
@@ -121,6 +122,38 @@ def test_help_only_commands_skip_runtime_imports(
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("argv", "expected"),
|
||||
[
|
||||
(["auth", "path"], "/auth.json"),
|
||||
(["config", "path", "--json"], '"command": "config path"'),
|
||||
(["doctor", "--json"], '"command": "doctor"'),
|
||||
],
|
||||
)
|
||||
def test_lightweight_commands_skip_settings_bootstrap(
|
||||
argv: list[str], expected: str
|
||||
) -> None:
|
||||
"""Lightweight diagnostics commands should avoid settings bootstrap."""
|
||||
result = _run_cli_main(argv)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert expected in result.stdout
|
||||
|
||||
bootstrap_done = _read_marker(result.stderr, "BOOTSTRAP_DONE=")
|
||||
assert bootstrap_done in (None, False), (
|
||||
f"settings bootstrap must not run for {argv}; got {bootstrap_done!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_auth_credential_resolution_commands_run_settings_bootstrap() -> None:
|
||||
"""Auth commands that resolve credentials must see dotenv-loaded values."""
|
||||
result = _run_cli_main(["auth", "status", "anthropic"])
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
bootstrap_done = _read_marker(result.stderr, "BOOTSTRAP_DONE=")
|
||||
assert bootstrap_done is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"argv",
|
||||
[
|
||||
|
||||
@@ -138,7 +138,9 @@ async def test_version_slash_command_sdk_unavailable() -> None:
|
||||
app = DeepAgentsApp()
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
with patch("importlib.metadata.version", side_effect=patched_version):
|
||||
with patch(
|
||||
"deepagents_code.extras_info.pkg_version", side_effect=patched_version
|
||||
):
|
||||
await app._handle_command("/version")
|
||||
await pilot.pause()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user