pi-qmem/extensions/local-command.ts

/**
 * 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 <query>    → 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 <query> | 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 <query> [--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 <query>|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");
			}
		},
	});
}