WSL path translation and ACP cwd handling fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #18233 fix(acp): translate Windows cwd for WSL sessions (salvage #18128)

  merged to main

  files: acp_adapter/session.py, tests/acp/test_session.py

  Show the diff

  • acp_adapter/session.py

    @@ -26,6 +26,33 @@
     logger = logging.getLogger(__name__)
     
     
    +def _win_path_to_wsl(path: str) -> str | None:
    +    """Convert a Windows drive path to its WSL /mnt/<drive>/... equivalent."""
    +    match = re.match(r"^([A-Za-z]):[\\/](.*)$", path)
    +    if not match:
    +        return None
    +    drive = match.group(1).lower()
    +    tail = match.group(2).replace("\\", "/")
    +    return f"/mnt/{drive}/{tail}"
    +
    +
    +def _translate_acp_cwd(cwd: str) -> str:
    +    """Translate Windows ACP cwd values when Hermes itself is running in WSL.
    +
    +    Windows ACP clients can launch ``hermes acp`` inside WSL while still sending
    +    editor workspaces as Windows drive paths such as ``E:\\Projects``. Store
    +    and execute against the WSL mount path so agents, tools, and persisted ACP
    +    sessions all agree on the usable workspace. Native Linux/macOS keeps the
    +    original cwd unchanged.
    +    """
    +    from hermes_constants import is_wsl
    +
    +    if not is_wsl():
    +        return cwd
    +    translated = _win_path_to_wsl(str(cwd))
    +    return translated if translated is not None else cwd
    +
    +
     def _normalize_cwd_for_compare(cwd: str | None) -> str:
         raw = str(cwd or ".").strip()
         if not raw:
    @@ -34,11 +61,9 @@ def _normalize_cwd_for_compare(cwd: str | None) -> str:
     
         # Normalize Windows drive paths into the equivalent WSL mount form so
         # ACP history filters match the same workspace across Windows and WSL.
    -    match = re.match(r"^([A-Za-z]):[\\/](.*)$", expanded)
    -    if match:
    -        drive = match.group(1).lower()
    -        tail = match.group(2).replace("\\", "/")
    -        expanded = f"/mnt/{drive}/{tail}"
    +    translated = _win_path_to_wsl(expanded)
    +    if translated is not None:
    +        expanded = translated
         elif re.match(r"^/mnt/[A-Za-z]/", expanded):
             expanded = f"/mnt/{expanded[5].lower()}/{expanded[7:]}"
     
    @@ -96,12 +121,18 @@ def _acp_stderr_print(*args, **kwargs) -> None:
     
     
     def _register_task_cwd(task_id: str, cwd: str) -> None:
    -    """Bind a task/session id to the editor's working directory for tools."""
    +    """Bind a task/session id to the editor's working directory for tools.
    +
    +    Zed can launch Hermes from a Windows workspace while the ACP process runs
    +    inside WSL. In that case ACP sends cwd as e.g. ``E:\\Projects\\POTI``;
    +    local tools need the WSL mount equivalent or subprocess creation fails
    +    before the command can run.
    +    """
         if not task_id:
             return
         try:
             from tools.terminal_tool import register_task_env_overrides
    -        register_task_env_overrides(task_id, {"cwd": cwd})
    +        register_task_env_overrides(task_id, {"cwd": _translate_acp_cwd(cwd)})
         except Exception:
             logger.debug("Failed to register ACP task cwd override", exc_info=True)
     
    @@ -175,6 +206,7 @@ def create_session(self, cwd: str = ".") -> SessionState:
             """Create a new session with a unique ID and a fresh AIAgent."""
             import threading
     
    +        cwd = _translate_acp_cwd(cwd)
             session_id = str(uuid.uuid4())
             agent = self._make_agent(session_id=session_id, cwd=cwd)
             state = SessionState(
    @@ -217,6 +249,7 @@ def fork_session(self, session_id: str, cwd: str = ".") -> Optional[SessionState
             """Deep-copy a session's history into a new session."""
             import threading
     
    +        cwd = _translate_acp_cwd(cwd)
             original = self.get_session(session_id)  # checks DB too
             if original is None:
                 return None
    @@ -318,6 +351,7 @@ def list_sessions(self, cwd: str | None = None) -> List[Dict[str, Any]]:
     
         def update_cwd(self, session_id: str, cwd: str) -> Optional[SessionState]:
             """Update the working directory for a session and its tool overrides."""
    +        cwd = _translate_acp_cwd(cwd)
             state = self.get_session(session_id)  # checks DB too
             if state is None:
                 return None

  • tests/acp/test_session.py

    @@ -8,6 +8,7 @@
     import pytest
     from unittest.mock import MagicMock, patch
     
    +from acp_adapter import session as acp_session
     from acp_adapter.session import SessionManager, SessionState
     from hermes_state import SessionDB
     
    @@ -42,6 +43,27 @@ def test_create_session_registers_task_cwd(self, manager, monkeypatch):
             state = manager.create_session(cwd="/tmp/work")
             assert calls == [(state.session_id, "/tmp/work")]
     
    +
    +    def test_register_task_cwd_translates_windows_drive_for_wsl_tools(self, monkeypatch):
    +        captured = {}
    +
    +        def fake_register_task_env_overrides(task_id, overrides):
    +            captured["task_id"] = task_id
    +            captured["overrides"] = overrides
    +
    +        monkeypatch.setattr("hermes_constants._wsl_detected", True)
    +        monkeypatch.setattr(
    +            "tools.terminal_tool.register_task_env_overrides",
    +            fake_register_task_env_overrides,
    +        )
    +
    +        acp_session._register_task_cwd("session-1", r"E:\Projects\AI\paperclip")
    +
    +        assert captured == {
    +            "task_id": "session-1",
    +            "overrides": {"cwd": "/mnt/e/Projects/AI/paperclip"},
    +        }
    +
         def test_session_ids_are_unique(self, manager):
             s1 = manager.create_session()
             s2 = manager.create_session()
    @@ -56,6 +78,59 @@ def test_get_nonexistent_session_returns_none(self, manager):
             assert manager.get_session("does-not-exist") is None
     
     
    +
    +
    +# ---------------------------------------------------------------------------
    +# WSL cwd translation
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestWslCwdTranslation:
    +    def test_translate_acp_cwd_converts_windows_drive_path_when_wsl(self, monkeypatch):
    +        monkeypatch.setattr("hermes_constants._wsl_detected", True)
    +
    +        assert acp_session._translate_acp_cwd(r"E:\Projects\AI\paperclip") == "/mnt/e/Projects/AI/paperclip"
    +
    +    def test_translate_acp_cwd_handles_forward_slashes_when_wsl(self, monkeypatch):
    +        monkeypatch.setattr("hermes_constants._wsl_detected", True)
    +
    +        assert acp_session._translate_acp_cwd("D:/work/project") == "/mnt/d/work/project"
    +
    +    def test_translate_acp_cwd_leaves_windows_drive_path_unchanged_off_wsl(self, monkeypatch):
    +        monkeypatch.setattr("hermes_constants._wsl_detected", False)
    +
    +        assert acp_session._translate_acp_cwd(r"E:\Projects\AI\paperclip") == r"E:\Projects\AI\paperclip"
    +
    +    def test_translate_acp_cwd_leaves_posix_path_unchanged_on_wsl(self, monkeypatch):
    +        monkeypatch.setattr("hermes_constants._wsl_detected", True)
    +
    +        assert acp_session._translate_acp_cwd("/mnt/e/Projects/AI/paperclip") == "/mnt/e/Projects/AI/paperclip"
    +
    +    def test_create_session_stores_translated_cwd_on_wsl(self, manager, monkeypatch):
    +        monkeypatch.setattr("hermes_constants._wsl_detected", True)
    +
    +        state = manager.create_session(cwd=r"E:\Projects\AI\paperclip")
    +
    +        assert state.cwd == "/mnt/e/Projects/AI/paperclip"
    +
    +    def test_fork_session_stores_translated_cwd_on_wsl(self, manager, monkeypatch):
    +        monkeypatch.setattr("hermes_constants._wsl_detected", True)
    +        original = manager.create_session(cwd="/tmp/base")
    +
    +        forked = manager.fork_session(original.session_id, cwd=r"D:\work\project")
    +
    +        assert forked is not None
    +        assert forked.cwd == "/mnt/d/work/project"
    +
    +    def test_update_cwd_stores_translated_cwd_on_wsl(self, manager, monkeypatch):
    +        monkeypatch.setattr("hermes_constants._wsl_detected", True)
    +        state = manager.create_session(cwd="/tmp/old")
    +
    +        updated = manager.update_cwd(state.session_id, cwd=r"C:\Users\foo\project")
    +
    +        assert updated is not None
    +        assert updated.cwd == "/mnt/c/Users/foo/project"
    +
     # ---------------------------------------------------------------------------
     # fork
     # ---------------------------------------------------------------------------