hermes-recovery/scripts/hermes-consolidate.py

#!/usr/bin/env python3
"""hermes-consolidate.py — Dump/prune Hermes memory.

Usage:
  hermes-consolidate.py --dump-only   # dump MEMORY.md entries to JSON (stdout)
  hermes-consolidate.py --prune-facts # prune MEMORY.md + holographic fact store
"""
import json, os, re, sys, sqlite3, time
from pathlib import Path

HERMES_HOME = Path(os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")))
MEMORY_MD = HERMES_HOME / "memories" / "MEMORY.md"
FACT_DB = HERMES_HOME / "memory_store.db"
TARGET = 5000  # target chars for MEMORY.md after prune
PROTECT = re.compile(
    r"(Germaine's #1 rule|never make up|fabrication|dealbreaker|"
    r"Sho'Nuff Brown|shonuff@|germainebrown\\.com|itpp-infra|"
    r"DR issue log|NETCUP_PASSWORD|LoveMyBoys|"
    r"Core: netcup|STANDING PRACTICE|app1-bu|"
    r"admin-ai\\.itpropartner|RS 4000|"
    r"EMAIL: proper Sho'Nuff formatting|EMAIL FORMATTING RULE|"
    r"#1 rule|Germaine's email|Super Search|Exa API key)"
)


def load_entries():
    raw = MEMORY_MD.read_text()
    entries = [e.strip() for e in raw.split("\n§\n") if e.strip()]
    return entries


def save_entries(entries):
    text = "\n§\n".join(entries) + "\n"
    tmp = MEMORY_MD.with_suffix(".md.tmp")
    tmp.write_text(text)
    tmp.replace(MEMORY_MD)


def dump_only():
    entries = load_entries()
    char_count = sum(len(e) for e in entries)
    data = {
        "backed_up_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "source": str(MEMORY_MD),
        "char_limit": 10000,
        "target": TARGET,
        "entry_count": len(entries),
        "char_count": char_count,
        "entries": entries,
    }
    json.dump(data, sys.stdout, indent=2)
    print()


def prune_entries(entries):
    """Pattern-prune + size safety valve. Returns kept entries."""
    # Pattern pruning: drop obviously stale entries
    stale_patterns = [
        r"\bcomplete\b",
        r"\bdone\b",
        r"\bshipped\b",
        r"\bresolved\b",
        r"\[STALE\]",
        r"\[EXPIRED\]",
        r"\[OBSOLETE\]",
    ]

    kept = []
    for e in entries:
        # Always keep PROTECTed entries
        if PROTECT.search(e):
            kept.append(e)
            continue

        # Drop if matches any stale pattern (single-line quick check)
        is_stale = False
        for pat in stale_patterns:
            if re.search(pat, e, re.IGNORECASE):
                is_stale = True
                break

        # Also check for task-completion indicators
        if re.search(r"^(✅|❌|complete|done)", e.strip()):
            is_stale = True

        if not is_stale:
            kept.append(e)

    # Size safety valve: if still over target, drop oldest entries
    char_count = sum(len(e) for e in kept)
    if char_count > TARGET:
        # Drop from the top (oldest-appended) first, but never PROTECTed
        safe = [e for e in kept if PROTECT.search(e)]
        pruneable = [e for e in kept if not PROTECT.search(e)]

        while pruneable and char_count > TARGET:
            removed = pruneable.pop(0)  # oldest first
            char_count -= len(removed)

        kept = safe + pruneable

    return kept


def prune_fact_store():
    """Remove facts older than 7 days that were never retrieved and have trust < 0.5."""
    try:
        conn = sqlite3.connect(str(FACT_DB))
        c = conn.cursor()
        cutoff = time.time() - (7 * 86400)
        c.execute(
            """DELETE FROM facts
               WHERE created_at < ?
                 AND retrieval_count = 0
                 AND trust_score < 0.5""",
            (cutoff,),
        )
        deleted = c.rowcount
        conn.commit()
        conn.close()
        return deleted
    except (sqlite3.Error, FileNotFoundError):
        return 0


def prune_facts():
    """Main prune entry point."""
    entries = load_entries()
    if not entries:
        print("RESULT:OK|No entries to prune")
        return

    original_count = len(entries)
    original_chars = sum(len(e) for e in entries)

    kept = prune_entries(entries)

    if not kept:
        print("RESULT:FAIL|Prune would result in empty memory — ABORTED")
        sys.exit(1)

    save_entries(kept)
    kept_count = len(kept)
    kept_chars = sum(len(e) for e in kept)

    # Prune fact store
    facts_deleted = prune_fact_store()

    print(f"RESULT:OK|Entries: {original_count}→{kept_count}, "
          f"Chars: {original_chars}→{kept_chars}, "
          f"Facts pruned: {facts_deleted}")


if __name__ == "__main__":
    if "--dump-only" in sys.argv:
        dump_only()
    elif "--prune-facts" in sys.argv:
        prune_facts()
    else:
        print("Usage: hermes-consolidate.py --dump-only | --prune-facts", file=sys.stderr)
        sys.exit(1)