Gateway overwriting computed TERMINAL_CWD with raw config value

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #11029 fix: enforce config.yaml as sole CWD source + deprecate .env CWD vars + add hermes memory reset

  merged to main

  files: cli.py, gateway/run.py, hermes_cli/config.py, hermes_cli/main.py, tests/cli/test_cwd_env_respect.py, tests/gateway/test_config_cwd_bridge.py, tests/hermes_cli/test_deprecated_cwd_warning.py, tests/hermes_cli/test_memory_reset.py

  Show the diff

  • 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",

  • gateway/run.py

    @@ -131,6 +131,12 @@ def _ensure_ssl_certs() -> 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 (".", "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 @@ def _ensure_ssl_certs() -> None:
     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 @@ def _ensure_ssl_certs() -> None:
     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 @@ async def _prepare_inbound_message_text(
                     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 "",
    @@ -5620,7 +5635,7 @@ async def _handle_rollback_command(self, event: MessageEvent) -> str:
                 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:

  • hermes_cli/config.py

    @@ -1597,13 +1597,8 @@ def _ensure_hermes_home_managed(home: Path):
         },
     
         # ── Agent settings ──
    -    "MESSAGING_CWD": {
    -        "description": "Working directory for terminal commands via messaging",
    -        "prompt": "Messaging working directory (default: home)",
    -        "url": None,
    -        "password": [redacted:secret-assignment]
    -        "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.

  • hermes_cli/main.py

    @@ -5659,6 +5659,18 @@ def cmd_plugins(args):
         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 @@ def cmd_memory(args):
                 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)

  • tests/cli/test_cwd_env_respect.py

    @@ -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"

  • tests/gateway/test_config_cwd_bridge.py

    @@ -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 @@ def test_top_level_cwd_beats_messaging_cwd(self):
             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"

  • tests/hermes_cli/test_deprecated_cwd_warning.py

    @@ -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

  • tests/hermes_cli/test_memory_reset.py

    @@ -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
    +
    +
    [redacted:email]
    +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"