fix(code): clarify managed rg install failures (#4578)

`dcode` now gives clearer messages when managed ripgrep is unavailable
for the current system or missing from the pinned upstream artifacts,
and POSIX managed ripgrep installs now use a relative symlink to a
versioned binary for better portability.

---

Managed ripgrep setup used to collapse permanent gaps and transient
download failures into the same generic failure path. This makes those
cases distinct and makes POSIX managed installs more portable.

## Before

- As a user on an unsupported system, I saw a generic managed ripgrep
install failure and could not tell whether retrying would help.
- As a user whose pinned managed ripgrep artifact was missing upstream,
I saw the same kind of failure as a network outage.
- As a user who moved, restored, or bind-mounted `~/.deepagents`, a
managed helper link could bake in the old absolute path or become stale.
- As a user with a broken managed `rg` symlink, dcode could fall back to
another `rg` on `PATH` instead of repairing the managed install.

## After

- As a user on an unsupported system, I get a direct message that
managed ripgrep is not available for my platform and can install ripgrep
manually or opt into the system installer path.
- As a user whose pinned managed ripgrep artifact returns HTTP 404, I
get a message that the artifact is unavailable for the pinned version,
instead of a generic network failure.
- As a user with transient download trouble, I still get the normal
download-failure path and can retry.
- As a user who moves or bind-mounts `~/.deepagents`, the `rg`
entrypoint is a relative symlink to a versioned managed binary, so it is
less fragile.
- As a user with a dangling managed `rg` symlink, dcode treats it as
managed state and repairs it rather than silently substituting a system
binary.
This commit is contained in:
Mason Daugherty
2026-07-08 20:06:53 -04:00
committed by GitHub
parent f6ee70c00a
commit 434c84ae14
8 changed files with 389 additions and 27 deletions
+9
View File
@@ -3427,6 +3427,7 @@ class DeepAgentsApp(App):
from deepagents_code.main import _should_ensure_managed_ripgrep
from deepagents_code.managed_tools import (
ChecksumMismatchError,
ManagedToolUnavailableError,
ensure_ripgrep,
managed_rg_path,
prepend_managed_bin_to_path,
@@ -3463,6 +3464,14 @@ class DeepAgentsApp(App):
timeout=15,
markup=False,
)
except ManagedToolUnavailableError as exc:
logger.info("ripgrep auto-install unavailable: %s", exc.reason)
self.notify(
exc.message,
severity="warning",
timeout=15,
markup=False,
)
except Exception:
logger.warning(
"ripgrep auto-install failed unexpectedly", exc_info=True
@@ -283,6 +283,7 @@ def _run_tools_install(args: argparse.Namespace) -> int:
from deepagents_code.managed_tools import (
RIPGREP_VERSION,
ChecksumMismatchError,
ManagedToolUnavailableError,
ensure_ripgrep,
is_offline,
managed_rg_path,
@@ -307,6 +308,13 @@ def _run_tools_install(args: argparse.Namespace) -> int:
"verification. Refusing to install."
),
)
except ManagedToolUnavailableError as exc:
logger.info("ripgrep install unavailable: %s", exc.reason)
return _emit_install_result(
output_format,
status="error",
message=exc.message,
)
except Exception:
# Backstop for a clean exit instead of a raw traceback.
# `ensure_ripgrep` is defensive internally, but this is the only
@@ -359,8 +367,8 @@ def _run_tools_install(args: argparse.Namespace) -> int:
output_format,
status="error",
message=(
"Could not install ripgrep (unsupported platform or download failure). "
"See logs, or install ripgrep manually."
"Could not install ripgrep (download failure). See logs, or install "
"ripgrep manually."
),
)
+5
View File
@@ -876,6 +876,7 @@ def _auto_install_ripgrep_cli(
"""
from deepagents_code.managed_tools import (
ChecksumMismatchError,
ManagedToolUnavailableError,
ensure_ripgrep,
managed_rg_path,
prepend_managed_bin_to_path,
@@ -893,6 +894,10 @@ def _auto_install_ripgrep_cli(
"archive failed SHA-256 verification. Refusing to install."
)
return missing_tools
except ManagedToolUnavailableError as exc:
logger.info("ripgrep auto-install unavailable: %s", exc.reason)
warn_console.print(f"[yellow]Warning:[/yellow] {exc.message}")
return missing_tools
except Exception:
logger.warning("ripgrep auto-install failed unexpectedly", exc_info=True)
warn_console.print(
+116 -15
View File
@@ -88,6 +88,33 @@ class ChecksumMismatchError(Exception):
"""
UnavailableReason = Literal["unsupported", "artifact_not_found"]
"""Stable reason token for logging and telemetry."""
class ManagedToolUnavailableError(Exception):
"""Raised when no managed helper binary is available for this system.
Distinct from transient network/download failures so callers can tell users
whether retrying can help or whether they need a different install path.
"""
def __init__(
self, *, tool: Literal["ripgrep"], reason: UnavailableReason, message: str
) -> None:
"""Initialize the unavailable-tool error.
Args:
tool: Managed helper name.
reason: Stable reason token for logging and telemetry.
message: User-facing remediation message.
"""
super().__init__(message)
self.tool = tool
self.reason = reason
self.message = message
def _normalized_arch() -> str | None:
"""Return a normalized arch key matching `RIPGREP_ASSETS`.
@@ -99,6 +126,36 @@ def _normalized_arch() -> str | None:
return _ARCH_ALIASES.get(raw)
def _unsupported_ripgrep_error(
platform_name: str, arch: str | None
) -> ManagedToolUnavailableError:
"""Return a clear unsupported-platform error for managed ripgrep."""
target = platform_name if arch is None else f"{platform_name}/{arch}"
return ManagedToolUnavailableError(
tool="ripgrep",
reason="unsupported",
message=(
f"Managed ripgrep is not available for this system ({target}). "
"Install ripgrep manually, or set DEEPAGENTS_CODE_RIPGREP_INSTALLER=system."
),
)
def _artifact_not_found_error(
platform_name: str, arch: str
) -> ManagedToolUnavailableError:
"""Return a clear missing-artifact error for managed ripgrep."""
return ManagedToolUnavailableError(
tool="ripgrep",
reason="artifact_not_found",
message=(
f"Managed ripgrep artifact for {platform_name}/{arch} was not found "
f"in pinned ripgrep {RIPGREP_VERSION}. Install ripgrep manually, or "
"try a newer dcode version."
),
)
def managed_rg_path() -> Path:
"""Return the managed ripgrep binary path (`.exe` on Windows)."""
name = "rg.exe" if sys.platform == "win32" else "rg"
@@ -396,15 +453,17 @@ def _install_ripgrep_sync(asset: str, sha256: str) -> Path:
"""Download, verify, extract, and install ripgrep atomically.
Staging happens *inside* `BIN_DIR` so the final rename is on the same
filesystem and therefore atomic on POSIX. On Windows it is also atomic
when the destination is not in use; a process holding the existing
`rg.exe` open will see `PermissionError`, which the caller surfaces
rather than silently corrupts. `_verify_sha256` propagates
filesystem and therefore atomic on POSIX. Windows keeps replacing the
user-facing `rg.exe` directly because symlink support varies by developer
mode and policy. POSIX installs use a versioned real binary plus a relative
`rg` symlink so moving or bind-mounting `~/.deepagents` does not bake in
the original home directory path. `_verify_sha256` propagates
`ChecksumMismatchError` to abort install before any move.
Returns:
Absolute path to the installed `rg` binary.
Absolute path to the installed `rg` entrypoint.
"""
import os
import tempfile
BIN_DIR.mkdir(parents=True, exist_ok=True)
@@ -418,7 +477,15 @@ def _install_ripgrep_sync(asset: str, sha256: str) -> Path:
if sys.platform != "win32":
extracted.chmod(0o755)
dest = managed_rg_path()
extracted.replace(dest)
if sys.platform == "win32":
extracted.replace(dest)
return dest
real = BIN_DIR / f"rg-{RIPGREP_VERSION}"
extracted.replace(real)
link = tmp / "rg-link"
link.symlink_to(os.path.relpath(real, start=BIN_DIR))
link.replace(dest)
return dest
@@ -436,13 +503,17 @@ async def ensure_ripgrep() -> Path | None:
version always wins, so a stale managed binary is re-fetched
rather than deferring to a system `rg` and the resolved version
stays deterministic.
4. If offline, on an unsupported platform, or no asset matches the
platform/arch, return `None` so callers fall back to the existing
4. If offline, return `None` so callers fall back to the existing
notification + slow path.
5. Otherwise download → SHA-256 verify → extract → install →
5. If no managed asset matches the platform/arch, return a non-managed
`rg` on `PATH` when one exists; otherwise raise
`ManagedToolUnavailableError` so callers can explain that retrying will
not help.
6. Otherwise download → SHA-256 verify → extract → install →
prepend `BIN_DIR` to `PATH` → return the installed path. On a
checksum mismatch, raises `ChecksumMismatchError` so callers can
surface a loud notice; other failures log and return `None`.
surface a loud notice. On a 404, raises `ManagedToolUnavailableError`;
other failures log and return `None`.
A stale managed binary is never proactively deleted. The atomic
replace in `_install_ripgrep_sync` overwrites it on success, and on
@@ -461,11 +532,18 @@ async def ensure_ripgrep() -> Path | None:
import zipfile
managed = managed_rg_path()
managed_exists = managed.exists()
if prefers_system_ripgrep():
managed_exists = managed.exists() or managed.is_symlink()
def non_managed_rg() -> Path | None:
system_rg = shutil.which("rg", path=_path_without_managed_bin())
if system_rg is not None:
return Path(system_rg)
return None
if prefers_system_ripgrep():
system_rg_path = non_managed_rg()
if system_rg_path is not None:
return system_rg_path
logger.debug(
"Skipping managed ripgrep download: %s=%s",
RIPGREP_INSTALLER,
@@ -486,21 +564,33 @@ async def ensure_ripgrep() -> Path | None:
return None
if sys.platform == "android":
logger.debug("Skipping ripgrep install: unsupported platform 'android'")
return None
system_rg_path = non_managed_rg()
if system_rg_path is not None:
return system_rg_path
error = _unsupported_ripgrep_error(sys.platform, None)
raise error
arch = _normalized_arch()
if arch is None:
logger.debug(
"Skipping ripgrep install: unsupported arch %r", platform.machine()
)
return None
system_rg_path = non_managed_rg()
if system_rg_path is not None:
return system_rg_path
error = _unsupported_ripgrep_error(sys.platform, platform.machine())
raise error
asset_entry = RIPGREP_ASSETS.get((sys.platform, arch))
if asset_entry is None:
logger.debug(
"Skipping ripgrep install: no asset for (%s, %s)", sys.platform, arch
)
return None
system_rg_path = non_managed_rg()
if system_rg_path is not None:
return system_rg_path
error = _unsupported_ripgrep_error(sys.platform, arch)
raise error
asset, sha256 = asset_entry
if managed_exists:
@@ -516,6 +606,17 @@ async def ensure_ripgrep() -> Path | None:
# until the verified replacement is ready. A failed download must
# not strand the user with no `rg` at all.
installed = await asyncio.to_thread(_install_ripgrep_sync, asset, sha256)
except urllib.error.HTTPError as exc:
if exc.code == 404: # noqa: PLR2004 # HTTP 404 Not Found
logger.warning(
"Managed ripgrep artifact was not found: %s/%s", sys.platform, arch
)
error = _artifact_not_found_error(sys.platform, arch)
raise error from exc
logger.warning(
"Could not download ripgrep from %s", _RELEASE_URL_PREFIX, exc_info=True
)
return None
except (urllib.error.URLError, TimeoutError):
logger.warning(
"Could not download ripgrep from %s", _RELEASE_URL_PREFIX, exc_info=True
@@ -127,6 +127,22 @@ class TestToolsInstall:
assert code == 1
assert "SHA-256" in output
def test_install_unavailable_returns_specific_message(self, tmp_path: Path) -> None:
args = argparse.Namespace(tools_command="install", output_format="text")
error = managed_tools.ManagedToolUnavailableError(
tool="ripgrep",
reason="artifact_not_found",
message="Managed ripgrep artifact for linux/x86_64 was not found.",
)
with (
patch.object(managed_tools, "ensure_ripgrep", side_effect=error),
patch.object(managed_tools, "managed_rg_path", return_value=tmp_path / "x"),
):
code, output = _run_text(args)
assert code == 1
assert "linux/x86_64" in output
assert "unexpectedly" not in output
def test_install_unexpected_error_returns_nonzero(self, tmp_path: Path) -> None:
"""An unexpected exception degrades to a clean error, not a traceback."""
args = argparse.Namespace(tools_command="install", output_format="text")
+36
View File
@@ -22106,6 +22106,42 @@ class TestEnsureManagedRipgrep:
assert any(note.get("severity") == "error" for note in notices), notices
assert app._ripgrep_install_failed is True
async def test_managed_tool_unavailable_notifies_message(self) -> None:
"""Permanent managed-tool gaps surface their remediation message."""
from deepagents_code.managed_tools import ManagedToolUnavailableError
app = DeepAgentsApp(agent=MagicMock(), thread_id="t")
notices: list[dict[str, Any]] = []
message = (
"Managed ripgrep is not available for this system. "
"Set DEEPAGENTS_CODE_RIPGREP_INSTALLER=system."
)
error = ManagedToolUnavailableError(
tool="ripgrep",
reason="unsupported",
message=message,
)
def _capture(message: str, **kwargs: Any) -> None:
notices.append({"message": message, **kwargs})
with (
patch(
"deepagents_code.main._should_ensure_managed_ripgrep",
return_value=True,
),
patch(
"deepagents_code.managed_tools.ensure_ripgrep",
AsyncMock(side_effect=error),
),
patch.object(app, "notify", _capture),
):
assert await app._ensure_managed_ripgrep() is False
assert any(note["message"] == message for note in notices), notices
assert any(note.get("severity") == "warning" for note in notices), notices
assert app._ripgrep_install_failed is True
async def test_unexpected_failure_notifies_warning(self) -> None:
"""An unexpected install error warns the user, not just the log.
+24
View File
@@ -1682,6 +1682,30 @@ class TestAutoInstallRipgrepCli:
printed = " ".join(str(c.args[0]) for c in console.print.call_args_list)
assert "SHA-256" in printed
def test_managed_tool_unavailable_keeps_ripgrep_and_reports(self) -> None:
"""Permanent managed-tool gaps report remediation and keep fallback active."""
from deepagents_code.managed_tools import ManagedToolUnavailableError
message = (
"Managed ripgrep is not available for this system. "
"Set DEEPAGENTS_CODE_RIPGREP_INSTALLER=system."
)
error = ManagedToolUnavailableError(
tool="ripgrep",
reason="unsupported",
message=message,
)
console = MagicMock()
with patch(
"deepagents_code.managed_tools.ensure_ripgrep",
AsyncMock(side_effect=error),
):
result = _auto_install_ripgrep_cli(console, ["ripgrep"])
assert result == ["ripgrep"]
printed = " ".join(str(c.args[0]) for c in console.print.call_args_list)
assert f"[yellow]Warning:[/yellow] {message}" in printed
def test_unexpected_failure_keeps_ripgrep(self) -> None:
"""An unexpected error degrades gracefully to the missing-tool path."""
console = MagicMock()
+173 -10
View File
@@ -9,6 +9,7 @@ import subprocess
import tarfile
import warnings
import zipfile
from email.message import Message
from pathlib import Path
from unittest import mock
@@ -16,7 +17,10 @@ import pytest
from deepagents_code import managed_tools
from deepagents_code._env_vars import OFFLINE, RIPGREP_INSTALLER
from deepagents_code.managed_tools import ChecksumMismatchError
from deepagents_code.managed_tools import (
ChecksumMismatchError,
ManagedToolUnavailableError,
)
_EXPECTED_PLATFORM_ARCHS = {
("darwin", "arm64"),
@@ -269,20 +273,25 @@ def test_path_without_managed_bin_preserves_empty_entries(
assert managed_tools._path_without_managed_bin() == f"{os.pathsep}{other}"
async def test_ensure_ripgrep_short_circuits_on_android(
async def test_ensure_ripgrep_reports_unsupported_android(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setattr(managed_tools, "managed_rg_path", lambda: tmp_path / "absent")
monkeypatch.delenv(OFFLINE, raising=False)
monkeypatch.setattr(managed_tools.sys, "platform", "android")
with mock.patch("shutil.which", return_value=None):
assert await managed_tools.ensure_ripgrep() is None
with (
mock.patch("shutil.which", return_value=None),
pytest.raises(ManagedToolUnavailableError) as exc_info,
):
await managed_tools.ensure_ripgrep()
assert exc_info.value.reason == "unsupported"
assert "android" in exc_info.value.message
async def test_ensure_ripgrep_short_circuits_on_unsupported_arch(
async def test_ensure_ripgrep_reports_unsupported_arch(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Unsupported arch (e.g. s390x) returns `None` before any download."""
"""Unsupported arch (e.g. s390x) raises before any download."""
monkeypatch.setattr(managed_tools, "managed_rg_path", lambda: tmp_path / "absent")
monkeypatch.delenv(OFFLINE, raising=False)
monkeypatch.setattr(managed_tools, "_normalized_arch", lambda: None)
@@ -292,8 +301,60 @@ async def test_ensure_ripgrep_short_circuits_on_unsupported_arch(
raise AssertionError(msg)
monkeypatch.setattr(managed_tools, "_download_to", _no_download)
with mock.patch("shutil.which", return_value=None):
assert await managed_tools.ensure_ripgrep() is None
with (
mock.patch("shutil.which", return_value=None),
pytest.raises(ManagedToolUnavailableError) as exc_info,
):
await managed_tools.ensure_ripgrep()
assert exc_info.value.reason == "unsupported"
async def test_ensure_ripgrep_uses_system_rg_when_managed_symlink_unsupported(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
if os.name == "nt":
pytest.skip("creating symlinks is not reliably available on Windows")
bin_dir = tmp_path / "managed-bin"
system_bin = tmp_path / "system-bin"
bin_dir.mkdir()
system_bin.mkdir()
managed = bin_dir / "rg"
system_rg = system_bin / "rg"
managed.symlink_to(bin_dir / "missing-rg")
system_rg.write_bytes(b"system-rg")
monkeypatch.setattr(managed_tools, "BIN_DIR", bin_dir)
monkeypatch.setattr(managed_tools, "managed_rg_path", lambda: managed)
monkeypatch.delenv(OFFLINE, raising=False)
monkeypatch.setattr(managed_tools, "_normalized_arch", lambda: None)
monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{system_bin}")
def _which(cmd: str, path: str | None = None) -> str | None:
assert cmd == "rg"
assert path == str(system_bin)
return str(system_rg)
with mock.patch("shutil.which", side_effect=_which):
assert await managed_tools.ensure_ripgrep() == system_rg
async def test_ensure_ripgrep_reports_missing_asset_entry(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A supported arch with no pinned asset is a permanent unavailable state."""
monkeypatch.setattr(managed_tools, "managed_rg_path", lambda: tmp_path / "absent")
monkeypatch.delenv(OFFLINE, raising=False)
monkeypatch.setattr(managed_tools.sys, "platform", "linux")
monkeypatch.setattr(managed_tools, "_normalized_arch", lambda: "x86_64")
monkeypatch.delitem(managed_tools.RIPGREP_ASSETS, ("linux", "x86_64"))
with (
mock.patch("shutil.which", return_value=None),
pytest.raises(ManagedToolUnavailableError) as exc_info,
):
await managed_tools.ensure_ripgrep()
assert exc_info.value.reason == "unsupported"
assert "linux/x86_64" in exc_info.value.message
async def test_ensure_ripgrep_preserves_stale_when_offline(
@@ -337,8 +398,12 @@ async def test_ensure_ripgrep_preserves_stale_on_unsupported_arch(
fake = mock.Mock()
fake.returncode = 0
fake.stdout = "ripgrep 1.0.0 (rev stale)\n"
with mock.patch.object(subprocess, "run", return_value=fake):
assert await managed_tools.ensure_ripgrep() is None
with (
mock.patch.object(subprocess, "run", return_value=fake),
mock.patch("shutil.which", return_value=None),
pytest.raises(ManagedToolUnavailableError),
):
await managed_tools.ensure_ripgrep()
assert managed.exists()
@@ -580,9 +645,48 @@ async def test_ensure_ripgrep_downloads_when_missing(
assert result is not None
assert result == bin_dir / "rg"
assert result.exists()
assert result.is_symlink()
assert result.readlink() == Path(f"rg-{managed_tools.RIPGREP_VERSION}")
versioned = bin_dir / f"rg-{managed_tools.RIPGREP_VERSION}"
assert versioned.read_bytes() == rg_payload
assert os.environ["PATH"].split(os.pathsep)[0] == str(bin_dir)
async def test_ensure_ripgrep_repairs_dangling_managed_symlink(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
bin_dir = tmp_path / "bin"
bin_dir.mkdir()
managed = bin_dir / "rg"
managed.symlink_to("missing-rg")
monkeypatch.setattr(managed_tools, "BIN_DIR", bin_dir)
monkeypatch.setattr(managed_tools, "managed_rg_path", lambda: managed)
monkeypatch.delenv(OFFLINE, raising=False)
monkeypatch.setattr(managed_tools.sys, "platform", "linux")
monkeypatch.setattr(managed_tools, "_normalized_arch", lambda: "x86_64")
rg_payload = b"repaired-binary"
tar_bytes = _make_fake_tarball(rg_payload)
sha = hashlib.sha256(tar_bytes).hexdigest()
monkeypatch.setitem(
managed_tools.RIPGREP_ASSETS,
("linux", "x86_64"),
("ripgrep-test.tar.gz", sha),
)
monkeypatch.setattr(
managed_tools,
"_download_to",
lambda _url, dest: dest.write_bytes(tar_bytes),
)
with mock.patch("shutil.which", return_value="/usr/bin/rg"):
result = await managed_tools.ensure_ripgrep()
assert result == managed
assert managed.read_bytes() == rg_payload
assert managed.is_symlink()
async def test_ensure_ripgrep_redownloads_stale_managed_binary(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
@@ -628,6 +732,36 @@ async def test_ensure_ripgrep_redownloads_stale_managed_binary(
assert result == managed
assert managed.read_bytes() == new_payload
assert managed.is_symlink()
async def test_ensure_ripgrep_reports_missing_artifact_on_404(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
bin_dir = tmp_path / "bin"
monkeypatch.setattr(managed_tools, "BIN_DIR", bin_dir)
monkeypatch.setattr(managed_tools, "managed_rg_path", lambda: bin_dir / "rg")
monkeypatch.delenv(OFFLINE, raising=False)
monkeypatch.setattr(managed_tools.sys, "platform", "linux")
monkeypatch.setattr(managed_tools, "_normalized_arch", lambda: "x86_64")
import urllib.error
err = urllib.error.HTTPError(
"https://example.test/rg.tar.gz", 404, "Not Found", hdrs=Message(), fp=None
)
def _boom(_url: str, _dest: Path) -> None:
raise err
monkeypatch.setattr(managed_tools, "_download_to", _boom)
with (
mock.patch("shutil.which", return_value=None),
pytest.raises(ManagedToolUnavailableError) as exc_info,
):
await managed_tools.ensure_ripgrep()
assert exc_info.value.reason == "artifact_not_found"
assert "linux/x86_64" in exc_info.value.message
async def test_ensure_ripgrep_returns_none_on_download_failure(
@@ -652,6 +786,35 @@ async def test_ensure_ripgrep_returns_none_on_download_failure(
assert await managed_tools.ensure_ripgrep() is None
async def test_ensure_ripgrep_returns_none_on_http_download_failure(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Non-404 HTTP failures are transient download failures, not unavailability."""
bin_dir = tmp_path / "bin"
monkeypatch.setattr(managed_tools, "BIN_DIR", bin_dir)
monkeypatch.setattr(managed_tools, "managed_rg_path", lambda: bin_dir / "rg")
monkeypatch.delenv(OFFLINE, raising=False)
monkeypatch.setattr(managed_tools.sys, "platform", "linux")
monkeypatch.setattr(managed_tools, "_normalized_arch", lambda: "x86_64")
import urllib.error
err = urllib.error.HTTPError(
"https://example.test/rg.tar.gz",
503,
"Service Unavailable",
hdrs=Message(),
fp=None,
)
def _boom(_url: str, _dest: Path) -> None:
raise err
monkeypatch.setattr(managed_tools, "_download_to", _boom)
with mock.patch("shutil.which", return_value=None):
assert await managed_tools.ensure_ripgrep() is None
async def test_ensure_ripgrep_preserves_stale_on_download_failure(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None: