fix: enforce config.yaml as sole CWD source + deprecate .env CWD vars + add hermes memory reset (#11029)
config.yaml terminal.cwd is now the single source of truth for working directory. MESSAGING_CWD and TERMINAL_CWD in .env are deprecated with a migration warning. Changes: 1. config.py: Remove MESSAGING_CWD from OPTIONAL_ENV_VARS (setup wizard no longer prompts for it). Add warn_deprecated_cwd_env_vars() that prints a migration hint when deprecated env vars are detected. 2. gateway/run.py: Replace all MESSAGING_CWD reads with TERMINAL_CWD (which is bridged from config.yaml terminal.cwd). MESSAGING_CWD is still accepted as a backward-compat fallback with deprecation warning. Config bridge skips cwd placeholder values so they don't clobber the resolved TERMINAL_CWD. 3. cli.py: Guard against lazy-import clobbering — when cli.py is imported lazily during gateway runtime (via delegate_tool), don't let load_cli_config() overwrite an already-resolved TERMINAL_CWD with os.getcwd() of the service's working directory. (#10817) 4. hermes_cli/main.py: Add 'hermes memory reset' command with --target all/memory/user and --yes flags. Profile-scoped via HERMES_HOME. Migration path for users with .env settings: Remove MESSAGING_CWD / TERMINAL_CWD from .env Add to config.yaml: terminal: cwd: /your/project/path Addresses: #10225, #4672, #10817, #7663
This commit is contained in:
107
tests/cli/test_cwd_env_respect.py
Normal file
107
tests/cli/test_cwd_env_respect.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""Tests that load_cli_config() guards against lazy-import TERMINAL_CWD clobbering.
|
||||
|
||||
When the gateway resolves TERMINAL_CWD at startup and cli.py is later
|
||||
imported lazily (via delegate_tool → CLI_CONFIG), load_cli_config() must
|
||||
not overwrite the already-resolved value with os.getcwd().
|
||||
|
||||
config.yaml terminal.cwd is the canonical source of truth.
|
||||
.env TERMINAL_CWD and MESSAGING_CWD are deprecated.
|
||||
See issue #10817.
|
||||
"""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
|
||||
|
||||
# The sentinel values that mean "resolve at runtime"
|
||||
_CWD_PLACEHOLDERS = (".", "auto", "cwd")
|
||||
|
||||
|
||||
def _resolve_terminal_cwd(terminal_config: dict, defaults: dict, env: dict):
|
||||
"""Simulate the CWD resolution logic from load_cli_config().
|
||||
|
||||
This mirrors the code in cli.py that checks for a pre-resolved
|
||||
TERMINAL_CWD before falling back to os.getcwd().
|
||||
"""
|
||||
if terminal_config.get("cwd") in _CWD_PLACEHOLDERS:
|
||||
_existing_cwd = env.get("TERMINAL_CWD", "")
|
||||
if _existing_cwd and _existing_cwd not in _CWD_PLACEHOLDERS and os.path.isabs(_existing_cwd):
|
||||
terminal_config["cwd"] = _existing_cwd
|
||||
defaults["terminal"]["cwd"] = _existing_cwd
|
||||
else:
|
||||
effective_backend = terminal_config.get("env_type", "local")
|
||||
if effective_backend == "local":
|
||||
terminal_config["cwd"] = "/fake/getcwd" # stand-in for os.getcwd()
|
||||
defaults["terminal"]["cwd"] = terminal_config["cwd"]
|
||||
else:
|
||||
terminal_config.pop("cwd", None)
|
||||
|
||||
# Simulate the bridging loop: write terminal_config["cwd"] to env
|
||||
_file_has_terminal = defaults.get("_file_has_terminal", False)
|
||||
if "cwd" in terminal_config:
|
||||
if _file_has_terminal or "TERMINAL_CWD" not in env:
|
||||
env["TERMINAL_CWD"] = str(terminal_config["cwd"])
|
||||
|
||||
return env.get("TERMINAL_CWD", "")
|
||||
|
||||
|
||||
class TestLazyImportGuard:
|
||||
"""TERMINAL_CWD resolved by gateway must survive a lazy cli.py import."""
|
||||
|
||||
def test_gateway_resolved_cwd_survives(self):
|
||||
"""Gateway set TERMINAL_CWD → lazy cli import must not clobber."""
|
||||
env = {"TERMINAL_CWD": "/home/user/workspace"}
|
||||
terminal_config = {"cwd": ".", "env_type": "local"}
|
||||
defaults = {"terminal": {"cwd": "."}, "_file_has_terminal": False}
|
||||
|
||||
result = _resolve_terminal_cwd(terminal_config, defaults, env)
|
||||
assert result == "/home/user/workspace"
|
||||
|
||||
def test_gateway_resolved_cwd_survives_with_file_terminal(self):
|
||||
"""Even when config.yaml has a terminal: section, resolved CWD survives."""
|
||||
env = {"TERMINAL_CWD": "/home/user/workspace"}
|
||||
terminal_config = {"cwd": ".", "env_type": "local"}
|
||||
defaults = {"terminal": {"cwd": "."}, "_file_has_terminal": True}
|
||||
|
||||
result = _resolve_terminal_cwd(terminal_config, defaults, env)
|
||||
assert result == "/home/user/workspace"
|
||||
|
||||
|
||||
class TestConfigCwdResolution:
|
||||
"""config.yaml terminal.cwd is the canonical source of truth."""
|
||||
|
||||
def test_explicit_config_cwd_wins(self):
|
||||
"""terminal.cwd: /explicit/path always wins."""
|
||||
env = {"TERMINAL_CWD": "/old/gateway/value"}
|
||||
terminal_config = {"cwd": "/explicit/path"}
|
||||
defaults = {"terminal": {"cwd": "/explicit/path"}, "_file_has_terminal": True}
|
||||
|
||||
result = _resolve_terminal_cwd(terminal_config, defaults, env)
|
||||
assert result == "/explicit/path"
|
||||
|
||||
def test_dot_cwd_resolves_to_getcwd_when_no_prior(self):
|
||||
"""With no pre-set TERMINAL_CWD, "." resolves to os.getcwd()."""
|
||||
env = {}
|
||||
terminal_config = {"cwd": "."}
|
||||
defaults = {"terminal": {"cwd": "."}, "_file_has_terminal": False}
|
||||
|
||||
result = _resolve_terminal_cwd(terminal_config, defaults, env)
|
||||
assert result == "/fake/getcwd"
|
||||
|
||||
def test_remote_backend_pops_cwd(self):
|
||||
"""Remote backend + placeholder cwd → popped for backend default."""
|
||||
env = {}
|
||||
terminal_config = {"cwd": ".", "env_type": "docker"}
|
||||
defaults = {"terminal": {"cwd": "."}, "_file_has_terminal": False}
|
||||
|
||||
result = _resolve_terminal_cwd(terminal_config, defaults, env)
|
||||
assert result == "" # cwd popped, no env var set
|
||||
|
||||
def test_remote_backend_with_prior_cwd_preserves(self):
|
||||
"""Remote backend + pre-resolved TERMINAL_CWD → adopted."""
|
||||
env = {"TERMINAL_CWD": "/project"}
|
||||
terminal_config = {"cwd": ".", "env_type": "docker"}
|
||||
defaults = {"terminal": {"cwd": "."}, "_file_has_terminal": False}
|
||||
|
||||
result = _resolve_terminal_cwd(terminal_config, defaults, env)
|
||||
assert result == "/project"
|
||||
@@ -37,6 +37,10 @@ def _simulate_config_bridge(cfg: dict, initial_env: dict | None = None):
|
||||
for cfg_key, env_var in terminal_env_map.items():
|
||||
if cfg_key in terminal_cfg:
|
||||
val = terminal_cfg[cfg_key]
|
||||
# Skip cwd placeholder values — don't overwrite already-resolved
|
||||
# TERMINAL_CWD. Mirrors the fix in gateway/run.py.
|
||||
if cfg_key == "cwd" and str(val) in (".", "auto", "cwd"):
|
||||
continue
|
||||
if isinstance(val, list):
|
||||
env[env_var] = json.dumps(val)
|
||||
else:
|
||||
@@ -146,3 +150,58 @@ class TestTopLevelCwdAlias:
|
||||
cfg = {"cwd": "/from/config"}
|
||||
result = _simulate_config_bridge(cfg, {"MESSAGING_CWD": "/from/env"})
|
||||
assert result["TERMINAL_CWD"] == "/from/config"
|
||||
|
||||
|
||||
class TestNestedTerminalCwdPlaceholderSkip:
|
||||
"""terminal.cwd placeholder values must not clobber TERMINAL_CWD.
|
||||
|
||||
When config.yaml has terminal.cwd: "." (or "auto"/"cwd"), the gateway
|
||||
config bridge should NOT write that placeholder to TERMINAL_CWD.
|
||||
This prevents .env or MESSAGING_CWD values from being overwritten.
|
||||
See issues #10225, #4672, #10817.
|
||||
"""
|
||||
|
||||
def test_terminal_dot_cwd_does_not_clobber_env(self):
|
||||
"""terminal.cwd: '.' should not overwrite a pre-set TERMINAL_CWD."""
|
||||
cfg = {"terminal": {"cwd": "."}}
|
||||
result = _simulate_config_bridge(cfg, {"TERMINAL_CWD": "/my/project"})
|
||||
assert result["TERMINAL_CWD"] == "/my/project"
|
||||
|
||||
def test_terminal_auto_cwd_does_not_clobber_env(self):
|
||||
cfg = {"terminal": {"cwd": "auto"}}
|
||||
result = _simulate_config_bridge(cfg, {"TERMINAL_CWD": "/my/project"})
|
||||
assert result["TERMINAL_CWD"] == "/my/project"
|
||||
|
||||
def test_terminal_cwd_keyword_does_not_clobber_env(self):
|
||||
cfg = {"terminal": {"cwd": "cwd"}}
|
||||
result = _simulate_config_bridge(cfg, {"TERMINAL_CWD": "/my/project"})
|
||||
assert result["TERMINAL_CWD"] == "/my/project"
|
||||
|
||||
def test_terminal_explicit_cwd_does_override(self):
|
||||
"""terminal.cwd: '/explicit/path' SHOULD override TERMINAL_CWD."""
|
||||
cfg = {"terminal": {"cwd": "/explicit/path"}}
|
||||
result = _simulate_config_bridge(cfg, {"TERMINAL_CWD": "/old/value"})
|
||||
assert result["TERMINAL_CWD"] == "/explicit/path"
|
||||
|
||||
def test_terminal_dot_cwd_falls_back_to_messaging_cwd(self):
|
||||
"""terminal.cwd: '.' with no TERMINAL_CWD should fall to MESSAGING_CWD."""
|
||||
cfg = {"terminal": {"cwd": "."}}
|
||||
result = _simulate_config_bridge(cfg, {"MESSAGING_CWD": "/from/env"})
|
||||
assert result["TERMINAL_CWD"] == "/from/env"
|
||||
|
||||
def test_terminal_dot_cwd_and_messaging_cwd_both_set(self):
|
||||
"""Pre-set TERMINAL_CWD from .env wins over terminal.cwd: '.'."""
|
||||
cfg = {"terminal": {"cwd": ".", "backend": "local"}}
|
||||
result = _simulate_config_bridge(cfg, {
|
||||
"TERMINAL_CWD": "/my/project",
|
||||
"MESSAGING_CWD": "/fallback",
|
||||
})
|
||||
assert result["TERMINAL_CWD"] == "/my/project"
|
||||
|
||||
def test_non_cwd_terminal_keys_still_bridge(self):
|
||||
"""Other terminal config keys (backend, timeout) should still bridge normally."""
|
||||
cfg = {"terminal": {"cwd": ".", "backend": "docker", "timeout": "300"}}
|
||||
result = _simulate_config_bridge(cfg, {"MESSAGING_CWD": "/from/env"})
|
||||
assert result["TERMINAL_ENV"] == "docker"
|
||||
assert result["TERMINAL_TIMEOUT"] == "300"
|
||||
assert result["TERMINAL_CWD"] == "/from/env"
|
||||
|
||||
64
tests/hermes_cli/test_deprecated_cwd_warning.py
Normal file
64
tests/hermes_cli/test_deprecated_cwd_warning.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""Tests for warn_deprecated_cwd_env_vars() migration warning."""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
|
||||
|
||||
class TestDeprecatedCwdWarning:
|
||||
"""Warn when MESSAGING_CWD or TERMINAL_CWD is set in .env."""
|
||||
|
||||
def test_messaging_cwd_triggers_warning(self, monkeypatch, capsys):
|
||||
monkeypatch.setenv("MESSAGING_CWD", "/some/path")
|
||||
monkeypatch.delenv("TERMINAL_CWD", raising=False)
|
||||
|
||||
from hermes_cli.config import warn_deprecated_cwd_env_vars
|
||||
warn_deprecated_cwd_env_vars(config={})
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "MESSAGING_CWD" in captured.err
|
||||
assert "deprecated" in captured.err.lower()
|
||||
assert "config.yaml" in captured.err
|
||||
|
||||
def test_terminal_cwd_triggers_warning_when_config_placeholder(self, monkeypatch, capsys):
|
||||
monkeypatch.setenv("TERMINAL_CWD", "/project")
|
||||
monkeypatch.delenv("MESSAGING_CWD", raising=False)
|
||||
|
||||
from hermes_cli.config import warn_deprecated_cwd_env_vars
|
||||
# config has placeholder cwd → TERMINAL_CWD likely from .env
|
||||
warn_deprecated_cwd_env_vars(config={"terminal": {"cwd": "."}})
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "TERMINAL_CWD" in captured.err
|
||||
assert "deprecated" in captured.err.lower()
|
||||
|
||||
def test_no_warning_when_config_has_explicit_cwd(self, monkeypatch, capsys):
|
||||
monkeypatch.setenv("TERMINAL_CWD", "/project")
|
||||
monkeypatch.delenv("MESSAGING_CWD", raising=False)
|
||||
|
||||
from hermes_cli.config import warn_deprecated_cwd_env_vars
|
||||
# config has explicit cwd → TERMINAL_CWD could be from config bridge
|
||||
warn_deprecated_cwd_env_vars(config={"terminal": {"cwd": "/project"}})
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "TERMINAL_CWD" not in captured.err
|
||||
|
||||
def test_no_warning_when_env_clean(self, monkeypatch, capsys):
|
||||
monkeypatch.delenv("MESSAGING_CWD", raising=False)
|
||||
monkeypatch.delenv("TERMINAL_CWD", raising=False)
|
||||
|
||||
from hermes_cli.config import warn_deprecated_cwd_env_vars
|
||||
warn_deprecated_cwd_env_vars(config={})
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert captured.err == ""
|
||||
|
||||
def test_both_deprecated_vars_warn(self, monkeypatch, capsys):
|
||||
monkeypatch.setenv("MESSAGING_CWD", "/msg/path")
|
||||
monkeypatch.setenv("TERMINAL_CWD", "/term/path")
|
||||
|
||||
from hermes_cli.config import warn_deprecated_cwd_env_vars
|
||||
warn_deprecated_cwd_env_vars(config={})
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "MESSAGING_CWD" in captured.err
|
||||
assert "TERMINAL_CWD" in captured.err
|
||||
157
tests/hermes_cli/test_memory_reset.py
Normal file
157
tests/hermes_cli/test_memory_reset.py
Normal file
@@ -0,0 +1,157 @@
|
||||
"""Tests for the `hermes memory reset` CLI command.
|
||||
|
||||
Covers:
|
||||
- Reset both stores (MEMORY.md + USER.md)
|
||||
- Reset individual stores (--target memory / --target user)
|
||||
- Skip confirmation with --yes
|
||||
- Graceful handling when no memory files exist
|
||||
- Profile-scoped reset (uses HERMES_HOME)
|
||||
"""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def memory_env(tmp_path, monkeypatch):
|
||||
"""Set up a fake HERMES_HOME with memory files."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
memories = hermes_home / "memories"
|
||||
memories.mkdir(parents=True)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
# Create sample memory files
|
||||
(memories / "MEMORY.md").write_text(
|
||||
"§\nHermes repo is at ~/.hermes/hermes-agent\n§\nUser prefers dark themes",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(memories / "USER.md").write_text(
|
||||
"§\nUser is Teknium\n§\nTimezone: US Pacific",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return hermes_home, memories
|
||||
|
||||
|
||||
def _run_memory_reset(target="all", yes=False, monkeypatch=None, confirm_input="no"):
|
||||
"""Invoke the memory reset logic from cmd_memory in main.py.
|
||||
|
||||
Simulates what happens when `hermes memory reset` is run.
|
||||
"""
|
||||
from hermes_constants import get_hermes_home, display_hermes_home
|
||||
|
||||
mem_dir = get_hermes_home() / "memories"
|
||||
files_to_reset = []
|
||||
if target in ("all", "memory"):
|
||||
files_to_reset.append(("MEMORY.md", "agent notes"))
|
||||
if target in ("all", "user"):
|
||||
files_to_reset.append(("USER.md", "user profile"))
|
||||
|
||||
existing = [(f, desc) for f, desc in files_to_reset if (mem_dir / f).exists()]
|
||||
if not existing:
|
||||
return "nothing"
|
||||
|
||||
if not yes:
|
||||
if confirm_input != "yes":
|
||||
return "cancelled"
|
||||
|
||||
for f, desc in existing:
|
||||
(mem_dir / f).unlink()
|
||||
|
||||
return "deleted"
|
||||
|
||||
|
||||
class TestMemoryReset:
|
||||
"""Tests for `hermes memory reset` subcommand."""
|
||||
|
||||
def test_reset_all_with_yes_flag(self, memory_env):
|
||||
"""--yes flag should skip confirmation and delete both files."""
|
||||
hermes_home, memories = memory_env
|
||||
assert (memories / "MEMORY.md").exists()
|
||||
assert (memories / "USER.md").exists()
|
||||
|
||||
result = _run_memory_reset(target="all", yes=True)
|
||||
assert result == "deleted"
|
||||
assert not (memories / "MEMORY.md").exists()
|
||||
assert not (memories / "USER.md").exists()
|
||||
|
||||
def test_reset_memory_only(self, memory_env):
|
||||
"""--target memory should only delete MEMORY.md."""
|
||||
hermes_home, memories = memory_env
|
||||
|
||||
result = _run_memory_reset(target="memory", yes=True)
|
||||
assert result == "deleted"
|
||||
assert not (memories / "MEMORY.md").exists()
|
||||
assert (memories / "USER.md").exists()
|
||||
|
||||
def test_reset_user_only(self, memory_env):
|
||||
"""--target user should only delete USER.md."""
|
||||
hermes_home, memories = memory_env
|
||||
|
||||
result = _run_memory_reset(target="user", yes=True)
|
||||
assert result == "deleted"
|
||||
assert (memories / "MEMORY.md").exists()
|
||||
assert not (memories / "USER.md").exists()
|
||||
|
||||
def test_reset_no_files_exist(self, tmp_path, monkeypatch):
|
||||
"""Should return 'nothing' when no memory files exist."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
(hermes_home / "memories").mkdir(parents=True)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
result = _run_memory_reset(target="all", yes=True)
|
||||
assert result == "nothing"
|
||||
|
||||
def test_reset_confirmation_denied(self, memory_env):
|
||||
"""Without --yes and without typing 'yes', should be cancelled."""
|
||||
hermes_home, memories = memory_env
|
||||
|
||||
result = _run_memory_reset(target="all", yes=False, confirm_input="no")
|
||||
assert result == "cancelled"
|
||||
# Files should still exist
|
||||
assert (memories / "MEMORY.md").exists()
|
||||
assert (memories / "USER.md").exists()
|
||||
|
||||
def test_reset_confirmation_accepted(self, memory_env):
|
||||
"""Typing 'yes' should proceed with deletion."""
|
||||
hermes_home, memories = memory_env
|
||||
|
||||
result = _run_memory_reset(target="all", yes=False, confirm_input="yes")
|
||||
assert result == "deleted"
|
||||
assert not (memories / "MEMORY.md").exists()
|
||||
assert not (memories / "USER.md").exists()
|
||||
|
||||
def test_reset_profile_scoped(self, tmp_path, monkeypatch):
|
||||
"""Reset should work on the active profile's HERMES_HOME."""
|
||||
profile_home = tmp_path / "profiles" / "myprofile"
|
||||
memories = profile_home / "memories"
|
||||
memories.mkdir(parents=True)
|
||||
(memories / "MEMORY.md").write_text("profile memory", encoding="utf-8")
|
||||
(memories / "USER.md").write_text("profile user", encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_HOME", str(profile_home))
|
||||
|
||||
result = _run_memory_reset(target="all", yes=True)
|
||||
assert result == "deleted"
|
||||
assert not (memories / "MEMORY.md").exists()
|
||||
assert not (memories / "USER.md").exists()
|
||||
|
||||
def test_reset_partial_files(self, memory_env):
|
||||
"""Reset should work when only one memory file exists."""
|
||||
hermes_home, memories = memory_env
|
||||
(memories / "USER.md").unlink()
|
||||
|
||||
result = _run_memory_reset(target="all", yes=True)
|
||||
assert result == "deleted"
|
||||
assert not (memories / "MEMORY.md").exists()
|
||||
|
||||
def test_reset_empty_memories_dir(self, tmp_path, monkeypatch):
|
||||
"""No memories dir at all should report nothing."""
|
||||
hermes_home = tmp_path / ".hermes"
|
||||
hermes_home.mkdir(parents=True)
|
||||
# No memories dir
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
|
||||
# The memories dir won't exist; get_hermes_home() / "memories" won't have files
|
||||
result = _run_memory_reset(target="all", yes=True)
|
||||
assert result == "nothing"
|
||||
Reference in New Issue
Block a user