Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions tests/test_tui_gateway_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -5694,3 +5694,62 @@ def test_notification_event_dedup_key_keeps_completions_one_shot():
assert server._notification_event_dedup_key(first) == server._notification_event_dedup_key(
replay
)


# ── Deleted-CWD tolerance (slash worker + path completion) ───────────


def _raise_cwd_gone():
raise FileNotFoundError("[Errno 2] No such file or directory")


def test_safe_getcwd_falls_back_when_cwd_deleted(monkeypatch, tmp_path):
"""os.getcwd() raises FileNotFoundError on a deleted CWD; the helper must
fall back to TERMINAL_CWD, then the home directory."""
monkeypatch.setattr(server.os, "getcwd", _raise_cwd_gone)

monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
assert server._safe_getcwd() == str(tmp_path)

monkeypatch.delenv("TERMINAL_CWD", raising=False)
monkeypatch.setattr(server.os.path, "expanduser", lambda p: str(tmp_path / "home"))
assert server._safe_getcwd() == str(tmp_path / "home")


def test_completion_cwd_tolerates_deleted_cwd(monkeypatch, tmp_path):
"""Path completion must not crash when the launch directory was removed
out from under the gateway mid-session."""
monkeypatch.setattr(server.os, "getcwd", _raise_cwd_gone)
monkeypatch.delenv("TERMINAL_CWD", raising=False)
monkeypatch.setattr(server.os.path, "expanduser", lambda p: str(tmp_path))

# No params, no session cwd, no TERMINAL_CWD → would have hit os.getcwd().
assert server._completion_cwd({}) == str(tmp_path)


def test_slash_worker_tolerates_deleted_cwd(monkeypatch, tmp_path):
"""The slash worker subprocess must launch from a safe cwd instead of
crashing when os.getcwd() raises on a deleted directory."""
monkeypatch.setattr(server.os, "getcwd", _raise_cwd_gone)
monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))

captured: dict[str, object] = {}

class _FakeProc:
stdout: list[str] = []
stderr: list[str] = []

def poll(self):
return None

def _fake_popen(argv, **kwargs):
captured["cwd"] = kwargs.get("cwd")
return _FakeProc()

monkeypatch.setattr(server.subprocess, "Popen", _fake_popen)

worker = server._SlashWorker("sess", "model")
try:
assert captured["cwd"] == str(tmp_path)
finally:
worker.close()
21 changes: 18 additions & 3 deletions tui_gateway/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,21 @@ def _thread_panic_hook(args):
_stdio_transport = StdioTransport(lambda: _real_stdout, _stdout_lock)


def _safe_getcwd() -> str:
"""Return the current working directory, tolerating a deleted CWD.

``os.getcwd()`` raises FileNotFoundError when the process's working
directory has been removed out from under it (e.g. a scratch workspace
cleaned up mid-session). Fall back to TERMINAL_CWD, then the user's home
directory, so the slash worker and path completion never crash on a stale
CWD. Mirrors ``tools.terminal_tool._safe_getcwd``.
"""
try:
return os.getcwd()
except FileNotFoundError:
return os.getenv("TERMINAL_CWD") or os.path.expanduser("~")


class _SlashWorker:
"""Persistent HermesCLI subprocess for slash commands."""

Expand All @@ -234,7 +249,7 @@ def __init__(self, session_key: str, model: str):
stderr=subprocess.PIPE,
text=True,
bufsize=1,
cwd=os.getcwd(),
cwd=_safe_getcwd(),

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When salvaging this helper onto current main, also route the later-added _default_session_cwd() fallback through it (tui_gateway/server.py:1116). Otherwise a deleted launch CWD still fails the default/resume session path.

env=os.environ.copy(),
)
threading.Thread(target=self._drain_stdout, daemon=True).start()
Expand Down Expand Up @@ -793,15 +808,15 @@ def _completion_cwd(params: dict | None = None) -> str:
(params or {}).get("cwd")
or _sessions.get((params or {}).get("session_id") or "", {}).get("cwd")
or os.environ.get("TERMINAL_CWD")
or os.getcwd()
or _safe_getcwd()
)
try:
resolved = os.path.abspath(os.path.expanduser(str(raw)))
if os.path.isdir(resolved):
return resolved
except Exception:
pass
return os.getcwd()
return _safe_getcwd()


def _git_branch_for_cwd(cwd: str) -> str:
Expand Down
Loading