refactor(memory,skills): replace tri-state write_mode with boolean write_approval (default off) (#43354)
The shipped tri-state write_mode (on|off|approve) conflated two concepts —
whether writes are enabled and whether they're gated — so 'on' (writes flow
freely, gate inactive) read like 'gating is on'. Replace it with a single
clear boolean gate that defaults off.
memory.write_approval / skills.write_approval:
false (default) — write freely; the approval gate is off (pre-gate behaviour)
true — require approval: memory foreground prompts inline, memory
background-review + all skill writes stage for review
The old 'off = block all writes' mode is dropped; memory_enabled: false already
disables memory entirely, so a third 'block' state was redundant.
- tools/write_approval.py: get_write_mode/MODE_* → write_approval_enabled() bool;
evaluate_gate() loses the config-driven 'blocked' path (blocked now only comes
from an interactive user denial).
- tools/memory_tool.py, tools/skill_manager_tool.py: comment + behaviour follow.
- hermes_cli/config.py: memory/skills write_mode → write_approval (False);
_config_version 28→29 with a 28→29 migration that renames any persisted
write_mode (approve→true, on/off/unset→false) and drops the old key.
- slash commands: '/memory|/skills mode <on|off|approve>' → 'approval <on|off>'
('mode' kept as a back-compat alias); set_mode_fn callback now takes a bool.
- write_approval_commands.py, cli_commands_mixin.py, gateway/slash_commands.py,
commands.py: handlers + registry args/subcommands updated.
- docs + tests rewritten for the boolean model; added migration tests.
This commit is contained in:
Teknium
@@ -1,9 +1,10 @@
|
||||
"""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 tri-state write_mode (on/off/approve) for both subsystems, the
|
||||
foreground-vs-background staging split, pending store CRUD, and the
|
||||
list/approve/reject/diff/mode subcommand dispatch.
|
||||
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
|
||||
@@ -24,64 +25,64 @@ def hermes_home(monkeypatch):
|
||||
shutil.rmtree(d, ignore_errors=True)
|
||||
|
||||
|
||||
def _set_mode(subsystem, mode):
|
||||
def _set_approval(subsystem, enabled):
|
||||
import hermes_cli.config as cfg
|
||||
c = cfg.load_config()
|
||||
c.setdefault(subsystem, {})["write_mode"] = mode
|
||||
c.setdefault(subsystem, {})["write_approval"] = enabled
|
||||
cfg.save_config(c)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mode resolution
|
||||
# Config resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_default_write_mode_is_on(hermes_home):
|
||||
def test_default_gate_is_off(hermes_home):
|
||||
from tools import write_approval as wa
|
||||
assert wa.get_write_mode("memory") == "on"
|
||||
assert wa.get_write_mode("skills") == "on"
|
||||
# Default: gate off → writes flow freely.
|
||||
assert wa.write_approval_enabled("memory") is False
|
||||
assert wa.write_approval_enabled("skills") is False
|
||||
|
||||
|
||||
def test_invalid_subsystem_returns_on(hermes_home):
|
||||
def test_invalid_subsystem_is_off(hermes_home):
|
||||
from tools import write_approval as wa
|
||||
assert wa.get_write_mode("bogus") == "on"
|
||||
assert wa.write_approval_enabled("bogus") is False
|
||||
|
||||
|
||||
def test_normalize_mode_handles_yaml_bool():
|
||||
def test_normalize_enabled_coerces_values():
|
||||
from tools import write_approval as wa
|
||||
assert wa._normalize_mode(False) == "off"
|
||||
assert wa._normalize_mode(True) == "on"
|
||||
assert wa._normalize_mode("approve") == "approve"
|
||||
assert wa._normalize_mode("garbage") == "on"
|
||||
# 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_off_blocks_write(hermes_home):
|
||||
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
|
||||
_set_mode("memory", "off")
|
||||
store = MemoryStore(); store.load_from_disk()
|
||||
r = json.loads(memory_tool("add", "user", "should not save", store=store))
|
||||
assert r["success"] is False
|
||||
assert "disabled" in r["error"].lower()
|
||||
assert store.user_entries == []
|
||||
|
||||
|
||||
def test_memory_on_allows_write(hermes_home):
|
||||
from tools.memory_tool import memory_tool, MemoryStore
|
||||
_set_mode("memory", "on")
|
||||
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_memory_approve_no_interactive_stages(hermes_home):
|
||||
# No approval callback registered and not a gateway context → stage.
|
||||
def test_memory_gate_on_no_interactive_stages(hermes_home):
|
||||
# Gate on, no approval callback / not a gateway context → stage.
|
||||
from tools.memory_tool import memory_tool, MemoryStore
|
||||
from tools import write_approval as wa
|
||||
_set_mode("memory", "approve")
|
||||
_set_approval("memory", True)
|
||||
store = MemoryStore(); store.load_from_disk()
|
||||
r = json.loads(memory_tool("add", "memory", "stage me", store=store))
|
||||
assert r.get("staged") is True
|
||||
@@ -93,10 +94,10 @@ def test_memory_approve_no_interactive_stages(hermes_home):
|
||||
assert pend[0]["id"] == r["pending_id"]
|
||||
|
||||
|
||||
def test_memory_approve_then_apply(hermes_home):
|
||||
def test_memory_gate_on_then_apply(hermes_home):
|
||||
from tools.memory_tool import memory_tool, MemoryStore, apply_memory_pending
|
||||
from tools import write_approval as wa
|
||||
_set_mode("memory", "approve")
|
||||
_set_approval("memory", True)
|
||||
store = MemoryStore(); store.load_from_disk()
|
||||
r = json.loads(memory_tool("add", "user", "approved entry", store=store))
|
||||
pid = r["pending_id"]
|
||||
@@ -116,33 +117,36 @@ _SKILL = (
|
||||
)
|
||||
|
||||
|
||||
def test_skill_off_blocks_create(hermes_home):
|
||||
from tools.skill_manager_tool import skill_manage
|
||||
_set_mode("skills", "off")
|
||||
r = json.loads(skill_manage("create", "blocked-skill", content=_SKILL))
|
||||
assert r["success"] is False
|
||||
assert "disabled" in r["error"].lower()
|
||||
def test_skill_gate_off_allows_create(hermes_home):
|
||||
# Default (gate off) → skill is created normally, not staged.
|
||||
import importlib
|
||||
import tools.skill_manager_tool as smt
|
||||
importlib.reload(smt)
|
||||
from tools import write_approval as wa
|
||||
r = json.loads(smt.skill_manage("create", "free-skill", content=_SKILL))
|
||||
assert r.get("success") is True
|
||||
assert wa.pending_count("skills") == 0
|
||||
|
||||
|
||||
def test_skill_approve_always_stages(hermes_home):
|
||||
def test_skill_gate_on_always_stages(hermes_home):
|
||||
# Skills stage even in the foreground (too big to review inline).
|
||||
from tools.skill_manager_tool import skill_manage
|
||||
from tools import write_approval as wa
|
||||
_set_mode("skills", "approve")
|
||||
_set_approval("skills", True)
|
||||
r = json.loads(skill_manage("create", "staged-skill", content=_SKILL))
|
||||
assert r.get("staged") is True
|
||||
assert "staged-skill" in r.get("gist", "")
|
||||
assert wa.pending_count("skills") == 1
|
||||
|
||||
|
||||
def test_skill_approve_then_apply_writes_file(hermes_home):
|
||||
def test_skill_gate_on_then_apply_writes_file(hermes_home):
|
||||
# SKILLS_DIR is resolved at import time, so reload the skill module under
|
||||
# this test's HERMES_HOME to exercise the real on-disk write path.
|
||||
import importlib
|
||||
import tools.skill_manager_tool as smt
|
||||
importlib.reload(smt)
|
||||
from tools import write_approval as wa
|
||||
_set_mode("skills", "approve")
|
||||
_set_approval("skills", True)
|
||||
r = json.loads(smt.skill_manage("create", "applied-skill", content=_SKILL))
|
||||
rec = wa.get_pending("skills", r["pending_id"])
|
||||
res = json.loads(smt.apply_skill_pending(rec["payload"]))
|
||||
@@ -153,7 +157,7 @@ def test_skill_approve_then_apply_writes_file(hermes_home):
|
||||
def test_skill_create_diff_is_full_content(hermes_home):
|
||||
from tools.skill_manager_tool import skill_manage
|
||||
from tools import write_approval as wa
|
||||
_set_mode("skills", "approve")
|
||||
_set_approval("skills", True)
|
||||
r = json.loads(skill_manage("create", "diff-skill", content=_SKILL))
|
||||
rec = wa.get_pending("skills", r["pending_id"])
|
||||
diff = wa.skill_pending_diff(rec)
|
||||
@@ -212,24 +216,49 @@ def test_handle_reject(hermes_home):
|
||||
assert wa.pending_count("skills") == 0
|
||||
|
||||
|
||||
def test_handle_mode_set(hermes_home):
|
||||
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, ["mode", "approve"],
|
||||
set_mode_fn=lambda m: captured.update(mode=m),
|
||||
wa.MEMORY, ["approval", "on"],
|
||||
set_mode_fn=lambda enabled: captured.update(enabled=enabled),
|
||||
)
|
||||
assert captured["mode"] == "approve"
|
||||
assert "approve" in out
|
||||
assert captured["enabled"] is True
|
||||
assert "on" in out
|
||||
|
||||
|
||||
def test_handle_mode_invalid(hermes_home):
|
||||
def test_handle_approval_off(hermes_home):
|
||||
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
||||
from tools import write_approval as wa
|
||||
out = handle_pending_subcommand(wa.MEMORY, ["mode", "bogus"],
|
||||
set_mode_fn=lambda m: None)
|
||||
assert "Invalid mode" in out
|
||||
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
|
||||
|
||||
|
||||
def test_handle_mode_alias_still_works(hermes_home):
|
||||
# 'mode' is kept as a back-compat alias for 'approval'.
|
||||
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
||||
from tools import write_approval as wa
|
||||
captured = {}
|
||||
out = handle_pending_subcommand(
|
||||
wa.MEMORY, ["mode", "on"],
|
||||
set_mode_fn=lambda enabled: captured.update(enabled=enabled),
|
||||
)
|
||||
assert captured["enabled"] is True
|
||||
assert "on" in out
|
||||
|
||||
|
||||
def test_handle_approval_invalid(hermes_home):
|
||||
from hermes_cli.write_approval_commands import handle_pending_subcommand
|
||||
from tools import write_approval as wa
|
||||
out = handle_pending_subcommand(wa.MEMORY, ["approval", "bogus"],
|
||||
set_mode_fn=lambda enabled: None)
|
||||
assert "Invalid value" in out
|
||||
|
||||
|
||||
def test_handle_unknown_subcommand_returns_none(hermes_home):
|
||||
|
||||
Reference in New Issue
Block a user