gemsearch.py — a deterministic dedup gate so your AI agent never repeats the same 'knowledge gem' web search twice

gemsearch.py

#!/usr/bin/env python3
"""gemsearch.py - deterministic dedup ledger for retro "knowledge gems".

Problem it solves: the retro gem step kept re-surfacing the same topics
(chezmoi encryption, gitleaks hook, ...) because it had no memory of what was
already covered. This tool is that memory - a SQLite ledger with fuzzy dedup.

The agent NEVER reads the whole ledger into context. It proposes a topic and
calls `claim`; the tool either records it (exit 0) or REFUSES it as a duplicate
(exit 2, printing the prior match). Only the refusal returns to context, so a
repeat costs one short line, not the whole history.

Usage:
  gemsearch.py search "<topic>" [--num N] [--note N] [--segment S]
        PRIMARY one-step command for retros: gate for novelty, and if novel,
        run the web search, store the results, and print them. Exit 2 if the
        topic was already covered (pick a farther-afield topic and retry).
  gemsearch.py claim "<topic>" [--url U] [--note N] [--segment S] [--date D]
        Record a gem without searching: refuse if duplicate (exit 2), else insert.
  gemsearch.py check "<topic>"
        Dry-run: exit 0 if novel, exit 2 if duplicate. No write.
  gemsearch.py list [--recent N] [--json]   Human review of covered gems.
  gemsearch.py import <file.md|->           Bulk-add topics (line list or md table).
  gemsearch.py export [--out PATH]          Regenerate human-readable md mirror.
  gemsearch.py stats                        Count + newest/oldest.

Search backend: Exa API if $EXA_API_KEY set, else ~/tools/search.py (ddgr).
DB: ~/Notes/knowledge-gems.db  (override with $GEMSEARCH_DB)
Duplicate = exact normalized match, subset of tokens, or SequenceMatcher>=0.82
against any existing signature. Conservative + deterministic (no false blocks).
Upgrade path for semantic near-dups: reuse OMP mnemopi's embedding model.
"""
import argparse
import datetime as _dt
import json
import os
import re
import shutil
import sqlite3
import subprocess
import sys
from difflib import SequenceMatcher

DB_PATH = os.environ.get(
    "GEMSEARCH_DB",
    os.path.join(os.path.expanduser("~"), "Notes", "knowledge-gems.db"),
)

# words too generic to carry dedup signal
_STOP = {
    "the", "a", "an", "of", "for", "to", "in", "on", "and", "or", "with",
    "how", "what", "why", "is", "are", "vs", "via", "using", "use", "tool",
    "technique", "guide", "tutorial", "intro", "introduction", "overview",
}
JACCARD_DUP = 0.55
OVERLAP_DUP = 0.6    # shared distinctive tokens / smaller signature
SEQRATIO_DUP = 0.82
ANCHOR_DF_ABS = 8    # a token in <= this many rows is "distinctive"...
ANCHOR_DF_FRAC = 0.03  # ...or <= this fraction of the corpus, whichever is larger

# Low-signal domains to drop from results (mailing lists, trackers, listicles) —
# Ankit wants deep-dives/blogs/primary sources, not these.
LOW_SIGNAL_DOMAINS = (
    "lists.", "marc.info", "mail-archive.com", "groups.google.com",
    "pipermail", "sourceforge.net/p/", "narkive.com", "quora.com",
    "pinterest.", "slideshare.net", "w3schools.com",
)


def _norm_tokens(text):
    text = text.lower()
    text = re.sub(r"[^a-z0-9\s]", " ", text)
    toks = [t for t in text.split() if t and t not in _STOP]
    return toks


def _signature(text):
    return " ".join(sorted(set(_norm_tokens(text))))


def _connect():
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    con = sqlite3.connect(DB_PATH)
    con.execute(
        """CREATE TABLE IF NOT EXISTS gems (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            topic TEXT NOT NULL,
            signature TEXT NOT NULL,
            note TEXT,
            url TEXT,
            segment TEXT,
            date TEXT NOT NULL,
            results TEXT
        )"""
    )
    # migrate older DBs lacking the results column
    cols = {r[1] for r in con.execute("PRAGMA table_info(gems)").fetchall()}
    if "results" not in cols:
        con.execute("ALTER TABLE gems ADD COLUMN results TEXT")
    con.commit()
    return con


def _find_dup(con, topic):
    """Return the first existing row that duplicates `topic`, else None.

    CONSERVATIVE by design — only high-confidence signals, so it never blocks a
    genuinely-novel topic (false positives are worse than the occasional miss):
      1. identical normalized signature
      2. subset either direction ("chezmoi" vs "chezmoi encryption")
      3. SequenceMatcher ratio >= SEQRATIO_DUP on the sorted signature
    Semantic near-duplicates that share few tokens but the same meaning
    (e.g. "SONiC architecture" vs "SONiC dataplane deep-dive") are NOT caught
    here — that needs embeddings; see the mnemopi-backed upgrade path in the
    module docstring. Loose token heuristics (Jaccard/overlap/IDF-anchor) were
    removed after they over-blocked unrelated topics sharing a common word.
    """
    sig = _signature(topic)
    sig_set = set(sig.split())
    if not sig_set:
        sig_set = set(_norm_tokens(topic))
    for row in con.execute(
        "SELECT id, topic, signature, url, date FROM gems"
    ).fetchall():
        ex_set = set(row[2].split())
        if not ex_set:
            continue
        if sig == row[2]:
            return row
        if sig_set and (sig_set <= ex_set or ex_set <= sig_set):
            return row
        if SequenceMatcher(None, sig, row[2]).ratio() >= SEQRATIO_DUP:
            return row
    return None


def _fmt(row):
    # row: id, topic, signature, url, date  (or subset)
    rid, topic = row[0], row[1]
    url = row[3] if len(row) > 3 else ""
    date = row[4] if len(row) > 4 else ""
    tail = f"  [{date}]" if date else ""
    tail += f"  {url}" if url else ""
    return f"#{rid} {topic}{tail}"


def cmd_claim(args, check_only=False):
    con = _connect()
    dup = _find_dup(con, args.topic)
    if dup:
        print(f"DUPLICATE - already covered: {_fmt(dup)}", file=sys.stderr)
        print("Pick a genuinely new topic (go farther afield).", file=sys.stderr)
        return 2
    if check_only:
        print(f"NOVEL - '{args.topic}' is not in the ledger.")
        return 0
    date = args.date or _dt.date.today().isoformat()
    con.execute(
        "INSERT INTO gems(topic, signature, note, url, segment, date)"
        " VALUES(?,?,?,?,?,?)",
        (args.topic, _signature(args.topic), args.note, args.url,
         args.segment, date),
    )
    con.commit()
    print(f"RECORDED - {args.topic}")
    return 0


def cmd_check(args):
    return cmd_claim(args, check_only=True)


def _web_search(query, num=10):
    """Run the actual search via the best available backend.

    Preference order (DRW - reuse existing infra):
      1. Exa API if $EXA_API_KEY is set (best quality)
      2. ~/tools/search.py (ddgr) - already installed, always works on AMD net
    Returns (backend_name, text) or (None, error_message).
    """
    key = os.environ.get("EXA_API_KEY")
    if key:
        try:
            import urllib.request
            body = json.dumps({
                "query": query, "numResults": num, "type": "auto",
                "contents": {"text": {"maxCharacters": 600},
                             "highlights": {"numSentences": 2}},
            }).encode()
            req = urllib.request.Request(
                "https://api.exa.ai/search", data=body,
                headers={"x-api-key": key, "Content-Type": "application/json"})
            with urllib.request.urlopen(req, timeout=30) as resp:
                data = json.load(resp)
            lines = []
            for r in data.get("results", []):
                url = r.get("url", "")
                if any(d in url for d in LOW_SIGNAL_DOMAINS):
                    continue
                title = r.get("title") or url
                snippet = (r.get("text") or " ".join(r.get("highlights", [])) or "").strip()
                lines.append(f"- {title}\n  {url}\n  {snippet[:280]}")
            return "exa", "\n".join(lines) if lines else "(no results)"
        except Exception as e:  # fall through to ddgr
            pass
    search_py = os.path.join(os.path.dirname(os.path.abspath(__file__)), "search.py")
    if os.path.exists(search_py):
        try:
            out = subprocess.run(
                [sys.executable, search_py, query, "--num", str(num)],
                capture_output=True, text=True, timeout=45)
            text = out.stdout.strip()
            # drop low-signal lines
            kept = [ln for ln in text.splitlines()
                    if not any(d in ln for d in LOW_SIGNAL_DOMAINS)]
            return "ddgr", "\n".join(kept) if kept else text
        except Exception as e:
            return None, f"search backend failed: {e}"
    return None, "no search backend available (set EXA_API_KEY or install ~/tools/search.py)"


def cmd_search(args):
    """One-step: gate for novelty, and if novel, SEARCH + store + return results.

    This is the command retros should call. If the topic was already covered it
    exits 2 and the agent must pick a farther-afield topic and call again.
    """
    con = _connect()
    dup = _find_dup(con, args.topic)
    if dup:
        print(f"DUPLICATE - already covered: {_fmt(dup)}", file=sys.stderr)
        print("Pick a genuinely new/farther-afield topic and search again.",
              file=sys.stderr)
        return 2
    backend, results = _web_search(args.topic, args.num)
    if backend is None:
        print(results, file=sys.stderr)
        return 1
    date = args.date or _dt.date.today().isoformat()
    con.execute(
        "INSERT INTO gems(topic, signature, note, url, segment, date, results)"
        " VALUES(?,?,?,?,?,?,?)",
        (args.topic, _signature(args.topic), args.note, args.url,
         args.segment, date, results),
    )
    con.commit()
    print(f"NOVEL - recorded & searched via {backend}:\n")
    print(results)
    return 0


def cmd_list(args):
    con = _connect()
    q = "SELECT id, topic, signature, url, date FROM gems ORDER BY date DESC, id DESC"
    rows = con.execute(q).fetchall()
    if args.recent:
        rows = rows[: args.recent]
    if args.json:
        print(json.dumps(
            [{"id": r[0], "topic": r[1], "url": r[3], "date": r[4]} for r in rows],
            indent=2))
        return 0
    for r in rows:
        print(_fmt(r))
    return 0


def _iter_import_lines(text):
    """Yield (topic, url, date) from plain lines or a markdown table."""
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("|"):
            cells = [c.strip() for c in line.strip("|").split("|")]
            # skip header / separator rows
            if not cells or cells[0].lower() in ("date", "") or set(cells[0]) <= set("-: "):
                continue
            # heuristic: table is date|topic|one-line|url|segment
            date = cells[0] if re.match(r"\d{4}-\d{2}-\d{2}", cells[0]) else None
            topic = cells[1] if len(cells) > 1 else cells[0]
            url = ""
            for c in cells:
                if c.startswith("http"):
                    url = c
                    break
            yield topic, url, date
        else:
            yield line, "", None


def cmd_import(args):
    con = _connect()
    if args.file == "-":
        text = sys.stdin.read()
    else:
        with open(os.path.expanduser(args.file), encoding="utf-8") as fh:
            text = fh.read()
    # Bulk import dedups on EXACT signature only (in-memory set -> O(n));
    # fuzzy dedup is reserved for the interactive `claim` path.
    seen = {r[0] for r in con.execute("SELECT signature FROM gems").fetchall()}
    added = skipped = 0
    batch = []
    for topic, url, date in _iter_import_lines(text):
        if not topic:
            continue
        sig = _signature(topic)
        if not sig or sig in seen:
            skipped += 1
            continue
        seen.add(sig)
        batch.append((topic, sig, None, url, "import",
                      date or _dt.date.today().isoformat()))
        added += 1
    con.executemany(
        "INSERT INTO gems(topic, signature, note, url, segment, date)"
        " VALUES(?,?,?,?,?,?)", batch)
    con.commit()
    print(f"imported {added}, skipped {skipped} (already covered)")
    return 0


def cmd_export(args):
    con = _connect()
    rows = con.execute(
        "SELECT date, topic, note, url, segment FROM gems ORDER BY date, id"
    ).fetchall()
    out = ["# Knowledge Gems - covered ledger (auto-generated by gemsearch.py)",
           "",
           "Do NOT hand-edit; source of truth is knowledge-gems.db.",
           "Regenerate: python ~/tools/gemsearch.py export",
           "",
           "| date | topic | one-line | URL | source-segment |",
           "|------|-------|----------|-----|----------------|"]
    for d, topic, note, url, seg in rows:
        out.append(f"| {d} | {topic} | {note or ''} | {url or ''} | {seg or ''} |")
    text = "\n".join(out) + "\n"
    path = args.out or os.path.join(os.path.dirname(DB_PATH), "knowledge-gems.md")
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    print(f"wrote {len(rows)} gems -> {path}")
    return 0


def cmd_stats(args):
    con = _connect()
    n = con.execute("SELECT COUNT(*) FROM gems").fetchone()[0]
    if not n:
        print("0 gems")
        return 0
    lo = con.execute("SELECT MIN(date) FROM gems").fetchone()[0]
    hi = con.execute("SELECT MAX(date) FROM gems").fetchone()[0]
    print(f"{n} gems  ({lo} .. {hi})  db={DB_PATH}")
    return 0


def main(argv=None):
    p = argparse.ArgumentParser(description="deterministic dedup ledger for retro knowledge gems")
    sub = p.add_subparsers(dest="cmd", required=True)

    def add_topic(sp):
        sp.add_argument("topic")
        sp.add_argument("--url", default=None)
        sp.add_argument("--note", default=None)
        sp.add_argument("--segment", default=None)
        sp.add_argument("--date", default=None)

    sp = sub.add_parser("claim", help="record a gem, refusing duplicates (exit 2)")
    add_topic(sp)
    sp.set_defaults(func=cmd_claim)

    sp = sub.add_parser("search",
                        help="ONE-STEP: gate for novelty, then search+store+return (exit 2 if dup)")
    add_topic(sp)
    sp.add_argument("--num", type=int, default=10, help="number of results")
    sp.set_defaults(func=cmd_search)

    sp = sub.add_parser("check", help="dry-run: novel (0) or duplicate (2)")
    add_topic(sp)
    sp.set_defaults(func=cmd_check)

    sp = sub.add_parser("list", help="print covered gems (human review)")
    sp.add_argument("--recent", type=int, default=0)
    sp.add_argument("--json", action="store_true")
    sp.set_defaults(func=cmd_list)

    sp = sub.add_parser("import", help="bulk-add topics from file or stdin")
    sp.add_argument("file")
    sp.set_defaults(func=cmd_import)

    sp = sub.add_parser("export", help="regenerate markdown mirror")
    sp.add_argument("--out", default=None)
    sp.set_defaults(func=cmd_export)

    sp = sub.add_parser("stats", help="count + date range")
    sp.set_defaults(func=cmd_stats)

    args = p.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())