diff --git a/package.json b/package.json index 27560ef..6a3e304 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@four-bytes/four-opencode-brain", - "version": "1.7.2", + "version": "1.7.3", "description": "Unified brain plugin — single SQLite DB for RAG search, memory, and knowledge base", "license": "Apache-2.0", "type": "module", diff --git a/src/four-opencode-brain.ts b/src/four-opencode-brain.ts index c72a2a6..595d57c 100644 --- a/src/four-opencode-brain.ts +++ b/src/four-opencode-brain.ts @@ -343,22 +343,25 @@ const _serverPlugin = async (input: PluginInput) => { updateStatus("busy", { text: "Rebuilding vector index…" }); const db = initBrainDatabase(); try { - db.run("DROP TABLE IF EXISTS chunks_vec"); - db.run(` - CREATE VIRTUAL TABLE IF NOT EXISTS chunks_vec USING vec0( - chunk_id TEXT PRIMARY KEY, - embedding FLOAT[384] - ) - `); - - const chunkRows = db.query<{ id: string }, []>( - "SELECT id FROM chunks", - ).all(); - const totalChunks = chunkRows.length; + // Sync DB operations wrapped with retry for SQLITE_BUSY resilience + const chunkIds = await withDbRetry(() => { + db.run("DROP TABLE IF EXISTS chunks_vec"); + db.run(` + CREATE VIRTUAL TABLE IF NOT EXISTS chunks_vec USING vec0( + chunk_id TEXT PRIMARY KEY, + embedding FLOAT[384] + ) + `); + + const chunkRows = db.query<{ id: string }, []>( + "SELECT id FROM chunks", + ).all(); + return chunkRows.map((r) => r.id); + }); + const totalChunks = chunkIds.length; let embedded = 0; if (totalChunks > 0 && loadVec0(db)) { - const chunkIds = chunkRows.map((r) => r.id); embedded = await embedChunks(db, chunkIds); } @@ -514,7 +517,7 @@ const _serverPlugin = async (input: PluginInput) => { const db = initBrainDatabase(); try { updateStatus("busy", { text: "Saving knowledge entry…" }); - const result = kbAdd(db, { + const result = await withDbRetry(() => kbAdd(db, { entry_key: (args.entry_key as string) ?? deriveEntryKey(args.title as string), kind: (args.kind as string) ?? "problem", title: args.title as string, @@ -525,7 +528,7 @@ const _serverPlugin = async (input: PluginInput) => { tags: args.tags as string | undefined, confidence: args.confidence as number | undefined, review_state: args.review_state as string | undefined, - } satisfies KbAddInput); + } satisfies KbAddInput)); updateStatus("success", { text: "Knowledge entry saved", toast: "Knowledge entry saved" }); return JSON.stringify(result); } catch (err) { @@ -555,7 +558,7 @@ const _serverPlugin = async (input: PluginInput) => { const db = initBrainDatabase(); try { updateStatus("busy", { text: "Recording occurrence…" }); - const occurrence = kbRecord(db, { + const occurrence = await withDbRetry(() => kbRecord(db, { entry_key: args.entry_key as string, kind: args.kind as string, project_ref: args.project_ref as string | undefined, @@ -564,7 +567,7 @@ const _serverPlugin = async (input: PluginInput) => { commit_ref: args.commit_ref as string | undefined, observed_symptoms: args.observed_symptoms as string | undefined, outcome: args.outcome as "fixed" | "failed" | "workaround" | "observed", - } satisfies KbRecordInput); + } satisfies KbRecordInput)); updateStatus("success", { text: "Occurrence recorded", toast: "Occurrence recorded" }); return JSON.stringify(occurrence); } catch (err) { @@ -590,12 +593,12 @@ const _serverPlugin = async (input: PluginInput) => { const db = initBrainDatabase(); try { updateStatus("busy", { text: "Updating review…" }); - const entry = kbReview(db, { + const entry = await withDbRetry(() => kbReview(db, { entry_key: args.entry_key as string, kind: args.kind as string, review_state: args.review_state as "draft" | "reviewed" | "accepted" | "rejected" | "superseded", confidence: args.confidence as number | undefined, - } satisfies KbReviewInput); + } satisfies KbReviewInput)); updateStatus("success", { text: "Review updated", toast: "Review updated" }); return JSON.stringify(entry); } catch (err) { diff --git a/src/ingest/index.ts b/src/ingest/index.ts index f95fe9d..66a7e8b 100644 --- a/src/ingest/index.ts +++ b/src/ingest/index.ts @@ -11,7 +11,7 @@ // 8. Extract symbols (code files only, 10s timeout) // 9. Chunk content (token-based) // 10. Insert chunks (dedup via BEFORE INSERT trigger, includes token_count) -// 11. Wrap in SAVEPOINT for atomicity +// 11. Per-file SAVEPOINT for atomicity (individual files, not the entire ingest) // --------------------------------------------------------------------------- import { stat } from "fs/promises"; @@ -80,8 +80,8 @@ function emitProgressEvent(event: string, data: Record): void { * Implements content-hash dedup: files with unchanged content are skipped. * New/updated files are inserted into `files`, `documents`, and `chunks` tables. * - * The entire operation is wrapped in a SAVEPOINT. On error, all changes - * are rolled back. + * Each file is wrapped in a per-file SAVEPOINT for atomicity. On per-file error, + * only that file's changes are rolled back and processing continues to the next file. */ export async function ingestPath( db: Database, @@ -144,15 +144,10 @@ export async function ingestPath( return result; } - // ── 3. SAVEPOINT: wrap the entire ingest ──────────────────────────── - const sp = "brain_ingest_" + generateId(); - db.exec(`SAVEPOINT ${sp}`); - // Yield so event loop can handle HTTP requests + tool calls before ingest starts await new Promise(r => setTimeout(r, 0)); - try { - for (const [i, walked] of walkedFiles.entries()) { + for (const [i, walked] of walkedFiles.entries()) { const filePath = walked.path; const language = walked.language; const fileStart = Date.now(); @@ -195,6 +190,11 @@ export async function ingestPath( continue; } + // ── Per-file SAVEPOINT for atomic DB writes ──────────────────────── + const spFile = "brain_file_" + generateId(); + db.exec(`SAVEPOINT ${spFile}`); + let fileFailed = false; + const contentHash = hashBuffer(new Uint8Array(buf)); const size = buf.byteLength; @@ -207,6 +207,7 @@ export async function ingestPath( .get(filePath); if (existing && existing.content_hash === contentHash) { + db.exec(`RELEASE SAVEPOINT ${spFile}`); result.filesSkipped++; result.filesProcessed++; options?.progressCallback?.({ current: result.filesProcessed, total: walkedFiles.length }); @@ -234,6 +235,7 @@ export async function ingestPath( [fileId, filePath, contentHash, mtime, language, size], ); } catch (err) { + db.exec(`ROLLBACK TO SAVEPOINT ${spFile}`); result.errors.push(`Failed to insert file ${filePath}: ${String(err)}`); continue; } @@ -261,6 +263,7 @@ export async function ingestPath( result.documentsCreated++; } } catch (err) { + db.exec(`ROLLBACK TO SAVEPOINT ${spFile}`); result.errors.push(`Failed to insert document ${filePath}: ${String(err)}`); continue; } @@ -279,6 +282,7 @@ export async function ingestPath( totalLines, }); } catch (err) { + db.exec(`ROLLBACK TO SAVEPOINT ${spFile}`); result.errors.push(`Failed to chunk ${filePath}: ${String(err)}`); continue; } @@ -313,22 +317,37 @@ export async function ingestPath( result.chunksCreated++; newChunkIds.push(chunk.id); } catch (err) { + db.exec(`ROLLBACK TO SAVEPOINT ${spFile}`); result.errors.push( `Failed to insert chunk ${chunk.chunkIndex} for ${filePath}: ${String(err)}`, ); + fileFailed = true; + break; } } + if (fileFailed) { + result.filesProcessed++; + continue; + } + // ── 10. Embed chunks (vec0) — non-fatal if unavailable ───────── if (newChunkIds.length > 0 && loadVec0(db)) { try { const embedded = await embedChunks(db, newChunkIds); result.chunksEmbedded += embedded; } catch (err) { - result.errors.push(`Embedding failed (non-fatal): ${String(err)}`); + db.exec(`ROLLBACK TO SAVEPOINT ${spFile}`); + result.errors.push(`Embedding failed: ${String(err)}`); + fileFailed = true; } } + if (fileFailed) { + result.filesProcessed++; + continue; + } + // ── 11. Populate symbols table (global symbol store) ───────────── for (const chunk of chunks) { if (chunk.symbol) { @@ -353,6 +372,9 @@ export async function ingestPath( } } + // ── RELEASE per-file SAVEPOINT ───────────────────────────── + db.exec(`RELEASE SAVEPOINT ${spFile}`); + result.filesIndexed++; result.filesProcessed++; options?.progressCallback?.({ current: result.filesProcessed, total: walkedFiles.length }); @@ -366,14 +388,6 @@ export async function ingestPath( await new Promise(r => setTimeout(r, 0)); } - // ── Commit ──────────────────────────────────────────────────────── - db.exec(`RELEASE SAVEPOINT ${sp}`); - } catch (err) { - // ── Rollback on error ──────────────────────────────────────────── - db.exec(`ROLLBACK TO SAVEPOINT ${sp}`); - result.errors.push(`Ingest failed: ${String(err)}`); - } - // Checkpoint WAL to keep file manageable after large ingests checkpointDatabase(db); diff --git a/src/schema.ts b/src/schema.ts index 74c06cc..d3ef5ad 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -243,13 +243,38 @@ function migrateV2toV3(db: Database): void { } } +/** + * Read the stored schema_version from the metadata table. + * Returns 0 if the metadata table does not exist yet (fresh database). + */ +export function getSchemaVersion(db: Database): number { + try { + const row = db + .query<{ value: string }, []>("SELECT value FROM metadata WHERE key = 'schema_version'") + .get(); + return row ? parseInt(row.value, 10) || 0 : 0; + } catch { + // metadata table may not exist yet on first run + return 0; + } +} + /** * Open the brain database, load the vec0 extension, create the schema, * run pending migrations, and perform integrity checks. * This is the single entry point — ensures vec0 is loaded before schema creation. + * Uses fast-path if schema is already up-to-date. */ export function initBrainDatabase(dbPath?: string): Database { const db = openDatabase(dbPath); + const currentVersion = getSchemaVersion(db); + if (currentVersion >= SCHEMA_VERSION) { + // Fast path: schema is up-to-date, just load vec0 + integrity checks + loadVec0(db); + runIntegrityChecks(db); + return db; + } + // Cold path: load vec0, create schema, run migrations, integrity checks loadVec0(db); createSchema(db); runMigrations(db); @@ -366,13 +391,6 @@ export function createSchema(db: Database): void { ) `); - // ---- Migration: add entity_type CHECK for existing databases ----------- - migrateConfidenceCheck(db); - migrateReviewStateCheck(db); - migrateOccurrenceOutcomeCheck(db); - migrateKnowledgeFts(db); - migrateEntityTypeCheck(db); - db.exec(` CREATE TABLE IF NOT EXISTS knowledge_occurrences ( id TEXT PRIMARY KEY,