Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a7e0e7db4 |
@@ -13,9 +13,11 @@ import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
HERMES_HOME = Path.home() / ".hermes"
|
||||
HERMES_HOME = get_hermes_home()
|
||||
CHECKPOINT_DIR = HERMES_HOME / "checkpoints"
|
||||
CHARS_PER_TOKEN = 4
|
||||
|
||||
|
||||
@@ -50,78 +50,6 @@ def sanitize_context(text: str) -> str:
|
||||
return _FENCE_TAG_RE.sub('', text)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prefetch filtering helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Meta-instruction debris that memory providers sometimes echo back.
|
||||
# These are prompts/instructions, not user-generated content.
|
||||
_META_INSTRUCTION_PATTERNS = [
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*Focus on:\s*", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*Note:\s*", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*System\s+(note|prompt|instruction):", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*You are\s+", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*Please\s+(provide|respond|answer|write)", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*Do not\s+", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*Always\s+", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*Consider\s+(the following|these|this)\b", re.IGNORECASE),
|
||||
re.compile(r"^\s*[\-\*]?\s*>?\s*Here\s+(is|are)\s+(some|the|a few)\b", re.IGNORECASE),
|
||||
]
|
||||
|
||||
|
||||
def _is_meta_instruction_line(line: str) -> bool:
|
||||
"""Return True if the line looks like a prompt/template instruction, not memory content."""
|
||||
for pat in _META_INSTRUCTION_PATTERNS:
|
||||
if pat.search(line):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_low_signal_line(line: str) -> bool:
|
||||
"""Return True for very short or content-free lines."""
|
||||
stripped = line.strip()
|
||||
# Empty or just punctuation/list marker
|
||||
if not stripped or stripped in {"-", "*", ">", "•", "—", "--"}:
|
||||
return True
|
||||
# Too short to be meaningful (< 15 chars after stripping markers)
|
||||
cleaned = re.sub(r"^[\-\*•>\s]+", "", stripped)
|
||||
if len(cleaned) < 15:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _filter_prefetch_lines(text: str) -> str:
|
||||
"""Filter and deduplicate prefetch result lines.
|
||||
|
||||
Removes:
|
||||
- exact duplicate lines
|
||||
- meta-instruction debris (prompts, templates)
|
||||
- very short / content-free lines
|
||||
|
||||
Returns cleaned text, preserving original line grouping.
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
return ""
|
||||
|
||||
seen: set = set()
|
||||
kept: list = []
|
||||
for line in text.splitlines(keepends=False):
|
||||
stripped = line.strip()
|
||||
# Deduplicate exact lines
|
||||
if stripped in seen:
|
||||
continue
|
||||
# Skip meta-instructions
|
||||
if _is_meta_instruction_line(line):
|
||||
continue
|
||||
# Skip low-signal lines
|
||||
if _is_low_signal_line(line):
|
||||
continue
|
||||
seen.add(stripped)
|
||||
kept.append(line)
|
||||
|
||||
return "\n".join(kept)
|
||||
|
||||
|
||||
def build_memory_context_block(raw_context: str) -> str:
|
||||
"""Wrap prefetched memory in a fenced block with system note.
|
||||
|
||||
@@ -252,14 +180,7 @@ class MemoryManager:
|
||||
"Memory provider '%s' prefetch failed (non-fatal): %s",
|
||||
provider.name, e,
|
||||
)
|
||||
raw = "\n\n".join(parts)
|
||||
if not raw:
|
||||
return ""
|
||||
# Apply line-level filtering: dedupe, strip meta-instructions,
|
||||
# remove very short fragments. This prevents noisy providers
|
||||
# (e.g. MemPalace transcript recall) from bloating context.
|
||||
filtered = _filter_prefetch_lines(raw)
|
||||
return filtered
|
||||
return "\n\n".join(parts)
|
||||
|
||||
def queue_prefetch_all(self, query: str, *, session_id: str = "") -> None:
|
||||
"""Queue background prefetch on all providers for the next turn."""
|
||||
|
||||
@@ -56,7 +56,7 @@ VIOLATIONS = [
|
||||
"id": "expanduser-hermes",
|
||||
"name": "os.path.expanduser ~/.hermes (non-fallback)",
|
||||
"pattern": r'os\.path\.expanduser\(["\']~/.hermes',
|
||||
"exclude_with": r'#',
|
||||
"exclude_with": r'#|HERMES_HOME',
|
||||
"message": "Use `os.environ.get('HERMES_HOME', os.path.expanduser('~/.hermes'))` instead",
|
||||
},
|
||||
]
|
||||
|
||||
@@ -198,14 +198,14 @@ class TestMemoryManager:
|
||||
def test_prefetch_skips_empty(self):
|
||||
mgr = MemoryManager()
|
||||
p1 = FakeMemoryProvider("builtin")
|
||||
p1._prefetch_result = "This provider has meaningful memories with enough length"
|
||||
p1._prefetch_result = "Has memories"
|
||||
p2 = FakeMemoryProvider("external")
|
||||
p2._prefetch_result = ""
|
||||
mgr.add_provider(p1)
|
||||
mgr.add_provider(p2)
|
||||
|
||||
result = mgr.prefetch_all("query")
|
||||
assert result == "This provider has meaningful memories with enough length"
|
||||
assert result == "Has memories"
|
||||
|
||||
def test_queue_prefetch_all(self):
|
||||
mgr = MemoryManager()
|
||||
@@ -695,92 +695,3 @@ class TestMemoryContextFencing:
|
||||
fence_end = combined.index("</memory-context>")
|
||||
assert "Alice" in combined[fence_start:fence_end]
|
||||
assert combined.index("weather") < fence_start
|
||||
|
||||
|
||||
class TestPrefetchFiltering:
|
||||
"""Tests for _filter_prefetch_lines and related helpers."""
|
||||
|
||||
def test_deduplicates_exact_lines(self):
|
||||
from agent.memory_manager import _filter_prefetch_lines
|
||||
raw = "- This is line one with enough characters\n- This is line two with enough characters\n- This is line one with enough characters\n- This is line three with enough characters"
|
||||
result = _filter_prefetch_lines(raw)
|
||||
lines = [l for l in result.splitlines() if l.strip()]
|
||||
assert len(lines) == 3
|
||||
assert "- This is line one with enough characters" in result
|
||||
assert "- This is line two with enough characters" in result
|
||||
assert "- This is line three with enough characters" in result
|
||||
|
||||
def test_removes_meta_instruction_debris(self):
|
||||
from agent.memory_manager import _filter_prefetch_lines
|
||||
raw = (
|
||||
"## Fleet Memories\n"
|
||||
"- > Focus on: was a non-trivial approach used\n"
|
||||
"- > Focus on: was a non-trivial approach used\n"
|
||||
"- Actual memory content about fleet ops\n"
|
||||
"- Note: this is just a note\n"
|
||||
)
|
||||
result = _filter_prefetch_lines(raw)
|
||||
assert "Focus on" not in result
|
||||
assert "Note:" not in result
|
||||
assert "Actual memory content about fleet ops" in result
|
||||
assert "Fleet Memories" in result
|
||||
|
||||
def test_removes_low_signal_short_lines(self):
|
||||
from agent.memory_manager import _filter_prefetch_lines
|
||||
raw = (
|
||||
"- \n"
|
||||
"- x\n"
|
||||
"- This is a meaningful memory entry with enough length\n"
|
||||
)
|
||||
result = _filter_prefetch_lines(raw)
|
||||
assert "- x" not in result
|
||||
assert "meaningful memory entry" in result
|
||||
|
||||
def test_preserves_structured_facts(self):
|
||||
from agent.memory_manager import _filter_prefetch_lines
|
||||
raw = (
|
||||
"## Local Facts (Hologram)\n"
|
||||
"- ALEXANDER: Prefers Gitea for reports and deliverables.\n"
|
||||
"- Telegram home channel is Timmy Time.\n"
|
||||
)
|
||||
result = _filter_prefetch_lines(raw)
|
||||
assert "ALEXANDER" in result
|
||||
assert "Gitea" in result
|
||||
assert "Telegram" in result
|
||||
|
||||
def test_is_meta_instruction_line(self):
|
||||
from agent.memory_manager import _is_meta_instruction_line
|
||||
assert _is_meta_instruction_line("- > Focus on: something") is True
|
||||
assert _is_meta_instruction_line("- Focus on: something") is True
|
||||
assert _is_meta_instruction_line("* Focus on: something") is True
|
||||
assert _is_meta_instruction_line("- Actual user memory content") is False
|
||||
assert _is_meta_instruction_line("ALEXANDER: Prefers Gitea") is False
|
||||
|
||||
def test_is_low_signal_line(self):
|
||||
from agent.memory_manager import _is_low_signal_line
|
||||
assert _is_low_signal_line("- ") is True
|
||||
assert _is_low_signal_line("*") is True
|
||||
assert _is_low_signal_line("- x") is True
|
||||
assert _is_low_signal_line("- Short line") is True
|
||||
assert _is_low_signal_line("- This is a long meaningful memory entry") is False
|
||||
|
||||
def test_prefetch_all_applies_filtering(self):
|
||||
from agent.memory_manager import MemoryManager
|
||||
mgr = MemoryManager()
|
||||
fake = FakeMemoryProvider(name="test")
|
||||
fake._prefetch_result = (
|
||||
"- > Focus on: was a non-trivial approach\n"
|
||||
"- > Focus on: was a non-trivial approach\n"
|
||||
"- Real memory fact\n"
|
||||
)
|
||||
mgr.add_provider(fake)
|
||||
result = mgr.prefetch_all("query")
|
||||
assert "Focus on" not in result
|
||||
assert "Real memory fact" in result
|
||||
assert result.count("Real memory fact") == 1
|
||||
|
||||
def test_empty_prefetch_returns_empty(self):
|
||||
from agent.memory_manager import _filter_prefetch_lines
|
||||
assert _filter_prefetch_lines("") == ""
|
||||
assert _filter_prefetch_lines(" ") == ""
|
||||
assert _filter_prefetch_lines("\n\n") == ""
|
||||
|
||||
@@ -13,9 +13,11 @@ from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Tuple
|
||||
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
HERMES_HOME = Path.home() / ".hermes"
|
||||
HERMES_HOME = get_hermes_home()
|
||||
AUDIT_DIR = HERMES_HOME / "audit"
|
||||
|
||||
# Credential patterns to detect and redact
|
||||
@@ -32,14 +34,14 @@ CREDENTIAL_PATTERNS = [
|
||||
(r"bearer\s+[a-zA-Z0-9._-]{20,}", "[REDACTED: Bearer token]"),
|
||||
|
||||
# Generic tokens/passwords
|
||||
(r"(?:token|TOKEN|Token)[:=]\s*["']?[a-zA-Z0-9._-]{20,}["']?", "[REDACTED: Token]"),
|
||||
(r"(?:password|PASSWORD|Password)[:=]\s*["']?[^\s"']{8,}["']?", "[REDACTED: Password]"),
|
||||
(r"(?:secret|SECRET|Secret)[:=]\s*["']?[a-zA-Z0-9._-]{20,}["']?", "[REDACTED: Secret]"),
|
||||
(r"(?:api_key|API_KEY|apiKey|ApiKey)[:=]\s*["']?[a-zA-Z0-9._-]{20,}["']?", "[REDACTED: API key]"),
|
||||
("(?:token|TOKEN|Token)[:=]\\s*['\"]?[a-zA-Z0-9._-]{20,}['\"]?", "[REDACTED: Token]"),
|
||||
("(?:password|PASSWORD|Password)[:=]\\s*['\"]?[^\\s\"']{8,}['\"]?", "[REDACTED: Password]"),
|
||||
("(?:secret|SECRET|Secret)[:=]\\s*['\"]?[a-zA-Z0-9._-]{20,}['\"]?", "[REDACTED: Secret]"),
|
||||
("(?:api_key|API_KEY|apiKey|ApiKey)[:=]\\s*['\"]?[a-zA-Z0-9._-]{20,}['\"]?", "[REDACTED: API key]"),
|
||||
|
||||
# AWS keys
|
||||
(r"AKIA[0-9A-Z]{16}", "[REDACTED: AWS access key]"),
|
||||
(r"(?:aws_secret_access_key|AWS_SECRET_ACCESS_KEY)[:=]\s*["']?[a-zA-Z0-9/+=]{40}["']?", "[REDACTED: AWS secret]"),
|
||||
("(?:aws_secret_access_key|AWS_SECRET_ACCESS_KEY)[:=]\\s*['\"]?[a-zA-Z0-9/+=]{40}['\"]?", "[REDACTED: AWS secret]"),
|
||||
|
||||
# Private keys
|
||||
(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----", "[REDACTED: Private key header]"),
|
||||
|
||||
@@ -249,7 +249,8 @@ def detect_crisis(text: str) -> CrisisDetectionResult:
|
||||
# ── Escalation Logging ────────────────────────────────────────────────────
|
||||
|
||||
BRIDGE_URL = os.environ.get("CRISIS_BRIDGE_URL", "")
|
||||
LOG_PATH = os.path.expanduser("~/.hermes/crisis_escalations.jsonl")
|
||||
_HERMES_HOME = os.environ.get("HERMES_HOME")
|
||||
LOG_PATH = os.path.join(_HERMES_HOME or os.path.expanduser("~/.hermes"), "crisis_escalations.jsonl")
|
||||
|
||||
|
||||
def _log_escalation(result: CrisisDetectionResult, text_preview: str = ""):
|
||||
|
||||
@@ -10,10 +10,10 @@ Usage:
|
||||
from tools.hardcoded_path_guard import check_path, validate_tool_args
|
||||
|
||||
# Check a single path
|
||||
err = check_path("/Users/apayne/.hermes/config.yaml")
|
||||
err = check_path("/Users/apayne/.hermes/config.yaml") # noqa: hardcoded-path-ok
|
||||
|
||||
# Validate all path-like args in a tool call
|
||||
clean_args, warnings = validate_tool_args("read_file", {"path": "/home/user/file.txt"})
|
||||
clean_args, warnings = validate_tool_args("read_file", {"path": "/home/user/file.txt"}) # noqa: hardcoded-path-ok
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
@@ -14,9 +14,11 @@ from typing import Dict, List, Optional, Any
|
||||
from dataclasses import dataclass, asdict, field
|
||||
from enum import Enum
|
||||
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TEMPLATE_DIR = Path.home() / ".hermes" / "session-templates"
|
||||
TEMPLATE_DIR = get_hermes_home() / "session-templates"
|
||||
|
||||
|
||||
class TaskType(Enum):
|
||||
@@ -106,7 +108,7 @@ class Templates:
|
||||
return TaskType.MIXED
|
||||
|
||||
def extract(self, session_id, max_n=10):
|
||||
db = Path.home() / ".hermes" / "state.db"
|
||||
db = get_hermes_home() / "state.db"
|
||||
if not db.exists():
|
||||
return []
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user