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:
parent
fe12042e50
commit
3c42064efc
27
cli.py
27
cli.py
@ -401,14 +401,27 @@ def load_cli_config() -> Dict[str, Any]:
|
||||
# filesystem is directly accessible. For ALL remote/container backends
|
||||
# (ssh, docker, modal, singularity), the host path doesn't exist on the
|
||||
# target -- remove the key so terminal_tool.py uses its per-backend default.
|
||||
if terminal_config.get("cwd") in (".", "auto", "cwd"):
|
||||
effective_backend = terminal_config.get("env_type", "local")
|
||||
if effective_backend == "local":
|
||||
terminal_config["cwd"] = os.getcwd()
|
||||
defaults["terminal"]["cwd"] = terminal_config["cwd"]
|
||||
#
|
||||
# GUARD: If TERMINAL_CWD is already set to a real absolute path (by the
|
||||
# gateway's config bridge earlier in the process), don't clobber it.
|
||||
# This prevents a lazy import of cli.py during gateway runtime from
|
||||
# rewriting TERMINAL_CWD to the service's working directory.
|
||||
# See issue #10817.
|
||||
_CWD_PLACEHOLDERS = (".", "auto", "cwd")
|
||||
if terminal_config.get("cwd") in _CWD_PLACEHOLDERS:
|
||||
_existing_cwd = os.environ.get("TERMINAL_CWD", "")
|
||||
if _existing_cwd and _existing_cwd not in _CWD_PLACEHOLDERS and os.path.isabs(_existing_cwd):
|
||||
# Gateway (or earlier startup) already resolved a real path — keep it
|
||||
terminal_config["cwd"] = _existing_cwd
|
||||
defaults["terminal"]["cwd"] = _existing_cwd
|
||||
else:
|
||||
# Remove so TERMINAL_CWD stays unset → tool picks backend default
|
||||
terminal_config.pop("cwd", None)
|
||||
effective_backend = terminal_config.get("env_type", "local")
|
||||
if effective_backend == "local":
|
||||
terminal_config["cwd"] = os.getcwd()
|
||||
defaults["terminal"]["cwd"] = terminal_config["cwd"]
|
||||
else:
|
||||
# Remove so TERMINAL_CWD stays unset → tool picks backend default
|
||||
terminal_config.pop("cwd", None)
|
||||
|
||||
env_mappings = {
|
||||
"env_type": "TERMINAL_ENV",
|
||||
|
||||
@ -131,6 +131,12 @@ if _config_path.exists():
|
||||
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 (".", "auto", "cwd") — the
|
||||
# gateway resolves these to Path.home() later (line ~255).
|
||||
# Writing the raw placeholder here would just be noise.
|
||||
# Only bridge explicit absolute paths from config.yaml.
|
||||
if _cfg_key == "cwd" and str(_val) in (".", "auto", "cwd"):
|
||||
continue
|
||||
if isinstance(_val, list):
|
||||
os.environ[_env_var] = json.dumps(_val)
|
||||
else:
|
||||
@ -225,6 +231,13 @@ try:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Warn if user has deprecated MESSAGING_CWD / TERMINAL_CWD in .env
|
||||
try:
|
||||
from hermes_cli.config import warn_deprecated_cwd_env_vars
|
||||
warn_deprecated_cwd_env_vars()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Gateway runs in quiet mode - suppress debug output and use cwd directly (no temp dirs)
|
||||
os.environ["HERMES_QUIET"] = "1"
|
||||
|
||||
@ -232,12 +245,14 @@ os.environ["HERMES_QUIET"] = "1"
|
||||
os.environ["HERMES_EXEC_ASK"] = "1"
|
||||
|
||||
# Set terminal working directory for messaging platforms.
|
||||
# If the user set an explicit path in config.yaml (not "." or "auto"),
|
||||
# respect it. Otherwise use MESSAGING_CWD or default to home directory.
|
||||
# config.yaml terminal.cwd is the canonical source (bridged to TERMINAL_CWD
|
||||
# by the config bridge above). When it's unset or a placeholder, default
|
||||
# to home directory. MESSAGING_CWD is accepted as a backward-compat
|
||||
# fallback (deprecated — the warning above tells users to migrate).
|
||||
_configured_cwd = os.environ.get("TERMINAL_CWD", "")
|
||||
if not _configured_cwd or _configured_cwd in (".", "auto", "cwd"):
|
||||
messaging_cwd = os.getenv("MESSAGING_CWD") or str(Path.home())
|
||||
os.environ["TERMINAL_CWD"] = messaging_cwd
|
||||
_fallback = os.getenv("MESSAGING_CWD") or str(Path.home())
|
||||
os.environ["TERMINAL_CWD"] = _fallback
|
||||
|
||||
from gateway.config import (
|
||||
Platform,
|
||||
@ -3403,7 +3418,7 @@ class GatewayRunner:
|
||||
from agent.context_references import preprocess_context_references_async
|
||||
from agent.model_metadata import get_model_context_length
|
||||
|
||||
_msg_cwd = os.environ.get("MESSAGING_CWD", os.path.expanduser("~"))
|
||||
_msg_cwd = os.environ.get("TERMINAL_CWD", os.path.expanduser("~"))
|
||||
_msg_ctx_len = get_model_context_length(
|
||||
self._model,
|
||||
base_url=self._base_url or "",
|
||||
@ -5614,7 +5629,7 @@ class GatewayRunner:
|
||||
max_snapshots=cp_cfg.get("max_snapshots", 50),
|
||||
)
|
||||
|
||||
cwd = os.getenv("MESSAGING_CWD", str(Path.home()))
|
||||
cwd = os.getenv("TERMINAL_CWD", str(Path.home()))
|
||||
arg = event.get_command_args().strip()
|
||||
|
||||
if not arg:
|
||||
|
||||
@ -1597,13 +1597,8 @@ OPTIONAL_ENV_VARS = {
|
||||
},
|
||||
|
||||
# ── Agent settings ──
|
||||
"MESSAGING_CWD": {
|
||||
"description": "Working directory for terminal commands via messaging",
|
||||
"prompt": "Messaging working directory (default: home)",
|
||||
"url": None,
|
||||
"password": False,
|
||||
"category": "setting",
|
||||
},
|
||||
# NOTE: MESSAGING_CWD was removed here — use terminal.cwd in config.yaml
|
||||
# instead. The gateway reads TERMINAL_CWD (bridged from terminal.cwd).
|
||||
"SUDO_PASSWORD": {
|
||||
"description": "Sudo password for terminal commands requiring root access; set to an explicit empty string to try empty without prompting",
|
||||
"prompt": "Sudo password",
|
||||
@ -2082,6 +2077,52 @@ def print_config_warnings(config: Optional[Dict[str, Any]] = None) -> None:
|
||||
sys.stderr.write("\n".join(lines) + "\n\n")
|
||||
|
||||
|
||||
def warn_deprecated_cwd_env_vars(config: Optional[Dict[str, Any]] = None) -> None:
|
||||
"""Warn if MESSAGING_CWD or TERMINAL_CWD is set in .env instead of config.yaml.
|
||||
|
||||
These env vars are deprecated — the canonical setting is terminal.cwd
|
||||
in config.yaml. Prints a migration hint to stderr.
|
||||
"""
|
||||
import os, sys
|
||||
messaging_cwd = os.environ.get("MESSAGING_CWD")
|
||||
terminal_cwd_env = os.environ.get("TERMINAL_CWD")
|
||||
|
||||
if config is None:
|
||||
try:
|
||||
config = load_config()
|
||||
except Exception:
|
||||
return
|
||||
|
||||
terminal_cfg = config.get("terminal", {})
|
||||
config_cwd = terminal_cfg.get("cwd", ".") if isinstance(terminal_cfg, dict) else "."
|
||||
# Only warn if config.yaml doesn't have an explicit path
|
||||
config_has_explicit_cwd = config_cwd not in (".", "auto", "cwd", "")
|
||||
|
||||
lines: list[str] = []
|
||||
if messaging_cwd:
|
||||
lines.append(
|
||||
f" \033[33m⚠\033[0m MESSAGING_CWD={messaging_cwd} found in .env — "
|
||||
f"this is deprecated."
|
||||
)
|
||||
if terminal_cwd_env and not config_has_explicit_cwd:
|
||||
# TERMINAL_CWD in env but not from config bridge — likely from .env
|
||||
lines.append(
|
||||
f" \033[33m⚠\033[0m TERMINAL_CWD={terminal_cwd_env} found in .env — "
|
||||
f"this is deprecated."
|
||||
)
|
||||
if lines:
|
||||
hint_path = os.environ.get("HERMES_HOME", "~/.hermes")
|
||||
lines.insert(0, "\033[33m⚠ Deprecated .env settings detected:\033[0m")
|
||||
lines.append(
|
||||
f" \033[2mMove to config.yaml instead: "
|
||||
f"terminal:\\n cwd: /your/project/path\033[0m"
|
||||
)
|
||||
lines.append(
|
||||
f" \033[2mThen remove the old entries from {hint_path}/.env\033[0m"
|
||||
)
|
||||
sys.stderr.write("\n".join(lines) + "\n\n")
|
||||
|
||||
|
||||
def migrate_config(interactive: bool = True, quiet: bool = False) -> Dict[str, Any]:
|
||||
"""
|
||||
Migrate config to latest version, prompting for new required fields.
|
||||
|
||||
@ -5659,6 +5659,18 @@ Examples:
|
||||
memory_sub.add_parser("setup", help="Interactive provider selection and configuration")
|
||||
memory_sub.add_parser("status", help="Show current memory provider config")
|
||||
memory_sub.add_parser("off", help="Disable external provider (built-in only)")
|
||||
_reset_parser = memory_sub.add_parser(
|
||||
"reset",
|
||||
help="Erase all built-in memory (MEMORY.md and USER.md)",
|
||||
)
|
||||
_reset_parser.add_argument(
|
||||
"--yes", "-y", action="store_true",
|
||||
help="Skip confirmation prompt",
|
||||
)
|
||||
_reset_parser.add_argument(
|
||||
"--target", choices=["all", "memory", "user"], default="all",
|
||||
help="Which store to reset: 'all' (default), 'memory', or 'user'",
|
||||
)
|
||||
|
||||
def cmd_memory(args):
|
||||
sub = getattr(args, "memory_command", None)
|
||||
@ -5671,6 +5683,44 @@ Examples:
|
||||
save_config(config)
|
||||
print("\n ✓ Memory provider: built-in only")
|
||||
print(" Saved to config.yaml\n")
|
||||
elif sub == "reset":
|
||||
from hermes_constants import get_hermes_home, display_hermes_home
|
||||
mem_dir = get_hermes_home() / "memories"
|
||||
target = getattr(args, "target", "all")
|
||||
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"))
|
||||
|
||||
# Check what exists
|
||||
existing = [(f, desc) for f, desc in files_to_reset if (mem_dir / f).exists()]
|
||||
if not existing:
|
||||
print(f"\n Nothing to reset — no memory files found in {display_hermes_home()}/memories/\n")
|
||||
return
|
||||
|
||||
print(f"\n This will permanently erase the following memory files:")
|
||||
for f, desc in existing:
|
||||
path = mem_dir / f
|
||||
size = path.stat().st_size
|
||||
print(f" ◆ {f} ({desc}) — {size:,} bytes")
|
||||
|
||||
if not getattr(args, "yes", False):
|
||||
try:
|
||||
answer = input("\n Type 'yes' to confirm: ").strip().lower()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\n Cancelled.\n")
|
||||
return
|
||||
if answer != "yes":
|
||||
print(" Cancelled.\n")
|
||||
return
|
||||
|
||||
for f, desc in existing:
|
||||
(mem_dir / f).unlink()
|
||||
print(f" ✓ Deleted {f} ({desc})")
|
||||
|
||||
print(f"\n Memory reset complete. New sessions will start with a blank slate.")
|
||||
print(f" Files were in: {display_hermes_home()}/memories/\n")
|
||||
else:
|
||||
from hermes_cli.memory_setup import memory_command
|
||||
memory_command(args)
|
||||
|
||||
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"
|
||||
Loading…
Reference in New Issue
Block a user