diff --git a/agent/prompt_builder.py b/agent/prompt_builder.py index aec348a8d1a159f069d919415d1db3c28e6307b1..16fb7ad45c00d56ad7c79fadb09885dea8020628 100644 --- a/agent/prompt_builder.py +++ b/agent/prompt_builder.py @@ -154,7 +154,10 @@ MEMORY_GUIDANCE = ( "will still matter later.\n" "Prioritize what reduces future user steering — the most valuable memory is one " "that prevents the user from having to correct or remind you again. " - "User preferences and recurring corrections matter more than procedural task details.\n" + "User preferences and recurring corrections matter more than procedural task details. " + "Treat preferred names, nicknames, and forms of address as durable user-profile " + "preferences: if the user says 'call me X', 'address me as X', 'refer to me as X', " + "'请称呼我为X', or '以后叫我X', save a compact declarative fact to the user store.\n" "Do NOT save task progress, session outcomes, completed-work logs, or temporary TODO " "state to memory; use session_search to recall those from past transcripts. " "Specifically: do not record PR numbers, issue numbers, commit SHAs, 'fixed bug X', " diff --git a/tools/memory_tool.py b/tools/memory_tool.py index fc889f59037e1f81d7f75eb071838447fac1a91f..f86ef549cf243b98dfabc6b3c0295b5302279fbb 100644 --- a/tools/memory_tool.py +++ b/tools/memory_tool.py @@ -17,7 +17,7 @@ Entry delimiter: § (section sign). Entries can be multiline. Character limits (not tokens) because char counts are model-independent. Design: -- Single `memory` tool with action parameter: add, replace, remove +- Single `memory` tool with action parameter: add, replace, remove, read - replace/remove use short unique substring matching (not full text or IDs) - Behavioral guidance lives in the tool schema description - Frozen snapshot pattern: system prompt is stable, tool responses show live state @@ -26,8 +26,8 @@ Design: import json import logging import os +import re import tempfile -import time from contextlib import contextmanager from pathlib import Path from hermes_constants import get_hermes_home @@ -62,52 +62,46 @@ ENTRY_DELIMITER = "\n§\n" # --------------------------------------------------------------------------- # Memory content scanning — lightweight check for injection/exfiltration # in content that gets injected into the system prompt. -# -# Patterns live in ``tools/threat_patterns.py`` — the single source of truth -# shared with the context-file scanner and the tool-result delimiter system. -# Memory uses the "strict" scope (broadest pattern set) because: -# - memory entries are user-curated; the user can rewrite a flagged entry -# - memory enters the system prompt as a FROZEN snapshot, so a poisoned -# entry persists for the entire session and across sessions until -# explicitly removed. # --------------------------------------------------------------------------- -from tools.threat_patterns import first_threat_message as _first_threat_message +_MEMORY_THREAT_PATTERNS = [ + # Prompt injection + (r'ignore\s+(previous|all|above|prior)\s+instructions', "prompt_injection"), + (r'you\s+are\s+now\s+', "role_hijack"), + (r'do\s+not\s+tell\s+the\s+user', "deception_hide"), + (r'system\s+prompt\s+override', "sys_prompt_override"), + (r'disregard\s+(your|all|any)\s+(instructions|rules|guidelines)', "disregard_rules"), + (r'act\s+as\s+(if|though)\s+you\s+(have\s+no|don\'t\s+have)\s+(restrictions|limits|rules)', "bypass_restrictions"), + # Exfiltration via curl/wget with secrets + (r'curl\s+[^\n]*\$\{?\w*(KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)', "exfil_curl"), + (r'wget\s+[^\n]*\$\{?\w*(KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)', "exfil_wget"), + (r'cat\s+[^\n]*(\.env|credentials|\.netrc|\.pgpass|\.npmrc|\.pypirc)', "read_secrets"), + # Persistence via shell rc + (r'authorized_keys', "ssh_backdoor"), + (r'\$HOME/\.ssh|\~/\.ssh', "ssh_access"), + (r'\$HOME/\.hermes/\.env|\~/\.hermes/\.env', "hermes_env"), +] + +# Subset of invisible chars for injection detection +_INVISIBLE_CHARS = { + '\u200b', '\u200c', '\u200d', '\u2060', '\ufeff', + '\u202a', '\u202b', '\u202c', '\u202d', '\u202e', +} def _scan_memory_content(content: str) -> Optional[str]: """Scan memory content for injection/exfil patterns. Returns error string if blocked.""" - return _first_threat_message(content, scope="strict") - + # Check invisible unicode + for char in _INVISIBLE_CHARS: + if char in content: + return f"Blocked: content contains invisible unicode character U+{ord(char):04X} (possible injection)." -def _drift_error(path: "Path", bak_path: str) -> Dict[str, Any]: - """Build the error dict returned when external drift is detected. + # Check threat patterns + for pattern, pid in _MEMORY_THREAT_PATTERNS: + if re.search(pattern, content, re.IGNORECASE): + return f"Blocked: content matches threat pattern '{pid}'. Memory entries are injected into the system prompt and must not contain injection or exfiltration payloads." - The on-disk memory file contains content that wouldn't round-trip - through the tool's parser/serializer — flushing would discard the - appended/edited content from a patch tool, shell append, manual edit, - or sister-session write. We refuse the mutation, point the operator at - the .bak. snapshot we took, and tell them what to do next. - """ - return { - "success": False, - "error": ( - f"Refusing to write {path.name}: file on disk has content that " - f"wouldn't round-trip through the memory tool (likely added by " - f"the patch tool, a shell append, a manual edit, or a " - f"concurrent session). A snapshot was saved to {bak_path}. " - f"Resolve the drift first — either rewrite the file as a clean " - f"§-delimited list of entries, or move the extra content out — " - f"then retry. This guard exists to prevent silent data loss " - f"(issue #26045)." - ), - "drift_backup": bak_path, - "remediation": ( - "Open the .bak file, integrate the missing entries into the " - "memory tool one at a time via memory(action=add, content=...), " - "then remove or rewrite the original file to a clean state." - ), - } + return None class MemoryStore: @@ -121,12 +115,6 @@ class MemoryStore: Tool responses always reflect this live state. """ - # After this many failed consolidation attempts (overflow / zero-match) in - # ONE turn, stop instructing the model to "retry in this turn" and return a - # terminal "save skipped" result so a fragile replace/add can't loop the - # turn to budget exhaustion and suppress the user's reply (issue #42405). - _MAX_CONSOLIDATION_FAILURES_PER_TURN = 3 - def __init__(self, memory_char_limit: int = 2200, user_char_limit: int = 1375): self.memory_entries: List[str] = [] self.user_entries: List[str] = [] @@ -134,54 +122,10 @@ class MemoryStore: self.user_char_limit = user_char_limit # Frozen snapshot for system prompt -- set once at load_from_disk() self._system_prompt_snapshot: Dict[str, str] = {"memory": "", "user": ""} - # Per-turn counter of failed at-capacity consolidation attempts; reset - # at each turn boundary by reset_consolidation_failures() (#42405). - self._consolidation_failures = 0 - - def reset_consolidation_failures(self) -> None: - """Reset the per-turn consolidation-failure counter (call at turn start).""" - self._consolidation_failures = 0 - - def _consolidation_failure(self, response: Dict[str, Any]) -> Dict[str, Any]: - """Count an at-capacity consolidation failure and degrade gracefully. - - Under the per-turn cap, return ``response`` unchanged (it already tells - the model how to self-correct + retry in this turn). Once the cap is - exceeded, drop the retry instruction and return a TERMINAL result so the - model stops looping memory calls and proceeds to answer the user — a - failed memory side effect must never block the turn's reply (#42405). - """ - self._consolidation_failures += 1 - if self._consolidation_failures <= self._MAX_CONSOLIDATION_FAILURES_PER_TURN: - return response - return { - "success": False, - "done": True, - "error": ( - f"Memory consolidation failed {self._consolidation_failures} times " - "this turn. Stop retrying memory calls — leave memory unchanged for " - "now and continue with your reply to the user. The fact can be saved " - "in a later turn." - ), - } def load_from_disk(self): - """Load entries from MEMORY.md and USER.md, capture system prompt snapshot. - - The frozen snapshot is what enters the system prompt. We scan each - entry for injection/promptware patterns at snapshot-build time — - ANY hit replaces the entry text in the snapshot with a placeholder - like ``[BLOCKED: …]``, so a poisoned-on-disk memory file (supply - chain, compromised tool, sister-session write) cannot inject into - the system prompt. - - The live ``memory_entries`` / ``user_entries`` lists keep the - original text so the user can still SEE poisoned entries via - see poisoned entries by inspecting the source files directly, and remove them — silently dropping them would hide the attack from the user. - - Scanning is deterministic from disk bytes, so the snapshot remains - stable for the entire session (prefix-cache invariant holds). - """ + """Load entries from MEMORY.md and USER.md, capture system prompt snapshot.""" + print("====此处从磁盘加载MEMORY.md和USER.md") mem_dir = get_memory_dir() mem_dir.mkdir(parents=True, exist_ok=True) @@ -192,53 +136,12 @@ class MemoryStore: self.memory_entries = list(dict.fromkeys(self.memory_entries)) self.user_entries = list(dict.fromkeys(self.user_entries)) - # Sanitize entries for the system-prompt snapshot only. Live state - # (memory_entries / user_entries) keeps the raw text so the user - # can see + remove poisoned entries via the memory tool. - sanitized_memory = self._sanitize_entries_for_snapshot(self.memory_entries, "MEMORY.md") - sanitized_user = self._sanitize_entries_for_snapshot(self.user_entries, "USER.md") - # Capture frozen snapshot for system prompt injection self._system_prompt_snapshot = { - "memory": self._render_block("memory", sanitized_memory), - "user": self._render_block("user", sanitized_user), + "memory": self._render_block("memory", self.memory_entries), + "user": self._render_block("user", self.user_entries), } - - @staticmethod - def _sanitize_entries_for_snapshot(entries: List[str], filename: str) -> List[str]: - """Return ``entries`` with any threat-matching entry replaced by a placeholder. - - Each entry is scanned with the shared threat-pattern library at the - ``"strict"`` scope (same as memory writes). On match, the entry is - replaced in the returned list with ``"[BLOCKED: entry - contained threat pattern: . Removed from system prompt.]"`` — - the placeholder enters the snapshot, the original entry stays in - live state for the user to inspect and delete. - - Empty or already-block-marker entries pass through unchanged. - """ - from tools.threat_patterns import scan_for_threats - - sanitized: List[str] = [] - for entry in entries: - if not entry or entry.startswith("[BLOCKED:"): - sanitized.append(entry) - continue - findings = scan_for_threats(entry, scope="strict") - if findings: - logger.warning( - "Memory entry from %s blocked at load time: %s", - filename, ", ".join(findings), - ) - sanitized.append( - f"[BLOCKED: {filename} entry contained threat pattern(s): " - f"{', '.join(findings)}. Removed from system prompt; " - f"use memory(action=remove) " - f"to delete the original.]" - ) - else: - sanitized.append(entry) - return sanitized + print("====此处MEMORY.md和USER.md已加载并冻结为system prompt快照") @staticmethod @contextmanager @@ -284,27 +187,14 @@ class MemoryStore: return mem_dir / "USER.md" return mem_dir / "MEMORY.md" - def _reload_target(self, target: str, *, skip_drift: bool = False) -> Optional[str]: + def _reload_target(self, target: str): """Re-read entries from disk into in-memory state. Called under file lock to get the latest state before mutating. - Returns the backup path if external drift was detected (the on-disk - file contains content that wouldn't round-trip through our - parser/serializer, OR an entry larger than the store's char limit). - When drift is detected the caller must abort the mutation — - flushing would discard the un-roundtrippable content. - Returns None on clean reload. - - When *skip_drift* is True the round-trip / entry-size check is - bypassed. Used by the ``add`` action which appends without - rewriting, so existing content is never clobbered. """ - path = self._path_for(target) - bak = None if skip_drift else self._detect_external_drift(target) - fresh = self._read_file(path) + fresh = self._read_file(self._path_for(target)) fresh = list(dict.fromkeys(fresh)) # deduplicate self._set_entries(target, fresh) - return bak def save_to_disk(self, target: str): """Persist entries to the appropriate file. Called after every mutation.""" @@ -345,13 +235,8 @@ class MemoryStore: return {"success": False, "error": scan_error} with self._file_lock(self._path_for(target)): - # Re-read from disk under lock to pick up writes from other sessions. - # For add (append-only), we skip the drift guard — appending never - # clobbers existing content, so round-trip mismatches from prior - # tool-written entries in the same session are harmless. The drift - # guard remains active for replace/remove where full-file rewrite - # would discard un-roundtrippable content (issue #26045). - self._reload_target(target, skip_drift=True) + # Re-read from disk under lock to pick up writes from other sessions + self._reload_target(target) entries = self._entries_for(target) limit = self._char_limit(target) @@ -366,18 +251,16 @@ class MemoryStore: if new_total > limit: current = self._char_count(target) - return self._consolidation_failure({ + return { "success": False, "error": ( f"Memory at {current:,}/{limit:,} chars. " f"Adding this entry ({len(content)} chars) would exceed the limit. " - f"Consolidate now: use 'replace' to merge overlapping entries into " - f"shorter ones or 'remove' stale or less important entries (see " - f"current_entries below), then retry this add — all in this turn." + f"Replace or remove existing entries first." ), "current_entries": entries, "usage": f"{current:,}/{limit:,}", - }) + } entries.append(content) self._set_entries(target, entries) @@ -400,25 +283,19 @@ class MemoryStore: return {"success": False, "error": scan_error} with self._file_lock(self._path_for(target)): - bak = self._reload_target(target) - if bak: - return _drift_error(self._path_for(target), bak) + self._reload_target(target) entries = self._entries_for(target) matches = [(i, e) for i, e in enumerate(entries) if old_text in e] if not matches: - return self._consolidation_failure({ - "success": False, - "error": f"No entry matched '{old_text}'. Check current_entries below and retry with the exact text of the entry you want to replace.", - "current_entries": entries, - }) + return {"success": False, "error": f"No entry matched '{old_text}'."} if len(matches) > 1: # If all matches are identical (exact duplicates), operate on the first one unique_texts = {e for _, e in matches} if len(unique_texts) > 1: - previews = self._previews([e for _, e in matches]) + previews = [e[:80] + ("..." if len(e) > 80 else "") for _, e in matches] return { "success": False, "error": f"Multiple entries matched '{old_text}'. Be more specific.", @@ -435,18 +312,13 @@ class MemoryStore: new_total = len(ENTRY_DELIMITER.join(test_entries)) if new_total > limit: - current = self._char_count(target) - return self._consolidation_failure({ + return { "success": False, "error": ( f"Replacement would put memory at {new_total:,}/{limit:,} chars. " - f"Shorten the new content, or 'remove' other stale or less important " - f"entries to make room (see current_entries below), then retry — all " - f"in this turn." + f"Shorten the new content or remove other entries first." ), - "current_entries": entries, - "usage": f"{current:,}/{limit:,}", - }) + } entries[idx] = new_content self._set_entries(target, entries) @@ -461,25 +333,19 @@ class MemoryStore: return {"success": False, "error": "old_text cannot be empty."} with self._file_lock(self._path_for(target)): - bak = self._reload_target(target) - if bak: - return _drift_error(self._path_for(target), bak) + self._reload_target(target) entries = self._entries_for(target) matches = [(i, e) for i, e in enumerate(entries) if old_text in e] if not matches: - return self._consolidation_failure({ - "success": False, - "error": f"No entry matched '{old_text}'. Check current_entries below and retry with the exact text of the entry you want to remove.", - "current_entries": entries, - }) + return {"success": False, "error": f"No entry matched '{old_text}'."} if len(matches) > 1: # If all matches are identical (exact duplicates), remove the first one unique_texts = {e for _, e in matches} if len(unique_texts) > 1: - previews = self._previews([e for _, e in matches]) + previews = [e[:80] + ("..." if len(e) > 80 else "") for _, e in matches] return { "success": False, "error": f"Multiple entries matched '{old_text}'. Be more specific.", @@ -494,124 +360,6 @@ class MemoryStore: return self._success_response(target, "Entry removed.") - def apply_batch(self, target: str, operations: List[Dict[str, Any]]) -> Dict[str, Any]: - """Apply a sequence of add/replace/remove ops to one target atomically. - - All operations are validated and applied against the FINAL budget -- - intermediate overflow is irrelevant. This lets the model free space - (remove/replace) and add new entries in a SINGLE tool call instead of - the multi-turn consolidate-then-retry dance that re-sends the whole - conversation context several times. - - Semantics: all-or-nothing. If any op is malformed, doesn't match, or - the net result would exceed the char limit, NOTHING is written and an - error is returned describing the first failure plus the live state. - """ - if not operations: - return {"success": False, "error": "operations list is empty."} - - # Scan every add/replace content for injection/exfil BEFORE touching - # disk -- a single poisoned op rejects the whole batch. - for i, op in enumerate(operations): - act = (op or {}).get("action") - new_content = (op or {}).get("content") - if act in {"add", "replace"} and new_content: - scan_error = _scan_memory_content(new_content) - if scan_error: - return {"success": False, "error": f"Operation {i + 1}: {scan_error}"} - - with self._file_lock(self._path_for(target)): - bak = self._reload_target(target) - if bak: - return _drift_error(self._path_for(target), bak) - - # Work on a copy; only commit if the whole batch validates. - working: List[str] = list(self._entries_for(target)) - limit = self._char_limit(target) - - for i, op in enumerate(operations): - op = op or {} - act = op.get("action") - content = (op.get("content") or "").strip() - old_text = (op.get("old_text") or "").strip() - pos = f"Operation {i + 1} ({act or 'unknown'})" - - if act == "add": - if not content: - return self._batch_error(target, f"{pos}: content is required.") - if content in working: - continue # idempotent -- skip duplicate, don't fail the batch - working.append(content) - - elif act == "replace": - if not old_text: - return self._batch_error(target, f"{pos}: old_text is required.") - if not content: - return self._batch_error( - target, - f"{pos}: content is required (use action='remove' to delete).", - ) - matches = [j for j, e in enumerate(working) if old_text in e] - if not matches: - return self._batch_error(target, f"{pos}: no entry matched '{old_text}'.") - if len({working[j] for j in matches}) > 1: - return self._batch_error( - target, - f"{pos}: '{old_text}' matched multiple distinct entries -- be more specific.", - ) - working[matches[0]] = content - - elif act == "remove": - if not old_text: - return self._batch_error(target, f"{pos}: old_text is required.") - matches = [j for j, e in enumerate(working) if old_text in e] - if not matches: - return self._batch_error(target, f"{pos}: no entry matched '{old_text}'.") - if len({working[j] for j in matches}) > 1: - return self._batch_error( - target, - f"{pos}: '{old_text}' matched multiple distinct entries -- be more specific.", - ) - working.pop(matches[0]) - - else: - return self._batch_error( - target, - f"{pos}: unknown action. Use add, replace, or remove.", - ) - - # Budget check against the FINAL state only. - new_total = len(ENTRY_DELIMITER.join(working)) if working else 0 - if new_total > limit: - current = self._char_count(target) - return self._consolidation_failure({ - "success": False, - "error": ( - f"After applying all {len(operations)} operations, memory would be at " - f"{new_total:,}/{limit:,} chars -- over the limit. Remove or shorten more " - f"entries in the same batch (see current_entries below), then retry." - ), - "current_entries": self._entries_for(target), - "usage": f"{current:,}/{limit:,}", - }) - - # Commit. - self._set_entries(target, working) - self.save_to_disk(target) - - return self._success_response(target, f"Applied {len(operations)} operation(s).") - - def _batch_error(self, target: str, message: str) -> Dict[str, Any]: - """Build a batch-abort error that reports live (uncommitted) state.""" - current = self._char_count(target) - limit = self._char_limit(target) - return self._consolidation_failure({ - "success": False, - "error": message + " No operations were applied (batch is all-or-nothing).", - "current_entries": self._entries_for(target), - "usage": f"{current:,}/{limit:,}", - }) - def format_for_system_prompt(self, target: str) -> Optional[str]: """ Return the frozen snapshot for system prompt injection. @@ -622,49 +370,34 @@ class MemoryStore: Returns None if the snapshot is empty (no entries at load time). """ + print(f"====此处格式化/注入{target.upper()}信息到system prompt") block = self._system_prompt_snapshot.get(target, "") return block if block else None # -- Internal helpers -- - @staticmethod - def _previews(entries: List[str], width: int = 80) -> List[str]: - """Truncated one-line previews of entries for error feedback.""" - return [e[:width] + ("..." if len(e) > width else "") for e in entries] - def _success_response(self, target: str, message: str = None) -> Dict[str, Any]: - # A successful write means the consolidation loop made progress, so the - # per-turn failure budget resets (the cap counts consecutive failures, - # not lifetime ones within a turn) (#42405). - self._consolidation_failures = 0 entries = self._entries_for(target) current = self._char_count(target) limit = self._char_limit(target) pct = min(100, int((current / limit) * 100)) if limit > 0 else 0 - # The success response is intentionally TERMINAL: it confirms the write - # landed and tells the model to stop. We do NOT echo the full entries - # list here -- dumping it invites the model to "find more to fix" and - # re-issue the same operations (observed thrash: the correct batch on - # call 1, then 5 redundant repeats). Entries are only shown on the - # error/over-budget paths, where the model genuinely needs them to - # decide what to consolidate. resp = { "success": True, - "done": True, "target": target, + "entries": entries, "usage": f"{pct}% — {current:,}/{limit:,} chars", "entry_count": len(entries), } if message: resp["message"] = message - resp["note"] = "Write saved. This update is complete — do not repeat it." return resp def _render_block(self, target: str, entries: List[str]) -> str: """Render a system prompt block with header and usage indicator.""" if not entries: return "" + print(f"====此处渲染{target.upper()}信息块") limit = self._char_limit(target) content = ENTRY_DELIMITER.join(entries) @@ -701,61 +434,6 @@ class MemoryStore: entries = [e.strip() for e in raw.split(ENTRY_DELIMITER)] return [e for e in entries if e] - def _detect_external_drift(self, target: str) -> Optional[str]: - """Return a backup-path string if on-disk content shows external drift. - - The memory file is supposed to be a list of small entries the tool - wrote, joined by §. Detect drift via two signals: - - 1. Round-trip mismatch — re-parsing and re-serializing the file - doesn't produce identical bytes (rare; would catch oddly-encoded - delimiters). - 2. Entry-size overflow — any single parsed entry exceeds the - store's whole-file char limit. The tool budgets the ENTIRE store - against that limit; no single tool-written entry can exceed it. - When we see one entry larger than the limit, an external writer - (patch tool, shell append, manual edit, sister session) appended - free-form content into what the tool will treat as one entry. - Flushing would then truncate that entry to the model's new - content, discarding the appended bytes — issue #26045. - - Returns the absolute path of the .bak file when drift was found and - backed up; returns None when the file looks tool-shaped. - - Note: this is an INSTANCE method (not static) because we need the - per-target char_limit for signal #2. - """ - path = self._path_for(target) - if not path.exists(): - return None - try: - raw = path.read_text(encoding="utf-8") - except (OSError, IOError): - return None - if not raw.strip(): - return None - - parsed = [e.strip() for e in raw.split(ENTRY_DELIMITER) if e.strip()] - roundtrip = ENTRY_DELIMITER.join(parsed) - - char_limit = self._char_limit(target) - max_entry_len = max((len(e) for e in parsed), default=0) - - drift_detected = (raw.strip() != roundtrip) or (max_entry_len > char_limit) - if not drift_detected: - return None - - # Drift confirmed — snapshot the file so the operator can recover - # whatever the external writer added, then return the .bak path so - # the caller can refuse the mutation. - ts = int(time.time()) - bak_path = path.with_suffix(path.suffix + f".bak.{ts}") - try: - bak_path.write_text(raw, encoding="utf-8") - except (OSError, IOError): - return str(bak_path) + " (BACKUP FAILED — file unchanged on disk)" - return str(bak_path) - @staticmethod def _write_file(path: Path, entries: List[str]): """Write entries to a memory file using atomic temp-file + rename. @@ -788,244 +466,39 @@ class MemoryStore: raise RuntimeError(f"Failed to write memory file {path}: {e}") -def load_on_disk_store() -> "MemoryStore": - """Build a fresh on-disk :class:`MemoryStore`, honoring configured char limits. - - Use this from any context that has no live agent (the messaging gateway, the - Desktop GUI, the bare CLI ``/memory`` handler) but still needs to read or - apply approved memory writes. Mirrors how the live agent constructs its store - in ``agent/agent_init.py`` — including the user's ``memory.memory_char_limit`` - / ``memory.user_char_limit`` overrides — so an approval applied without a live - agent enforces the SAME caps as one applied with one. - - Falls back to the built-in defaults if config can't be loaded, so this can - never raise on a missing/unreadable config. - """ - memory_char_limit = 2200 - user_char_limit = 1375 - try: - from hermes_cli.config import load_config - - mem_cfg = (load_config() or {}).get("memory", {}) or {} - memory_char_limit = int(mem_cfg.get("memory_char_limit", memory_char_limit)) - user_char_limit = int(mem_cfg.get("user_char_limit", user_char_limit)) - except Exception: - pass # config optional — fall back to defaults rather than break /memory - - store = MemoryStore( - memory_char_limit=memory_char_limit, - user_char_limit=user_char_limit, - ) - store.load_from_disk() - return store - - -def _apply_write_gate(action: str, target: str, content: Optional[str], - old_text: Optional[str]) -> Optional[str]: - """Evaluate the memory write gate. Returns a JSON tool-result string when - the write should NOT proceed normally (blocked or staged), or None when the - caller should perform the real write. - - Only the mutating actions (add/replace/remove) are gated. - """ - if action not in {"add", "replace", "remove"}: - return None - - try: - from tools import write_approval as wa - except Exception: - # If the gate module can't load, fail open (current behaviour) rather - # than blocking all memory writes. - return None - - # Build a small inline summary/detail for the foreground approval prompt. - label = "user profile" if target == "user" else "memory" - if action == "add": - summary = f"add to {label}" - detail = content or "" - elif action == "replace": - summary = f"replace in {label}" - detail = f"old: {old_text}\nnew: {content}" - else: # remove - summary = f"remove from {label}" - detail = old_text or "" - - decision = wa.evaluate_gate(wa.MEMORY, inline_summary=summary, inline_detail=detail) - - if decision.allow: - return None - - if decision.blocked: - return tool_error(decision.message, success=False) - - # stage - payload = { - "action": action, - "target": target, - "content": content, - "old_text": old_text, - } - record = wa.stage_write( - wa.MEMORY, payload, - summary=f"{summary}: {detail[:120]}", - origin=wa.current_origin(), - ) - return json.dumps( - {"success": True, "staged": True, "pending_id": record["id"], - "message": decision.message}, - ensure_ascii=False, - ) - - -def _apply_batch_write_gate(target: str, operations: List[Dict[str, Any]]) -> Optional[str]: - """Evaluate the write gate for a batch of memory operations. - - Returns a JSON tool-result string when the batch should NOT proceed - (blocked or staged), or None when the caller should perform the real - batch write. The whole batch is gated as a single unit. - """ - try: - from tools import write_approval as wa - except Exception: - return None - - label = "user profile" if target == "user" else "memory" - summary = f"apply {len(operations)} op(s) to {label}" - detail_lines = [] - for op in operations: - op = op or {} - act = op.get("action", "?") - if act == "remove": - detail_lines.append(f"- remove: {op.get('old_text', '')}") - elif act == "replace": - detail_lines.append(f"- replace: {op.get('old_text', '')} -> {op.get('content', '')}") - else: - detail_lines.append(f"- {act}: {op.get('content', '')}") - detail = "\n".join(detail_lines) - - decision = wa.evaluate_gate(wa.MEMORY, inline_summary=summary, inline_detail=detail) - - if decision.allow: - return None - - if decision.blocked: - return tool_error(decision.message, success=False) - - payload = {"action": "batch", "target": target, "operations": operations} - record = wa.stage_write( - wa.MEMORY, payload, - summary=f"{summary}: {detail[:120]}", - origin=wa.current_origin(), - ) - return json.dumps( - {"success": True, "staged": True, "pending_id": record["id"], - "message": decision.message}, - ensure_ascii=False, - ) - - -def _missing_old_text_error(store: "MemoryStore", target: str, action: str) -> str: - """Build a recoverable error for a replace/remove call that arrived without - ``old_text``. - - ``replace``/``remove`` are inherently targeted -- without ``old_text`` there - is no entry to act on, so we cannot fulfil the call. But returning a bare - "old_text is required" is a dead-end: some structured-output clients omit the - optional ``old_text`` field (it isn't, and can't be, schema-required without - a top-level combinator the Codex backend rejects -- see - tests/tools/test_memory_tool_schema.py). So instead we return the current - entry inventory plus an explicit retry instruction, letting the model reissue - the call with ``old_text`` set to a unique substring of the entry it means. - Mirrors the batch path's ``_batch_error`` shape. (issues #43412, #49466) - """ - entries = store._entries_for(target) - current = store._char_count(target) - limit = store._char_limit(target) - return json.dumps( - { - "success": False, - "error": ( - f"'{action}' needs old_text -- a short unique substring of the entry " - f"to {action}. None was provided. Reissue the {action} with old_text " - f"set to part of one of the current_entries below." - ), - "current_entries": entries, - "usage": f"{current:,}/{limit:,}", - }, - ensure_ascii=False, - ) - - def memory_tool( - action: str = None, + action: str, target: str = "memory", content: str = None, old_text: str = None, - operations: Optional[List[Dict[str, Any]]] = None, store: Optional[MemoryStore] = None, ) -> str: """ Single entry point for the memory tool. Dispatches to MemoryStore methods. - Two shapes: - - Single op: action + (content / old_text). - - Batch: operations=[{action, content?, old_text?}, ...] applied - atomically against the final char budget in ONE call. - Returns JSON string with results. """ if store is None: return tool_error("Memory is not available. It may be disabled in config or this environment.", success=False) - # Some strict providers fill optional schema fields with JSON null rather - # than omitting them. Treat ``target: null`` as omitted so memory writes - # still use the documented default store instead of failing validation. - if target is None: - target = "memory" - if target not in {"memory", "user"}: return tool_error(f"Invalid target '{target}'. Use 'memory' or 'user'.", success=False) - # --- Batch path ------------------------------------------------------- - if operations: - if not isinstance(operations, list): - return tool_error("operations must be a list of {action, content?, old_text?} objects.", success=False) - gate_result = _apply_batch_write_gate(target, operations) - if gate_result is not None: - return gate_result - result = store.apply_batch(target, operations) - return json.dumps(result, ensure_ascii=False) - - # --- Single-op path --------------------------------------------------- - # Validate required params BEFORE the gate so an invalid write is rejected - # immediately instead of being staged and only failing at approve time. - if action == "add" and not content: - return tool_error("Content is required for 'add' action.", success=False) - if action == "replace" and (not old_text or not content): - missing = "old_text" if not old_text else "content" - if not old_text: - # The client/model omitted old_text. Replace is inherently targeted - # -- we can't guess which entry. Return the current inventory plus a - # retry instruction so the model can reissue with old_text set, - # instead of hitting a dead-end error. (issues #43412, #49466) - return _missing_old_text_error(store, target, "replace") - return tool_error(f"{missing} is required for 'replace' action.", success=False) - if action == "remove" and not old_text: - return _missing_old_text_error(store, target, "remove") - - # Approval gate: when on, stages the write (background/gateway) or prompts - # inline (interactive CLI); when off (default) passes straight through. - gate_result = _apply_write_gate(action, target, content, old_text) - if gate_result is not None: - return gate_result - if action == "add": + if not content: + return tool_error("Content is required for 'add' action.", success=False) result = store.add(target, content) elif action == "replace": + if not old_text: + return tool_error("old_text is required for 'replace' action.", success=False) + if not content: + return tool_error("content is required for 'replace' action.", success=False) result = store.replace(target, old_text, content) elif action == "remove": + if not old_text: + return tool_error("old_text is required for 'remove' action.", success=False) result = store.remove(target, old_text) else: @@ -1039,51 +512,34 @@ def check_memory_requirements() -> bool: return True -def apply_memory_pending(payload: Dict[str, Any], store: "MemoryStore") -> Dict[str, Any]: - """Replay a staged memory write directly against the store, bypassing the - write gate. Called by the /memory approve handler. - - Returns the store's result dict. - """ - action = payload.get("action") - target = payload.get("target", "memory") - content = payload.get("content") or "" - old_text = payload.get("old_text") or "" - if action == "batch": - return store.apply_batch(target, payload.get("operations") or []) - if action == "add": - return store.add(target, content) - if action == "replace": - return store.replace(target, old_text, content) - if action == "remove": - return store.remove(target, old_text) - return {"success": False, "error": f"Unknown staged action '{action}'."} +# ============================================================================= # OpenAI Function-Calling Schema # ============================================================================= MEMORY_SCHEMA = { "name": "memory", "description": ( - "Save durable facts to persistent memory that survive across sessions. Memory is " - "injected into every future turn, so keep entries compact and high-signal.\n\n" - "HOW: make ALL your changes in ONE call via an 'operations' array (each item: " - "{action, content?, old_text?}). The batch applies atomically and the char limit is " - "checked only on the FINAL result — so a single call can remove/replace stale entries " - "to free room AND add new ones, even when an add alone would overflow. The response " - "reports current/limit chars and confirms completion; one batch call finishes the " - "update, so don't repeat it. Use the bare action/content/old_text fields only for a " - "single lone change.\n\n" - "WHEN: save proactively when the user states a preference, correction, or personal " - "detail, or you learn a stable fact about their environment, conventions, or workflow. " - "Priority: user preferences & corrections > environment facts > procedures. The best " - "memory stops the user repeating themselves.\n\n" - "IF FULL: an add is rejected with the current entries shown. Reissue as ONE batch that " - "removes or shortens enough stale entries and adds the new one together.\n\n" - "TARGETS: 'user' = who the user is (name, role, preferences, style). 'memory' = your " - "notes (environment, conventions, tool quirks, lessons).\n\n" - "SKIP: trivial/obvious info, easily re-discovered facts, raw data dumps, task progress, " - "completed-work logs, temporary TODO state (use session_search for those). Reusable " - "procedures belong in a skill, not memory." + "Save durable information to persistent memory that survives across sessions. " + "Memory is injected into future turns, so keep it compact and focused on facts " + "that will still matter later.\n\n" + "WHEN TO SAVE (do this proactively, don't wait to be asked):\n" + "- User corrects you or says 'remember this' / 'don't do that again'\n" + "- User shares a preference, habit, or personal detail (name, role, timezone, coding style)\n" + "- You discover something about the environment (OS, installed tools, project structure)\n" + "- You learn a convention, API quirk, or workflow specific to this user's setup\n" + "- You identify a stable fact that will be useful again in future sessions\n\n" + "PRIORITY: User preferences and corrections > environment facts > procedural knowledge. " + "The most valuable memory prevents the user from having to repeat themselves.\n\n" + "Do NOT save task progress, session outcomes, completed-work logs, or temporary TODO " + "state to memory; use session_search to recall those from past transcripts.\n" + "If you've discovered a new way to do something, solved a problem that could be " + "necessary later, save it as a skill with the skill tool.\n\n" + "TWO TARGETS:\n" + "- 'user': who the user is -- name, role, preferences, communication style, pet peeves\n" + "- 'memory': your notes -- environment facts, project conventions, tool quirks, lessons learned\n\n" + "ACTIONS: add (new entry), replace (update existing -- old_text identifies it, When the new information duplicates an existing one, replace the old text with the new information), " + "remove (delete -- old_text identifies it).\n\n" + "SKIP: trivial/obvious info, things easily re-discovered, raw data dumps, and temporary task state." ), "parameters": { "type": "object", @@ -1091,7 +547,7 @@ MEMORY_SCHEMA = { "action": { "type": "string", "enum": ["add", "replace", "remove"], - "description": "The action to perform (single-op shape). Omit when using 'operations'." + "description": "The action to perform." }, "target": { "type": "string", @@ -1100,31 +556,14 @@ MEMORY_SCHEMA = { }, "content": { "type": "string", - "description": "The entry content. Required for 'add' and 'replace' (single-op shape)." + "description": "The entry content. Required for 'add' and 'replace'." }, "old_text": { "type": "string", - "description": "REQUIRED for 'replace' and 'remove' (single-op shape): a short unique substring identifying the existing entry to modify. Omit only for 'add'." - }, - "operations": { - "type": "array", - "description": ( - "Batch shape: a list of operations applied atomically in one call " - "against the final char budget. Preferred when making multiple changes " - "or consolidating to make room. Each item is {action, content?, old_text?}." - ), - "items": { - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["add", "replace", "remove"]}, - "content": {"type": "string", "description": "Entry content for add/replace."}, - "old_text": {"type": "string", "description": "Substring identifying the entry for replace/remove."}, - }, - "required": ["action"], - }, + "description": "Short unique substring identifying the entry to replace or remove." }, }, - "required": ["target"], + "required": ["action", "target"], }, } @@ -1141,7 +580,6 @@ registry.register( target=args.get("target", "memory"), content=args.get("content"), old_text=args.get("old_text"), - operations=args.get("operations"), store=kw.get("store")), check_fn=check_memory_requirements, emoji="🧠",