fix(local): normalize invalid cwd to home

tests/tools/test_local_env_blocklist.py

@@ -10,6 +10,7 @@
 
 import os
 import threading
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 from tools.environments.local import (
@@ -111,6 +112,18 @@ def test_non_registry_provider_vars_are_stripped(self):
         for var in extra_provider_vars:
             assert var not in result_env, f"{var} leaked into subprocess env"
 
+    def test_execute_expands_tilde_cwd(self):
+        env = LocalEnvironment(cwd="/tmp", timeout=10, env={})
+        result = env.execute("pwd", cwd="~")
+        assert result["returncode"] == 0
+        assert result["output"].strip() == str(Path.home())
+
+    def test_invalid_cwd_falls_back_to_home(self):
+        env = LocalEnvironment(cwd="/tmp", timeout=10, env={})
+        result = env.execute("pwd", cwd="/definitely-missing-hermes-cwd")
+        assert result["returncode"] == 0
+        assert result["output"].strip() == str(Path.home())
+
     def test_tool_and_gateway_vars_are_stripped(self):
         """Tool and gateway secrets/config must not leak into subprocess env."""
         leaked_vars = {

tests/tools/test_process_registry.py

@@ -298,6 +298,50 @@ def test_prune_over_max_removes_oldest(self, registry):
 # =========================================================================
 
 class TestSpawnEnvSanitization:
+    def test_spawn_local_normalizes_tilde_cwd(self, registry):
+        captured = {}
+
+        def fake_popen(cmd, **kwargs):
+            captured["cwd"] = kwargs["cwd"]
+            proc = MagicMock()
+            proc.pid = 4321
+            proc.stdout = iter([])
+            proc.stdin = MagicMock()
+            proc.poll.return_value = None
+            return proc
+
+        fake_thread = MagicMock()
+
+        with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
+            patch("subprocess.Popen", side_effect=fake_popen), \
+            patch("threading.Thread", return_value=fake_thread), \
+            patch.object(registry, "_write_checkpoint"):
+            registry.spawn_local("echo hello", cwd="~")
+
+        assert captured["cwd"] == str(Path.home())
+
+    def test_spawn_local_falls_back_when_cwd_missing(self, registry):
+        captured = {}
+
+        def fake_popen(cmd, **kwargs):
+            captured["cwd"] = kwargs["cwd"]
+            proc = MagicMock()
+            proc.pid = 4321
+            proc.stdout = iter([])
+            proc.stdin = MagicMock()
+            proc.poll.return_value = None
+            return proc
+
+        fake_thread = MagicMock()
+
+        with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
+            patch("subprocess.Popen", side_effect=fake_popen), \
+            patch("threading.Thread", return_value=fake_thread), \
+            patch.object(registry, "_write_checkpoint"):
+            registry.spawn_local("echo hello", cwd="/definitely-missing-hermes-cwd")
+
+        assert captured["cwd"] == str(Path.home())
+
     def test_spawn_local_strips_blocked_vars_from_background_env(self, registry):
         captured = {}
 

tools/environments/local.py

@@ -1,5 +1,6 @@
 """Local execution environment — spawn-per-call with session snapshot."""
 
+import logging
 import os
 import platform
 import shutil
@@ -10,6 +11,7 @@
 from tools.environments.base import BaseEnvironment, _pipe_stdin
 
 _IS_WINDOWS = platform.system() == "Windows"
+logger = logging.getLogger(__name__)
 
 
 # Hermes-internal env vars that should NOT leak into terminal subprocesses.
@@ -138,6 +140,32 @@ def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = Non
     return sanitized
 
 
+def _normalize_local_cwd(cwd: str | None, *, fallback: str | None = None) -> str:
+    """Resolve user-relative / invalid working directories for local subprocesses."""
+    home = os.path.expanduser("~")
+
+    base_dir = os.path.expanduser(fallback or "") if fallback else ""
+    if base_dir and not os.path.isabs(base_dir):
+        base_dir = os.path.abspath(base_dir)
+    if not base_dir or not os.path.isdir(base_dir):
+        base_dir = home if os.path.isdir(home) else os.getcwd()
+
+    raw = (cwd or "").strip()
+    if not raw:
+        return base_dir
+
+    candidate = os.path.expanduser(raw)
+    if not os.path.isabs(candidate):
+        candidate = os.path.abspath(os.path.join(base_dir, candidate))
+
+    if os.path.isdir(candidate):
+        return candidate
+
+    fallback_dir = home if os.path.isdir(home) else base_dir
+    logger.warning("Invalid local cwd %r, falling back to %s", cwd, fallback_dir)
+    return fallback_dir
+
+
 def _find_bash() -> str:
     """Find bash for command execution."""
     if not _IS_WINDOWS:
@@ -222,9 +250,21 @@ class LocalEnvironment(BaseEnvironment):
     """
 
     def __init__(self, cwd: str = "", timeout: int = 60, env: dict = None):
-        super().__init__(cwd=cwd or os.getcwd(), timeout=timeout, env=env)
+        normalized_cwd = _normalize_local_cwd(cwd, fallback=os.getcwd())
+        super().__init__(cwd=normalized_cwd, timeout=timeout, env=env)
         self.init_session()
 
+    def execute(self, command: str, cwd: str = "", *,
+                timeout: int | None = None,
+                stdin_data: str | None = None) -> dict:
+        normalized_cwd = _normalize_local_cwd(cwd, fallback=self.cwd)
+        return super().execute(
+            command,
+            cwd=normalized_cwd,
+            timeout=timeout,
+            stdin_data=stdin_data,
+        )
+
     def get_temp_dir(self) -> str:
         """Return a shell-safe writable temp dir for local execution.
 

tools/process_registry.py

@@ -41,7 +41,11 @@
 import uuid
 
 _IS_WINDOWS = platform.system() == "Windows"
-from tools.environments.local import _find_shell, _sanitize_subprocess_env
+from tools.environments.local import (
+    _find_shell,
+    _normalize_local_cwd,
+    _sanitize_subprocess_env,
+)
 from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional
 
@@ -330,7 +334,7 @@ def spawn_local(
             command=command,
             task_id=task_id,
             session_key=session_key,
-            cwd=cwd or os.getcwd(),
+            cwd=_normalize_local_cwd(cwd, fallback=os.getcwd()),
             started_at=time.time(),
         )