Skip to content
Closed
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
14 changes: 10 additions & 4 deletions plugins/memory/hindsight/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -496,7 +496,9 @@ def _load_simple_env(path) -> dict[str, str]:
return {}

values: dict[str, str] = {}
for line in path.read_text(encoding="utf-8").splitlines():
# utf-8-sig, not plain utf-8: this is also used on the Hermes .env during
# post_setup, and a Notepad BOM would otherwise stick to the first key.
for line in path.read_text(encoding="utf-8-sig", errors="replace").splitlines():
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
Expand Down Expand Up @@ -895,7 +897,9 @@ def post_setup(self, hermes_home: str, config: dict) -> None:
env_path = Path(hermes_home) / ".env"
existing_llm_key = ""
if env_path.exists():
for line in env_path.read_text().splitlines():
for line in env_path.read_text(
encoding="utf-8-sig"
).splitlines():
if line.startswith("HINDSIGHT_LLM_API_KEY="):
existing_llm_key = line.split("=", 1)[1]
break
Expand Down Expand Up @@ -925,7 +929,9 @@ def post_setup(self, hermes_home: str, config: dict) -> None:
env_path.parent.mkdir(parents=True, exist_ok=True)
existing_lines = []
if env_path.exists():
existing_lines = env_path.read_text().splitlines()
existing_lines = env_path.read_text(
encoding="utf-8-sig"
).splitlines()
updated_keys = set()
new_lines = []
for line in existing_lines:
Expand All @@ -938,7 +944,7 @@ def post_setup(self, hermes_home: str, config: dict) -> None:
for k, v in env_writes.items():
if k not in updated_keys:
new_lines.append(f"{k}={v}")
env_path.write_text("\n".join(new_lines) + "\n")
env_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8")

if mode == "local_embedded":
materialized_config = dict(provider_config)
Expand Down
18 changes: 15 additions & 3 deletions plugins/memory/mem0/_setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -189,7 +189,14 @@ def _write_env(env_path: Path, env_writes: dict[str, str]) -> None:
env_path.parent.mkdir(parents=True, exist_ok=True)
existing_lines: list[str] = []
if env_path.exists():
existing_lines = env_path.read_text().splitlines()
# Read as UTF-8 (BOM-tolerant), matching the canonical .env readers in
# hermes_cli/config.py. read_text() with no encoding falls back to the
# system locale (cp1252/GBK on Windows): it mangles or crashes on
# non-ASCII values while copying existing lines through, and a BOM'd
# first line would fail the key match and get duplicated.
existing_lines = env_path.read_text(
encoding="utf-8-sig"
).splitlines()

updated_keys: set[str] = set()
new_lines: list[str] = []
Expand All @@ -204,7 +211,7 @@ def _write_env(env_path: Path, env_writes: dict[str, str]) -> None:
if k not in updated_keys:
new_lines.append(f"{k}={v}")

env_path.write_text("\n".join(new_lines) + "\n")
env_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8")


def _save_mem0_json(hermes_home: str, data: dict) -> None:
Expand Down Expand Up @@ -375,7 +382,12 @@ def _prompt_api_key(label: str, env_var: str, hermes_home: str) -> str:
if not existing:
env_path = Path(hermes_home) / ".env"
if env_path.exists():
for line in env_path.read_text().splitlines():
# BOM-tolerant read matching the canonical .env readers in
# hermes_cli/config.py; a Notepad BOM on the first line would
# otherwise defeat the startswith() key match below.
for line in env_path.read_text(
encoding="utf-8-sig", errors="replace"
).splitlines():
if line.startswith(f"{env_var}="):
existing = line.split("=", 1)[1].strip()
break
Expand Down
60 changes: 60 additions & 0 deletions tests/plugins/memory/test_hindsight_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
REFLECT_SCHEMA,
RETAIN_SCHEMA,
_load_config,
_load_simple_env,
_build_embedded_profile_env,
_normalize_observation_scopes,
_normalize_retain_tags,
Expand Down Expand Up @@ -1819,3 +1820,62 @@ def test_save_config_sets_owner_only_permissions(tmp_path):
assert config_file.exists()
mode = stat.S_IMODE(config_file.stat().st_mode)
assert mode == 0o600, f"Expected 0o600 (owner-only), got {oct(mode)}"


class TestLoadSimpleEnv:
def test_bom_first_key_is_recognized(self, tmp_path):
"""A Notepad-edited .env carries a BOM; the first key must still parse
instead of becoming '\ufeffHINDSIGHT_LLM_API_KEY'."""
env_path = tmp_path / ".env"
env_path.write_bytes("HINDSIGHT_LLM_API_KEY=sk-test\n".encode("utf-8"))
values = _load_simple_env(env_path)
assert values.get("HINDSIGHT_LLM_API_KEY") == "sk-test"

def test_non_ascii_values_read_intact(self, tmp_path):
env_path = tmp_path / ".env"
env_path.write_bytes("PROXY_NOTE=café-zürich-完了\n".encode("utf-8"))
values = _load_simple_env(env_path)
assert values["PROXY_NOTE"] == "café-zürich-完了"


class TestPostSetupEnvEncoding:
def _run_cloud_post_setup(self, tmp_path, monkeypatch):
"""Drive post_setup through the cloud path with piped stdin."""
import io
import shutil as shutil_mod

monkeypatch.setattr("hermes_cli.memory_setup._curses_select",
lambda *a, **kw: 0) # cloud mode
monkeypatch.setattr("hermes_cli.config.save_config", lambda c: None)
monkeypatch.setattr(shutil_mod, "which", lambda *_: None) # skip uv install
# First line: API key prompt (readline). Second line: API URL (input).
monkeypatch.setattr(sys, "stdin", io.StringIO("sk-new\n\n"))

provider = HindsightMemoryProvider()
provider.post_setup(str(tmp_path), {"memory": {}})

def test_bom_first_key_updated_in_place(self, tmp_path, monkeypatch):
"""The setup writer reads the existing .env BOM-tolerantly, so a
BOM'd first key is matched and rewritten, not duplicated."""
env_path = tmp_path / ".env"
env_path.write_bytes("HINDSIGHT_API_KEY=old\n".encode("utf-8"))

self._run_cloud_post_setup(tmp_path, monkeypatch)

content = env_path.read_text(encoding="utf-8")
assert content.count("HINDSIGHT_API_KEY=") == 1
assert "HINDSIGHT_API_KEY=sk-new" in content
assert "old" not in content
assert "" not in content

def test_non_ascii_lines_survive_round_trip(self, tmp_path, monkeypatch):
"""Unrelated non-ASCII .env content must be copied through as UTF-8
(the locale codec would crash or mangle it on Windows)."""
env_path = tmp_path / ".env"
env_path.write_bytes("PROXY_NOTE=café-zürich-完了\n".encode("utf-8"))

self._run_cloud_post_setup(tmp_path, monkeypatch)

content = env_path.read_text(encoding="utf-8")
assert "PROXY_NOTE=café-zürich-完了" in content
assert "HINDSIGHT_API_KEY=sk-new" in content
43 changes: 43 additions & 0 deletions tests/plugins/memory/test_mem0_setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
parse_flags,
build_oss_config,
_write_env,
_prompt_api_key,
post_setup,
_check_qdrant_path,
_check_ollama,
Expand Down Expand Up @@ -164,6 +165,48 @@ def test_update_existing_var(self, tmp_path):
assert "OTHER=keep" in content
assert "old" not in content

def test_preserves_non_ascii_existing_lines(self, tmp_path):
"""Existing non-ASCII .env content must survive the read-modify-write
as UTF-8 (the locale codec would crash/mangle it on Windows)."""
env_path = tmp_path / ".env"
env_path.write_bytes("PROXY_NOTE=café-zürich-完了\n".encode("utf-8"))
_write_env(env_path, {"OPENAI_API_KEY": "sk-test"})
content = env_path.read_text(encoding="utf-8")
assert "PROXY_NOTE=café-zürich-完了" in content
assert "OPENAI_API_KEY=sk-test" in content

def test_updates_first_key_with_bom(self, tmp_path):
"""A Notepad-edited .env carries a BOM; the first key must still be
matched/updated in place, not duplicated."""
env_path = tmp_path / ".env"
env_path.write_bytes("OPENAI_API_KEY=old\n".encode("utf-8"))
_write_env(env_path, {"OPENAI_API_KEY": "new"})
content = env_path.read_text(encoding="utf-8")
assert content.count("OPENAI_API_KEY=") == 1
assert "OPENAI_API_KEY=new" in content


class TestPromptApiKey:

def test_existing_key_found_behind_bom(self, tmp_path, monkeypatch):
"""The masked-current-value lookup must see a key on the BOM'd first
line of a Notepad-edited .env instead of prompting from scratch."""
env_path = tmp_path / ".env"
env_path.write_bytes("OPENAI_API_KEY=sk-existing\n".encode("utf-8"))
monkeypatch.delenv("OPENAI_API_KEY", raising=False)

prompts: list[str] = []

def _fake_getpass(prompt):
prompts.append(prompt)
return ""

monkeypatch.setattr("plugins.memory.mem0._setup.getpass.getpass", _fake_getpass)
_prompt_api_key("OpenAI", "OPENAI_API_KEY", str(tmp_path))

assert len(prompts) == 1
assert "current: ...ting" in prompts[0]


class TestPostSetup:

Expand Down
Loading