5c3afeee45
"The replace/remove ops of a staged payload, single or batch" was spelled three ways across apply_memory_pending, the CLI pending list and the pin step, one of them re-typing _BG_DELETE_ACTIONS as a literal tuple. One destructive_ops() helper beside the constant replaces them, and the CLI drops isinstance guards on records this code itself wrote (the apply path never checked them either). Tests: the legacy/unpinned case had its own outcome and an early return inside the parametrised "names the removed entry" test; it is its own test now, so each name states what it asserts.
352 lines
15 KiB
Python
352 lines
15 KiB
Python
"""Tests for the memory/skill write-approval gate (tools/write_approval.py)
|
|
and the shared slash-command handlers (hermes_cli/write_approval_commands.py).
|
|
|
|
Covers the boolean write_approval gate (off by default = write freely; on =
|
|
require approval) for both subsystems, the foreground-vs-background staging
|
|
split, pending store CRUD, and the list/approve/reject/diff/approval
|
|
subcommand dispatch.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import shutil
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def hermes_home(monkeypatch):
|
|
d = tempfile.mkdtemp(prefix="hermes_wa_test_")
|
|
home = os.path.join(d, ".hermes")
|
|
os.makedirs(home)
|
|
monkeypatch.setenv("HERMES_HOME", home)
|
|
yield home
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
|
|
def _set_approval(subsystem, enabled):
|
|
import hermes_cli.config as cfg
|
|
c = cfg.load_config()
|
|
c.setdefault(subsystem, {})["write_approval"] = enabled
|
|
cfg.save_config(c)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
def test_list_pending_skips_non_dict_record(hermes_home):
|
|
"""A parseable-but-non-object pending file must be skipped, not crash the sort."""
|
|
from tools import write_approval as wa
|
|
wa.stage_write("memory", {"action": "add", "target": "user", "content": "ok"},
|
|
summary="ok", origin="foreground")
|
|
pending_dir = wa._pending_path("memory", "").parent
|
|
(pending_dir / "bad.json").write_text('"not a record"', encoding="utf-8")
|
|
records = wa.list_pending("memory")
|
|
assert len(records) == 1 and records[0]["payload"]["content"] == "ok"
|
|
assert wa.get_pending("memory", "bad") is None
|
|
|
|
|
|
def test_normalize_enabled_coerces_values():
|
|
from tools import write_approval as wa
|
|
# Real bools pass through.
|
|
assert wa._normalize_enabled(True) is True
|
|
assert wa._normalize_enabled(False) is False
|
|
# Truthy strings → True (incl. legacy 'approve').
|
|
assert wa._normalize_enabled("on") is True
|
|
assert wa._normalize_enabled("approve") is True
|
|
assert wa._normalize_enabled("true") is True
|
|
# Everything else → False (gate off is the safe default).
|
|
assert wa._normalize_enabled("off") is False
|
|
assert wa._normalize_enabled("garbage") is False
|
|
assert wa._normalize_enabled(None) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Memory gate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_memory_gate_off_allows_write(hermes_home):
|
|
# Default (gate off) → write straight through, no staging.
|
|
from tools.memory_tool import memory_tool, MemoryStore
|
|
from tools import write_approval as wa
|
|
store = MemoryStore(); store.load_from_disk()
|
|
r = json.loads(memory_tool("add", "user", "save me", store=store))
|
|
assert r["success"] is True
|
|
assert r["entry_count"] == 1
|
|
assert wa.pending_count("memory") == 0
|
|
|
|
|
|
def test_cli_memory_approve_without_live_agent_uses_fresh_store(hermes_home, capsys):
|
|
"""#46783: ``/memory approve`` from a context with no live agent (e.g. the
|
|
Desktop GUI) passed ``memory_store=None`` into the shared handler, which
|
|
returned "memory store unavailable" and applied nothing. The CLI handler must
|
|
fall back to a freshly loaded on-disk store, like the gateway path does."""
|
|
import json
|
|
from tools.memory_tool import memory_tool, MemoryStore
|
|
from tools import write_approval as wa
|
|
from hermes_cli.cli_commands_mixin import CLICommandsMixin
|
|
|
|
_set_approval("memory", True)
|
|
staging = MemoryStore(); staging.load_from_disk()
|
|
r = json.loads(memory_tool("add", "memory", "remember the launch date", store=staging))
|
|
assert r.get("pending_id"), r
|
|
assert wa.pending_count("memory") == 1
|
|
|
|
# Bare CLI handler with no live agent → store resolves to None pre-fix.
|
|
handler = CLICommandsMixin.__new__(CLICommandsMixin)
|
|
handler.agent = None
|
|
handler._handle_memory_command("/memory approve all")
|
|
|
|
out = capsys.readouterr().out
|
|
assert "memory store unavailable" not in out, out
|
|
assert "Approved 1" in out, out
|
|
assert wa.pending_count("memory") == 0
|
|
# The approved write landed in a freshly loaded on-disk store (MEMORY.md).
|
|
reloaded = MemoryStore(); reloaded.load_from_disk()
|
|
assert any("remember the launch date" in e for e in reloaded.memory_entries)
|
|
|
|
|
|
def test_load_on_disk_store_honors_configured_limits_and_permissions(hermes_home, monkeypatch):
|
|
"""Fresh approval stores must match the live agent's limits and target gates."""
|
|
from tools.memory_tool import MemoryStore, load_on_disk_store
|
|
|
|
# Config override path: helper picks up configured limits and store flags.
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {
|
|
"memory": {
|
|
"memory_char_limit": 999,
|
|
"user_char_limit": 444,
|
|
"memory_enabled": False,
|
|
"user_profile_enabled": True,
|
|
}
|
|
},
|
|
)
|
|
store = load_on_disk_store()
|
|
assert store.memory_char_limit == 999
|
|
assert store.user_char_limit == 444
|
|
assert store.memory_enabled is False
|
|
assert store.user_profile_enabled is True
|
|
|
|
# Failure path: config raises → defaults, never blows up.
|
|
def _boom():
|
|
raise RuntimeError("no config")
|
|
|
|
monkeypatch.setattr("hermes_cli.config.load_config", _boom)
|
|
fallback = load_on_disk_store()
|
|
defaults = MemoryStore()
|
|
assert fallback.memory_char_limit == defaults.memory_char_limit
|
|
assert fallback.user_char_limit == defaults.user_char_limit
|
|
assert fallback.memory_enabled is True
|
|
assert fallback.user_profile_enabled is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared command handler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_handle_approve_all(hermes_home):
|
|
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
|
from tools.memory_tool import MemoryStore
|
|
from tools import write_approval as wa
|
|
store = MemoryStore(); store.load_from_disk()
|
|
wa.stage_write("memory", {"action": "add", "target": "user", "content": "a"},
|
|
summary="a", origin="foreground")
|
|
wa.stage_write("memory", {"action": "add", "target": "user", "content": "b"},
|
|
summary="b", origin="foreground")
|
|
out = handle_pending_subcommand(wa.MEMORY, ["approve", "all"], memory_store=store)
|
|
assert "Approved 2" in out
|
|
assert wa.pending_count("memory") == 0
|
|
assert len(store.user_entries) == 2
|
|
|
|
|
|
def test_handle_approve_surfaces_overwritten_entry(hermes_home):
|
|
"""#117952: on the /memory approve surface a partial-entry replace must show the
|
|
approver the FULL entry it overwrote — the store's replaced_entries field used to be
|
|
dropped by _apply_one, so the incident path stayed silent."""
|
|
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
|
from tools.memory_tool import MemoryStore
|
|
from tools import write_approval as wa
|
|
store = MemoryStore(); store.load_from_disk()
|
|
entry = "RULE A: gate merges. RULE B: ci per HEAD. RULE C: never squash."
|
|
store.add("memory", entry)
|
|
wa.stage_write("memory", {"action": "batch", "target": "memory", "operations": [
|
|
{"action": "replace", "old_text": "RULE B: ci per HEAD.", "content": "RULE B: CI is per-head.",
|
|
"matched_entry": entry}]},
|
|
summary="batch", origin="background_review")
|
|
out = handle_pending_subcommand(wa.MEMORY, ["approve", "all"], memory_store=store)
|
|
assert "Approved 1" in out and entry in out
|
|
assert store.memory_entries == ["RULE B: CI is per-head."]
|
|
|
|
|
|
_KEPT = "Repo lives in ~/src/app; tests via make test"
|
|
_REVIEWED = "Staging DB: pg-staging-2 (old cluster, retiring)"
|
|
|
|
|
|
def _review_stages_remove(shape):
|
|
"""Seed memory, then stage a remove the way the unattended background review does."""
|
|
from tools.memory_tool import MemoryStore, memory_tool
|
|
from tools.skill_provenance import (reset_current_write_origin, reset_review_attended,
|
|
set_current_write_origin, set_review_attended)
|
|
store = MemoryStore(); store.load_from_disk()
|
|
for entry in (_KEPT, _REVIEWED):
|
|
assert store.add("memory", entry)["success"]
|
|
op = {"action": "remove", "old_text": "Staging DB"}
|
|
kwargs = op if shape == "single" else {"operations": [op, {"action": "add", "content": "Deploys via make ship"}]}
|
|
origin, attended = set_current_write_origin("background_review"), set_review_attended(False)
|
|
try:
|
|
staged = json.loads(memory_tool(target="memory", store=store, **kwargs))
|
|
finally:
|
|
reset_review_attended(attended)
|
|
reset_current_write_origin(origin)
|
|
assert staged["staged"] is True, staged
|
|
return store, staged["pending_id"]
|
|
|
|
|
|
@pytest.mark.parametrize("shape", ["single", "batch"])
|
|
def test_approve_refuses_staged_remove_whose_entry_changed(hermes_home, shape):
|
|
"""Approval re-ran the staged old_text search against the file as it is THEN, so it
|
|
deleted the newer entry the live agent had written in place, which the approver never saw."""
|
|
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
|
from tools.memory_tool import load_on_disk_store, memory_tool
|
|
from tools import write_approval as wa
|
|
store, pid = _review_stages_remove(shape)
|
|
newer = "Staging DB: pg-staging-3 (migrated 2026-09-20, creds in vault 'stg')"
|
|
assert json.loads(memory_tool(action="replace", old_text="pg-staging-2", content=newer, store=store))["success"]
|
|
|
|
out = handle_pending_subcommand(wa.MEMORY, ["approve", pid], memory_store=load_on_disk_store())
|
|
|
|
assert load_on_disk_store().memory_entries == [_KEPT, newer], out
|
|
assert "changed since it was staged" in out
|
|
assert wa.get_pending(wa.MEMORY, pid) is not None
|
|
# The pending list shows the whole entry the write targets, not just its search string.
|
|
assert _REVIEWED in handle_pending_subcommand(wa.MEMORY, ["pending"])
|
|
|
|
|
|
@pytest.mark.parametrize("shape", ["single", "batch"])
|
|
def test_approve_names_the_entry_a_remove_deleted(hermes_home, shape):
|
|
"""Approve listed what a replace overwrote but was silent about what a remove deleted."""
|
|
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
|
from tools.memory_tool import load_on_disk_store
|
|
from tools import write_approval as wa
|
|
_store, pid = _review_stages_remove(shape)
|
|
out = handle_pending_subcommand(wa.MEMORY, ["approve", pid], memory_store=load_on_disk_store())
|
|
assert _REVIEWED not in load_on_disk_store().memory_entries, out
|
|
assert _REVIEWED in out
|
|
|
|
|
|
def test_approve_refuses_unpinned_legacy_remove(hermes_home):
|
|
"""A record staged before removes were pinned to their full entry has no verifiable target,
|
|
so approve refuses it (keeping the record) instead of replaying its old_text search."""
|
|
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
|
from tools.memory_tool import load_on_disk_store
|
|
from tools import write_approval as wa
|
|
_store, pid = _review_stages_remove("single")
|
|
path = wa._pending_path(wa.MEMORY, pid)
|
|
record = json.loads(path.read_text(encoding="utf-8"))
|
|
record["payload"].pop("matched_entry", None)
|
|
path.write_text(json.dumps(record), encoding="utf-8")
|
|
assert "unpinned legacy target" in handle_pending_subcommand(wa.MEMORY, ["pending"])
|
|
|
|
out = handle_pending_subcommand(wa.MEMORY, ["approve", pid], memory_store=load_on_disk_store())
|
|
|
|
assert "Approved 0" in out and "predates entry pinning" in out, out
|
|
assert _REVIEWED in load_on_disk_store().memory_entries
|
|
assert wa.get_pending(wa.MEMORY, pid) is not None
|
|
|
|
|
|
def test_handle_approval_on(hermes_home):
|
|
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
|
from tools import write_approval as wa
|
|
captured = {}
|
|
out = handle_pending_subcommand(
|
|
wa.MEMORY, ["approval", "on"],
|
|
set_mode_fn=lambda enabled: captured.update(enabled=enabled),
|
|
)
|
|
assert captured["enabled"] is True
|
|
assert "on" in out
|
|
|
|
|
|
def test_handle_approval_off(hermes_home):
|
|
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
|
from tools import write_approval as wa
|
|
captured = {}
|
|
out = handle_pending_subcommand(
|
|
wa.SKILLS, ["approval", "off"],
|
|
set_mode_fn=lambda enabled: captured.update(enabled=enabled),
|
|
)
|
|
assert captured["enabled"] is False
|
|
assert "off" in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Inline (interactive CLI) approval path — regression for the bug where the
|
|
# per-thread approval callback was never passed to prompt_dangerous_approval,
|
|
# so every gated foreground memory write was silently denied.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def approval_callback_cleanup():
|
|
yield
|
|
from tools.terminal_tool import set_approval_callback
|
|
set_approval_callback(None)
|
|
|
|
|
|
def test_memory_inline_approve_writes(hermes_home, approval_callback_cleanup):
|
|
from tools.memory_tool import memory_tool, MemoryStore
|
|
from tools.terminal_tool import set_approval_callback
|
|
from tools import write_approval as wa
|
|
_set_approval("memory", True)
|
|
|
|
calls = []
|
|
def approve_cb(command, description, **kw):
|
|
calls.append((command, description))
|
|
return "once"
|
|
set_approval_callback(approve_cb)
|
|
|
|
store = MemoryStore(); store.load_from_disk()
|
|
r = json.loads(memory_tool("add", "memory", "approved fact", store=store))
|
|
assert r["success"] is True
|
|
assert r.get("staged") is None # real write, not staged
|
|
assert store.memory_entries == ["approved fact"]
|
|
assert wa.pending_count("memory") == 0
|
|
# The registered callback must actually be invoked (not the input() path).
|
|
assert len(calls) == 1
|
|
assert "approved fact" in calls[0][0]
|
|
|
|
|
|
def test_memory_inline_deny_blocks(hermes_home, approval_callback_cleanup):
|
|
from tools.memory_tool import memory_tool, MemoryStore
|
|
from tools.terminal_tool import set_approval_callback
|
|
from tools import write_approval as wa
|
|
_set_approval("memory", True)
|
|
set_approval_callback(lambda command, description, **kw: "deny")
|
|
|
|
store = MemoryStore(); store.load_from_disk()
|
|
r = json.loads(memory_tool("add", "memory", "denied fact", store=store))
|
|
assert r["success"] is False
|
|
assert "denied" in r["error"].lower()
|
|
assert store.memory_entries == []
|
|
assert wa.pending_count("memory") == 0 # denied, not staged
|
|
|
|
|
|
def test_memory_invalid_params_rejected_before_staging(hermes_home):
|
|
# Param validation must run BEFORE the gate so a broken write is rejected
|
|
# immediately instead of staged and failing at approve time.
|
|
from tools.memory_tool import memory_tool, MemoryStore
|
|
from tools import write_approval as wa
|
|
_set_approval("memory", True)
|
|
store = MemoryStore(); store.load_from_disk()
|
|
r = json.loads(memory_tool("add", "memory", None, store=store))
|
|
assert r["success"] is False
|
|
assert wa.pending_count("memory") == 0
|
|
|
|
|