#!/usr/bin/env python3 """drift_compression.py — Drift-aware compression. Compression is proportional to wash count: - High wash_count (proven): less compression, preserve detail - Low wash_count (untested): more compression, might get washed anyway - Rock status: no compression (immutable) This ensures proven knowledge retains its detail while untested knowledge gets compressed aggressively. Research basis: - AMV-L (arXiv:2603.04443): value-driven lifecycle management - SmartVector (arXiv:2604.20598): progressive embedding compression Run under ~/.venvs/memory/bin/python3. """ import math import os import sys from typing import Optional SCRIPTS_DIR = os.path.expanduser("~/.config/opencode/scripts") sys.path.insert(0, SCRIPTS_DIR) # Compression ratios by rock_status COMPRESSION_ROCK = 0.0 # no compression COMPRESSION_ESTABLISHED = 0.2 # 20% compression COMPRESSION_WASHABLE = 0.5 # 50% compression (default) # Wash count modifiers (reduce compression as wash_count increases) WASH_THRESHOLD_FULL = 20 # above this: minimal compression WASH_THRESHOLD_PARTIAL = 10 # above this: reduced compression def get_compression_ratio(wash_count: int = 0, rock_status: str = "washable") -> float: """Get the compression ratio for an entry. Returns 0.0-1.0 where 0.0 = no compression, 1.0 = maximum compression. """ # Rock status overrides everything if rock_status == "rock": return COMPRESSION_ROCK if rock_status == "established": return COMPRESSION_ESTABLISHED # Base compression for washable entries base = COMPRESSION_WASHABLE # Reduce compression based on wash_count if wash_count >= WASH_THRESHOLD_FULL: # Proven entries: minimal compression return max(0.1, base * 0.2) elif wash_count >= WASH_THRESHOLD_PARTIAL: # Partially proven: reduced compression return max(0.2, base * 0.5) else: # Unproven: full compression return base def compress_text(text: str, ratio: float) -> str: """Compress text by the given ratio. Strategy: take first (1-ratio) proportion of the text. For ratio=0.5, take first 50%. For ratio=0.0, return full text. For ratio=1.0, return empty string. """ if ratio <= 0.0: return text if ratio >= 1.0: return "" keep_chars = int(len(text) * (1.0 - ratio)) if keep_chars < 20: keep_chars = min(20, len(text)) compressed = text[:keep_chars] if len(compressed) < len(text): compressed = compressed.rstrip() + "..." return compressed def compress_entry(entry: dict) -> str: """Compress an entry's text based on its wash_count and rock_status. Returns the compressed text. """ text = entry.get("text", "") wash_count = entry.get("wash_count", 0) or 0 rock_status = entry.get("rock_status", "washable") ratio = get_compression_ratio(wash_count, rock_status) return compress_text(text, ratio) def should_warn(entry: dict) -> bool: """Check if compressing this entry should trigger a warning. Warn when: - Entry has wash_count > 0 (proven content being compressed) - Entry is established or rock status (important content) - Entry text is long (>200 chars) and would lose significant detail """ wash_count = entry.get("wash_count", 0) or 0 rock_status = entry.get("rock_status", "washable") text = entry.get("text", "") # Proven content should warn before compression if wash_count >= 10: return True # Established/rock content should warn if rock_status in ("established", "rock"): return True # Long content losing significant detail ratio = get_compression_ratio(wash_count, rock_status) if len(text) > 200 and ratio > 0.3: return True return False def warn_before_compress(entry: dict) -> dict: """Generate a warning before compressing an entry. Returns dict with warning details and recommendation. """ text = entry.get("text", "") wash_count = entry.get("wash_count", 0) or 0 rock_status = entry.get("rock_status", "washable") ratio = get_compression_ratio(wash_count, rock_status) compressed = compress_text(text, ratio) chars_lost = len(text) - len(compressed) warning = { "should_warn": should_warn(entry), "entry_id": entry.get("id"), "wash_count": wash_count, "rock_status": rock_status, "compression_ratio": ratio, "chars_lost": chars_lost, "original_length": len(text), "compressed_length": len(compressed), } if rock_status == "rock": warning["recommendation"] = "DO NOT COMPRESS — rock status, immutable" elif wash_count >= 20: warning["recommendation"] = "CONSIDER PROMOTION before compression — high wash count" elif wash_count >= 10: warning["recommendation"] = "LOW compression recommended — proven content" else: warning["recommendation"] = "OK to compress — untested content" return warning def get_compression_stats(db) -> dict: """Get compression statistics for the current store.""" rows = db.execute( "SELECT rock_status, wash_count, LENGTH(text) as text_len FROM entries WHERE status='active'" ).fetchall() total_entries = len(rows) total_chars = sum(r[2] or 0 for r in rows) compressed_chars = 0 for row in rows: rock_status = row[0] or "washable" wash_count = row[1] or 0 text_len = row[2] or 0 ratio = get_compression_ratio(wash_count, rock_status) compressed_chars += int(text_len * (1.0 - ratio)) savings = total_chars - compressed_chars savings_pct = (savings / total_chars * 100) if total_chars > 0 else 0 return { "total_entries": total_entries, "total_chars": total_chars, "compressed_chars": compressed_chars, "savings_chars": savings, "savings_pct": round(savings_pct, 2), } # ═══════════════════════════════════════════════════════════════════════════════ # CLI # ═══════════════════════════════════════════════════════════════════════════════ def main(): import argparse parser = argparse.ArgumentParser(description="Drift-aware compression") sub = parser.add_subparsers(dest="cmd") # ratio p_ratio = sub.add_parser("ratio", help="Get compression ratio for an entry") p_ratio.add_argument("entry_id", type=int) # stats sub.add_parser("stats", help="Get compression statistics") # compress p_compress = sub.add_parser("compress", help="Compress an entry") p_compress.add_argument("entry_id", type=int) args = parser.parse_args() if not args.cmd: parser.print_help() sys.exit(1) import memory_store db = memory_store.connect() if args.cmd == "ratio": row = db.execute( "SELECT wash_count, rock_status FROM entries WHERE id = ?", (args.entry_id,) ).fetchone() if row: ratio = get_compression_ratio(row[0] or 0, row[1] or "washable") print(f"Entry {args.entry_id}: compression_ratio={ratio:.2f}") print(f" wash_count={row[0] or 0}") print(f" rock_status={row[1] or 'washable'}") else: print(f"Entry {args.entry_id} not found") elif args.cmd == "stats": stats = get_compression_stats(db) print("=== Compression Statistics ===") print(f"Total entries: {stats['total_entries']}") print(f"Total chars: {stats['total_chars']:,}") print(f"Compressed chars: {stats['compressed_chars']:,}") print(f"Savings: {stats['savings_chars']:,} chars ({stats['savings_pct']:.1f}%)") elif args.cmd == "compress": row = db.execute( "SELECT id, text, wash_count, rock_status FROM entries WHERE id = ?", (args.entry_id,) ).fetchone() if row: entry = {"text": row[1], "wash_count": row[2] or 0, "rock_status": row[3] or "washable"} compressed = compress_entry(entry) print(f"Original ({len(row[1])} chars): {row[1][:100]}...") print(f"Compressed ({len(compressed)} chars): {compressed[:100]}") else: print(f"Entry {args.entry_id} not found") db.close() if __name__ == "__main__": main()