260 lines
8.5 KiB
Python
260 lines
8.5 KiB
Python
#!/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 <entry_id>
|
|
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 <entry_id>
|
|
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()
|