From ec1443b9f106bf0c4e83669d9abea8ecf934fb3d Mon Sep 17 00:00:00 2001 From: Henkey Date: Fri, 1 May 2026 00:51:31 +0100 Subject: [PATCH] fix(acp): normalize Windows cwd for WSL tool execution --- acp_adapter/session.py | 48 +++++++++++++++++++++---- tests/acp/test_session.py | 75 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+), 7 deletions(-) diff --git a/acp_adapter/session.py b/acp_adapter/session.py index d1fb1a874e..d6dace66b4 100644 --- a/acp_adapter/session.py +++ b/acp_adapter/session.py @@ -26,6 +26,33 @@ from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) +def _win_path_to_wsl(path: str) -> str | None: + """Convert a Windows drive path to its WSL /mnt//... 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) @@ -180,6 +211,7 @@ class SessionManager: """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( @@ -222,6 +254,7 @@ class SessionManager: """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 @@ -323,6 +356,7 @@ class SessionManager: 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 diff --git a/tests/acp/test_session.py b/tests/acp/test_session.py index c86819f6df..03d5f3f658 100644 --- a/tests/acp/test_session.py +++ b/tests/acp/test_session.py @@ -8,6 +8,7 @@ from types import SimpleNamespace 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 @@ class TestCreateSession: 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 @@ class TestCreateSession: 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 # ---------------------------------------------------------------------------