/notebook-query

description: Query the notebook knowledge base (SQLite) built by /agy:notebook — precise, grounded, cited. Ask in natural language ("sum the amounts by category", "which docs mention 'Acme Corp'", "build a project timeline") or pass raw SQL. Read-only. Use this when you need exact aggregates/lookups across a document corpus instead of re-reading prose.
argument-hint: "<folder> | <pregunta o SQL>"
context: fork
allowed-tools: Bash, Read

Run a precise, **read-only** query over the `notebook.db` that `/agy:notebook` compiled from a folder of documents. Every answer is grounded in the documents and **cited** (each row carries its source `doc_ref` / `basename`). There is **no `sqlite3` CLI** on this machine — all DB access is Python.

Raw user request: $ARGUMENTS

Phase 0 — Resolve the DB (ONE Bash call)

Split `$ARGUMENTS` on the first `|`: left = folder (or its notebook slug), right = the question/SQL. Resolve `OUTDIR = docs/agy/notebook/<slug>` (same `slug()` rule the notebook uses: lowercase, non -alnum→`-`). Confirm `notebook.db` exists; if missing, tell the user to run `/agy:notebook <folder> | <objetivo>` first and stop. If the DB is older than the newest `*.facts.json`, rebuild it first: `python "<plugin>/scripts/notebook_db.py" "$OUTDIR" "$OBJETIVO"`.

Phase 1 — Two query modes

**A) Raw SQL** — if the text after `|` starts with `SELECT` or `WITH`, run it verbatim.

**B) Natural language** — write a `SELECT` against the schema below (use the recetas as templates), then run it. Prefer the `v_*` views (they dedup + carry citations). For aggregates over money, sum `monto_cents` and divide by 100.0 only for display.

ALWAYS execute via this read-only Python heredoc (never a `sqlite3` shell):

python - "$OUTDIR/notebook.db" "$SQL" <<'PY'
import sqlite3, sys, json
con = sqlite3.connect("file:%s?mode=ro" % sys.argv[1], uri=True); con.row_factory = sqlite3.Row
try:
    print(json.dumps([dict(r) for r in con.execute(sys.argv[2])], ensure_ascii=False, indent=2, default=str))
except Exception as e:
    print("SQL_ERROR: %s" % e)
PY

Phase 1b — Hybrid semantic retrieval (only if the DB was built with `--semantic`)

If `meta` has an `embedder` row (i.e. `/agy:notebook … --semantic` ran and `sqlite-vec` is installed), a fuzzy/conceptual question can use **hybrid retrieval**: FTS5 keyword ranking + vector KNN, fused with Reciprocal Rank Fusion (RRF, k=60). The `vec0` KNN needs its `LIMIT` inside a CTE (not through a JOIN). Use this to FIND the relevant documents, then answer with the structured queries above.

python - "$OUTDIR/notebook.db" "$PREGUNTA" <<'PY'
import sqlite3, sys, struct, re, hashlib, math, json
db, q = sys.argv[1], sys.argv[2]
con = sqlite3.connect("file:%s?mode=ro" % db, uri=True); con.row_factory = sqlite3.Row
emb = con.execute("SELECT v FROM meta WHERE k='embedder'").fetchone()
if not emb:
    print("NO_SEMANTIC: build with /agy:notebook … --semantic first (FTS5 keyword search still works)"); raise SystemExit
import sqlite_vec
con.enable_load_extension(True); sqlite_vec.load(con)
dim = int(con.execute("SELECT v FROM meta WHERE k='embed_dim'").fetchone()[0])
# query vector: Gemini if a real key + 768-dim, else the lexical-hash fallback (matches the embedder)
def hash_embed(t, d):
    v=[0.0]*d
    for tok in re.findall(r"\w+", t.lower()): v[int(hashlib.md5(tok.encode()).hexdigest(),16)%d]+=1.0
    n=math.sqrt(sum(x*x for x in v)) or 1.0; return [x/n for x in v]
key=__import__('os').environ.get('GEMINI_API_KEY','')
if emb[0].startswith('gemini') and key and not key.startswith('$'):
    import urllib.request
    u=f"https://generativelanguage.googleapis.com/v1beta/models/text-embedding-004:embedContent?key={key}"
    r=urllib.request.Request(u,data=json.dumps({"model":"models/text-embedding-004","content":{"parts":[{"text":q}]}}).encode(),headers={"Content-Type":"application/json"})
    qvec=json.load(urllib.request.urlopen(r,timeout=30))["embedding"]["values"]
else:
    qvec=hash_embed(q, dim)
qv=struct.pack("%df"%dim, *qvec)
fts=[r["doc_id"] for r in con.execute("SELECT c.doc_id FROM chunks_fts f JOIN chunks c ON c.id=f.rowid WHERE chunks_fts MATCH ? ORDER BY rank LIMIT 10",(re.sub(r'[^\w ]',' ',q),))]
vec=[r["doc_id"] for r in con.execute("WITH knn AS (SELECT chunk_id,distance FROM vec_chunks WHERE embedding MATCH ? ORDER BY distance LIMIT 10) SELECT c.doc_id FROM knn JOIN chunks c ON c.id=knn.chunk_id ORDER BY knn.distance",(qv,))]
s={}
for rl in (fts,vec):
    for rank,d in enumerate(dict.fromkeys(rl),1): s[d]=s.get(d,0)+1.0/(60+rank)
order=sorted(s,key=lambda d:-s[d])
docs=[dict(con.execute("SELECT id,doc_ref,tipo,basename,relevancia FROM documents WHERE id=?",(d,)).fetchone()) for d in order[:8]]
print(json.dumps(docs, ensure_ascii=False, indent=2))
PY

Phase 2 — Present (grounded + cited)

Narrate the rows. **Cite every claim** by `doc_ref`/`basename`. For a SUM, list the contributing rows so it's auditable. If a query returns 0 rows, say *"no aparece en el corpus"* (and note any `estado='no_procesado'` docs as a coverage gap) — **never invent** a value.

Schema (notebook.db, schema_ver=1)

documents(id, basename, nn, slug, doc_name, tipo, doc_ref, fecha, emisor, relevancia, estado, summary_md_path, cache_key)
chunks(id, doc_id→documents, ord, seccion, texto)          chunks_fts(texto)  -- FTS5, diacritic-folded
entities(id, doc_id→documents, clase, ent_key, valor, detalle, monto_cents, fecha_iso, quote)
        clase ∈ persona|organizacion|monto|fecha|referencia
events(id, doc_id→documents, fecha_iso, hecho, monto_cents, quote)
relations(id, doc_id→documents, sujeto, predicado, objeto, quote)
citations(id, doc_id→documents, tabla, fila_id, cita)
views: v_personas(id,nombre,n_docs,docs)  v_organizaciones(org_key,nombre,n_docs)  v_montos(concepto,n,total_cents,total)
       v_referencias(ref_key,valor,detalle,n_docs)  v_fechas  v_timeline(fecha_iso,hecho,monto_cents,doc_ref,basename,quote)

Recetas (NL → SQL)