Nick
8b3665f468
The checkout was missing the dedupe+supersede gate (retirement-map row 24, parity/memory-ops.json): 'No module named dedupe_policy' on every default add; only --no-dedupe worked. Module restored with the 2026-08-13 policy (created/duplicate/supersede + receipts ledger), pinned red-first by tests/test_dedupe_policy.py (6 tests incl. consumer-contract + one-record law); live CLI add now returns receipted decisions.
289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""scripts/dedupe_policy.py — the repo copy of the dedupe + supersede gate.
|
|
|
|
memory_store.py's `add` CLI imports `dedupe_policy` (the gate) but the
|
|
module was absent from this checkout: the default add path failed with
|
|
"No module named dedupe_policy" and only --no-dedupe worked (recorded in
|
|
verification/retirement-map.json and parity/memory-ops.json). The module
|
|
is restored here, satisfying the consumer's contract:
|
|
|
|
dedupe_apply(db, text, topic, importance, source, confidence)
|
|
-> {"decision": "created"|"duplicate"|"superseded",
|
|
"id": ..., "prior_id": ..., "receipt": {"reason": ...}}
|
|
|
|
Policy (2026-08-13 rule): one record per data point. Exact or
|
|
semantically-similar text with no new information -> duplicate (touch
|
|
last_accessed). Similar text with NEW info (meaningfully longer or
|
|
stronger confidence) -> supersede (old retired with valid_until, new
|
|
active). Every decision is receipted into the dedupe_receipts ledger.
|
|
"""
|
|
|
|
import hashlib
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
# Similarity threshold: above this, two texts describe the same data point.
|
|
SIM_THRESHOLD = 0.72
|
|
# A supersede is justified when the new fact is meaningfully more specific
|
|
# or more confident than the existing one (new info replaces old).
|
|
SUPERSEDE_LENGTH_BONUS = 20
|
|
|
|
|
|
def now_iso():
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def _norm(text):
|
|
t = re.sub(r"^[-*#>\s]+", "", text.strip())
|
|
t = re.sub(r"[^a-z0-9 ]", " ", t.lower())
|
|
return re.sub(r"\s+", " ", t).strip()
|
|
|
|
|
|
def _embed_pair(a, b):
|
|
"""Best-effort embeddings for cosine similarity. Falls back to token
|
|
overlap when no embedder is available (test/CLI contexts)."""
|
|
try:
|
|
import memory_store as ms
|
|
|
|
vecs = ms._embed([a, b])
|
|
va, vb = vecs.get(a), vecs.get(b)
|
|
if va and vb and len(va) == len(vb):
|
|
dot = sum(x * y for x, y in zip(va, vb))
|
|
na = (sum(x * x for x in va)) ** 0.5 or 1
|
|
nb = (sum(y * y for y in vb)) ** 0.5 or 1
|
|
return dot / (na * nb)
|
|
except Exception:
|
|
pass
|
|
ta = set(_norm(a).split())
|
|
tb = set(_norm(b).split())
|
|
if not ta or not tb:
|
|
return 0.0
|
|
return len(ta & tb) / len(ta | tb)
|
|
|
|
|
|
def _find_similar(db, text, topic=None):
|
|
"""Return (entry_row, score) for the best existing active entry on the
|
|
same data point, or None."""
|
|
norm = _norm(text)
|
|
if not norm:
|
|
return None
|
|
row = db.execute(
|
|
"SELECT id, text, topic, confidence, valid_until FROM entries "
|
|
"WHERE status='active' AND lower(trim(text))=lower(trim(?)) "
|
|
"ORDER BY id LIMIT 1",
|
|
(text.strip(),),
|
|
).fetchone()
|
|
if row:
|
|
return (row, 1.0)
|
|
if topic:
|
|
cands = db.execute(
|
|
"SELECT id, text, topic, confidence, valid_until FROM entries "
|
|
"WHERE status='active' AND topic=? ORDER BY last_accessed DESC "
|
|
"LIMIT 60",
|
|
(topic,),
|
|
).fetchall()
|
|
else:
|
|
cands = db.execute(
|
|
"SELECT id, text, topic, confidence, valid_until FROM entries "
|
|
"WHERE status='active' ORDER BY last_accessed DESC LIMIT 60"
|
|
).fetchall()
|
|
best, best_score = None, 0.0
|
|
for c in cands:
|
|
s = _embed_pair(text, c[1])
|
|
if s > best_score:
|
|
best, best_score = c, s
|
|
if best and best_score >= SIM_THRESHOLD:
|
|
return (best, best_score)
|
|
return None
|
|
|
|
|
|
def _receipt(db, decision, text, topic, evidence, reason, prior_id=None):
|
|
ts = now_iso()
|
|
record = {
|
|
"when": ts,
|
|
"decision": decision,
|
|
"topic": topic,
|
|
"text": text[:200],
|
|
"prior_id": prior_id,
|
|
"evidence": (evidence or "")[:200],
|
|
"reason": reason[:200],
|
|
"hash": hashlib.sha256(f"{ts}|{decision}|{topic}|{text}".encode()).hexdigest()[:16],
|
|
}
|
|
try:
|
|
db.execute(
|
|
"CREATE TABLE IF NOT EXISTS dedupe_receipts ("
|
|
"id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, decision TEXT, "
|
|
"topic TEXT, text TEXT, prior_id INTEGER, evidence TEXT, "
|
|
"reason TEXT, hash TEXT)"
|
|
)
|
|
db.execute(
|
|
"INSERT INTO dedupe_receipts (ts, decision, topic, text, "
|
|
"prior_id, evidence, reason, hash) VALUES (?,?,?,?,?,?,?,?)",
|
|
(
|
|
record["when"],
|
|
record["decision"],
|
|
record["topic"],
|
|
record["text"],
|
|
record["prior_id"],
|
|
record["evidence"],
|
|
record["reason"],
|
|
record["hash"],
|
|
),
|
|
)
|
|
db.commit()
|
|
except Exception:
|
|
pass
|
|
return record
|
|
|
|
|
|
def _add_via_store(db, text, importance, topic, source, confidence):
|
|
try:
|
|
import memory_store as ms
|
|
|
|
return ms.add_entry(
|
|
db, text, importance, topic or "", source=source,
|
|
confidence=confidence, method="dedupe_policy",
|
|
)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def apply(db, text, topic="", importance=0.5, source="", confidence=None):
|
|
"""The dedupe policy gate. Returns the decision taken:
|
|
{"decision": "created" | "duplicate" | "superseded",
|
|
"id": <entry id>, "receipt": {...}}
|
|
"""
|
|
text = (text or "").strip()
|
|
if not text:
|
|
return {"decision": "rejected", "reason": "empty text"}
|
|
hit = _find_similar(db, text, topic or None)
|
|
if not hit:
|
|
ok = _add_via_store(db, text, importance, topic, source, confidence)
|
|
if not ok:
|
|
return {"decision": "rejected", "reason": "insert failed"}
|
|
eid = db.execute(
|
|
"SELECT id FROM entries WHERE text=? ORDER BY id DESC LIMIT 1",
|
|
(text,),
|
|
).fetchone()
|
|
rec = _receipt(db, "created", text, topic, source,
|
|
"first record for this data point",
|
|
prior_id=eid[0] if eid else None)
|
|
return {"decision": "created", "id": eid[0] if eid else None, "receipt": rec}
|
|
|
|
existing, sim = hit
|
|
eid, etxt, econf = existing[0], existing[1], existing[3]
|
|
new_longer = len(text) - len(etxt) >= SUPERSEDE_LENGTH_BONUS
|
|
new_stronger = (confidence or 0) > (econf or 0) + 0.15
|
|
if not new_longer and not new_stronger:
|
|
try:
|
|
db.execute("UPDATE entries SET last_accessed=? WHERE id=?",
|
|
(now_iso(), eid))
|
|
db.commit()
|
|
except Exception:
|
|
pass
|
|
rec = _receipt(db, "duplicate", text, topic, source,
|
|
f"same data point (sim {sim:.2f}); no new info — "
|
|
f"touched existing #{eid}")
|
|
return {"decision": "duplicate", "id": eid, "receipt": rec}
|
|
|
|
try:
|
|
db.execute("UPDATE entries SET status='superseded', valid_until=? "
|
|
"WHERE id=?", (now_iso(), eid))
|
|
db.execute("UPDATE entries_vec SET embedding=NULL WHERE rowid=?", (eid,))
|
|
db.commit()
|
|
except Exception:
|
|
pass
|
|
ok = _add_via_store(db, text, importance, topic, source, confidence)
|
|
if not ok:
|
|
return {"decision": "rejected", "reason": "insert failed", "prior_id": eid}
|
|
rec = _receipt(db, "superseded", text, topic, source,
|
|
f"new info replaces #{eid} (sim {sim:.2f}, "
|
|
f"+{len(text)-len(etxt)} chars, conf "
|
|
f"{confidence or 0} vs {econf or 0})", prior_id=eid)
|
|
new_id = db.execute(
|
|
"SELECT id FROM entries WHERE text=? ORDER BY id DESC LIMIT 1",
|
|
(text,),
|
|
).fetchone()
|
|
return {"decision": "superseded", "id": new_id[0] if new_id else None,
|
|
"prior_id": eid, "receipt": rec}
|
|
|
|
|
|
def receipts(db_path):
|
|
try:
|
|
db = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
|
|
rows = db.execute(
|
|
"SELECT ts, decision, topic, text, prior_id, reason, hash "
|
|
"FROM dedupe_receipts ORDER BY id DESC LIMIT 50"
|
|
).fetchall()
|
|
db.close()
|
|
return rows
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _connect_repo():
|
|
"""Connect through the repo's own memory_store (src/rune), preferring
|
|
RUNE_STORE_DB when set."""
|
|
here = Path(__file__).resolve()
|
|
for cand in (here.parents[1] / "src" / "rune", here.parents[1] / "src"):
|
|
if (cand / "memory_store.py").exists() or cand.name == "rune":
|
|
if str(cand) not in sys.path:
|
|
sys.path.insert(0, str(cand))
|
|
break
|
|
import memory_store as ms
|
|
|
|
return ms.connect()
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
import json as _json
|
|
import os
|
|
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("cmd", choices=["add", "check", "receipts"])
|
|
ap.add_argument("text", nargs="*", default=[])
|
|
ap.add_argument("--topic", default="")
|
|
ap.add_argument("--importance", type=float, default=0.5)
|
|
ap.add_argument("--source", default="")
|
|
ap.add_argument("--confidence", type=float, default=None)
|
|
args = ap.parse_args()
|
|
|
|
db = _connect_repo()
|
|
if args.cmd == "receipts":
|
|
db_path = os.environ.get("RUNE_STORE_DB") or str(
|
|
Path.home() / ".local/share/rune/memory/store/memory.db")
|
|
rows = receipts(db_path)
|
|
if not rows:
|
|
print("(no dedupe receipts yet — the ledger is empty)")
|
|
return
|
|
for r in rows:
|
|
print(f"[{r[0]}] {r[1]:10s} {r[2]:10s} #{(r[3] or '')[:50]} "
|
|
f"prior=#{r[4]} :: {(r[5] or '')[:80]}")
|
|
return
|
|
|
|
text = " ".join(args.text).strip()
|
|
if not text:
|
|
print("empty text")
|
|
sys.exit(1)
|
|
if args.cmd == "check":
|
|
hit = _find_similar(db, text, args.topic or None)
|
|
if not hit:
|
|
print("VERDICT: new")
|
|
elif hit[1] >= 1.0:
|
|
print(f"VERDICT: duplicate (exact) of #{hit[0][0]}")
|
|
else:
|
|
print(f"VERDICT: similar (sim {hit[1]:.2f}) to #{hit[0][0]}: "
|
|
f"{hit[0][1][:60]}")
|
|
else:
|
|
res = apply(db, text, args.topic, args.importance, args.source,
|
|
args.confidence)
|
|
print(_json.dumps(res, indent=2, default=str))
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|