Skip to content
Merged
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
83 changes: 83 additions & 0 deletions tests/tools/test_terminal_task_cwd.py
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,89 @@ class FakeEnv:
assert fake_env.cwd == "/workspace/keep"


def test_stale_env_cwd_from_different_session_is_ignored(monkeypatch):
"""A different session's `cd` left env.cwd pointing at its checkout.

The terminal env is shared (collapsed to "default"), so env.cwd tracks the
LAST session that ran a command. When session B claims the env after
session A left it in A's worktree, the first command must NOT run in A's
leftover cwd — it must fall through to the config/override cwd (this
session's own workspace).
"""
calls = []

class FakeEnv:
env = {}
cwd = "/home/user/src/hermes-desktop-tipc/apps/desktop"
cwd_owner = "session-A-key"

def execute(self, command, **kwargs):
calls.append((command, kwargs))
return {"output": "ok", "returncode": 0}

task_id = "session-B"
monkeypatch.setattr(terminal_tool, "_active_environments", {"default": FakeEnv()})
monkeypatch.setattr(terminal_tool, "_last_activity", {})
monkeypatch.setattr(terminal_tool, "_task_env_overrides", {})
monkeypatch.setattr(terminal_tool, "_get_env_config", lambda: _minimal_terminal_config(cwd="/home/user/src/hermes-agent"))
monkeypatch.setattr(terminal_tool, "_start_cleanup_thread", lambda: None)
monkeypatch.setattr(terminal_tool, "_resolve_container_task_id", lambda value: "default")
monkeypatch.setattr(
terminal_tool,
"_check_all_guards",
lambda command, env_type, **kwargs: {"approved": True},
)

result = json.loads(terminal_tool.terminal_tool(command="pwd", task_id=task_id))

assert result["exit_code"] == 0
# The command must run in the config cwd (hermes-agent), NOT the stale
# env.cwd left by session A (hermes-desktop-tipc).
assert calls == [("pwd", {"timeout": 60, "cwd": "/home/user/src/hermes-agent"})]


def test_same_session_env_cwd_is_trusted_after_first_claim(monkeypatch):
"""Once a session has claimed the env, subsequent commands trust env.cwd.

The prev_owner check only rejects env.cwd when a DIFFERENT session owned it
before this call. After the first command (which claims ownership),
subsequent calls in the same session should trust the live env.cwd so that
in-session `cd` state survives.
"""
calls = []

class FakeEnv:
env = {}
cwd = "/workspace/deep"
cwd_owner = "session-X"

def execute(self, command, **kwargs):
calls.append((command, kwargs))
return {"output": "ok", "returncode": 0}

env = FakeEnv()
task_id = "session-X"
monkeypatch.setattr(terminal_tool, "_active_environments", {"default": env})
monkeypatch.setattr(terminal_tool, "_last_activity", {})
monkeypatch.setattr(terminal_tool, "_task_env_overrides", {})
monkeypatch.setattr(terminal_tool, "_get_env_config", lambda: _minimal_terminal_config(cwd="/workspace/config"))
monkeypatch.setattr(terminal_tool, "_start_cleanup_thread", lambda: None)
monkeypatch.setattr(terminal_tool, "_resolve_container_task_id", lambda value: "default")
monkeypatch.setattr(
terminal_tool,
"_check_all_guards",
lambda command, env_type, **kwargs: {"approved": True},
)

# First call: env was owned by "session-X" (same session_key since
# get_current_session_key falls back to task_id). prev_owner == current
# session, so env.cwd is trusted.
result = json.loads(terminal_tool.terminal_tool(command="pwd", task_id=task_id))

assert result["exit_code"] == 0
assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/deep"})]


def test_safe_getcwd_returns_real_cwd(monkeypatch):
monkeypatch.setattr(terminal_tool.os, "getcwd", lambda: "/home/user/project")
assert terminal_tool._safe_getcwd() == "/home/user/project"
Expand Down
24 changes: 24 additions & 0 deletions tools/terminal_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -1988,6 +1988,7 @@ def _resolve_command_cwd(
workdir: Optional[str],
env: Any,
default_cwd: str,
prev_owner: Optional[str] = None,
) -> str:
"""Return the cwd for a command, preferring the live session cwd.

Expand All @@ -1996,12 +1997,29 @@ def _resolve_command_cwd(
new directory in ``env.cwd``, but foreground/background calls kept forcing
the old cwd back through ``env.execute(..., cwd=...)``. Explicit
``workdir=`` must still override everything.

When ``prev_owner`` is provided and differs from the current session,
``env.cwd`` was mutated by a *different* session's ``cd`` and must NOT be
trusted — fall through to ``default_cwd`` (the config/override cwd) so
the command runs in this session's own workspace, not the previous
session's leftover checkout. This mirrors the ``_live_cwd_if_owned``
guard file_tools uses for the same shared-env problem.
"""
if workdir:
return workdir

live_cwd = getattr(env, "cwd", None)
if isinstance(live_cwd, str) and live_cwd.strip():
# The env is shared (collapsed to "default"); its cwd tracks the LAST
# session that ran a command. If a different session owned the env
# before this call claimed it, env.cwd is that session's leftover `cd`
# — not ours. Don't use it.
if prev_owner is not None:
session_key = getattr(env, "cwd_owner", "")
# cwd_owner was already overwritten to the current session at the
# call site, so compare against the captured previous owner.
if prev_owner and prev_owner != "default" and session_key != prev_owner:
return default_cwd
return live_cwd

return default_cwd
Expand Down Expand Up @@ -2352,6 +2370,10 @@ def terminal_tool(
from tools.approval import get_current_session_key

session_key = get_current_session_key(default="") or (task_id or "")
# Capture the env's previous owner BEFORE claiming it — _resolve_command_cwd
# needs to know whether env.cwd was left by a *different* session's `cd`
# (in which case it's stale for this session and must be ignored).
prev_cwd_owner = getattr(env, "cwd_owner", "") or ""
try:
env.cwd_owner = session_key
except Exception:
Expand All @@ -2367,6 +2389,7 @@ def terminal_tool(
workdir=workdir,
env=env,
default_cwd=cwd,
prev_owner=prev_cwd_owner,
)
try:
if env_type == "local":
Expand Down Expand Up @@ -2627,6 +2650,7 @@ def terminal_tool(
workdir=workdir,
env=env,
default_cwd=cwd,
prev_owner=prev_cwd_owner,
)
execute_kwargs = {
"timeout": effective_timeout,
Expand Down
Loading