From 322b4cf44661fefd88880111d03367d8e4761855 Mon Sep 17 00:00:00 2001 From: Matteo Benedetto Date: Sun, 13 Sep 2026 17:29:09 +0200 Subject: [PATCH] feat(outbox): store offline con coda locale e sincronizzazione al ritorno della rete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prima qmem_store falliva se il gateway non era raggiungibile: la conoscenza andava persa. Ora il record entra in una coda locale persistente e viene inviato automaticamente quando la connessione torna. Core (extensions/local-db.ts): - tabella `pending` (local_id, payload JSON, attempts, last_error, status, remote_id) + colonna `records.pending` (migrazione automatica dei DB esistenti) - queueStore(): accoda e crea subito il placeholder locale ricercabile (⏳) - flushQueue(): POST /v1/memories con Idempotency-Key = local_id (retry senza duplicati), FIFO, pacing sotto il rate limit, timeout 12s per richiesta - esiti: synced (il record locale adotta l'ID remoto, niente duplicati) · duplicate (409: registra l'ID del match e NON sovrascrive il testo locale autorevole) · failed (4xx di validazione, non ritentato) · 0/429/5xx: resta in coda e il flush si ferma - supersede offline: supersedes_id che punta a un local_id viene rimappato al remote_id al flush (se il genitore non è sincronizzato → failed esplicito) - submitOrQueue(): online → gateway + indicizzazione locale; offline → coda - maybeBackgroundFlush() (single-flight) e flushQueueIfPending() per session_start - stato/report: queued/synced/duplicate/failed, più vecchio, ultimo errore, last_flush, record pendenti in indice Estensione: - qmem_store: gateway giù → accoda e risponde con id locale, dimensione coda e spiegazione (details.queued/local_id/queue_size) - fallback offline di session_start: flush in background (non blocca l'avvio) - /qmem:local queue|flush; status con la coda; marker "⏳ in coda" nei risultati locali di qmem_search/qmem_get - regole e skill: un record in coda NON è ancora nella memoria condivisa CLI: store [--queue-only], queue, flush (+ status con la coda). Test: scripts/test-local.mjs ora copre anche outbox → 24 controlli (flush con 2 sync + 1 duplicato 409 + 1 fallito 422, Idempotency-Key, rimappatura del supersede, ricerca del record con l'ID remoto dopo il sync). Verifiche: 24/24 test superati; demo reale su DB temporaneo: store accodato, queue con local_id, flush con gateway giù → "fermato: HTTP 0" e voce che resta in coda con l'errore registrato. --- README.md | 29 ++- extensions/index.ts | 23 ++ extensions/local-command.ts | 36 ++- extensions/local-db.ts | 452 +++++++++++++++++++++++++++++++++++- extensions/rules.ts | 4 +- extensions/shared.ts | 3 + extensions/tools/get.ts | 2 +- extensions/tools/search.ts | 2 +- extensions/tools/store.ts | 71 +++--- scripts/qmem-sqlite.mjs | 62 ++++- scripts/test-local.mjs | 71 +++++- skills/qmem/SKILL.md | 17 +- 12 files changed, 705 insertions(+), 67 deletions(-) diff --git a/README.md b/README.md index fe828b8..78476f0 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ Oppure copia `extensions/index.ts` in `~/.pi/agent/extensions/pi-qmem/`. | Tool | Descrizione | |---|---| -| `qmem_store` | Salva un record di memoria (text, kind, agent_id, **project_id obbligatorio**, scope, source, expires_at, supersedes_id, supersede_reason) | +| `qmem_store` | Salva un record di memoria (text, kind, agent_id, **project_id obbligatorio**, scope, source, expires_at, supersedes_id, supersede_reason). **Se il gateway è giù il record viene accodato localmente (outbox)** e inviato automaticamente al ritorno della connessione | | `qmem_search` | Ricerca semantica su tutta la conoscenza condivisa (query, kind, project_id, scope, top_k, include_superseded, min_score). **Se il gateway non risponde degrada all'indice locale SQLite/FTS5** (testuale, etichettato `fallback: local_sqlite`) | | `qmem_correct` | Corregge una memoria falsa: crea un nuovo record che **supersede** il vecchio (che resta in archivio marcato superseded) | | `qmem_meta` | Discovery: panoramica di scope×kind, progetti, agenti e superseduti (per scegliere i filtri di ricerca) | @@ -43,12 +43,14 @@ Config salvata in `~/.config/pi-qmem/config.json` (0600): "url": "https://qmem.enne2.net", "apiKey": "...", "localDbPath": "~/.local/share/pi-qmem/qmem.sqlite", - "localFallback": true + "localFallback": true, + "offlineQueue": true } ``` -`localFallback: false` disabilita il fallback sull'indice locale (utile per -misurare il comportamento "solo gateway"). +`localFallback: false` disabilita il fallback in lettura; `offlineQueue: false` +disabilita l'accodamento offline in scrittura (lo store torna a fallire come +prima). ## Indice locale (fallback offline) @@ -67,6 +69,14 @@ testualmente la conoscenza **senza gateway, senza modelli, senza dipendenze**: trova nulla si ripiega su OR (match parziale, dichiarato) - **Fallback automatico**: `qmem_search`/`qmem_get` usano l'indice locale quando il gateway risponde 0/429/5xx, etichettando i risultati come **non neurali** +- **Outbox (store offline)**: `qmem_store` con gateway irraggiungibile accoda il + record in SQLite: è subito ricercabile (marcato ⏳ `pending`) e viene inviato a + `POST /v1/memories` al ritorno della connessione — `Idempotency-Key` = id locale + (retry senza duplicati), poi il record locale **adotta l'ID del gateway**. + Esiti: `synced` · `duplicate` (409, con l'ID del match) · `failed` (4xx di + validazione). Le correzioni che puntano a un record ancora locale vengono + rimappate all'ID remoto al flush. Trigger: `session_start` (background, non + blocca l'avvio), dopo uno store riuscito, o `/qmem:local flush` - **Arricchimento**: quando il gateway torna online, `enrich` completa testo/`project_id`/`private`/stato supersede via `GET /v1/memories/{id}`, e `pull` sincronizza dall'`export` (endpoint previsto lato gateway) @@ -77,18 +87,23 @@ Comandi (TUI) e CLI standalone: /qmem:local status # record, copertura, lag, duplicati /qmem:local import # ricostruisce/aggiorna dalle sessioni pi /qmem:local find "circuit breaker" # ricerca testuale locale +/qmem:local queue # stato della coda (in attesa/sync/dup/fallite) +/qmem:local flush # invia subito la coda al gateway /qmem:local enrich [--all] # arricchisce dal gateway /qmem:local pull # pull incrementale dall'export # equivalente standalone (stesso core, nessuna dipendenza) node scripts/qmem-sqlite.mjs status|import|find "query"|enrich|pull -node scripts/test-local.mjs # suite di test (14 controlli, HOME temporanea) +node scripts/qmem-sqlite.mjs store --project P --text "..." [--queue-only] +node scripts/qmem-sqlite.mjs queue|flush +node scripts/test-local.mjs # suite di test (24 controlli, HOME temporanea) ``` Limiti dichiarati: è uno **storico osservato** (più vecchio del gateway), la ricerca è **lessicale** (nessuno score 0.45/0.60: non applicare le soglie -semantiche) e `qmem_store` **non ha coda locale** — a gateway giù il record non -viene salvato. +semantiche) e un record in coda (⏳) **non è ancora nella memoria condivisa**: +sarà visibile agli altri agenti solo dopo il flush. La coda è locale alla +macchina (nessuna sincronizzazione tra macchine diverse). ## Regole comportamentali (autocontenute) diff --git a/extensions/index.ts b/extensions/index.ts index b19563b..c80115f 100644 --- a/extensions/index.ts +++ b/extensions/index.ts @@ -14,6 +14,8 @@ import { registerQmemStore } from "./tools/store"; import { registerQmemTree } from "./tools/tree"; import { registerQmemRules } from "./rules"; import { registerQmemLocal } from "./local-command"; +import { flushQueueIfPending, localDbPath } from "./local-db.ts"; +import { loadConfig } from "./shared.ts"; export default function qmemExtension(pi: ExtensionAPI) { registerQmemStore(pi); @@ -25,4 +27,25 @@ export default function qmemExtension(pi: ExtensionAPI) { registerQmemConfig(pi); registerQmemLocal(pi); registerQmemRules(pi); + + // Outbox: al ritorno della connessione (nuova sessione/reload) i record + // accodati offline vengono inviati al gateway. In background: l'avvio della + // sessione non deve mai attendere la rete. + pi.on("session_start", async (_event, ctx) => { + const cfg = loadConfig(); + if (cfg.offlineQueue === false) return; + void (async () => { + try { + const res = await flushQueueIfPending(cfg, localDbPath(cfg)); + if (res && (res.synced || res.duplicates || res.failed)) { + ctx.ui.notify( + `Outbox qmem: ${res.synced} sincronizzati, ${res.duplicates} duplicati, ${res.failed} falliti${res.remaining ? `, ${res.remaining} in coda` : ""}${res.stopped ? ` (fermato: ${res.stopped})` : ""}`, + res.synced ? "info" : "warning", + ); + } + } catch { + /* best effort: la coda resta e verrà ritentata */ + } + })(); + }); } diff --git a/extensions/local-command.ts b/extensions/local-command.ts index 7893818..50a6f70 100644 --- a/extensions/local-command.ts +++ b/extensions/local-command.ts @@ -10,11 +10,14 @@ import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { enrichFromGateway, + flushQueue, importFromSessions, localDbPath, localDbReport, localSearch, pullFromGatewayExport, + queueList, + queueStats, } from "./local-db.ts"; import { loadConfig } from "./shared.ts"; @@ -33,9 +36,10 @@ export function registerQmemLocal(pi: ExtensionAPI) { 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` : ""}` : ""; ctx.ui.notify( - `Indice locale: ${r.total} record (${r.withText} con testo, ${r.active} attivi, ${r.superseded} superseduti, ${r.private} privati) | project_id ${r.withProject}/${r.total} | ${r.sizeKb} KB${dup}\n` + - `ultimo import: ${r.lastImport ?? "-"} | enrich: ${r.lastEnrich ?? "-"} | export: ${r.lastExport ?? "-"}\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", ); @@ -82,7 +86,7 @@ export function registerQmemLocal(pi: ExtensionAPI) { } 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.superseded_by ? " superseduto" : ""}] ${h.snippet}\n (id: ${h.memory_id}, creato: ${h.created_at ?? "?"})`, + `${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; @@ -103,6 +107,32 @@ export function registerQmemLocal(pi: ExtensionAPI) { ); 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 === "pull") { ctx.ui.setStatus("pi-qmem", "Pull export dal gateway..."); const res = await pullFromGatewayExport(cfg, { dbFile }); diff --git a/extensions/local-db.ts b/extensions/local-db.ts index 2595157..942c927 100644 --- a/extensions/local-db.ts +++ b/extensions/local-db.ts @@ -24,7 +24,7 @@ import * as fs from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { gatewayRequest, loadConfig, type MemoryConfig } from "./shared.ts"; // --------------------------------------------------------------------------- @@ -90,7 +90,9 @@ CREATE TABLE IF NOT EXISTS records( text TEXT, kind TEXT, project_id TEXT, scope TEXT, agent_id TEXT, confidence TEXT, importance REAL, created_at TEXT, supersedes_id TEXT, superseded_by TEXT, parent_id TEXT, topic TEXT, level TEXT, private INTEGER NOT NULL DEFAULT 0, - text_hash TEXT, sources TEXT, synced_at TEXT, updated_at TEXT + text_hash TEXT, sources TEXT, synced_at TEXT, updated_at TEXT, + /** 1 = record creato offline, non ancora presente sul gateway */ + pending INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS records_project ON records(project_id); CREATE INDEX IF NOT EXISTS records_kind ON records(kind); @@ -110,7 +112,26 @@ CREATE TRIGGER IF NOT EXISTS records_ad AFTER DELETE ON records BEGIN INSERT INTO records_fts(records_fts, rowid, text, kind, project_id) VALUES('delete', old.rowid_, old.text, old.kind, old.project_id); END; CREATE TABLE IF NOT EXISTS meta(k TEXT PRIMARY KEY, v TEXT); +CREATE TABLE IF NOT EXISTS pending( + id INTEGER PRIMARY KEY AUTOINCREMENT, + local_id TEXT UNIQUE NOT NULL, + payload TEXT NOT NULL, + created_at TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt TEXT, + last_error TEXT, + status TEXT NOT NULL DEFAULT 'queued', -- queued | synced | duplicate | failed + remote_id TEXT, + synced_at TEXT +); +CREATE INDEX IF NOT EXISTS pending_status ON pending(status); `); + // migrazione per DB creati prima dell'introduzione della coda + const cols = db + .prepare("PRAGMA table_info(records)") + .all() + .map((c: any) => c.name); + if (!cols.includes("pending")) db.exec("ALTER TABLE records ADD COLUMN pending INTEGER NOT NULL DEFAULT 0"); return db; } @@ -133,12 +154,14 @@ export interface LocalRecordInput { topic?: string | null; level?: string | null; private?: boolean | null; - /** Fonte dell'osservazione: store | correct | get | search | export | enrich */ + /** true = creato offline (in coda), false = presente sul gateway */ + pending?: boolean | null; + /** Fonte dell'osservazione: store | correct | get | search | export | enrich | outbox | sync */ source: string; } /** Sorgenti ordinate per affidabilità del testo (maggiore = più completo). */ -const TEXT_PRIORITY: Record = { export: 5, enrich: 5, get: 4, store: 3, correct: 3, search: 2 }; +const TEXT_PRIORITY: Record = { export: 5, enrich: 5, sync: 5, get: 4, store: 3, correct: 3, outbox: 3, search: 2 }; function normalizeText(t: string): string { return t.replace(/\s+/g, " ").trim().toLowerCase(); @@ -155,11 +178,11 @@ export function upsertRecords(db: Db, records: LocalRecordInput[]): number { const find = db.prepare("SELECT * FROM records WHERE memory_id = ?"); const insert = db.prepare(`INSERT INTO records (memory_id, text, kind, project_id, scope, agent_id, confidence, importance, created_at, - supersedes_id, superseded_by, parent_id, topic, level, private, text_hash, sources, synced_at, updated_at) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`); + supersedes_id, superseded_by, parent_id, topic, level, private, text_hash, sources, synced_at, updated_at, pending) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`); const update = db.prepare(`UPDATE records SET text=?, kind=?, project_id=?, scope=?, agent_id=?, confidence=?, importance=?, created_at=?, - supersedes_id=?, superseded_by=?, parent_id=?, topic=?, level=?, private=?, text_hash=?, sources=?, synced_at=?, updated_at=? + supersedes_id=?, superseded_by=?, parent_id=?, topic=?, level=?, private=?, text_hash=?, sources=?, synced_at=?, updated_at=?, pending=? WHERE rowid_=?`); const now = new Date().toISOString(); let written = 0; @@ -193,17 +216,19 @@ export function upsertRecords(db: Db, records: LocalRecordInput[]): number { [...sources].sort().join(","), now, now, + r.pending === true ? 1 : r.pending === false ? 0 : (existing?.pending ?? 0), ]; if (existing) { // I campi già noti non vengono mai azzerati da osservazioni più povere // (es. un risultato di ricerca che non riporta project_id). const keys = ["text", "kind", "project_id", "scope", "agent_id", "confidence", "importance", "created_at", - "supersedes_id", "superseded_by", "parent_id", "topic", "level", "private", "text_hash", "sources", "synced_at", "updated_at"]; + "supersedes_id", "superseded_by", "parent_id", "topic", "level", "private", "text_hash", "sources", "synced_at", "updated_at", "pending"]; const merged = vals.map((v, i) => { const key = keys[i]; const cur = (existing as any)[key]; const curEmpty = cur === null || cur === undefined || cur === "" || cur === 0; if (key === "private") return cur === 1 ? 1 : v; // monotono: una volta privato resta privato + if (key === "pending") return v; // 0 = sincronizzato: prevale sullo stato locale if (key === "sources" || key === "synced_at" || key === "updated_at") return v; if (v === null || v === "" ) return curEmpty ? v : cur; // non azzerare return v; @@ -422,6 +447,8 @@ export interface LocalSearchHit { created_at: string | null; superseded_by: string | null; sources: string | null; + /** 1 = creato offline, non ancora sul gateway */ + pending?: number; snippet: string; rank: number; /** "and" = tutti i termini presenti; "or" = match parziale (AND senza risultati) */ @@ -466,7 +493,7 @@ async function runSearch(db: Db, params: LocalSearchParams, match: string): Prom if (!params.include_private) where.push("r.private = 0"); args.push(Math.min(Math.max(params.top_k ?? 5, 1), 50)); const sql = `SELECT r.memory_id, r.text, r.kind, r.project_id, r.scope, r.agent_id, r.created_at, - r.superseded_by, r.sources, bm25(records_fts) AS rank, + r.superseded_by, r.sources, r.pending, bm25(records_fts) AS rank, snippet(records_fts, 0, '«', '»', '…', 14) AS snippet FROM records_fts JOIN records r ON r.rowid_ = records_fts.rowid WHERE ${where.join(" AND ")} ORDER BY rank LIMIT ?`; @@ -495,17 +522,39 @@ export async function localSearch(params: LocalSearchParams, opts?: { dbFile?: s } /** Recupero locale di un singolo record (fallback di qmem_get). */ -export async function localGet(memoryId: string, opts?: { dbFile?: string }): Promise { +export async function localGet(memoryId: string, opts?: { dbFile?: string }): Promise<(LocalSearchHit & { queue_status?: string; remote_id?: string | null }) | null> { const db = await openLocalDb(opts?.dbFile); try { const row = db .prepare( - `SELECT memory_id, text, kind, project_id, scope, agent_id, created_at, superseded_by, sources, 0 AS rank, + `SELECT memory_id, text, kind, project_id, scope, agent_id, created_at, superseded_by, sources, pending, 0 AS rank, COALESCE(substr(text,1,200),'') AS snippet FROM records WHERE memory_id = ?`, ) .get(memoryId) as LocalSearchHit | undefined; - return row ?? null; + if (row) return row; + // non è nell'indice: potrebbe essere un id locale della coda + const q = db + .prepare("SELECT local_id, payload, status, remote_id FROM pending WHERE local_id = ?") + .get(memoryId) as any; + if (!q) return null; + const payload: QueuePayload = JSON.parse(q.payload); + return { + memory_id: q.local_id, + text: payload.text, + kind: payload.kind ?? "fact", + project_id: payload.project_id, + scope: payload.scope ?? "agent", + agent_id: payload.agent_id ?? null, + created_at: null, + superseded_by: null, + sources: "outbox", + pending: q.status === "queued" ? 1 : 0, + snippet: String(payload.text ?? "").slice(0, 200), + rank: 0, + queue_status: q.status, + remote_id: q.remote_id ?? null, + }; } finally { db.close(); } @@ -641,6 +690,364 @@ export async function pullFromGatewayExport( } } +// --------------------------------------------------------------------------- +// Coda offline (outbox): store quando il gateway non è raggiungibile +// --------------------------------------------------------------------------- +// Il record viene scritto subito nell'indice locale (marcato `pending`) e in +// `pending` come payload JSON. Al ritorno della connessione il flush lo invia a +// POST /v1/memories con `Idempotency-Key` = local_id (il gateway deduplica i +// retry), poi il record locale **adotta l'ID remoto** (niente duplicati). +// 409 (duplicato noto) → status `duplicate` con l'id del match; 4xx di +// validazione → `failed` (non ritentato); 0/429/5xx → resta `queued` e il flush +// si ferma. +export interface QueuePayload { + text: string; + kind?: string; + project_id: string; + scope?: string; + agent_id?: string; + source?: string; + confidence?: string; + expires_at?: string; + supersedes_id?: string; + supersede_reason?: string; + parent_id?: string; + level?: string; + topic?: string; + links?: unknown; + importance?: number; + private?: boolean; +} + +export interface QueueItem { + local_id: string; + payload: QueuePayload; + created_at: string; + attempts: number; + last_attempt: string | null; + last_error: string | null; + status: "queued" | "synced" | "duplicate" | "failed"; + remote_id: string | null; +} + +export interface QueueStats { + queued: number; + synced: number; + duplicate: number; + failed: number; + oldestQueued: string | null; + lastError: string | null; +} + +export interface FlushStats { + processed: number; + synced: number; + duplicates: number; + failed: number; + remaining: number; + stopped?: string; + remoteIds: Record; + errors: string[]; +} + +const nowIso = () => new Date().toISOString(); +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +function stripEmpty(obj: Record): Record { + const out: Record = {}; + for (const [k, v] of Object.entries(obj)) if (v !== undefined && v !== null && v !== "") out[k] = v; + return out; +} + +/** Accoda un record (crea anche il placeholder locale ricercabile). */ +export async function queueStore( + payload: QueuePayload, + opts?: { dbFile?: string }, +): Promise<{ local_id: string; queue_size: number }> { + const db = await openLocalDb(opts?.dbFile); + try { + const localId = randomUUID(); + const at = nowIso(); + db.prepare("INSERT INTO pending(local_id,payload,created_at,status) VALUES(?,?,?,'queued')").run( + localId, + JSON.stringify(stripEmpty(payload as unknown as Record)), + at, + ); + upsertRecords(db, [ + { + memory_id: localId, + text: payload.text, + kind: payload.kind ?? "fact", + project_id: payload.project_id, + scope: payload.scope ?? "agent", + agent_id: payload.agent_id ?? null, + confidence: payload.confidence ?? null, + importance: typeof payload.importance === "number" ? payload.importance : null, + created_at: at, + supersedes_id: payload.supersedes_id ?? null, + parent_id: payload.parent_id ?? null, + topic: payload.topic ?? null, + level: payload.level ?? null, + private: payload.private ?? null, + pending: true, + source: "outbox", + }, + ]); + const n = db.prepare("SELECT COUNT(*) n FROM pending WHERE status='queued'").get()?.n ?? 0; + return { local_id: localId, queue_size: n }; + } finally { + db.close(); + } +} + +export async function queueList(opts?: { dbFile?: string; status?: string; limit?: number }): Promise { + const db = await openLocalDb(opts?.dbFile); + try { + const where = opts?.status ? "WHERE status = ?" : ""; + const args: any[] = opts?.status ? [opts.status] : []; + args.push(opts?.limit ?? 50); + const rows = db + .prepare(`SELECT local_id, payload, created_at, attempts, last_attempt, last_error, status, remote_id FROM pending ${where} ORDER BY id DESC LIMIT ?`) + .all(...args) as any[]; + return rows.map((r) => ({ ...r, payload: JSON.parse(r.payload) })); + } finally { + db.close(); + } +} + +export async function queueStats(opts?: { dbFile?: string }): Promise { + const db = await openLocalDb(opts?.dbFile); + try { + const count = (s: string) => db.prepare("SELECT COUNT(*) n FROM pending WHERE status=?").get(s)?.n ?? 0; + const oldest = db.prepare("SELECT created_at FROM pending WHERE status='queued' ORDER BY id LIMIT 1").get()?.created_at ?? null; + const lastError = db.prepare("SELECT last_error FROM pending WHERE last_error IS NOT NULL ORDER BY id DESC LIMIT 1").get()?.last_error ?? null; + return { queued: count("queued"), synced: count("synced"), duplicate: count("duplicate"), failed: count("failed"), oldestQueued: oldest, lastError }; + } finally { + db.close(); + } +} + +/** Il record locale "diventa" quello remoto: adotta l'ID del gateway. */ +function adoptRemoteId(db: Db, localId: string, remoteId: string, payload: QueuePayload, data: any): void { + db.prepare("DELETE FROM records WHERE memory_id = ?").run(localId); + upsertRecords(db, [ + { + memory_id: remoteId, + text: data?.text ?? payload.text, + kind: data?.kind ?? payload.kind ?? "fact", + project_id: data?.project_id ?? payload.project_id, + scope: data?.scope ?? payload.scope ?? "agent", + agent_id: data?.agent_id ?? payload.agent_id ?? null, + confidence: data?.confidence ?? payload.confidence ?? null, + importance: typeof data?.importance === "number" ? data.importance : (payload.importance ?? null), + created_at: data?.created_at ?? null, + supersedes_id: data?.supersedes_id ?? payload.supersedes_id ?? null, + superseded_by: data?.superseded_by ?? null, + parent_id: data?.parent_id ?? payload.parent_id ?? null, + topic: data?.topic ?? payload.topic ?? null, + level: data?.level ?? payload.level ?? null, + private: data?.private ?? payload.private ?? null, + pending: false, + source: "sync", + }, + ]); +} + +/** Upload dei record in coda verso il gateway (idempotente, FIFO). */ +export async function flushQueue( + cfg: MemoryConfig, + opts?: { dbFile?: string; limit?: number; paceMs?: number }, +): Promise { + const db = await openLocalDb(opts?.dbFile); + const stats: FlushStats = { processed: 0, synced: 0, duplicates: 0, failed: 0, remaining: 0, remoteIds: {}, errors: [] }; + try { + const items = db + .prepare("SELECT local_id, payload FROM pending WHERE status='queued' ORDER BY id LIMIT ?") + .all(opts?.limit ?? 100) as Array<{ local_id: string; payload: string }>; + const pace = opts?.paceMs ?? 300; // < rate limit del gateway (120/min) + const perRequestMs = Math.min(cfg.timeoutMs ?? 30_000, 12_000); + for (const item of items) { + stats.processed++; + const payload: QueuePayload = JSON.parse(item.payload); + // supersede verso un record creato offline: rimappa local_id → remote_id + if (payload.supersedes_id) { + const parent = db.prepare("SELECT status, remote_id FROM pending WHERE local_id = ?").get(payload.supersedes_id) as any; + if (parent) { + if ((parent.status === "synced" || parent.status === "duplicate") && parent.remote_id) { + payload.supersedes_id = parent.remote_id; + } else { + const msg = `supersede di un record non ancora sincronizzato (${String(payload.supersedes_id).slice(0, 8)})`; + db.prepare("UPDATE pending SET status='failed', attempts=attempts+1, last_attempt=?, last_error=? WHERE local_id=?").run(nowIso(), msg, item.local_id); + stats.failed++; + stats.errors.push(`${item.local_id.slice(0, 8)}: ${msg}`); + continue; + } + } + } + const body = stripEmpty(payload as unknown as Record); + let res: { ok: boolean; status: number; data: any }; + try { + res = await gatewayRequest( + cfg, + "POST", + "/v1/memories", + body, + AbortSignal.timeout(perRequestMs), + item.local_id, // Idempotency-Key: retry sicuri, nessun duplicato + ); + } catch (e) { + const msg = `${e instanceof Error ? e.message : String(e)} (timeout ${perRequestMs}ms)`; + db.prepare("UPDATE pending SET attempts=attempts+1, last_attempt=?, last_error=? WHERE local_id=?").run(nowIso(), msg, item.local_id); + stats.errors.push(`${item.local_id.slice(0, 8)}: ${msg}`); + stats.stopped = msg; + break; + } + if (res.ok && res.data?.memory_id) { + db.prepare("UPDATE pending SET status='synced', remote_id=?, synced_at=?, attempts=attempts+1, last_attempt=?, last_error=NULL WHERE local_id=?").run( + res.data.memory_id, + nowIso(), + nowIso(), + item.local_id, + ); + adoptRemoteId(db, item.local_id, res.data.memory_id, payload, res.data); + stats.synced++; + stats.remoteIds[item.local_id] = res.data.memory_id; + } else if (res.status === 409) { + const detail = res.data?.detail ?? res.data ?? {}; + const remoteId: string | null = detail?.matches?.[0]?.memory_id ?? null; + db.prepare("UPDATE pending SET status='duplicate', remote_id=?, attempts=attempts+1, last_attempt=?, last_error=? WHERE local_id=?").run( + remoteId, + nowIso(), + `409 ${detail?.reason ?? "duplicate_memory"}`, + item.local_id, + ); + if (remoteId) { + // Il record è già sul gateway: NON sovrascrivere il testo locale + // autorevole con il payload appena inviato — si elimina solo il + // placeholder locale e, se il record remoto non è ancora in indice, + // lo si aggiunge con il testo appena scritto. + const known = db.prepare("SELECT memory_id FROM records WHERE memory_id = ?").get(remoteId); + db.prepare("DELETE FROM records WHERE memory_id = ?").run(item.local_id); + if (!known) { + upsertRecords(db, [ + { + memory_id: remoteId, + text: payload.text, + kind: payload.kind ?? "fact", + project_id: payload.project_id, + scope: payload.scope ?? "agent", + agent_id: payload.agent_id ?? null, + created_at: null, + topic: payload.topic ?? null, + level: payload.level ?? null, + private: payload.private ?? null, + pending: false, + source: "sync", + }, + ]); + } + } else { + db.prepare("UPDATE records SET pending=0 WHERE memory_id=?").run(item.local_id); + } + stats.duplicates++; + } else if (res.status >= 400 && res.status < 500) { + // errore permanente (422 validazione, 400, 401): non ritentare in automatico + const msg = `${res.status}: ${JSON.stringify(res.data).slice(0, 300)}`; + db.prepare("UPDATE pending SET status='failed', attempts=attempts+1, last_attempt=?, last_error=? WHERE local_id=?").run(nowIso(), msg, item.local_id); + stats.failed++; + stats.errors.push(`${item.local_id.slice(0, 8)}: HTTP ${res.status}`); + if (res.status === 401) { + stats.stopped = "chiave API non valida (401)"; + break; + } + } else { + // 0/429/5xx: gateway non utilizzabile, resta in coda + const msg = `HTTP ${res.status}`; + db.prepare("UPDATE pending SET attempts=attempts+1, last_attempt=?, last_error=? WHERE local_id=?").run(nowIso(), msg, item.local_id); + stats.errors.push(`${item.local_id.slice(0, 8)}: ${msg}`); + stats.stopped = msg; + break; + } + if (pace > 0) await sleep(pace); + } + stats.remaining = db.prepare("SELECT COUNT(*) n FROM pending WHERE status='queued'").get()?.n ?? 0; + db.prepare("INSERT INTO meta(k,v) VALUES('last_flush',?) ON CONFLICT(k) DO UPDATE SET v=excluded.v").run(nowIso()); + return stats; + } finally { + db.close(); + } +} + +/** + * Store con fallback offline: prova il gateway, e se non è raggiungibile accoda + * il record localmente (default `offlineQueue: true`). + */ +export async function submitOrQueue( + cfg: MemoryConfig, + payload: QueuePayload, + opts?: { dbFile?: string; queue?: boolean }, +): Promise<{ queued: boolean; remote_id?: string; local_id?: string; queue_size?: number; status: number; data?: any }> { + const queueAllowed = opts?.queue ?? cfg.offlineQueue !== false; + const body = stripEmpty(payload as unknown as Record); + const perRequestMs = Math.min(cfg.timeoutMs ?? 30_000, 15_000); + const res = await gatewayRequest(cfg, "POST", "/v1/memories", body, AbortSignal.timeout(perRequestMs), randomUUID()); + if (res.ok && res.data?.memory_id) { + // indicizza subito in locale (senza attendere enrich/export) + try { + const db = await openLocalDb(opts?.dbFile); + upsertRecords(db, [ + { + memory_id: res.data.memory_id, + text: res.data.text ?? payload.text, + kind: res.data.kind ?? payload.kind ?? "fact", + project_id: res.data.project_id ?? payload.project_id, + scope: res.data.scope ?? payload.scope ?? "agent", + agent_id: res.data.agent_id ?? payload.agent_id ?? null, + confidence: res.data.confidence ?? payload.confidence ?? null, + importance: typeof res.data.importance === "number" ? res.data.importance : (payload.importance ?? null), + created_at: res.data.created_at ?? null, + supersedes_id: res.data.supersedes_id ?? payload.supersedes_id ?? null, + parent_id: payload.parent_id ?? null, + topic: payload.topic ?? null, + level: payload.level ?? null, + private: payload.private ?? null, + pending: false, + source: "store", + }, + ]); + db.close(); + } catch { + /* l'indice locale è best-effort sul percorso online */ + } + maybeBackgroundFlush(cfg, opts?.dbFile); + return { queued: false, remote_id: res.data.memory_id, status: res.status, data: res.data }; + } + const down = res.status === 0 || res.status >= 500 || res.status === 429; + if (down && queueAllowed) { + const q = await queueStore(payload, { dbFile: opts?.dbFile }); + return { queued: true, local_id: q.local_id, queue_size: q.queue_size, status: res.status, data: res.data }; + } + return { queued: false, status: res.status, data: res.data }; +} + +let flushInFlight: Promise | null = null; + +/** Flush in background (single-flight): usato dopo operazioni riuscite e su session_start. */ +export function maybeBackgroundFlush(cfg: MemoryConfig, dbFile?: string): void { + if (flushInFlight) return; + flushInFlight = flushQueue(cfg, { dbFile, limit: 20, paceMs: 400 }) + .catch(() => undefined) + .finally(() => { + flushInFlight = null; + }); +} + +export async function flushQueueIfPending(cfg: MemoryConfig, dbFile?: string): Promise { + const stats = await queueStats({ dbFile }); + if (!stats.queued) return null; + return flushQueue(cfg, { dbFile, limit: 50, paceMs: 300 }); +} + // --------------------------------------------------------------------------- // Report / stato // --------------------------------------------------------------------------- @@ -657,6 +1064,14 @@ export interface LocalDbReport { lastImport?: string; lastEnrich?: string; lastExport?: string; + lastFlush?: string; + queued: number; + syncedQueue: number; + duplicateQueue: number; + failedQueue: number; + pendingInIndex: number; + oldestQueued?: string | null; + queueLastError?: string | null; topProjects: Array<{ project_id: string | null; n: number }>; duplicates: Array<{ text_hash: string; n: number; ids: string[] }>; } @@ -674,6 +1089,11 @@ export async function localDbReport(opts?: { dbFile?: string }): Promise 1 ORDER BY n DESC LIMIT 20") diff --git a/extensions/rules.ts b/extensions/rules.ts index 0f806f2..d76b414 100644 --- a/extensions/rules.ts +++ b/extensions/rules.ts @@ -16,8 +16,8 @@ Procedures (hierarchy L1/L2, scores, supersede, reflexion, consolidation): skill ### Indice locale (fallback offline) - Quando il gateway non risponde, qmem_search degrada all'INDICE LOCALE SQLite/FTS5 (ricerca testuale, NON neurale: nessuno score 0.45/0.60). Il risultato è etichettato 'fallback: local_sqlite'. - I risultati locali sono osservazioni più vecchie del gateway: verificali prima dell'uso e non applicare le soglie di score del gateway. -- qmem_store NON ha coda locale: a gateway giù il record non viene salvato → riprovalo quando torna raggiungibile. -- Gestione: /qmem:local status | import | find | enrich | pull (import = ricostruisce l'indice dalle sessioni pi; enrich/pull = allineamento dal gateway). +- OUTBOX: se il gateway è giù, qmem_store accoda il record in locale (non lo perde). Il record è subito ricercabile (marcato ⏳ in coda) e viene inviato automaticamente al ritorno della connessione (flush su session_start). Finché non è sincronizzato NON è nella memoria condivisa: trattalo come non condiviso. +- Gestione: /qmem:local status | import | find | queue | flush | enrich | pull. ### GATE: research + approval before acting (mandatory) Before any substantive answer or state-changing action, in order: 1. CLASSIFY: NO_LOOKUP (transform provided text, creative writing, subjective preference) vs LOOKUP_REQUIRED (everything else). diff --git a/extensions/shared.ts b/extensions/shared.ts index 534b3f0..f58c1d2 100644 --- a/extensions/shared.ts +++ b/extensions/shared.ts @@ -35,6 +35,8 @@ export interface MemoryConfig { localDbPath?: string; /** Usa l'indice locale come fallback quando il gateway non risponde (default true). */ localFallback?: boolean; + /** Accoda i record in locale quando il gateway non è raggiungibile (default true). */ + offlineQueue?: boolean; } const CONFIG_DEFAULTS: MemoryConfig = { @@ -43,6 +45,7 @@ const CONFIG_DEFAULTS: MemoryConfig = { timeoutMs: 30_000, correctMinScore: 0.6, localFallback: true, + offlineQueue: true, }; // Retry su errori transitori (429/5xx/timeout): backoff esponenziale + jitter diff --git a/extensions/tools/get.ts b/extensions/tools/get.ts index 308c165..5cede5a 100644 --- a/extensions/tools/get.ts +++ b/extensions/tools/get.ts @@ -36,7 +36,7 @@ export function registerQmemGet(pi: ExtensionAPI) { type: "text", text: `⚠️ Gateway non raggiungibile (HTTP ${status}): record dall'INDICE LOCALE (osservazione più vecchia del gateway, può essere incompleta).\n` + - `memory_id: ${local.memory_id}\n[${local.kind ?? "?"}/${local.scope ?? "?"}${local.project_id ? ` project=${local.project_id}` : ""}] agente: ${local.agent_id ?? "?"}, creato: ${local.created_at ?? "?"}${local.superseded_by ? `, ⚠️ superseduto da ${local.superseded_by}` : ""}\n\n${local.text ?? "(nessun testo)"}`, + `memory_id: ${local.memory_id}\n[${local.kind ?? "?"}/${local.scope ?? "?"}${local.project_id ? ` project=${local.project_id}` : ""}] agente: ${local.agent_id ?? "?"}, creato: ${local.created_at ?? "?"}${local.pending ? ", ⏳ creato offline: non ancora sul gateway" : ""}${local.superseded_by ? `, ⚠️ superseduto da ${local.superseded_by}` : ""}${local.remote_id ? `, sincronizzato come ${local.remote_id}` : ""}\n\n${local.text ?? "(nessun testo)"}`, }, ], details: { memory_id: local.memory_id, fallback: "local_sqlite", gateway_status: status }, diff --git a/extensions/tools/search.ts b/extensions/tools/search.ts index cebae90..931f20f 100644 --- a/extensions/tools/search.ts +++ b/extensions/tools/search.ts @@ -109,7 +109,7 @@ export function registerQmemSearch(pi: ExtensionAPI) { if (hits.length) { const lines = hits.map( (h: LocalSearchHit, i: number) => - `${i + 1}. [${h.kind ?? "?"}/${h.scope ?? "?"}${h.project_id ? ` project=${h.project_id}` : ""} locale${h.match_mode === "or" ? " match-parziale(OR)" : ""}${h.superseded_by ? " ⚠️ superseduto" : ""}] ${h.snippet}\n (id: ${h.memory_id}, creato: ${h.created_at ?? "?"}, agente: ${h.agent_id ?? "?"}, fonti: ${h.sources ?? "?"})`, + `${i + 1}. [${h.kind ?? "?"}/${h.scope ?? "?"}${h.project_id ? ` project=${h.project_id}` : ""} locale${h.match_mode === "or" ? " match-parziale(OR)" : ""}${h.pending ? " ⏳ in coda (non ancora sul gateway)" : ""}${h.superseded_by ? " ⚠️ superseduto" : ""}] ${h.snippet}\n (id: ${h.memory_id}, creato: ${h.created_at ?? "?"}, agente: ${h.agent_id ?? "?"}, fonti: ${h.sources ?? "?"})`, ); const header = `⚠️ Gateway non raggiungibile (HTTP ${status}): risultati dall'INDICE LOCALE (SQLite/FTS5).\n` + diff --git a/extensions/tools/store.ts b/extensions/tools/store.ts index 9b405f9..555e1af 100644 --- a/extensions/tools/store.ts +++ b/extensions/tools/store.ts @@ -1,6 +1,7 @@ import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { Type } from "typebox"; import { gatewayRequest, loadConfig } from "../shared"; +import { localDbPath, submitOrQueue } from "../local-db.ts"; export function registerQmemStore(pi: ExtensionAPI) { pi.registerTool({ @@ -81,45 +82,51 @@ export function registerQmemStore(pi: ExtensionAPI) { signal, ); if (dupCheck.ok) dupes = dupCheck.data.results ?? []; - // Idempotency: stessa key per tutta l'operazione (e per eventuali retry) - const idemKey = crypto.randomUUID(); - const { ok, status, data } = await gatewayRequest( - cfg, - "POST", - "/v1/memories", - { - text: p.text, - kind: p.kind ?? "fact", - agent_id: p.agent_id, - project_id: p.project_id, - scope: p.scope ?? "agent", - source: p.source, - confidence: p.confidence ?? "medium", - expires_at: p.expires_at, - supersedes_id: p.supersedes_id, - supersede_reason: p.supersede_reason, - parent_id: p.parent_id, - private: p.private ?? false, - level: p.level, - topic: p.topic, - links: p.links, - }, - signal, - idemKey, - ); - if (!ok) { - const down = status === 0 || status >= 500 || status === 429; + // Idempotency: la chiave è generata da submitOrQueue (Idempotency-Key) + const payload = { + text: p.text, + kind: p.kind ?? "fact", + agent_id: p.agent_id, + project_id: p.project_id, + scope: p.scope ?? "agent", + source: p.source, + confidence: p.confidence ?? "medium", + expires_at: p.expires_at, + supersedes_id: p.supersedes_id, + supersede_reason: p.supersede_reason, + parent_id: p.parent_id, + private: p.private ?? false, + level: p.level, + topic: p.topic, + links: p.links, + }; + // Online → gateway (con Idempotency-Key); offline → coda locale (outbox) + const submitted = await submitOrQueue(cfg, payload, { dbFile: localDbPath(cfg) }); + if (submitted.queued) { return { content: [ { type: "text", text: - `Errore ${status}: ${JSON.stringify(data)}` + - (down - ? "\n⚠️ Il record NON è stato salvato: nessuna coda locale (il gateway è la fonte di verità). Riprova quando è raggiungibile, oppure annota il contenuto e usa /qmem:local import per l'indice testuale." - : ""), + `⚠️ Gateway non raggiungibile (HTTP ${submitted.status}): record ACCODATO in locale (outbox).\n` + + `id locale: ${submitted.local_id}\n` + + `in coda: ${submitted.queue_size} record\n` + + `Il contenuto è già ricercabile offline (indice locale, marcato ⏳) e verrà caricato automaticamente al ritorno della connessione ` + + `(/qmem:local flush per forzare, /qmem:local queue per lo stato).`, }, ], + details: { + queued: true, + local_id: submitted.local_id, + queue_size: submitted.queue_size, + gateway_status: submitted.status, + }, + }; + } + const { ok, status, data } = submitted; + if (!ok) { + return { + content: [{ type: "text", text: `Errore ${status}: ${JSON.stringify(data)}` }], details: { error: "gateway_error", status }, }; } diff --git a/scripts/qmem-sqlite.mjs b/scripts/qmem-sqlite.mjs index 69898d0..2c311c3 100644 --- a/scripts/qmem-sqlite.mjs +++ b/scripts/qmem-sqlite.mjs @@ -7,6 +7,9 @@ * node scripts/qmem-sqlite.mjs import [--db FILE] * node scripts/qmem-sqlite.mjs find "query" [--kind K] [--project P] [--scope S] * [--top N] [--all] [--private] [--exact] [--json] + * node scripts/qmem-sqlite.mjs store --project P [--kind K] [--text "..."] [--queue-only] + * node scripts/qmem-sqlite.mjs queue [--status queued|synced|duplicate|failed] + * node scripts/qmem-sqlite.mjs flush [--limit N] * node scripts/qmem-sqlite.mjs enrich [--all] [--limit N] [--pace MS] * node scripts/qmem-sqlite.mjs pull [--limit N] * @@ -23,7 +26,7 @@ process.emitWarning = (warning, ...rest) => { return originalEmitWarning.call(process, warning, ...rest); }; const localDb = await import("../extensions/local-db.ts"); -const { DEFAULT_DB_FILE, enrichFromGateway, importFromSessions, localDbPath, localDbReport, localSearch, pullFromGatewayExport, sessionRoots } = localDb; +const { DEFAULT_DB_FILE, enrichFromGateway, flushQueue, importFromSessions, localDbPath, localDbReport, localSearch, pullFromGatewayExport, queueList, queueStats, queueStore, sessionRoots, submitOrQueue } = localDb; const { loadConfig } = await import("../extensions/shared.ts"); process.emitWarning = originalEmitWarning; @@ -55,7 +58,12 @@ if (cmd === "status") { console.log(` dimensione : ${r.sizeKb} KB`); console.log(` record : ${r.total} (${r.withText} con testo, ${r.active} attivi, ${r.superseded} superseduti, ${r.private} privati)`); console.log(` con project_id: ${r.withProject}/${r.total}`); - console.log(` ultimo import : ${r.lastImport ?? "-"} enrich: ${r.lastEnrich ?? "-"} export: ${r.lastExport ?? "-"}`); + console.log(` ultimo import : ${r.lastImport ?? "-"} enrich: ${r.lastEnrich ?? "-"} export: ${r.lastExport ?? "-"} flush: ${r.lastFlush ?? "-"}`); + if (r.queued || r.syncedQueue || r.failedQueue || r.duplicateQueue) { + console.log(` coda offline : ${r.queued} in attesa, ${r.syncedQueue} sincronizzati, ${r.duplicateQueue} duplicati, ${r.failedQueue} falliti${r.oldestQueued ? ` (più vecchio: ${r.oldestQueued})` : ""}`); + if (r.queueLastError) console.log(` ultimo errore : ${r.queueLastError.slice(0, 120)}`); + } + if (r.pendingInIndex) console.log(` in indice : ${r.pendingInIndex} record marcati ⏳ (creati offline, non ancora sul gateway)`); console.log(` top progetti : ${r.topProjects.map((p) => `${p.project_id ?? "(null)"}=${p.n}`).join(", ") || "-"}`); if (r.duplicates.length) { console.log(` possibili duplicati (testo identico): ${r.duplicates.length} gruppi — es. ${r.duplicates[0].ids.map((i) => i.slice(0, 8)).join(", ")}`); @@ -120,6 +128,54 @@ if (cmd === "status") { if (stats.errors.length) console.log(` errori : ${stats.errors.join(" | ")}`); if (stats.failed && !stats.ok) console.log(" (gateway non raggiungibile: riprova quando torna online)"); } +} else if (cmd === "store") { + // store con fallback offline: prova il gateway, altrimenti accoda + const cfg = loadConfig(); + const text = typeof opt("text", null) === "string" ? opt("text", null) : fs.readFileSync(String(opt("file", "/dev/stdin")), "utf8").trim(); + const project = typeof opt("project", null) === "string" ? opt("project", null) : ""; + if (!text || !project) { + console.error('Uso: store --project P [--kind K] [--scope S] [--text "..."] | --file FILE [--queue-only]'); + process.exit(2); + } + const payload = { + text, + project_id: project, + kind: typeof opt("kind", null) === "string" ? opt("kind", null) : "fact", + scope: typeof opt("scope", null) === "string" ? opt("scope", null) : "agent", + agent_id: typeof opt("agent", null) === "string" ? opt("agent", null) : undefined, + topic: typeof opt("topic", null) === "string" ? opt("topic", null) : undefined, + }; + if (has("queue-only")) { + const q = await queueStore(payload, { dbFile }); + out(json ? q : `Accodato localmente: ${q.local_id} (in coda: ${q.queue_size})`); + } else { + const res = await submitOrQueue(cfg, payload, { dbFile }); + if (json) out(res); + else if (res.queued) out(`Gateway non raggiungibile (HTTP ${res.status}): accodato localmente ${res.local_id} (in coda: ${res.queue_size}). Flush: qmem-sqlite flush`); + else if (res.remote_id) out(`Salvato sul gateway: ${res.remote_id}`); + else out(`Errore HTTP ${res.status}: ${JSON.stringify(res.data)}`); + } +} else if (cmd === "queue") { + const stats = await queueStats({ dbFile }); + const items = await queueList({ dbFile, status: typeof opt("status", null) === "string" ? opt("status", null) : undefined, limit: Number(opt("limit", 20)) || 20 }); + if (json) out({ stats, items }); + else { + console.log(`Coda offline (${dbFile})`); + console.log(` in attesa: ${stats.queued} sincronizzati: ${stats.synced} duplicati: ${stats.duplicate} falliti: ${stats.failed}`); + if (stats.oldestQueued) console.log(` più vecchio: ${stats.oldestQueued}`); + if (stats.lastError) console.log(` ultimo errore: ${stats.lastError.slice(0, 140)}`); + items.forEach((q, i) => console.log(` ${i + 1}. [${q.status}] ${q.local_id.slice(0, 8)} [${q.payload.kind ?? "fact"}/${q.payload.project_id}] ${String(q.payload.text).slice(0, 70)}`)); + } +} else if (cmd === "flush") { + const cfg = loadConfig(); + const res = await flushQueue(cfg, { dbFile, limit: Number(opt("limit", 200)) || 200, paceMs: Number(opt("pace", 300)) }); + if (json) out(res); + else { + console.log(`Flush outbox verso ${cfg.url}`); + console.log(` sincronizzati: ${res.synced} duplicati: ${res.duplicates} falliti: ${res.failed} ancora in coda: ${res.remaining}`); + if (res.stopped) console.log(` fermato: ${res.stopped}`); + if (res.errors.length) console.log(` errori: ${res.errors.join(" | ")}`); + } } else if (cmd === "pull") { const cfg = loadConfig(); const res = await pullFromGatewayExport(cfg, { dbFile, limit: Number(opt("limit", 500)) || 500 }); @@ -129,7 +185,7 @@ if (cmd === "status") { console.log(` supportato: ${res.supported} pagine: ${res.pages} record: ${res.fetched}${res.message ? ` — ${res.message}` : ""}`); } } else { - console.error(`Comando sconosciuto: ${cmd}\nComandi: status | import | find | enrich | pull`); + console.error(`Comando sconosciuto: ${cmd}\nComandi: status | import | find | store | queue | flush | enrich | pull`); process.exit(2); } diff --git a/scripts/test-local.mjs b/scripts/test-local.mjs index 61882ce..bfa387f 100644 --- a/scripts/test-local.mjs +++ b/scripts/test-local.mjs @@ -10,6 +10,7 @@ * Uso: node scripts/test-local.mjs (esce != 0 se un controllo fallisce) */ import { spawn, spawnSync } from "node:child_process"; +import * as crypto from "node:crypto"; import { createServer } from "node:http"; import * as fs from "node:fs"; import * as os from "node:os"; @@ -84,8 +85,15 @@ const runCliAsync = (args) => new Promise((resolve) => { const child = spawn("node", [path.join(HERE, "qmem-sqlite.mjs"), ...args], { env }); let out = ""; + let err = ""; child.stdout.on("data", (d) => (out += d)); - child.on("close", () => resolve(out)); + child.stderr.on("data", (d) => (err += d)); + child.on("close", (code) => { + if (process.env.DEBUG_CLI && (!out.trim() || code !== 0)) { + console.log(` [cli ${args.join(" ")}] code=${code} stdout=${out.trim().slice(0, 200)} stderr=${err.trim().slice(0, 200)}`); + } + resolve(out); + }); }); const imp = runCli(["import", "--json"]); let impJson = null; @@ -148,11 +156,30 @@ check("risultato locale etichettato come testuale", /INDICE LOCALE/i.test(search const getRes = await tools.get("qmem_get").execute("t2", { memory_id: ID_E }, undefined, undefined, ctx); if (process.env.DEBUG_GET) console.log("GET RESULT:", JSON.stringify(getRes, null, 1).slice(0, 900)); check("qmem_get → fallback locale per UUID", getRes.details?.fallback === "local_sqlite" && /nmap/.test(getRes.content[0].text)); -const storeRes = await tools.get("qmem_store").execute("t3", { text: "prova", project_id: "test-project" }, undefined, undefined, ctx); -check("qmem_store avvisa che il record NON è salvato", /NON è stato salvato/i.test(storeRes.content[0].text)); +const storeRes = await tools.get("qmem_store").execute("t3", { text: "Record creato offline durante un outage del gateway.", project_id: "test-project", kind: "fact" }, undefined, undefined, ctx); +const queuedLocalId = storeRes.details?.local_id; +check("qmem_store offline → accoda nell'outbox", storeRes.details?.queued === true && /ACCODATO/i.test(storeRes.content[0].text), `local_id=${String(queuedLocalId).slice(0, 8)} coda=${storeRes.details?.queue_size}`); // ---------------------------------------------------------------- 4) arricchimento dal gateway (stub) +const posted = []; const server = createServer((req, res) => { + if (req.method === "POST" && req.url === "/v1/memories") { + let raw = ""; + req.on("data", (c) => (raw += c)); + req.on("end", () => { + let body = {}; + try { body = JSON.parse(raw); } catch { /* ignore */ } + posted.push({ body, idem: req.headers["idempotency-key"] }); + if (/duplicato/i.test(body.text ?? "")) { + res.writeHead(409, { "content-type": "application/json" }).end(JSON.stringify({ detail: { error: "duplicate_memory", reason: "KNOWN_SOLUTION", matches: [{ memory_id: ID_A, score: 0.93 }] } })); + } else if (/invalido/i.test(body.text ?? "")) { + res.writeHead(422, { "content-type": "application/json" }).end(JSON.stringify({ detail: [{ loc: ["body", "project_id"], msg: "campo obbligatorio" }] })); + } else { + res.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify({ memory_id: crypto.randomUUID(), text: body.text, kind: body.kind, project_id: body.project_id, scope: body.scope, created_at: new Date().toISOString(), supersedes_id: body.supersedes_id })); + } + }); + return; + } if (req.url?.startsWith("/v1/memories:export")) { res.writeHead(404, { "content-type": "application/json" }).end(JSON.stringify({ detail: "non trovato" })); return; @@ -184,8 +211,44 @@ const pullOut = await runCliAsync(["pull", "--json"]); const pull = { stdout: pullOut }; const pullJson = JSON.parse(pull.stdout || "{}"); check("pull segnala endpoint export assente", pullJson.supported === false, pullJson.message ?? ""); -server.close(); +// ---------------------------------------------------------------- 5) outbox: flush, 409, 422, supersede +const q1 = JSON.parse((await runCliAsync(["queue", "--json"])) || "{}"); +check("coda: 1 record in attesa dopo lo store offline", q1.stats?.queued >= 1, `queued=${q1.stats?.queued} localId=${String(queuedLocalId).slice(0, 8)}`); +const findPending = JSON.parse(runCli(["find", "outage gateway", "--json"]).stdout || "[]"); +check("il record accodato è già ricercabile offline (pending=1)", findPending.some((h) => h.pending === 1 && h.memory_id === queuedLocalId), `${findPending.length} hit`); + +// seconda voce: duplicato (409) e terza: invalida (422), più un supersede verso un local_id +const dupStore = JSON.parse((await runCliAsync(["store", "--project", "test-project", "--text", "Questo è un duplicato noto del gateway", "--queue-only", "--json"])) || "{}"); +const badStore = JSON.parse((await runCliAsync(["store", "--project", "test-project", "--text", "Record invalido per test 422", "--queue-only", "--json"])) || "{}"); +const childStore = JSON.parse((await runCliAsync(["store", "--project", "test-project", "--text", "Correzione offline di un record locale", "--queue-only", "--json"])) || "{}"); +// supersede verso il record locale: il flush deve rimappare local_id → remote_id +const dbMod = spawnSync("node", ["-e", ` +const { DatabaseSync } = require("node:sqlite"); +const db = new DatabaseSync(process.env.QMEM_SQLITE); +const row = db.prepare("SELECT payload FROM pending WHERE local_id = ?").get(${JSON.stringify(childStore.local_id)}); +const p = JSON.parse(row.payload); p.supersedes_id = ${JSON.stringify(queuedLocalId)}; +db.prepare("UPDATE pending SET payload = ? WHERE local_id = ?").run(JSON.stringify(p), ${JSON.stringify(childStore.local_id)}); +`], { env, encoding: "utf8" }); +check("setup supersede offline (payload con supersedes_id locale)", dbMod.status === 0, dbMod.stderr?.slice(0, 80) ?? ""); + +const flush1 = JSON.parse((await runCliAsync(["flush", "--json"])) || "{}"); +check("flush: 2 sincronizzati, 1 duplicato (409), 1 fallito (422), coda vuota", flush1.synced === 2 && flush1.duplicates === 1 && flush1.failed === 1 && flush1.remaining === 0, `processed=${flush1.processed} synced=${flush1.synced} duplicates=${flush1.duplicates} failed=${flush1.failed} remaining=${flush1.remaining}`); +const remoteOfQueued = flush1.remoteIds?.[queuedLocalId]; +check("Idempotency-Key = local_id inviato al gateway", posted.every((p) => typeof p.idem === "string" && p.idem.length === 36), `${posted.length} POST`); +check("supersede rimappato da local_id a remote_id", posted.some((p) => p.body.supersedes_id === remoteOfQueued && remoteOfQueued), `remote=${String(remoteOfQueued).slice(0, 8)}`); +const afterFlush = JSON.parse((await runCliAsync(["status", "--json"])) || "{}"); +check("dopo il flush resta pendente solo il record fallito (422)", afterFlush.queued === 0 && afterFlush.pendingInIndex === 1 && afterFlush.failedQueue === 1, `queued=${afterFlush.queued} pending_in_index=${afterFlush.pendingInIndex} failed=${afterFlush.failedQueue}`); +const findRemote = JSON.parse(runCli(["find", "outage gateway", "--json"]).stdout || "[]"); +check("il record è ricercabile con l'ID remoto", findRemote.some((h) => h.memory_id === remoteOfQueued && h.pending === 0), `${findRemote.length} hit`); + +// duplicato (409) e invalido (422) +const dupFlush = JSON.parse((await runCliAsync(["flush", "--json"])) || "{}"); +const qFinal = JSON.parse((await runCliAsync(["queue", "--json"])) || "{}"); +const dupItem = (qFinal.items ?? []).find((i) => i.local_id === dupStore.local_id); +const badItem = (qFinal.items ?? []).find((i) => i.local_id === badStore.local_id); +check("duplicato marcato con remote_id del match", dupItem?.status === "duplicate" && dupItem?.remote_id === ID_A, `status=${dupItem?.status} remote=${String(dupItem?.remote_id).slice(0, 8)}`); +check("422 marcato failed con errore conservato", badItem?.status === "failed" && /422/.test(badItem?.last_error ?? ""), `err=${String(badItem?.last_error).slice(0, 40)}`); // ---------------------------------------------------------------- report const failed = results.filter((r) => !r.ok); console.log(`\n${results.length - failed.length}/${results.length} controlli superati — HOME di test: ${HOME}`); diff --git a/skills/qmem/SKILL.md b/skills/qmem/SKILL.md index 0191e99..1afbea3 100644 --- a/skills/qmem/SKILL.md +++ b/skills/qmem/SKILL.md @@ -42,7 +42,20 @@ In quel caso: - comandi: `/qmem:local status | import | find | enrich | pull` (`import` dalle sessioni, `enrich`/`pull` dal gateway quando torna online). -`qmem_store` non ha coda locale: con il gateway giù il record **non** viene -salvato. Annota il contenuto e riscrivilo quando il gateway è raggiungibile. +`qmem_store` accoda in locale: con il gateway giù il record entra nell'**outbox** +locale (SQLite), è subito ricercabile (marcato ⏳) e viene inviato al gateway al +ritorno della connessione (flush automatico su `session_start`, oppure +`/qmem:local flush`). Finché non è sincronizzato **non** è nella memoria +condivisa: i risultati locali marcati ⏳ non sono visibili agli altri agenti. + +Stato e gestione della coda: + +- `/qmem:local queue` — voci in attesa, tentativi, ultimo errore +- `/qmem:local flush` — invio immediato (idempotente: `Idempotency-Key` = id locale) +- esiti: `synced` (sul gateway, il record locale adotta l'ID remoto), `duplicate` + (409: era già presente, viene registrato l'ID del match), `failed` (4xx di + validazione: non ritentato in automatico) +- supersede offline: una correzione che punta a un record ancora locale viene + rimappata all'ID remoto al momento del flush