From 5c3afeee45ed867c23b4eeadedc3a30220265f0e Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 24 Sep 2026 14:41:52 +0530 Subject: [PATCH] refactor(memory): derive a payload's destructive ops in one place "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. --- hermes_cli/write_approval_commands.py | 9 ++---- tests/tools/test_write_approval.py | 41 +++++++++++++++------------ tools/memory_tool.py | 15 ++++++---- 3 files changed, 35 insertions(+), 30 deletions(-) diff --git a/hermes_cli/write_approval_commands.py b/hermes_cli/write_approval_commands.py index 54379ac2da..a1e73feb3e 100644 --- a/hermes_cli/write_approval_commands.py +++ b/hermes_cli/write_approval_commands.py @@ -24,7 +24,7 @@ def _fmt_pending_list(subsystem: str) -> str: tag = " [auto]" if origin == "background_review" else "" lines.append(f" {r['id']}{tag} {r.get('summary', '')}") if subsystem == wa.MEMORY: - lines.extend(f" {line}" for line in _matched_entries(r.get("payload"))) + lines.extend(f" {line}" for line in _matched_entries(r["payload"])) lines.append("") lines.append(f"Apply: /{subsystem} approve Reject: /{subsystem} reject ") if subsystem == wa.SKILLS: @@ -113,13 +113,10 @@ def _changed_entries(result: dict, kind: str) -> List[str]: def _matched_entries(payload) -> List[str]: """The full entry each staged memory replace/remove is pinned to: the summary shows only the old_text search string, and approval applies to this entry, not to that search.""" - if not isinstance(payload, dict): - return [] - ops = payload.get("operations") if payload.get("action") == "batch" else [payload] + from tools.memory_tool import destructive_ops return [f"{op['action']}s entry: {op['matched_entry']}" if op.get("matched_entry") else f"{op['action']}: unpinned legacy target \u2014 reject and recreate before approving" - for op in (ops if isinstance(ops, list) else []) - if isinstance(op, dict) and op.get("action") in ("replace", "remove")] + for op in destructive_ops(payload)] def _apply_one(subsystem: str, rec, memory_store): diff --git a/tests/tools/test_write_approval.py b/tests/tools/test_write_approval.py index 1cfdfe0ab1..061e0c54b3 100644 --- a/tests/tools/test_write_approval.py +++ b/tests/tools/test_write_approval.py @@ -230,31 +230,36 @@ def test_approve_refuses_staged_remove_whose_entry_changed(hermes_home, shape): assert _REVIEWED in handle_pending_subcommand(wa.MEMORY, ["pending"]) -@pytest.mark.parametrize("shape,legacy", [("single", False), ("batch", False), ("single", True)]) -def test_approve_names_the_entry_a_remove_deleted(hermes_home, shape, legacy): - """Approve listed what a replace overwrote but was silent about what a remove deleted. 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.""" +@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) - if legacy: - 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 _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()) - if legacy: - 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 - return - assert _REVIEWED not in load_on_disk_store().memory_entries, out - assert _REVIEWED in out + 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): diff --git a/tools/memory_tool.py b/tools/memory_tool.py index 9ffaaa61c0..5d679d2d9d 100644 --- a/tools/memory_tool.py +++ b/tools/memory_tool.py @@ -68,11 +68,10 @@ def _pin_matched_entries(store: "MemoryStore", payload: Dict[str, Any]) -> Optio contains it. Returns the JSON error when the search fails now, as the direct write would.""" target = payload.get("target", "memory") if payload.get("action") == "batch": - ops = [op or {} for op in payload["operations"]] - result = store.resolve_batch_entries(target, ops) + result = store.resolve_batch_entries(target, payload["operations"]) if result.get("success"): payload["operations"] = [op if entry is None else {**op, "matched_entry": entry} - for op, entry in zip(ops, result["matched_entries"])] + for op, entry in zip(payload["operations"], result["matched_entries"])] elif payload.get("action") in _BG_DELETE_ACTIONS: result = store.resolve_entry(target, payload.get("old_text") or "", payload["action"]) if result.get("success"): @@ -158,6 +157,12 @@ def _validate_single_op(store, action, target, content, old_text) -> Optional[st _BG_DELETE_ACTIONS = ("replace", "remove") +def destructive_ops(payload: Dict[str, Any]) -> List[Dict[str, Any]]: + """The replace/remove ops of a staged memory payload, single-op or batch shape.""" + ops = payload.get("operations") or [] if payload.get("action") == "batch" else [payload] + return [op for op in ops if op.get("action") in _BG_DELETE_ACTIONS] + + def _background_delete_gate(store, action, operations, target="memory", content=None, old_text=None) -> Optional[str]: """Fail-closed operation gate for unattended background-review forks (#105921): ``add`` @@ -288,9 +293,7 @@ def apply_memory_pending(payload: Dict[str, Any], store: "MemoryStore") -> Dict[ target_error = _memory_target_error(store, target) if target_error is not None: return target_error - ops = payload.get("operations") or [] if action == "batch" else [payload] - if any(isinstance(op, dict) and op.get("action") in _BG_DELETE_ACTIONS and not op.get("matched_entry") - for op in ops): + if any(not op.get("matched_entry") for op in destructive_ops(payload)): return {"success": False, "error": "This destructive pending write predates entry pinning and cannot be " "verified; nothing was applied. Reject it and recreate the change."} if action == "batch":