feat(terminal): graceful degradation for remote backend connection failures

Connection-class infrastructure failures on remote terminal backends (SSH
host unreachable/timed out, Docker daemon down or missing, remote file
sync failing on a dead link) previously surfaced to the model as raised
RuntimeError tracebacks. The model got a stack blob with no guidance and
the failure was indistinguishable from a tool bug.

Now:

- New EnvironmentConnectionError(RuntimeError) in tools/environments/base.py
  carrying a reason + retry_hint. Subclassing RuntimeError keeps every
  existing catcher working.
- ssh.py classifies connect-refused, connect-timeout, scp, remote mkdir,
  bulk upload/download, and remote rm failures as connection errors.
- docker.py classifies all four _ensure_docker_available() failure paths
  (missing exe, non-executable exe, daemon timeout, `docker version`
  failure).
- terminal_tool catches EnvironmentConnectionError and returns a
  structured tool result the model can act on:
    {"status": "degraded", "reason": ..., "retry_hint": ..., "exit_code": -1}
  The failed backend is evicted from the environment cache so a later
  call retries from scratch — recovery is automatic once the backend is
  reachable again.
- Config gate terminal.degraded_mode: warn|fail (default warn) in
  config.yaml, bridged as TERMINAL_DEGRADED_MODE across all four bridge
  sites (cli.py env_mappings, gateway/run.py _terminal_env_map,
  TERMINAL_CONFIG_ENV_MAP, DEFAULT_CONFIG). "fail" preserves the
  historical error+traceback tool result.
- Command failures (nonzero exit, command-not-found) are NOT touched —
  only infrastructure failures classify as degraded.

Tests: tests/tools/test_terminal_degraded_mode.py (15 tests) covering
exception classification for ssh+docker, structured degraded results,
no-caching of degraded envs, recovery after the backend returns,
nonzero-exit results unaffected, fail-mode preservation, invalid-mode
fallback to warn, and the four-site config bridge invariant.

Inspired by: Claude Cowork degraded-backend behavior (idea-level,
docs-only evidence).

cli.py

@@ -658,6 +658,7 @@ def load_cli_config() -> Dict[str, Any]:
     
     env_mappings = {
         "env_type": "TERMINAL_ENV",
+        "degraded_mode": "TERMINAL_DEGRADED_MODE",
         "cwd": "TERMINAL_CWD",
         "timeout": "TERMINAL_TIMEOUT",
         "home_mode": "TERMINAL_HOME_MODE",

gateway/run.py

@@ -2082,6 +2082,7 @@ if _config_path.exists():
             ).strip().lower()
             _terminal_env_map = {
                 "backend": "TERMINAL_ENV",
+                "degraded_mode": "TERMINAL_DEGRADED_MODE",
                 "cwd": "TERMINAL_CWD",
                 "timeout": "TERMINAL_TIMEOUT",
                 "home_mode": "TERMINAL_HOME_MODE",

hermes_cli/config.py

@@ -3183,6 +3183,7 @@ def write_platform_config_field(
 TERMINAL_CONFIG_ENV_MAP = {
     "backend": "TERMINAL_ENV",
     "modal_mode": "TERMINAL_MODAL_MODE",
+    "degraded_mode": "TERMINAL_DEGRADED_MODE",
     "cwd": "TERMINAL_CWD",
     "timeout": "TERMINAL_TIMEOUT",
     "lifetime_seconds": "TERMINAL_LIFETIME_SECONDS",

hermes_cli/config_defaults.py

@@ -279,6 +279,12 @@ DEFAULT_CONFIG = {
     "terminal": {
         "backend": "local",
         "modal_mode": "auto",
+        # Remote-backend graceful degradation: when a connection-class
+        # infrastructure failure occurs (SSH host unreachable, Docker daemon
+        # down), "warn" (default) returns a structured degraded tool result
+        # with a reason + retry hint so the model can act on it; "fail"
+        # preserves the historical error + traceback behavior.
+        "degraded_mode": "warn",
         "cwd": ".",  # Use current directory
         # Terminal font family for the desktop app's embedded xterm.js terminal.
         # When set (e.g. "'CaskaydiaCoveNerdFont', 'JetBrains Mono', monospace"),

tests/tools/test_terminal_degraded_mode.py

@@ -0,0 +1,223 @@
+"""Remote terminal backend graceful degradation (terminal.degraded_mode).
+
+Connection-class infrastructure failures (SSH unreachable, Docker daemon
+down) must come back to the model as a structured ``status: "degraded"``
+tool result with a reason and a retry hint — not as a raised traceback
+blob.  Command failures (nonzero exit codes) are NOT infrastructure
+failures and must stay untouched.  ``terminal.degraded_mode: fail``
+preserves the historical raise/traceback behavior for anyone relying
+on it.
+
+Inspired by: Claude Cowork degraded-backend behavior (idea-level,
+docs-only evidence).
+"""
+
+import json
+import subprocess
+from unittest.mock import MagicMock
+
+import pytest
+
+from tools.environments.base import EnvironmentConnectionError
+
+
+@pytest.fixture
+def isolated_env(tmp_path, monkeypatch):
+    """Isolated HERMES_HOME + a clean environment cache for terminal_tool."""
+    import tools.terminal_tool as tt
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    # The one-shot config bridge would overwrite our TERMINAL_* test vars
+    # from the developer's real config.yaml; mark it as already attempted.
+    monkeypatch.setattr(tt, "_terminal_config_bridge_attempted", True)
+
+    def _clear():
+        with tt._env_lock:
+            tt._active_environments.clear()
+            tt._last_activity.clear()
+
+    _clear()
+    yield tt
+    _clear()
+
+
+def _mock_ssh_unreachable(monkeypatch, stderr="ssh: connect to host unreachable.invalid port 22: Connection refused"):
+    """Make every ssh subprocess in the ssh backend fail like a dead host."""
+    monkeypatch.setattr("tools.environments.ssh.shutil.which", lambda _x: "/usr/bin/ssh")
+    monkeypatch.setattr(
+        "tools.environments.ssh.subprocess.run",
+        lambda *a, **k: subprocess.CompletedProcess([], 255, stdout="", stderr=stderr),
+    )
+
+
+def _ssh_backend_env(monkeypatch):
+    monkeypatch.setenv("TERMINAL_ENV", "ssh")
+    monkeypatch.setenv("TERMINAL_SSH_HOST", "unreachable.invalid")
+    monkeypatch.setenv("TERMINAL_SSH_USER", "nobody")
+    monkeypatch.delenv("TERMINAL_SSH_PORT", raising=False)
+    monkeypatch.delenv("TERMINAL_SSH_KEY", raising=False)
+
+
+class TestExceptionClassification:
+    """Backends raise EnvironmentConnectionError for connection-class failures."""
+
+    def test_ssh_connect_refused_raises_connection_error(self, monkeypatch):
+        from tools.environments.ssh import SSHEnvironment
+
+        _mock_ssh_unreachable(monkeypatch)
+        with pytest.raises(EnvironmentConnectionError):
+            SSHEnvironment(host="unreachable.invalid", user="nobody")
+
+    def test_ssh_connect_timeout_raises_connection_error(self, monkeypatch):
+        from tools.environments.ssh import SSHEnvironment
+
+        monkeypatch.setattr("tools.environments.ssh.shutil.which", lambda _x: "/usr/bin/ssh")
+
+        def _timeout(*a, **k):
+            raise subprocess.TimeoutExpired(cmd="ssh", timeout=15)
+
+        monkeypatch.setattr("tools.environments.ssh.subprocess.run", _timeout)
+        with pytest.raises(EnvironmentConnectionError):
+            SSHEnvironment(host="unreachable.invalid", user="nobody")
+
+    def test_docker_missing_executable_raises_connection_error(self, monkeypatch):
+        from tools.environments import docker as docker_env
+
+        monkeypatch.setattr(docker_env, "find_docker", lambda: None)
+        with pytest.raises(EnvironmentConnectionError):
+            docker_env._ensure_docker_available()
+
+    def test_docker_daemon_timeout_raises_connection_error(self, monkeypatch):
+        from tools.environments import docker as docker_env
+
+        monkeypatch.setattr(docker_env, "find_docker", lambda: "/usr/bin/docker")
+
+        def _timeout(*a, **k):
+            raise subprocess.TimeoutExpired(cmd="docker version", timeout=5)
+
+        monkeypatch.setattr(docker_env.subprocess, "run", _timeout)
+        with pytest.raises(EnvironmentConnectionError):
+            docker_env._ensure_docker_available()
+
+    def test_connection_error_is_a_runtime_error(self):
+        # Existing catchers of RuntimeError must keep working unchanged.
+        assert issubclass(EnvironmentConnectionError, RuntimeError)
+        err = EnvironmentConnectionError("boom")
+        assert err.reason == "boom"
+        assert err.retry_hint  # non-empty default hint
+
+
+class TestDegradedToolResult:
+    """terminal_tool returns structured degraded results in warn mode."""
+
+    def test_ssh_unreachable_returns_degraded_result(self, isolated_env, monkeypatch):
+        _ssh_backend_env(monkeypatch)
+        _mock_ssh_unreachable(monkeypatch)
+        monkeypatch.delenv("TERMINAL_DEGRADED_MODE", raising=False)
+
+        r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-ssh"))
+        assert r["status"] == "degraded"
+        assert r["exit_code"] == -1
+        assert "reason" in r and r["reason"]
+        assert "retry_hint" in r and r["retry_hint"]
+        assert "traceback" not in r
+
+    def test_docker_daemon_down_returns_degraded_result(self, isolated_env, monkeypatch):
+        monkeypatch.setenv("TERMINAL_ENV", "docker")
+        monkeypatch.delenv("TERMINAL_DEGRADED_MODE", raising=False)
+        monkeypatch.setattr(isolated_env, "_maybe_reap_docker_orphans", lambda _cc: None)
+        monkeypatch.setattr("tools.environments.docker.find_docker", lambda: None)
+
+        r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-docker"))
+        assert r["status"] == "degraded"
+        assert "reason" in r and r["reason"]
+        assert "retry_hint" in r and r["retry_hint"]
+        assert "traceback" not in r
+
+    def test_degraded_env_is_not_cached(self, isolated_env, monkeypatch):
+        """A degraded backend must not be cached — a later call must retry."""
+        _ssh_backend_env(monkeypatch)
+        _mock_ssh_unreachable(monkeypatch)
+
+        r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-cache"))
+        assert r["status"] == "degraded"
+        with isolated_env._env_lock:
+            assert not isolated_env._active_environments
+
+    def test_recovery_after_degraded(self, isolated_env, monkeypatch):
+        """When the backend comes back, the next call just works."""
+        import shutil as real_shutil
+
+        real_run = subprocess.run
+        real_which = real_shutil.which
+        _ssh_backend_env(monkeypatch)
+        _mock_ssh_unreachable(monkeypatch)
+        r1 = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-recover"))
+        assert r1["status"] == "degraded"
+
+        # Backend "recovers" — restore the real subprocess machinery (the
+        # ssh-module patch hits the shared subprocess/shutil modules) and
+        # switch to a reachable backend; the tool path must not be poisoned.
+        monkeypatch.setattr(subprocess, "run", real_run)
+        monkeypatch.setattr(real_shutil, "which", real_which)
+        monkeypatch.setenv("TERMINAL_ENV", "local")
+        r2 = json.loads(isolated_env.terminal_tool("echo back", task_id="t-degraded-recover"))
+        assert r2["exit_code"] == 0
+        assert "back" in r2["output"]
+
+
+class TestNonInfrastructureFailuresUntouched:
+    def test_nonzero_exit_is_not_degraded(self, isolated_env, monkeypatch):
+        monkeypatch.setenv("TERMINAL_ENV", "local")
+        r = json.loads(isolated_env.terminal_tool("exit 3", task_id="t-degraded-exit3"))
+        assert r["exit_code"] == 3
+        assert r.get("status") != "degraded"
+
+    def test_command_not_found_is_not_degraded(self, isolated_env, monkeypatch):
+        monkeypatch.setenv("TERMINAL_ENV", "local")
+        r = json.loads(isolated_env.terminal_tool(
+            "definitely_not_a_real_command_zzz_42", task_id="t-degraded-notfound"))
+        assert r["exit_code"] != 0
+        assert r.get("status") != "degraded"
+
+
+class TestFailModePreservesRaiseBehavior:
+    def test_fail_mode_returns_error_with_traceback(self, isolated_env, monkeypatch):
+        _ssh_backend_env(monkeypatch)
+        _mock_ssh_unreachable(monkeypatch)
+        monkeypatch.setenv("TERMINAL_DEGRADED_MODE", "fail")
+
+        r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-fail"))
+        assert r["status"] == "error"
+        assert "traceback" in r
+        assert "SSH connection failed" in r["error"]
+
+    def test_invalid_mode_falls_back_to_warn(self, isolated_env, monkeypatch):
+        _ssh_backend_env(monkeypatch)
+        _mock_ssh_unreachable(monkeypatch)
+        monkeypatch.setenv("TERMINAL_DEGRADED_MODE", "bogus-value")
+
+        r = json.loads(isolated_env.terminal_tool("echo hi", task_id="t-degraded-bogus"))
+        assert r["status"] == "degraded"
+
+
+class TestConfigBridging:
+    def test_degraded_mode_is_bridged_everywhere(self):
+        """terminal.degraded_mode must ride every config->env bridge path,
+        same four-site invariant as the docker_* keys."""
+        from tests.tools.test_terminal_config_env_sync import (
+            _cli_env_map_keys,
+            _gateway_env_map_keys,
+            _save_config_env_sync_keys,
+            _terminal_tool_env_var_names,
+        )
+
+        assert "degraded_mode" in _cli_env_map_keys()
+        assert "degraded_mode" in _gateway_env_map_keys()
+        assert "degraded_mode" in _save_config_env_sync_keys()
+        assert "TERMINAL_DEGRADED_MODE" in _terminal_tool_env_var_names()
+
+    def test_default_config_carries_degraded_mode(self):
+        from hermes_cli.config_defaults import DEFAULT_CONFIG
+
+        assert DEFAULT_CONFIG["terminal"].get("degraded_mode") == "warn"

tools/environments/base.py

@@ -52,6 +52,32 @@ _activity_callback_local = threading.local()
 _UNBOUNDED_CAPTURE_CHARS = 2**63 - 1
 
 
+class EnvironmentConnectionError(RuntimeError):
+    """Infrastructure/connection-class failure of a terminal backend.
+
+    Raised when the backend itself is unreachable (SSH host down, Docker
+    daemon not running, remote file sync failing on a dead link) — never
+    for a command that merely exited nonzero.  Subclassing RuntimeError
+    keeps every existing ``except RuntimeError`` catcher working.
+
+    ``terminal_tool`` turns this into a structured ``status: "degraded"``
+    tool result (config gate ``terminal.degraded_mode: warn|fail``) so the
+    model gets an actionable reason + retry hint instead of a traceback.
+    The failed backend is never cached, so a later call retries from
+    scratch and simply works once the backend is reachable again.
+    """
+
+    def __init__(self, reason: str, *, retry_hint: str = ""):
+        super().__init__(reason)
+        self.reason = reason
+        self.retry_hint = retry_hint or (
+            "This is an infrastructure failure, not a command failure. "
+            "Verify the backend is reachable (network, service running, "
+            "credentials), then retry the same command — recovery is "
+            "automatic once the backend is back."
+        )
+
+
 class _BoundedOutputCollector:
     """Retain a bounded 40/60 head-tail window of streamed text.
 

tools/environments/docker.py

@@ -18,7 +18,11 @@ import uuid
 from pathlib import Path
 from typing import Optional
 
-from tools.environments.base import BaseEnvironment, _popen_bash
+from tools.environments.base import (
+    BaseEnvironment,
+    EnvironmentConnectionError,
+    _popen_bash,
+)
 from tools.environments.local import (
     _HERMES_PROVIDER_ENV_BLOCKLIST,
     _is_hermes_internal_secret,
@@ -778,9 +782,13 @@ def _ensure_docker_available() -> None:
             "or known install locations. Install Docker Desktop and ensure the "
             "CLI is available."
         )
-        raise RuntimeError(
+        raise EnvironmentConnectionError(
             "Docker executable not found in PATH or known install locations. "
-            "Install Docker and ensure the 'docker' command is available."
+            "Install Docker and ensure the 'docker' command is available.",
+            retry_hint=(
+                "Install Docker (or fix PATH) and retry, or switch "
+                "terminal.backend to 'local'."
+            ),
         )
 
     try:
@@ -798,8 +806,9 @@ def _ensure_docker_available() -> None:
             docker_exe,
             exc_info=True,
         )
-        raise RuntimeError(
-            "Docker executable could not be executed. Check your Docker installation."
+        raise EnvironmentConnectionError(
+            "Docker executable could not be executed. Check your Docker installation.",
+            retry_hint="Repair the Docker installation and retry.",
         )
     except subprocess.TimeoutExpired:
         logger.error(
@@ -808,8 +817,12 @@ def _ensure_docker_available() -> None:
             docker_exe,
             exc_info=True,
         )
-        raise RuntimeError(
-            "Docker daemon is not responding. Ensure Docker is running and try again."
+        raise EnvironmentConnectionError(
+            "Docker daemon is not responding. Ensure Docker is running and try again.",
+            retry_hint=(
+                "Start the Docker daemon (e.g. `systemctl start docker` or "
+                "launch Docker Desktop), then retry the same command."
+            ),
         )
     except Exception:
         logger.error(
@@ -826,9 +839,13 @@ def _ensure_docker_available() -> None:
                 result.returncode,
                 result.stderr.strip(),
             )
-            raise RuntimeError(
+            raise EnvironmentConnectionError(
                 "Docker command is available but 'docker version' failed. "
-                "Check your Docker installation."
+                "Check your Docker installation.",
+                retry_hint=(
+                    "The Docker daemon may be down or the current user lacks "
+                    "permission (docker group). Fix and retry."
+                ),
             )
 
 

tools/environments/ssh.py

@@ -9,7 +9,11 @@ import subprocess
 import tempfile
 from pathlib import Path
 
-from tools.environments.base import BaseEnvironment, _popen_bash
+from tools.environments.base import (
+    BaseEnvironment,
+    EnvironmentConnectionError,
+    _popen_bash,
+)
 from tools.environments.file_sync import (
     FileSyncManager,
     iter_sync_files,
@@ -110,9 +114,22 @@ class SSHEnvironment(BaseEnvironment):
             )
             if result.returncode != 0:
                 error_msg = result.stderr.strip() or result.stdout.strip()
-                raise RuntimeError(f"SSH connection failed: {error_msg}")
+                raise EnvironmentConnectionError(
+                    f"SSH connection failed: {error_msg}",
+                    retry_hint=(
+                        f"Verify {self.user}@{self.host}:{self.port} is reachable "
+                        "(host up, sshd running, key/agent auth working), then "
+                        "retry — the connection is re-established automatically."
+                    ),
+                )
         except subprocess.TimeoutExpired:
-            raise RuntimeError(f"SSH connection to {self.user}@{self.host} timed out")
+            raise EnvironmentConnectionError(
+                f"SSH connection to {self.user}@{self.host} timed out",
+                retry_hint=(
+                    f"Check network connectivity to {self.host}:{self.port} "
+                    "and that sshd is accepting connections, then retry."
+                ),
+            )
 
     def _detect_remote_home(self) -> str:
         """Detect the remote user's home directory."""
@@ -183,7 +200,13 @@ class SSHEnvironment(BaseEnvironment):
             stdin=subprocess.DEVNULL,
         )
         if result.returncode != 0:
-            raise RuntimeError(f"scp failed: {result.stderr.strip()}")
+            raise EnvironmentConnectionError(
+                f"scp failed: {result.stderr.strip()}",
+                retry_hint=(
+                    f"File sync to {self.user}@{self.host} failed — verify the "
+                    "SSH connection is healthy, then retry."
+                ),
+            )
 
     def _ssh_bulk_upload(self, files: list[tuple[str, str]]) -> None:
         """Upload many files in a single tar-over-SSH stream.
@@ -212,7 +235,13 @@ class SSHEnvironment(BaseEnvironment):
                 stdin=subprocess.DEVNULL,
             )
             if result.returncode != 0:
-                raise RuntimeError(f"remote mkdir failed: {result.stderr.strip()}")
+                raise EnvironmentConnectionError(
+                    f"remote mkdir failed: {result.stderr.strip()}",
+                    retry_hint=(
+                        f"Remote directory setup on {self.host} failed — verify "
+                        "the SSH connection is healthy, then retry."
+                    ),
+                )
 
         # Symlink staging avoids fragile GNU tar --transform rules.
         # On Windows without Developer Mode, symlink creation raises
@@ -285,7 +314,13 @@ class SSHEnvironment(BaseEnvironment):
                 ssh_proc.kill()
                 tar_proc.wait()
                 ssh_proc.wait()
-                raise RuntimeError("SSH bulk upload timed out")
+                raise EnvironmentConnectionError(
+                    "SSH bulk upload timed out",
+                    retry_hint=(
+                        f"Bulk file sync to {self.host} timed out — check the "
+                        "connection and retry."
+                    ),
+                )
 
             if tar_proc.returncode != 0:
                 raise RuntimeError(
@@ -293,9 +328,13 @@ class SSHEnvironment(BaseEnvironment):
                     f"{tar_stderr_raw.decode(errors='replace').strip()}"
                 )
             if ssh_proc.returncode != 0:
-                raise RuntimeError(
+                raise EnvironmentConnectionError(
                     f"tar extract over SSH failed (rc={ssh_proc.returncode}): "
-                    f"{ssh_stderr.decode(errors='replace').strip()}"
+                    f"{ssh_stderr.decode(errors='replace').strip()}",
+                    retry_hint=(
+                        f"File sync over SSH to {self.host} failed — verify the "
+                        "connection is healthy, then retry."
+                    ),
                 )
 
         logger.debug("SSH: bulk-uploaded %d file(s) via tar pipe", len(files))
@@ -316,7 +355,13 @@ class SSHEnvironment(BaseEnvironment):
                 timeout=120,
             )
         if result.returncode != 0:
-            raise RuntimeError(f"SSH bulk download failed: {result.stderr.decode(errors='replace').strip()}")
+            raise EnvironmentConnectionError(
+                f"SSH bulk download failed: {result.stderr.decode(errors='replace').strip()}",
+                retry_hint=(
+                    f"File sync from {self.host} failed — verify the SSH "
+                    "connection is healthy, then retry."
+                ),
+            )
 
     def _ssh_delete(self, remote_paths: list[str]) -> None:
         """Batch-delete remote files in one SSH call."""
@@ -330,7 +375,13 @@ class SSHEnvironment(BaseEnvironment):
             stdin=subprocess.DEVNULL,
         )
         if result.returncode != 0:
-            raise RuntimeError(f"remote rm failed: {result.stderr.strip()}")
+            raise EnvironmentConnectionError(
+                f"remote rm failed: {result.stderr.strip()}",
+                retry_hint=(
+                    f"Remote file cleanup on {self.host} failed — verify the "
+                    "SSH connection is healthy, then retry."
+                ),
+            )
 
     def _before_execute(self) -> None:
         """Sync files to remote via FileSyncManager (rate-limited internally)."""

tools/terminal_tool.py

@@ -1055,6 +1055,7 @@ def _transform_sudo_command(command: str | None) -> tuple[str | None, str | None
 
 
 # Environment classes now live in tools/environments/
+from tools.environments.base import EnvironmentConnectionError
 from tools.environments.local import LocalEnvironment as _LocalEnvironment
 from tools.environments.singularity import SingularityEnvironment as _SingularityEnvironment
 from tools.environments.ssh import SSHEnvironment as _SSHEnvironment
@@ -3183,6 +3184,43 @@ def terminal_tool(
 
             return json.dumps(result_dict, ensure_ascii=False)
 
+    except EnvironmentConnectionError as e:
+        # Infrastructure/connection-class failure (SSH host down, Docker
+        # daemon unreachable) — distinct from a command failing with a
+        # nonzero exit code.  Config gate ``terminal.degraded_mode``:
+        #   warn (default) — return a structured degraded result the model
+        #                    can act on (reason + retry hint, no traceback).
+        #   fail           — preserve the historical error+traceback result.
+        degraded_mode = os.getenv("TERMINAL_DEGRADED_MODE", "warn").strip().lower()
+        if degraded_mode == "fail":
+            import traceback
+            tb_str = traceback.format_exc()
+            logger.error("terminal_tool exception:\n%s", tb_str)
+            return json.dumps({
+                "output": "",
+                "exit_code": -1,
+                "error": f"Failed to execute command: {str(e)}",
+                "traceback": tb_str,
+                "status": "error"
+            }, ensure_ascii=False)
+
+        logger.warning("terminal backend degraded: %s", e.reason)
+        # Never keep a possibly-broken backend cached: evict it so the next
+        # call re-creates the environment from scratch and simply works once
+        # the backend is reachable again.
+        try:
+            _evict_environment_for_task(task_id)
+        except Exception:
+            logger.debug("degraded-env eviction failed", exc_info=True)
+        return json.dumps({
+            "output": "",
+            "exit_code": -1,
+            "status": "degraded",
+            "reason": e.reason,
+            "retry_hint": e.retry_hint,
+            "error": f"Terminal backend degraded: {e.reason}",
+        }, ensure_ascii=False)
+
     except Exception as e:
         import traceback
         tb_str = traceback.format_exc()
@@ -3196,6 +3234,30 @@ def terminal_tool(
         }, ensure_ascii=False)
 
 
+def _evict_environment_for_task(task_id: Optional[str]) -> None:
+    """Drop any cached environment for *task_id* (and its collapsed key).
+
+    Used when a backend reports an infrastructure failure: keeping the dead
+    env cached would make every subsequent call fail against a stale
+    connection, defeating automatic recovery.
+    """
+    keys = {_resolve_container_task_id(task_id)}
+    if task_id:
+        keys.add(task_id)
+    evicted = []
+    with _env_lock:
+        for key in keys:
+            env = _active_environments.pop(key, None)
+            _last_activity.pop(key, None)
+            if env is not None:
+                evicted.append(env)
+    for env in evicted:
+        try:
+            env.cleanup()
+        except Exception:
+            logger.debug("cleanup of degraded environment failed", exc_info=True)
+
+
 def check_terminal_requirements() -> bool:
     """Check if all requirements for the terminal tool are met."""
     try: