Merge pull request #1560 from eren-karakus0/fix/singularity-preflight-check

fix(terminal): add Singularity/Apptainer preflight availability check

tests/tools/test_singularity_preflight.py

@@ -0,0 +1,77 @@
+"""Tests for Singularity/Apptainer preflight availability check.
+
+Verifies that a clear error is raised when neither apptainer nor
+singularity is installed, instead of a cryptic FileNotFoundError.
+
+See: https://github.com/NousResearch/hermes-agent/issues/1511
+"""
+
+import subprocess
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from tools.environments.singularity import (
+    _find_singularity_executable,
+    _ensure_singularity_available,
+)
+
+
+class TestFindSingularityExecutable:
+    """_find_singularity_executable resolution tests."""
+
+    def test_prefers_apptainer(self):
+        """When both are available, apptainer should be preferred."""
+        def which_both(name):
+            return f"/usr/bin/{name}" if name in ("apptainer", "singularity") else None
+
+        with patch("shutil.which", side_effect=which_both):
+            assert _find_singularity_executable() == "apptainer"
+
+    def test_falls_back_to_singularity(self):
+        """When only singularity is available, use it."""
+        def which_singularity_only(name):
+            return "/usr/bin/singularity" if name == "singularity" else None
+
+        with patch("shutil.which", side_effect=which_singularity_only):
+            assert _find_singularity_executable() == "singularity"
+
+    def test_raises_when_neither_found(self):
+        """Must raise RuntimeError with install instructions."""
+        with patch("shutil.which", return_value=None):
+            with pytest.raises(RuntimeError, match="Neither.*apptainer.*nor.*singularity"):
+                _find_singularity_executable()
+
+
+class TestEnsureSingularityAvailable:
+    """_ensure_singularity_available preflight tests."""
+
+    def test_returns_executable_on_success(self):
+        """Returns the executable name when version check passes."""
+        fake_result = MagicMock(returncode=0, stderr="")
+
+        with patch("shutil.which", side_effect=lambda n: "/usr/bin/apptainer" if n == "apptainer" else None), \
+             patch("subprocess.run", return_value=fake_result):
+            assert _ensure_singularity_available() == "apptainer"
+
+    def test_raises_on_version_failure(self):
+        """Raises RuntimeError when version command fails."""
+        fake_result = MagicMock(returncode=1, stderr="unknown flag")
+
+        with patch("shutil.which", side_effect=lambda n: "/usr/bin/apptainer" if n == "apptainer" else None), \
+             patch("subprocess.run", return_value=fake_result):
+            with pytest.raises(RuntimeError, match="version.*failed"):
+                _ensure_singularity_available()
+
+    def test_raises_on_timeout(self):
+        """Raises RuntimeError when version command times out."""
+        with patch("shutil.which", side_effect=lambda n: "/usr/bin/apptainer" if n == "apptainer" else None), \
+             patch("subprocess.run", side_effect=subprocess.TimeoutExpired("apptainer", 10)):
+            with pytest.raises(RuntimeError, match="timed out"):
+                _ensure_singularity_available()
+
+    def test_raises_when_not_installed(self):
+        """Raises RuntimeError when neither executable exists."""
+        with patch("shutil.which", return_value=None):
+            with pytest.raises(RuntimeError, match="Neither.*apptainer.*nor.*singularity"):
+                _ensure_singularity_available()

tools/environments/singularity.py

@@ -25,6 +25,57 @@ logger = logging.getLogger(__name__)
 _SNAPSHOT_STORE = get_hermes_home() / "singularity_snapshots.json"
 
 
+def _find_singularity_executable() -> str:
+    """Locate the apptainer or singularity CLI binary.
+
+    Returns the executable name (``"apptainer"`` or ``"singularity"``).
+    Raises ``RuntimeError`` with install instructions if neither is found.
+    """
+    if shutil.which("apptainer"):
+        return "apptainer"
+    if shutil.which("singularity"):
+        return "singularity"
+    raise RuntimeError(
+        "Neither 'apptainer' nor 'singularity' was found in PATH. "
+        "Install Apptainer (https://apptainer.org/docs/admin/main/installation.html) "
+        "or Singularity and ensure the CLI is available."
+    )
+
+
+def _ensure_singularity_available() -> str:
+    """Preflight check: resolve the executable and verify it responds.
+
+    Returns the executable name on success.
+    Raises ``RuntimeError`` with an actionable message on failure.
+    """
+    exe = _find_singularity_executable()
+
+    try:
+        result = subprocess.run(
+            [exe, "version"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except FileNotFoundError:
+        raise RuntimeError(
+            f"Singularity backend selected but the resolved executable '{exe}' "
+            "could not be executed. Check your installation."
+        )
+    except subprocess.TimeoutExpired:
+        raise RuntimeError(
+            f"'{exe} version' timed out. The runtime may be misconfigured."
+        )
+
+    if result.returncode != 0:
+        stderr = result.stderr.strip()[:200]
+        raise RuntimeError(
+            f"'{exe} version' failed (exit code {result.returncode}): {stderr}"
+        )
+
+    return exe
+
+
 def _load_snapshots() -> Dict[str, str]:
     if _SNAPSHOT_STORE.exists():
         try:
@@ -169,7 +220,7 @@ class SingularityEnvironment(BaseEnvironment):
         task_id: str = "default",
     ):
         super().__init__(cwd=cwd, timeout=timeout)
-        self.executable = "apptainer" if shutil.which("apptainer") else "singularity"
+        self.executable = _ensure_singularity_available()
         self.image = _get_or_build_sif(image, self.executable)
         self.instance_id = f"hermes_{uuid.uuid4().hex[:12]}"
         self._instance_started = False