/** * pi-qmem — comando /qmem:local: gestione dell'indice locale SQLite/FTS5. * * /qmem:local → stato (record, copertura, lag, duplicati) * /qmem:local import → ricostruisce/aggiorna l'indice dalle sessioni pi * /qmem:local find → ricerca testuale locale (anche con gateway giù) * /qmem:local enrich [--all] → arricchisce dal gateway (GET /v1/memories/{id}) * /qmem:local pull → pull incrementale dall'export del gateway */ import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { enrichFromGateway, flushQueue, importFromSessions, localDbPath, localDbReport, localSearch, pullFromGatewayExport, queueList, queueStats, } from "./local-db.ts"; import { breakerInfo, loadConfig, resetBreaker } from "./shared.ts"; export function registerQmemLocal(pi: ExtensionAPI) { pi.registerCommand("qmem:local", { description: "Indice locale SQLite/FTS5: status | import | find | queue | flush | breaker [reset] | enrich [--all] | pull", handler: async (args, ctx) => { const cfg = loadConfig(); const dbFile = localDbPath(cfg); const [sub = "status", ...rest] = (args ?? "").trim().split(/\s+/); try { if (sub === "status") { const r = await localDbReport({ dbFile }); if (!r.exists) { ctx.ui.notify(`Indice locale assente (${dbFile}): esegui /qmem:local import`, "warning"); return; } const dup = r.duplicates.length ? ` | duplicati: ${r.duplicates.length} gruppi` : ""; const coda = r.queued || r.failedQueue || r.syncedQueue ? ` | coda: ${r.queued} in attesa, ${r.syncedQueue} sincronizzati${r.failedQueue ? `, ${r.failedQueue} falliti` : ""}${r.duplicateQueue ? `, ${r.duplicateQueue} duplicati` : ""}` : ""; const br = breakerInfo(); ctx.ui.notify( `Circuit breaker: ${br.open ? `APERTO (riprova tra ${Math.ceil(br.remainingMs / 1000)}s)` : "chiuso"} | fallimenti: ${br.failures} | aperture: ${br.trips}\n` + `Indice locale: ${r.total} record (${r.withText} con testo, ${r.active} attivi, ${r.superseded} superseduti, ${r.private} privati, ${r.pendingInIndex} in coda) | project_id ${r.withProject}/${r.total} | ${r.sizeKb} KB${dup}${coda}\n` + `ultimo import: ${r.lastImport ?? "-"} | enrich: ${r.lastEnrich ?? "-"} | export: ${r.lastExport ?? "-"} | flush: ${r.lastFlush ?? "-"}\n` + `DB: ${r.path}`, "info", ); return; } if (sub === "import") { ctx.ui.setStatus("pi-qmem", "Import sessioni → indice locale..."); const stats = await importFromSessions({ dbFile }); const r = await localDbReport({ dbFile }); ctx.ui.setStatus("pi-qmem", ""); ctx.ui.notify( `Import completato: ${stats.files} sessioni, store=${stats.store} correct=${stats.correct} get=${stats.get} search_hit=${stats.searchHits} → ${stats.records} record unici.\n` + `Indice: ${r.total} record (${r.withText} con testo, ${r.active} attivi) in ${r.sizeKb} KB`, "info", ); return; } if (sub === "find") { const query = rest.filter((a) => !a.startsWith("--")).join(" ").trim(); if (!query) { ctx.ui.notify("Uso: /qmem:local find [--all] [--deleted] [--kind K] [--project P]", "warning"); return; } const has = (f: string) => rest.includes(`--${f}`); const val = (f: string) => { const i = rest.indexOf(`--${f}`); return i >= 0 && rest[i + 1] && !rest[i + 1].startsWith("--") ? rest[i + 1] : undefined; }; const hits = await localSearch( { query, kind: val("kind"), project_id: val("project"), scope: val("scope"), include_superseded: has("all"), include_private: has("private"), include_deleted: has("deleted"), top_k: Number(val("top")) || 5, }, { dbFile }, ); if (!hits.length) { ctx.ui.notify(`Nessun risultato locale per "${query}" (indice: ${dbFile})`, "warning"); return; } const lines = hits.map( (h, i) => `${i + 1}. [${h.kind ?? "?"}/${h.scope ?? "?"}${h.project_id ? ` project=${h.project_id}` : ""}${h.match_mode === "or" ? " OR" : ""}${h.pending ? " ⏳ in coda" : ""}${h.superseded_by ? " superseduto" : ""}] ${h.snippet}\n (id: ${h.memory_id}, creato: ${h.created_at ?? "?"})`, ); ctx.ui.notify(`Indice locale (ricerca testuale, non neurale) — ${hits.length} risultati:\n${lines.join("\n")}`, "info"); return; } if (sub === "enrich") { ctx.ui.setStatus("pi-qmem", "Arricchimento dal gateway..."); const stats = await enrichFromGateway(cfg, { dbFile, onlyIncomplete: !rest.includes("--all"), limit: 1000, }); ctx.ui.setStatus("pi-qmem", ""); ctx.ui.notify( stats.ok ? `Arricchimento: ${stats.updated} record aggiornati (richiesti ${stats.requested}, falliti ${stats.failed})` : `Arricchimento non possibile: gateway non raggiungibile (${stats.errors[0] ?? "errore di rete"})`, stats.ok ? "info" : "warning", ); return; } if (sub === "queue") { const stats = await queueStats({ dbFile }); const items = await queueList({ dbFile, status: "queued", limit: 8 }); const lines = items.map( (q, i) => `${i + 1}. [${q.payload.kind ?? "fact"}/${q.payload.project_id}] ${String(q.payload.text).slice(0, 70)}… (locale ${q.local_id.slice(0, 8)}${q.attempts ? `, tentativi ${q.attempts}` : ""}${q.last_error ? `, ultimo errore: ${q.last_error.slice(0, 60)}` : ""})`, ); ctx.ui.notify( `Coda offline: ${stats.queued} in attesa, ${stats.synced} sincronizzati, ${stats.duplicate} duplicati, ${stats.failed} falliti\n` + `più vecchio: ${stats.oldestQueued ?? "-"}${stats.lastError ? ` | ultimo errore: ${stats.lastError.slice(0, 80)}` : ""}` + (lines.length ? `\n${lines.join("\n")}` : "") + `\nFlush: /qmem:local flush`, stats.queued ? "info" : "warning", ); return; } if (sub === "flush") { ctx.ui.setStatus("pi-qmem", "Invio della coda offline al gateway..."); const res = await flushQueue(cfg, { dbFile, limit: 200, paceMs: 300 }); ctx.ui.setStatus("pi-qmem", ""); const msg = `Flush outbox: ${res.synced} sincronizzati, ${res.duplicates} duplicati già presenti, ${res.failed} falliti, ${res.remaining} ancora in coda` + (res.stopped ? ` — fermato: ${res.stopped}` : ""); ctx.ui.notify(msg, res.synced || res.duplicates ? "info" : "warning"); return; } if (sub === "breaker") { if (rest.includes("reset")) { resetBreaker(); ctx.ui.notify("Circuit breaker qmem: chiuso — il prossimo accesso al gateway sarà immediato", "info"); return; } const br = breakerInfo(); ctx.ui.notify( `Circuit breaker qmem: ${br.open ? `APERTO (riprova tra ${Math.ceil(br.remainingMs / 1000)}s)` : "chiuso"}\n` + `fallimenti consecutivi: ${br.failures} | aperture totali: ${br.trips}${br.lastError ? `\nultimo errore: ${br.lastError}` : ""}\n` + `file: ${br.file}\nUso: /qmem:local breaker reset`, "info", ); return; } if (sub === "pull") { ctx.ui.setStatus("pi-qmem", "Pull export dal gateway..."); const res = await pullFromGatewayExport(cfg, { dbFile }); ctx.ui.setStatus("pi-qmem", ""); ctx.ui.notify( res.supported ? `Pull export: ${res.fetched} record in ${res.pages} pagine` : `Pull export non disponibile: ${res.message ?? "endpoint assente sul gateway"}`, res.supported ? "info" : "warning", ); return; } ctx.ui.notify("Uso: /qmem:local [status|import|find |enrich [--all]|pull]", "warning"); } catch (e) { ctx.ui.setStatus("pi-qmem", ""); ctx.ui.notify(`Errore indice locale: ${e instanceof Error ? e.message : String(e)}`, "error"); } }, }); }