From 78bfd229092669783db179b02368e7cd8f90eb74 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 14:34:00 +0000 Subject: [PATCH 1/4] perf(mobile): keep chat history safe on failed loads, index and cache RAG vectors - Stores never open their write gate after a failed read, so an empty store can no longer overwrite saved history; a snapshot counts as saved only after its write commits, so a failed write is retried. - Android AsyncStorage size cap raised to 64 MB (history was at 5.79 MB of 6). - RAG: indexes on embeddings(doc_id), chunks(doc_id, position) and documents(project_id, enabled); per-document deletes are atomic; decoded vectors are cached (16 MB LRU) and checked against the live embedding ids, with the original tie order pinned. - Pro: sync store uses WAL; chat projection repair skips shown messages. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016JzAmMBhYRz9FHP1r3mEVr --- .../integration/rag/embeddingFlow.test.ts | 17 +- .../rag/embeddingVectorCache.test.ts | 414 ++++++++++++++++++ .../chatProjectionRepair.integration.test.ts | 186 ++++++++ __tests__/unit/services/rag/database.test.ts | 2 +- .../unit/utils/hydrationGatedStorage.test.ts | 149 ++++++- android/gradle.properties | 5 + pro | 2 +- src/services/rag/database.ts | 202 +++++++-- src/services/rag/vectorMath.ts | 9 +- src/stores/appStore.ts | 2 +- src/stores/authStore.ts | 2 +- src/stores/chatStore.ts | 2 +- src/stores/projectStore.ts | 2 +- src/stores/remoteServerStore.ts | 2 +- src/stores/whisperStore.ts | 2 +- src/utils/hydrationGatedStorage.ts | 40 +- 16 files changed, 982 insertions(+), 56 deletions(-) create mode 100644 __tests__/integration/rag/embeddingVectorCache.test.ts create mode 100644 __tests__/pro/sync/chatProjectionRepair.integration.test.ts diff --git a/__tests__/integration/rag/embeddingFlow.test.ts b/__tests__/integration/rag/embeddingFlow.test.ts index a016b222e..300e4a9a7 100644 --- a/__tests__/integration/rag/embeddingFlow.test.ts +++ b/__tests__/integration/rag/embeddingFlow.test.ts @@ -118,14 +118,17 @@ describe('Embedding Flow Integration', () => { const mlBuffer = new Float32Array(mlEmbed).buffer; const cookBuffer = new Float32Array(cookEmbed).buffer; - mockExecuteSync.mockImplementation((sql: string) => { + const rows = [ + { embedding_id: 1, chunk_rowid: 1, doc_id: 1, name: 'ml.pdf', content: 'Machine learning algorithms', position: 0, embedding: mlBuffer }, + { embedding_id: 2, chunk_rowid: 2, doc_id: 2, name: 'recipes.pdf', content: 'Chocolate cake recipe', position: 0, embedding: cookBuffer }, + ]; + mockExecuteSync.mockImplementation((sql: string, params?: unknown[]) => { if (typeof sql === 'string' && sql.includes('rag_embeddings') && sql.includes('SELECT')) { - return { - rows: [ - { chunk_rowid: 1, doc_id: 1, name: 'ml.pdf', content: 'Machine learning algorithms', position: 0, embedding: mlBuffer }, - { chunk_rowid: 2, doc_id: 2, name: 'recipes.pdf', content: 'Chocolate cake recipe', position: 0, embedding: cookBuffer }, - ], - }; + // The per-document vector read is scoped by doc_id; the project read is not. + if (sql.includes('e.doc_id = ?')) { + return { rows: rows.filter(row => row.doc_id === params?.[0]) }; + } + return { rows }; } return { rows: [], insertId: 0, rowsAffected: 0 }; }); diff --git a/__tests__/integration/rag/embeddingVectorCache.test.ts b/__tests__/integration/rag/embeddingVectorCache.test.ts new file mode 100644 index 000000000..bf0a309ce --- /dev/null +++ b/__tests__/integration/rag/embeddingVectorCache.test.ts @@ -0,0 +1,414 @@ +/** + * Knowledge-base search keeps decoded embedding vectors between chat turns. These tests pin what that must + * never change: the chunks a search returns, their scores and their order (ties included), and how quickly + * a write shows up in the next search. + * + * Everything runs over a REAL SQLite engine (node:sqlite), so documents, chunks and embedding BLOBs are + * written and read back with the module's actual SQL. The only stand-in is the embedding MODEL (native + * llama.rn), replaced by deterministic keyword vectors so cosine ranking is real rather than canned. + * + * The last block opens a database written by an older release (no indexes, optionally no sync_id column) + * from a file, the way op-sqlite opens rag.db by name on a device, and checks it upgrades in place. + */ +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { installRealSqlite } from '../../harness/sqliteFake'; + +const KEYWORDS = ['solar', 'battery', 'inverter', 'ledger']; +const toVec = (text: string): number[] => { + const lower = String(text).toLowerCase(); + return KEYWORDS.map(k => (lower.includes(k) ? 1 : 0)); +}; + +/** The embedding model boundary: same surface as embeddingService, deterministic vectors. */ +function installEmbeddingModel(): void { + jest.doMock('../../../src/services/rag/embedding', () => ({ + embeddingService: { + isLoaded: () => true, + load: async () => undefined, + unload: async () => undefined, + getDimension: () => KEYWORDS.length, + embed: async (text: string) => toVec(text), + embedBatch: async (texts: string[]) => texts.map(toVec), + }, + })); +} + +type Chunk = { doc_id: number; name: string; content: string; position: number; score: number }; +type RagDb = { + ensureReady: () => Promise; + insertDocument: (doc: { projectId: string; name: string; path: string; size: number }) => number; + insertChunks: (docId: number, chunks: { content: string; position: number }[]) => number[]; + insertEmbeddingsBatch: (entries: { chunkRowid: number; docId: number; embedding: number[] }[]) => void; + getDocument: (docId: number) => { sync_id: string } | undefined; +}; +type Rag = { + ragDatabase: RagDb; + ragService: { + deleteDocument: (docId: number) => Promise; + toggleDocument: (docId: number, enabled: boolean) => Promise; + deleteProjectDocuments: (projectId: string) => Promise; + backfillEmbeddings: (projectId: string) => Promise; + setSyncedDocumentEnabled: (syncId: string, enabled: boolean) => Promise; + deleteSyncedDocument: (syncId: string) => Promise; + getDocumentsByProject: (projectId: string) => Promise<{ name: string }[]>; + }; + search: (projectId: string, query: string, topK?: number) => Promise; +}; + +function requireRag(): Rag { + const { ragDatabase } = require('../../../src/services/rag/database'); + const { ragService } = require('../../../src/services/rag'); + const { retrievalService } = require('../../../src/services/rag/retrieval'); + return { + ragDatabase, + ragService, + search: async (projectId, query, topK = 10) => + (await retrievalService.search(projectId, query, topK)).chunks, + }; +} + +async function freshRag(): Promise { + installRealSqlite(); + installEmbeddingModel(); + const rag = requireRag(); + await rag.ragDatabase.ensureReady(); + return rag; +} + +/** Index a document through the real write path; `vectors` overrides the embedding per chunk. */ +function addDocument( + db: RagDb, + projectId: string, + name: string, + texts: string[], + vectors: number[][] = texts.map(toVec), +): number { + const docId = db.insertDocument({ projectId, name, path: `/docs/${name}`, size: texts.join('').length }); + const rowIds = db.insertChunks(docId, texts.map((content, position) => ({ content, position }))); + db.insertEmbeddingsBatch(rowIds.map((chunkRowid, i) => ({ chunkRowid, docId, embedding: vectors[i] }))); + return docId; +} + +const label = (chunks: Chunk[]) => chunks.map(c => `${c.name}#${c.position}`); + +describe('repeated knowledge-base searches', () => { + it('return exactly the same chunks, scores and tie order as the first search', async () => { + const { ragDatabase, search } = await freshRag(); + // Two documents whose embeddings are written interleaved, all tying on the query, plus one weaker + // match. Ties have always come back in the order the embeddings were stored, not grouped by document. + const a = ragDatabase.insertDocument({ projectId: 'p1', name: 'a.txt', path: '/a', size: 1 }); + const b = ragDatabase.insertDocument({ projectId: 'p1', name: 'b.txt', path: '/b', size: 1 }); + const [a0, a1] = ragDatabase.insertChunks(a, [ + { content: 'solar one', position: 0 }, + { content: 'solar two', position: 1 }, + ]); + const [b0, b1, b2] = ragDatabase.insertChunks(b, [ + { content: 'solar three', position: 0 }, + { content: 'solar four', position: 1 }, + { content: 'solar battery five', position: 2 }, + ]); + const solar = [1, 0, 0, 0]; + ragDatabase.insertEmbeddingsBatch([ + { chunkRowid: b2, docId: b, embedding: [1, 1, 0, 0] }, + { chunkRowid: a0, docId: a, embedding: solar }, + { chunkRowid: b0, docId: b, embedding: solar }, + { chunkRowid: a1, docId: a, embedding: solar }, + { chunkRowid: b1, docId: b, embedding: solar }, + ]); + + const first = await search('p1', 'solar'); + const second = await search('p1', 'solar'); + const third = await search('p1', 'solar'); + + expect(label(first)).toEqual(['a.txt#0', 'b.txt#0', 'a.txt#1', 'b.txt#1', 'b.txt#2']); + expect(first.slice(0, 4).map(c => c.score)).toEqual([1, 1, 1, 1]); + expect(first[4].score).toBeCloseTo(Math.SQRT1_2, 6); + expect(second).toEqual(first); + expect(third).toEqual(first); + }); + + it('respect topK the same way on every search', async () => { + const { ragDatabase, search } = await freshRag(); + addDocument(ragDatabase, 'p1', 'a.txt', ['solar a', 'battery a', 'ledger a']); + addDocument(ragDatabase, 'p1', 'b.txt', ['solar battery b', 'inverter b']); + + const first = await search('p1', 'solar battery', 2); + const second = await search('p1', 'solar battery', 2); + + expect(label(first)).toEqual(['b.txt#0', 'a.txt#0']); + expect(second).toEqual(first); + }); +}); + +describe('a write shows up in the very next search', () => { + it('a newly indexed document', async () => { + const { ragDatabase, search } = await freshRag(); + addDocument(ragDatabase, 'p1', 'a.txt', ['solar and battery']); + expect(label(await search('p1', 'inverter'))).toEqual(['a.txt#0']); + + addDocument(ragDatabase, 'p1', 'b.txt', ['inverter manual']); + + expect(label(await search('p1', 'inverter'))).toEqual(['b.txt#0', 'a.txt#0']); + }); + + it('embeddings added to a document that was already searched', async () => { + const { ragDatabase, search } = await freshRag(); + const docId = ragDatabase.insertDocument({ projectId: 'p1', name: 'a.txt', path: '/a', size: 1 }); + const [c0, c1] = ragDatabase.insertChunks(docId, [ + { content: 'solar intro', position: 0 }, + { content: 'ledger detail', position: 1 }, + ]); + ragDatabase.insertEmbeddingsBatch([{ chunkRowid: c0, docId, embedding: toVec('solar intro') }]); + expect(label(await search('p1', 'ledger'))).toEqual(['a.txt#0']); + + ragDatabase.insertEmbeddingsBatch([{ chunkRowid: c1, docId, embedding: toVec('ledger detail') }]); + + expect(label(await search('p1', 'ledger'))).toEqual(['a.txt#1', 'a.txt#0']); + }); + + it('a backfill that embeds a document indexed without embeddings', async () => { + const { ragDatabase, ragService, search } = await freshRag(); + addDocument(ragDatabase, 'p1', 'a.txt', ['solar notes']); + const legacy = ragDatabase.insertDocument({ projectId: 'p1', name: 'old.txt', path: '/old', size: 1 }); + ragDatabase.insertChunks(legacy, [{ content: 'battery notes', position: 0 }]); + expect(label(await search('p1', 'battery'))).toEqual(['a.txt#0']); + + await ragService.backfillEmbeddings('p1'); + + expect(label(await search('p1', 'battery'))).toEqual(['old.txt#0', 'a.txt#0']); + }); + + it('a document replaced by a new version', async () => { + const { ragDatabase, ragService, search } = await freshRag(); + const old = addDocument(ragDatabase, 'p1', 'spec.txt', ['solar spec v1']); + addDocument(ragDatabase, 'p1', 'other.txt', ['battery other']); + expect((await search('p1', 'solar'))[0].content).toBe('solar spec v1'); + + await ragService.deleteDocument(old); + addDocument(ragDatabase, 'p1', 'spec.txt', ['solar spec v2', 'inverter appendix']); + + const after = await search('p1', 'solar'); + expect(after.map(c => c.content)).toEqual(['solar spec v2', 'battery other', 'inverter appendix']); + }); + + it('a deleted document, down to an empty knowledge base', async () => { + const { ragDatabase, ragService, search } = await freshRag(); + const a = addDocument(ragDatabase, 'p1', 'a.txt', ['solar a']); + const b = addDocument(ragDatabase, 'p1', 'b.txt', ['solar b']); + expect(label(await search('p1', 'solar'))).toEqual(['a.txt#0', 'b.txt#0']); + + await ragService.deleteDocument(a); + expect(label(await search('p1', 'solar'))).toEqual(['b.txt#0']); + + await ragService.deleteDocument(b); + expect(await search('p1', 'solar')).toEqual([]); + expect(await ragService.getDocumentsByProject('p1')).toEqual([]); + }); + + it('a document disabled and enabled again', async () => { + const { ragDatabase, ragService, search } = await freshRag(); + const a = addDocument(ragDatabase, 'p1', 'a.txt', ['solar a', 'battery a']); + addDocument(ragDatabase, 'p1', 'b.txt', ['solar b']); + const before = await search('p1', 'solar'); + + await ragService.toggleDocument(a, false); + expect(label(await search('p1', 'solar'))).toEqual(['b.txt#0']); + + await ragService.toggleDocument(a, true); + expect(await search('p1', 'solar')).toEqual(before); + }); + + it('enable, disable and delete arriving from sync', async () => { + const { ragDatabase, ragService, search } = await freshRag(); + const a = addDocument(ragDatabase, 'p1', 'a.txt', ['solar a']); + addDocument(ragDatabase, 'p1', 'b.txt', ['solar b']); + const syncId = ragDatabase.getDocument(a)!.sync_id; + expect(label(await search('p1', 'solar'))).toEqual(['a.txt#0', 'b.txt#0']); + + await ragService.setSyncedDocumentEnabled(syncId, false); + expect(label(await search('p1', 'solar'))).toEqual(['b.txt#0']); + + await ragService.setSyncedDocumentEnabled(syncId, true); + expect(label(await search('p1', 'solar'))).toEqual(['a.txt#0', 'b.txt#0']); + + await ragService.deleteSyncedDocument(syncId); + expect(label(await search('p1', 'solar'))).toEqual(['b.txt#0']); + }); + + it('switching projects, and deleting one project\'s documents', async () => { + const { ragDatabase, ragService, search } = await freshRag(); + addDocument(ragDatabase, 'p1', 'one.txt', ['solar for p1']); + addDocument(ragDatabase, 'p2', 'two.txt', ['solar for p2']); + + const p1 = await search('p1', 'solar'); + expect(label(p1)).toEqual(['one.txt#0']); + expect(label(await search('p2', 'solar'))).toEqual(['two.txt#0']); + expect(await search('p1', 'solar')).toEqual(p1); + + await ragService.deleteProjectDocuments('p1'); + + expect(await search('p1', 'solar')).toEqual([]); + expect(label(await search('p2', 'solar'))).toEqual(['two.txt#0']); + }); +}); + +describe('opening rag.db written by an earlier release', () => { + /** op-sqlite opens rag.db by name from app storage; here the same name maps to a file in a temp dir. */ + function fileBackedOpSqlite(dir: string) { + const { DatabaseSync } = require('node:sqlite'); + const isArrayBuffer = (v: unknown) => Object.prototype.toString.call(v) === '[object ArrayBuffer]'; + return { + open: ({ name }: { name: string }) => { + const db = new DatabaseSync(path.join(dir, name)); + const executeSync = (sql: string, params: unknown[] = []) => { + const bound = params.map(p => (isArrayBuffer(p) ? new Uint8Array(p as ArrayBuffer) : p)); + if (/^\s*(BEGIN|COMMIT|ROLLBACK|CREATE|ALTER|DROP)\b/i.test(sql) && bound.length === 0) { + db.exec(sql); + return { rows: [], rowsAffected: 0 }; + } + const statement = db.prepare(sql); + if (/^\s*(SELECT|WITH|PRAGMA)\b/i.test(sql)) { + return { rows: statement.all(...bound), rowsAffected: 0 }; + } + const info = statement.run(...bound); + return { rows: [], insertId: Number(info.lastInsertRowid), rowsAffected: Number(info.changes) }; + }; + return { + executeSync, + execute: async (sql: string, params?: unknown[]) => executeSync(sql, params), + close: () => db.close(), + }; + }, + }; + } + + function writeLegacyDatabase(file: string, withSyncId: boolean): void { + const { DatabaseSync } = require('node:sqlite'); + const db = new DatabaseSync(file); + db.exec(` + CREATE TABLE rag_documents ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ${withSyncId ? 'sync_id TEXT,' : ''} + project_id TEXT NOT NULL, + name TEXT NOT NULL, + path TEXT NOT NULL, + size INTEGER NOT NULL, + created_at TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 1 + ); + ${withSyncId ? 'CREATE UNIQUE INDEX idx_rag_documents_sync_id ON rag_documents(sync_id);' : ''} + CREATE TABLE rag_chunks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + content TEXT NOT NULL, + doc_id INTEGER NOT NULL, + position INTEGER NOT NULL, + FOREIGN KEY (doc_id) REFERENCES rag_documents(id) + ); + CREATE TABLE rag_embeddings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + chunk_rowid INTEGER NOT NULL, + doc_id INTEGER NOT NULL, + embedding BLOB NOT NULL, + FOREIGN KEY (chunk_rowid) REFERENCES rag_chunks(id), + FOREIGN KEY (doc_id) REFERENCES rag_documents(id) + ); + `); + const docs: [string, string, number, string[]][] = [ + ['p1', 'manual.txt', 1, ['solar panel wiring', 'battery storage sizing']], + ['p1', 'hidden.txt', 0, ['solar secret']], + ['p2', 'books.txt', 1, ['ledger totals']], + ]; + for (const [projectId, name, enabled, texts] of docs) { + const docId = Number( + db.prepare( + withSyncId + ? 'INSERT INTO rag_documents (sync_id, project_id, name, path, size, created_at, enabled) VALUES (?, ?, ?, ?, ?, ?, ?)' + : 'INSERT INTO rag_documents (project_id, name, path, size, created_at, enabled) VALUES (?, ?, ?, ?, ?, ?)', + ).run(...(withSyncId ? [`sync-${name}`] : []), projectId, name, `/docs/${name}`, 10, '2025-01-01T00:00:00.000Z', enabled) + .lastInsertRowid, + ); + texts.forEach((text, position) => { + const chunkId = Number( + db.prepare('INSERT INTO rag_chunks (content, doc_id, position) VALUES (?, ?, ?)') + .run(text, docId, position).lastInsertRowid, + ); + db.prepare('INSERT INTO rag_embeddings (chunk_rowid, doc_id, embedding) VALUES (?, ?, ?)') + .run(chunkId, docId, new Uint8Array(new Float32Array(toVec(text)).buffer)); + }); + } + db.close(); + } + + /** Every stored row except the sync_id the upgrade assigns to legacy documents. */ + function readStoredRows(file: string) { + const { DatabaseSync } = require('node:sqlite'); + const db = new DatabaseSync(file); + const rows = { + documents: db.prepare('SELECT id, project_id, name, path, size, created_at, enabled FROM rag_documents ORDER BY id').all(), + chunks: db.prepare('SELECT id, content, doc_id, position FROM rag_chunks ORDER BY id').all(), + embeddings: db.prepare('SELECT id, chunk_rowid, doc_id, hex(embedding) AS embedding FROM rag_embeddings ORDER BY id').all(), + }; + db.close(); + return JSON.parse(JSON.stringify(rows)); + } + + /** The column lists of every index on `table`, e.g. [['doc_id'], ['doc_id', 'position']]. */ + function indexedColumns(file: string, table: string): string[][] { + const { DatabaseSync } = require('node:sqlite'); + const db = new DatabaseSync(file); + const indexes = db.prepare('SELECT name FROM pragma_index_list(?)').all(table) as { name: string }[]; + const columns = indexes.map(({ name }) => + (db.prepare('SELECT name FROM pragma_index_info(?) ORDER BY seqno').all(name) as { name: string }[]) + .map(c => c.name), + ); + db.close(); + return columns; + } + + async function openApp(dir: string): Promise { + jest.resetModules(); + jest.doMock('@op-engineering/op-sqlite', () => fileBackedOpSqlite(dir)); + installEmbeddingModel(); + const rag = requireRag(); + await rag.ragDatabase.ensureReady(); + return rag; + } + + it.each([ + ['without sync_id (stable)', false], + ['with sync_id (beta)', true], + ])('upgrades a database %s in place, keeping every row and gaining the lookup indexes', async (_label, withSyncId) => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'rag-upgrade-')); + const file = path.join(dir, 'rag.db'); + writeLegacyDatabase(file, withSyncId); + const before = readStoredRows(file); + expect(indexedColumns(file, 'rag_embeddings')).toEqual([]); + expect(indexedColumns(file, 'rag_chunks')).toEqual([]); + + const { search } = await openApp(dir); + + expect(readStoredRows(file)).toEqual(before); + expect(indexedColumns(file, 'rag_embeddings')).toContainEqual(['doc_id']); + expect(indexedColumns(file, 'rag_chunks')).toContainEqual(['doc_id', 'position']); + expect(indexedColumns(file, 'rag_documents')).toContainEqual(['project_id', 'enabled']); + + // The upgraded data still answers searches, scoped and ranked as before. + expect((await search('p1', 'battery')).map(c => c.content)).toEqual([ + 'battery storage sizing', + 'solar panel wiring', + ]); + expect((await search('p2', 'ledger')).map(c => c.content)).toEqual(['ledger totals']); + + // Opening again (next launch) is a no-op for the schema and the data. + const afterFirstOpen = readStoredRows(file); + const reopened = await openApp(dir); + expect(readStoredRows(file)).toEqual(afterFirstOpen); + expect(indexedColumns(file, 'rag_embeddings').filter(c => c.join() === 'doc_id')).toHaveLength(1); + expect((await reopened.search('p1', 'solar')).map(c => c.content)).toEqual([ + 'solar panel wiring', + 'battery storage sizing', + ]); + }); +}); diff --git a/__tests__/pro/sync/chatProjectionRepair.integration.test.ts b/__tests__/pro/sync/chatProjectionRepair.integration.test.ts new file mode 100644 index 000000000..0da118c3e --- /dev/null +++ b/__tests__/pro/sync/chatProjectionRepair.integration.test.ts @@ -0,0 +1,186 @@ +import { installRealSqlite } from '../../harness/sqliteFake'; + +/** + * Startup chat repair over the real op-log and the real SQLite op store (only the native binding is + * replaced). The chat list here is the projection the repair writes into: conversations holding the + * messages the op-log materialized, exactly what the user would see. + */ +type Conversation = { id: string; messages: Array<{ id: string; text: string }> }; + +const DESKTOP = 'desktop-device'; +const PHONE = 'phone-device'; + +function chatList() { + const conversations: Conversation[] = []; + const materializer = { + put(entity: string, entityId: string, fields: Record) { + if (entity === 'conversation') { + if (!conversations.some(({ id }) => id === entityId)) + conversations.push({ id: entityId, messages: [] }); + return; + } + const conversation = conversations.find( + ({ id }) => id === fields.conversation_id, + ); + if (!conversation) return false; + conversation.messages = conversation.messages.filter( + ({ id }) => id !== entityId, + ); + conversation.messages.push({ id: entityId, text: String(fields.content) }); + return true; + }, + remove(entity: string, entityId: string) { + for (const conversation of conversations) + conversation.messages = conversation.messages.filter( + ({ id }) => id !== entityId, + ); + }, + }; + return { conversations, materializer }; +} + +const conversation = (id: string, lamport: number) => ({ + opId: `op-${id}-${lamport}`, + entity: 'conversation', + entityId: id, + kind: 'put' as const, + fields: { title: id }, + lamport, + deviceId: DESKTOP, + ts: lamport, +}); + +const message = ( + id: string, + conversationId: string, + lamport: number, + over: Record = {}, +) => ({ + opId: `op-${id}-${lamport}`, + entity: 'message', + entityId: id, + kind: 'put' as const, + fields: { + conversation_id: conversationId, + role: 'assistant', + content: `reply ${id}`, + created_at: 1_700_000_000_000 + lamport, + }, + lamport, + deviceId: DESKTOP, + ts: lamport, + ...over, +}); + +const HISTORY = [ + conversation('c1', 1), + conversation('c2', 2), + message('m1', 'c1', 3), + message('m2', 'c1', 4), + message('m3', 'c2', 5), + message('m4', 'c1', 6), + // Resend on the phone: a concurrent, lower-clock delete still removes the reply for good. + { + opId: 'op-m4-delete', + entity: 'message', + entityId: 'm4', + kind: 'delete' as const, + lamport: 2, + deviceId: PHONE, + ts: 2, + }, + // A row the chat projection cannot display is never replayed. + message('m5', 'c2', 7, { + fields: { conversation_id: 'c2', role: 'narrator', content: 'x', created_at: 1 }, + }), +]; + +function openSync() { + const sync = require('@offgrid/sync'); + const { opStoreDriver } = require('../../../pro/sync/opStoreDriver'); + const { + repairMissingChatProjection, + } = require('../../../pro/sync/chatProjectionRepair'); + sync.migrateOpStore(opStoreDriver); + let n = 0; + const launch = (materializer: ReturnType['materializer']) => + new sync.OpLog({ + deviceId: PHONE, + materializer, + persist: (op: unknown) => + sync.persistOp(opStoreDriver, op, { lazyFields: true }), + persisted: sync.loadWinningOps(opStoreDriver, { lazyFields: true }), + persistedEntityVersionVector: sync.loadOpWatermarks(opStoreDriver), + uuid: () => `phone-op-${++n}`, + now: () => 1_700_000_000_000, + }); + return { launch, repairMissingChatProjection }; +} + +const shown = (conversations: Conversation[]) => + Object.fromEntries( + conversations.map(({ id, messages }) => [ + id, + messages.map(({ id: messageId, text }) => `${messageId}:${text}`).sort(), + ]), + ); + +describe('chat projection repair at startup', () => { + it('restores exactly the missing winners after a relaunch, duplicate delivery included', () => { + installRealSqlite(); + const { launch, repairMissingChatProjection } = openSync(); + + const before = chatList(); + const first = launch(before.materializer); + first.ingest(HISTORY); + first.ingest(HISTORY); + expect(shown(before.conversations)).toEqual({ + c1: ['m1:reply m1', 'm2:reply m2'], + c2: ['m3:reply m3', 'm5:x'], + }); + + // Relaunch with a chat list saved before most of the history arrived. + const after = chatList(); + after.conversations.push({ + id: 'c1', + messages: [{ id: 'm1', text: 'reply m1' }], + }); + const log = launch(after.materializer); + + expect( + repairMissingChatProjection(log, undefined, () => after.conversations), + ).toBe(3); + expect(shown(after.conversations)).toEqual({ + c1: ['m1:reply m1', 'm2:reply m2'], + c2: ['m3:reply m3'], + }); + expect( + repairMissingChatProjection(log, undefined, () => after.conversations), + ).toBe(0); + expect(shown(after.conversations)).toEqual({ + c1: ['m1:reply m1', 'm2:reply m2'], + c2: ['m3:reply m3'], + }); + }); + + it('repairs one conversation without touching the others', () => { + installRealSqlite(); + const { launch, repairMissingChatProjection } = openSync(); + launch(chatList().materializer).ingest(HISTORY); + + const after = chatList(); + after.conversations.push( + { id: 'c1', messages: [] }, + { id: 'c2', messages: [] }, + ); + const log = launch(after.materializer); + + expect( + repairMissingChatProjection(log, 'c1', () => after.conversations), + ).toBe(2); + expect(shown(after.conversations)).toEqual({ + c1: ['m1:reply m1', 'm2:reply m2'], + c2: [], + }); + }); +}); diff --git a/__tests__/unit/services/rag/database.test.ts b/__tests__/unit/services/rag/database.test.ts index 61801ca12..d84515bad 100644 --- a/__tests__/unit/services/rag/database.test.ts +++ b/__tests__/unit/services/rag/database.test.ts @@ -150,7 +150,7 @@ describe('RagDatabase', () => { const results = ragDatabase.getEmbeddingsByProject('proj1'); expect(results).toHaveLength(1); expect(results[0].content).toBe('hello'); - expect(results[0].embedding).toBeInstanceOf(Array); + expect(Array.from(results[0].embedding)).toHaveLength(2); }); }); diff --git a/__tests__/unit/utils/hydrationGatedStorage.test.ts b/__tests__/unit/utils/hydrationGatedStorage.test.ts index d49ad64db..1ee54fc95 100644 --- a/__tests__/unit/utils/hydrationGatedStorage.test.ts +++ b/__tests__/unit/utils/hydrationGatedStorage.test.ts @@ -92,12 +92,18 @@ describe('hydration-gated storage', () => { type Selection = { selected: string }; function createPersistedSelection(base: StateStorage) { - const gate = createHydrationGatedStorage(base); + const gate = createHydrationGatedStorage( + base, + (previous, next) => previous.selected === next.selected, + ); const store = createStore()( persist(() => ({ selected: 'default' }), { name: 'settings', storage: gate.storage, - onRehydrateStorage: () => () => gate.markHydrated(), + // Same callback as the app stores: a failed read must not open the write gate. + onRehydrateStorage: () => (_state, error) => { + if (!error) gate.markHydrated(); + }, }), ); return { gate, store }; @@ -111,3 +117,142 @@ function waitForHydration( store.persist.onFinishHydration(() => resolve()), ); } + +const stored = (state: Selection, version = 0) => + JSON.stringify({ state, version }); + +/** + * AsyncStorage-contract fake for the device boundary: reads and writes can fail. It records how + * many writes reached storage. + */ +function deviceStorage(initial: Record = {}) { + const memory = memoryStorage(initial); + const device = { + values: memory.values, + writes: 0, + failReads: false, + failNextWrite: false, + storage: { + getItem: async (name: string) => { + if (device.failReads) throw new Error('read failed'); + return memory.getItem(name); + }, + setItem: async (name: string, value: string) => { + if (device.failNextWrite) { + device.failNextWrite = false; + throw new Error('database or disk is full'); + } + device.writes += 1; + await memory.setItem(name, value); + }, + removeItem: (name: string) => memory.removeItem(name), + } satisfies StateStorage, + }; + return device; +} + +describe('hydration-gated storage load and write failures', () => { + it('keeps writes blocked after a failed read so saved history survives', async () => { + const device = deviceStorage({ settings: stored({ selected: 'saved' }) }); + device.failReads = true; + const { gate, store } = createPersistedSelection(device.storage); + await store.persist.rehydrate(); + + store.setState({ selected: 'empty-after-failed-read' }); + await gate.waitForWrites(); + + expect(store.persist.hasHydrated()).toBe(false); + expect(device.writes).toBe(0); + expect(device.values.get('settings')).toBe(stored({ selected: 'saved' })); + }); + + it('loads and then persists after a retried read succeeds', async () => { + const device = deviceStorage({ settings: stored({ selected: 'saved' }) }); + device.failReads = true; + const { gate, store } = createPersistedSelection(device.storage); + await store.persist.rehydrate(); + + device.failReads = false; + await store.persist.rehydrate(); + expect(store.getState().selected).toBe('saved'); + + store.setState({ selected: 'after-retry' }); + await gate.waitForWrites(); + expect(device.values.get('settings')).toBe( + stored({ selected: 'after-retry' }), + ); + }); + + it('retries the same content after a failed write', async () => { + const device = deviceStorage({ settings: stored({ selected: 'saved' }) }); + const gate = createHydrationGatedStorage( + device.storage, + (previous, next) => previous.selected === next.selected, + ); + await gate.storage.getItem('settings'); + gate.markHydrated(); + const next = { state: { selected: 'next' }, version: 0 }; + + device.failNextWrite = true; + await expect(gate.storage.setItem('settings', next)).rejects.toThrow( + 'database or disk is full', + ); + expect(device.values.get('settings')).toBe(stored({ selected: 'saved' })); + + await gate.storage.setItem('settings', next); + expect(device.values.get('settings')).toBe(stored({ selected: 'next' })); + }); + + it('reports a failed write to waiters', async () => { + const device = deviceStorage(); + const { gate, store } = createPersistedSelection(device.storage); + await waitForHydration(store); + + device.failNextWrite = true; + const write = Promise.resolve( + gate.storage.setItem('settings', { + state: { selected: 'lost' }, + version: 0, + }), + ).then( + () => 'stored', + (error: Error) => error.message, + ); + + await expect(gate.waitForWrites()).rejects.toThrow( + 'database or disk is full', + ); + expect(await write).toBe('database or disk is full'); + }); +}); + +describe('hydration-gated storage format and restart', () => { + it('loads the latest stored snapshot after a restart', async () => { + const device = deviceStorage(); + const first = createPersistedSelection(device.storage); + await waitForHydration(first.store); + + first.store.setState({ selected: 'draft' }); + first.store.setState({ selected: 'final' }); + await first.gate.waitForWrites(); + + const restarted = createPersistedSelection(device.storage); + await waitForHydration(restarted.store); + expect(restarted.store.getState().selected).toBe('final'); + }); + + it('keeps the stored format readable by both old and new builds', async () => { + const device = deviceStorage({ + settings: stored({ selected: 'written-by-old-build' }), + }); + const { gate, store } = createPersistedSelection(device.storage); + await waitForHydration(store); + expect(store.getState().selected).toBe('written-by-old-build'); + + store.setState({ selected: 'written-by-new-build' }); + await gate.waitForWrites(); + expect(device.values.get('settings')).toBe( + stored({ selected: 'written-by-new-build' }), + ); + }); +}); diff --git a/android/gradle.properties b/android/gradle.properties index 6da9b0a50..fb614558a 100644 --- a/android/gradle.properties +++ b/android/gradle.properties @@ -42,3 +42,8 @@ hermesEnabled=true # This allows your app to draw behind system bars for an immersive UI. # Note: Only works with ReactActivity and should not be used with custom Activity. edgeToEdgeEnabled=false + +# Raise AsyncStorage's Android SQLite cap (library default 6 MB, read by +# @react-native-async-storage/async-storage android/config.gradle). Chat history is +# stored as one value and writes fail once the database reaches the cap. +AsyncStorage_db_size_in_MB=64 diff --git a/pro b/pro index c15e454b0..661315f13 160000 --- a/pro +++ b/pro @@ -1 +1 @@ -Subproject commit c15e454b0b493d242ad76aaf510f6bea1c90781c +Subproject commit 661315f13088ce9146486d40ab77793da1f780c2 diff --git a/src/services/rag/database.ts b/src/services/rag/database.ts index 2d3e08088..defbc7e85 100644 --- a/src/services/rag/database.ts +++ b/src/services/rag/database.ts @@ -29,12 +29,84 @@ interface StoredEmbedding { name: string; content: string; position: number; - embedding: number[]; + embedding: Float32Array | number[]; +} + +/** Byte budget for decoded embedding vectors kept between searches; modest so phones stay comfortable. */ +const EMBEDDING_CACHE_MAX_BYTES = 16 * 1024 * 1024; + +const EMPTY_VECTOR = new Float32Array(0); + +/** Realm-safe: a BLOB can arrive as an ArrayBuffer created in another JS realm. */ +function isArrayBuffer(value: unknown): value is ArrayBuffer { + return Object.prototype.toString.call(value) === '[object ArrayBuffer]'; +} + +function sameIds(a: readonly number[], b: readonly number[]): boolean { + if (a.length !== b.length) return false; + for (let i = 0; i < a.length; i++) { + if (a[i] !== b[i]) return false; + } + return true; +} + +/** One document's decoded vectors, aligned with the rag_embeddings ids they were read from (ascending). */ +interface DocumentVectors { + ids: number[]; + vectors: Float32Array[]; + bytes: number; +} + +/** + * Least-recently-used, byte-bounded cache of decoded vectors per document. + * + * An entry is only served when its embedding ids match the ids the current search query returned for that + * document. rag_embeddings rows are only ever inserted or deleted, never updated, and AUTOINCREMENT ids are + * never reused, so a matching id list means identical vectors even if a write skipped invalidation. + */ +class DocumentVectorCache { + private entries = new Map(); + private bytes = 0; + + constructor(private readonly maxBytes: number) {} + + get(docId: number): DocumentVectors | undefined { + const entry = this.entries.get(docId); + if (!entry) return undefined; + this.entries.delete(docId); + this.entries.set(docId, entry); + return entry; + } + + set(docId: number, entry: DocumentVectors): void { + this.delete(docId); + if (entry.bytes > this.maxBytes) return; + this.entries.set(docId, entry); + this.bytes += entry.bytes; + while (this.bytes > this.maxBytes) { + const oldest = this.entries.keys().next(); + if (oldest.done) break; + this.delete(oldest.value); + } + } + + delete(docId: number): void { + const entry = this.entries.get(docId); + if (!entry) return; + this.entries.delete(docId); + this.bytes -= entry.bytes; + } + + clear(): void { + this.entries.clear(); + this.bytes = 0; + } } class RagDatabase { private db: DB | null = null; private ready = false; + private vectorCache = new DocumentVectorCache(EMBEDDING_CACHE_MAX_BYTES); async ensureReady(): Promise { if (this.ready) return; @@ -96,6 +168,15 @@ class RagDatabase { FOREIGN KEY (doc_id) REFERENCES rag_documents(id) )`, ); + this.db.executeSync( + 'CREATE INDEX IF NOT EXISTS idx_rag_embeddings_doc_id ON rag_embeddings(doc_id)', + ); + this.db.executeSync( + 'CREATE INDEX IF NOT EXISTS idx_rag_chunks_doc_id_position ON rag_chunks(doc_id, position)', + ); + this.db.executeSync( + 'CREATE INDEX IF NOT EXISTS idx_rag_documents_project_id_enabled ON rag_documents(project_id, enabled)', + ); this.ready = true; } catch (error) { logger.error('[RagDB] Failed to initialize:', error); @@ -140,6 +221,7 @@ class RagDatabase { insertChunks(docId: number, chunks: Chunk[]): number[] { const db = this.getDb(); + this.vectorCache.delete(docId); const rowIds: number[] = []; db.executeSync('BEGIN'); try { @@ -166,17 +248,18 @@ class RagDatabase { return new Float32Array(embedding).buffer; } - private blobToEmbedding(blob: any): number[] { - if (blob instanceof ArrayBuffer) return Array.from(new Float32Array(blob)); - if (blob?.buffer instanceof ArrayBuffer) - return Array.from(new Float32Array(blob.buffer)); - return []; + /** A view over the BLOB's bytes; no element copy. */ + private blobToEmbedding(blob: any): Float32Array { + if (isArrayBuffer(blob)) return new Float32Array(blob); + if (isArrayBuffer(blob?.buffer)) return new Float32Array(blob.buffer); + return EMPTY_VECTOR; } insertEmbeddingsBatch( entries: { chunkRowid: number; docId: number; embedding: number[] }[], ): void { const db = this.getDb(); + for (const entry of entries) this.vectorCache.delete(entry.docId); db.executeSync('BEGIN'); try { for (const entry of entries) { @@ -198,18 +281,68 @@ class RagDatabase { getEmbeddingsByProject(projectId: string): StoredEmbedding[] { const db = this.getDb(); + // ORDER BY e.id keeps the order the unindexed query scanned in, so equal scores stay in the same order. const result = db.executeSync( - `SELECT e.chunk_rowid, e.doc_id, d.name, c.content, c.position, e.embedding + `SELECT e.id AS embedding_id, e.chunk_rowid, e.doc_id, d.name, c.content, c.position FROM rag_embeddings e JOIN rag_chunks c ON e.chunk_rowid = c.id JOIN rag_documents d ON e.doc_id = d.id - WHERE d.project_id = ? AND d.enabled = 1`, + WHERE d.project_id = ? AND d.enabled = 1 + ORDER BY e.id`, [projectId], ); - return ((result.rows ?? []) as unknown as any[]).map(row => ({ - ...row, - embedding: this.blobToEmbedding(row.embedding), - })); + const rows = (result.rows ?? []) as unknown as (Omit< + StoredEmbedding, + 'embedding' + > & { embedding_id: number })[]; + + const idsByDoc = new Map(); + for (const row of rows) { + const ids = idsByDoc.get(row.doc_id); + if (ids) ids.push(row.embedding_id); + else idsByDoc.set(row.doc_id, [row.embedding_id]); + } + const vectorsByDoc = new Map(); + for (const [docId, ids] of idsByDoc) { + vectorsByDoc.set(docId, this.getDocumentVectors(docId, ids)); + } + + const nextIndexByDoc = new Map(); + return rows.map(({ embedding_id: _embeddingId, ...row }) => { + const index = nextIndexByDoc.get(row.doc_id) ?? 0; + nextIndexByDoc.set(row.doc_id, index + 1); + return { ...row, embedding: vectorsByDoc.get(row.doc_id)![index] }; + }); + } + + /** Decoded vectors for `ids` (ascending, all from `docId`), from the cache when it still matches. */ + private getDocumentVectors(docId: number, ids: number[]): Float32Array[] { + const cached = this.vectorCache.get(docId); + if (cached && sameIds(cached.ids, ids)) return cached.vectors; + + const result = this.getDb().executeSync( + `SELECT e.id AS embedding_id, e.embedding + FROM rag_embeddings e + JOIN rag_chunks c ON e.chunk_rowid = c.id + WHERE e.doc_id = ? + ORDER BY e.id`, + [docId], + ); + const rows = (result.rows ?? []) as unknown as { + embedding_id: number; + embedding: unknown; + }[]; + const loadedIds = rows.map(row => row.embedding_id); + const vectors = rows.map(row => this.blobToEmbedding(row.embedding)); + if (sameIds(loadedIds, ids)) { + const bytes = vectors.reduce((sum, vector) => sum + vector.byteLength, 0); + this.vectorCache.set(docId, { ids: loadedIds, vectors, bytes }); + return vectors; + } + // The rows changed between the two reads; align by id and leave the cache alone. + const byId = new Map(); + loadedIds.forEach((id, i) => byId.set(id, vectors[i])); + return ids.map(id => byId.get(id) ?? EMPTY_VECTOR); } hasEmbeddingsForDocument(docId: number): boolean { @@ -239,9 +372,17 @@ class RagDatabase { deleteDocument(docId: number): void { const db = this.getDb(); - db.executeSync('DELETE FROM rag_embeddings WHERE doc_id = ?', [docId]); - db.executeSync('DELETE FROM rag_chunks WHERE doc_id = ?', [docId]); - db.executeSync('DELETE FROM rag_documents WHERE id = ?', [docId]); + this.vectorCache.delete(docId); + db.executeSync('BEGIN'); + try { + db.executeSync('DELETE FROM rag_embeddings WHERE doc_id = ?', [docId]); + db.executeSync('DELETE FROM rag_chunks WHERE doc_id = ?', [docId]); + db.executeSync('DELETE FROM rag_documents WHERE id = ?', [docId]); + db.executeSync('COMMIT'); + } catch (e) { + db.executeSync('ROLLBACK'); + throw e; + } } getDocumentsByProject(projectId: string): RagDocument[] { @@ -281,6 +422,7 @@ class RagDatabase { toggleEnabled(docId: number, enabled: boolean): void { const db = this.getDb(); + this.vectorCache.delete(docId); db.executeSync('UPDATE rag_documents SET enabled = ? WHERE id = ?', [ enabled ? 1 : 0, docId, @@ -301,17 +443,25 @@ class RagDatabase { deleteDocumentsByProject(projectId: string): void { const db = this.getDb(); - db.executeSync( - 'DELETE FROM rag_embeddings WHERE doc_id IN (SELECT id FROM rag_documents WHERE project_id = ?)', - [projectId], - ); - db.executeSync( - 'DELETE FROM rag_chunks WHERE doc_id IN (SELECT id FROM rag_documents WHERE project_id = ?)', - [projectId], - ); - db.executeSync('DELETE FROM rag_documents WHERE project_id = ?', [ - projectId, - ]); + this.vectorCache.clear(); + db.executeSync('BEGIN'); + try { + db.executeSync( + 'DELETE FROM rag_embeddings WHERE doc_id IN (SELECT id FROM rag_documents WHERE project_id = ?)', + [projectId], + ); + db.executeSync( + 'DELETE FROM rag_chunks WHERE doc_id IN (SELECT id FROM rag_documents WHERE project_id = ?)', + [projectId], + ); + db.executeSync('DELETE FROM rag_documents WHERE project_id = ?', [ + projectId, + ]); + db.executeSync('COMMIT'); + } catch (e) { + db.executeSync('ROLLBACK'); + throw e; + } } } diff --git a/src/services/rag/vectorMath.ts b/src/services/rag/vectorMath.ts index 10a9af429..e741373fd 100644 --- a/src/services/rag/vectorMath.ts +++ b/src/services/rag/vectorMath.ts @@ -1,4 +1,7 @@ -export function dotProduct(a: number[], b: number[]): number { +/** Plain arrays and typed arrays (stored embeddings are Float32Array views) score identically. */ +type Vector = ArrayLike; + +export function dotProduct(a: Vector, b: Vector): number { let sum = 0; for (let i = 0; i < a.length; i++) { sum += a[i] * b[i]; @@ -6,7 +9,7 @@ export function dotProduct(a: number[], b: number[]): number { return sum; } -export function cosineSimilarity(a: number[], b: number[]): number { +export function cosineSimilarity(a: Vector, b: Vector): number { let dot = 0; let normA = 0; let normB = 0; @@ -25,7 +28,7 @@ export interface SimilarityResult { score: number; } -export function topKSimilar(queryVec: number[], candidates: number[][], k: number): SimilarityResult[] { +export function topKSimilar(queryVec: Vector, candidates: Vector[], k: number): SimilarityResult[] { const scored = candidates.map((vec, index) => ({ index, score: cosineSimilarity(queryVec, vec), diff --git a/src/stores/appStore.ts b/src/stores/appStore.ts index 93d5f3b59..b9faf0a0c 100644 --- a/src/stores/appStore.ts +++ b/src/stores/appStore.ts @@ -475,7 +475,7 @@ export const useAppStore = create()( { name: 'local-llm-app-storage', storage: appStorage.storage, - onRehydrateStorage: () => () => appStorage.markHydrated(), + onRehydrateStorage: () => (_state, error) => { if (!error) appStorage.markHydrated(); }, merge: (persisted, current) => migratePersistedState(persisted, current, { defaultSettings: DEFAULT_SETTINGS, diff --git a/src/stores/authStore.ts b/src/stores/authStore.ts index 32ebe419f..a2cbb45a6 100644 --- a/src/stores/authStore.ts +++ b/src/stores/authStore.ts @@ -90,7 +90,7 @@ export const useAuthStore = create()( { name: 'local-llm-auth-storage', storage: authStorage.storage, - onRehydrateStorage: () => () => authStorage.markHydrated(), + onRehydrateStorage: () => (_state, error) => { if (!error) authStorage.markHydrated(); }, partialize: (state): PersistedAuthState => ({ isEnabled: state.isEnabled, failedAttempts: state.failedAttempts, diff --git a/src/stores/chatStore.ts b/src/stores/chatStore.ts index e137cf5bc..3399dcd07 100644 --- a/src/stores/chatStore.ts +++ b/src/stores/chatStore.ts @@ -501,7 +501,7 @@ export const useChatStore = create()( { name: 'local-llm-chat-storage', storage: chatStorage.storage, - onRehydrateStorage: () => () => chatStorage.markHydrated(), + onRehydrateStorage: () => (_state, error) => { if (!error) chatStorage.markHydrated(); }, version: CHAT_STORAGE_VERSION, migrate: migratePersistedChatState, partialize: (state): PersistedChatState => ({ diff --git a/src/stores/projectStore.ts b/src/stores/projectStore.ts index 051f03e07..36332a411 100644 --- a/src/stores/projectStore.ts +++ b/src/stores/projectStore.ts @@ -193,7 +193,7 @@ export const useProjectStore = create()( { name: 'local-llm-project-storage', storage: projectStorage.storage, - onRehydrateStorage: () => () => projectStorage.markHydrated(), + onRehydrateStorage: () => (_state, error) => { if (!error) projectStorage.markHydrated(); }, partialize: (state): PersistedProjectState => ({ projects: state.projects }), }, ), diff --git a/src/stores/remoteServerStore.ts b/src/stores/remoteServerStore.ts index d1a4ef682..15cbb01cf 100644 --- a/src/stores/remoteServerStore.ts +++ b/src/stores/remoteServerStore.ts @@ -434,7 +434,7 @@ export const useRemoteServerStore = create()( version: 2, migrate: migrateRemoteServerState, storage: remoteServerStorage.storage, - onRehydrateStorage: () => () => remoteServerStorage.markHydrated(), + onRehydrateStorage: () => (_state, error) => { if (!error) remoteServerStorage.markHydrated(); }, partialize: (state): PersistedRemoteServerState => ({ servers: state.servers.map(({ apiKey: _apiKey, ...server }) => server), activeServerId: state.activeServerId, diff --git a/src/stores/whisperStore.ts b/src/stores/whisperStore.ts index ecce13a1f..e465c8875 100644 --- a/src/stores/whisperStore.ts +++ b/src/stores/whisperStore.ts @@ -281,7 +281,7 @@ export const useWhisperStore = create()( { name: 'local-llm-whisper-storage', storage: whisperStorage.storage, - onRehydrateStorage: () => () => whisperStorage.markHydrated(), + onRehydrateStorage: () => (_state, error) => { if (!error) whisperStorage.markHydrated(); }, partialize: (state): PersistedWhisperState => ({ downloadedModelId: state.downloadedModelId, transcriptionLanguage: state.transcriptionLanguage, diff --git a/src/utils/hydrationGatedStorage.ts b/src/utils/hydrationGatedStorage.ts index 9fe19bb25..6edefa64c 100644 --- a/src/utils/hydrationGatedStorage.ts +++ b/src/utils/hydrationGatedStorage.ts @@ -18,7 +18,10 @@ export function createHydrationGatedStorage( unchanged?: (previous: S, next: S) => boolean, ): HydrationGatedStorage { let hydrated = false; - let lastWritten: S | undefined; + // Last snapshot known to be stored: read at hydration or confirmed by a resolved write. + let lastSaved: S | undefined; + // Newest snapshot handed to storage whose write has not settled yet. + let lastIssued: { state: S } | undefined; const pendingWrites = new Set>(); const json = createJSONStorage(() => base); if (!json) throw new Error('JSON storage is unavailable'); @@ -27,27 +30,44 @@ export function createHydrationGatedStorage( storage: { getItem: async name => { const value = await json.getItem(name); - lastWritten = value?.state; + lastSaved = value?.state; return value; }, setItem: (name, value) => { if (!hydrated) return undefined; - if ( - lastWritten !== undefined && - unchanged?.(lastWritten, value.state) - ) { + const previous = lastIssued ? lastIssued.state : lastSaved; + if (previous !== undefined && unchanged?.(previous, value.state)) { return undefined; } - lastWritten = value.state; + const issued = { state: value.state }; + lastIssued = issued; + const settle = () => { + if (lastIssued === issued) lastIssued = undefined; + }; const result = json.setItem(name, value); - if (!result) return undefined; - const write = Promise.resolve(result).then(() => undefined); + if (!result) { + settle(); + lastSaved = value.state; + return undefined; + } + const write = Promise.resolve(result).then( + () => { + settle(); + lastSaved = value.state; + }, + error => { + // Only a committed snapshot counts as saved, so the same content is written again later. + settle(); + throw error; + }, + ); pendingWrites.add(write); return write.finally(() => pendingWrites.delete(write)); }, removeItem: name => { if (!hydrated) return undefined; - lastWritten = undefined; + lastIssued = undefined; + lastSaved = undefined; return json.removeItem(name); }, }, From 944555dc04aba9be4e41fb06aee0efc583a681dd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 14:47:53 +0000 Subject: [PATCH 2/4] fix(rag): decode a BLOB slice from its own bytes, not its whole buffer A typed-array BLOB can be a view into a larger (pooled) buffer; decoding blob.buffer read the wrong values and length, and the vector cache would keep them. Read byteOffset/byteLength, copying when the offset is unaligned. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016JzAmMBhYRz9FHP1r3mEVr --- __tests__/unit/services/rag/database.test.ts | 28 ++++++++++++++++++++ src/services/rag/database.ts | 19 ++++++++++--- 2 files changed, 44 insertions(+), 3 deletions(-) diff --git a/__tests__/unit/services/rag/database.test.ts b/__tests__/unit/services/rag/database.test.ts index d84515bad..98d8aeab1 100644 --- a/__tests__/unit/services/rag/database.test.ts +++ b/__tests__/unit/services/rag/database.test.ts @@ -152,6 +152,34 @@ describe('RagDatabase', () => { expect(results[0].content).toBe('hello'); expect(Array.from(results[0].embedding)).toHaveLength(2); }); + + it.each([ + ['aligned', 4], + ['unaligned', 3], + ])( + 'reads only its own bytes when a BLOB is a %s slice of a larger buffer', + async (_label, offset) => { + await ragDatabase.ensureReady(); + const own = new Uint8Array(new Float32Array([0.5, -0.25]).buffer); + const pooled = new Uint8Array(offset + own.length + 8).fill(0xff); + pooled.set(own, offset); + mockExecuteSync.mockReturnValue({ + rows: [ + { + chunk_rowid: 7, + doc_id: 99, + name: 'pooled.txt', + content: 'pooled', + position: 0, + embedding: pooled.subarray(offset, offset + own.length), + }, + ], + }); + + const results = ragDatabase.getEmbeddingsByProject(`pooled-${offset}`); + expect(Array.from(results[0].embedding)).toEqual([0.5, -0.25]); + }, + ); }); describe('hasEmbeddingsForDocument', () => { diff --git a/src/services/rag/database.ts b/src/services/rag/database.ts index defbc7e85..cdbdc28f0 100644 --- a/src/services/rag/database.ts +++ b/src/services/rag/database.ts @@ -248,10 +248,23 @@ class RagDatabase { return new Float32Array(embedding).buffer; } - /** A view over the BLOB's bytes; no element copy. */ + /** + * A view over the BLOB's bytes; no element copy. A typed-array BLOB may be a slice of a larger + * (pooled) buffer, so only its own byteOffset/byteLength are read; an unaligned slice is copied. + */ private blobToEmbedding(blob: any): Float32Array { - if (isArrayBuffer(blob)) return new Float32Array(blob); - if (isArrayBuffer(blob?.buffer)) return new Float32Array(blob.buffer); + if (isArrayBuffer(blob)) { + return new Float32Array(blob, 0, Math.floor(blob.byteLength / 4)); + } + if (isArrayBuffer(blob?.buffer) && typeof blob.byteOffset === 'number') { + const length = Math.floor(blob.byteLength / 4); + if (blob.byteOffset % 4 === 0) { + return new Float32Array(blob.buffer, blob.byteOffset, length); + } + return new Float32Array( + blob.buffer.slice(blob.byteOffset, blob.byteOffset + length * 4), + ); + } return EMPTY_VECTOR; } From 040867359d69f7b291886b485bf7225255da0975 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 14:52:55 +0000 Subject: [PATCH 3/4] fix(rag): copy a pooled BLOB slice so a cached vector owns only its bytes A zero-copy view of a slice kept the whole pooled buffer alive while the vector cache counted only the slice, so the 16 MB bound could be exceeded. Keep zero-copy only when the BLOB owns its whole buffer. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016JzAmMBhYRz9FHP1r3mEVr --- __tests__/unit/services/rag/database.test.ts | 4 ++++ src/services/rag/database.ts | 10 ++++++---- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/__tests__/unit/services/rag/database.test.ts b/__tests__/unit/services/rag/database.test.ts index 98d8aeab1..e717791c0 100644 --- a/__tests__/unit/services/rag/database.test.ts +++ b/__tests__/unit/services/rag/database.test.ts @@ -178,6 +178,10 @@ describe('RagDatabase', () => { const results = ragDatabase.getEmbeddingsByProject(`pooled-${offset}`); expect(Array.from(results[0].embedding)).toEqual([0.5, -0.25]); + // The vector owns exactly its own bytes, not the pooled buffer it was sliced from. + expect((results[0].embedding as Float32Array).buffer.byteLength).toBe( + own.length, + ); }, ); }); diff --git a/src/services/rag/database.ts b/src/services/rag/database.ts index cdbdc28f0..5b3deb5ac 100644 --- a/src/services/rag/database.ts +++ b/src/services/rag/database.ts @@ -249,8 +249,10 @@ class RagDatabase { } /** - * A view over the BLOB's bytes; no element copy. A typed-array BLOB may be a slice of a larger - * (pooled) buffer, so only its own byteOffset/byteLength are read; an unaligned slice is copied. + * A view over the BLOB's bytes; no element copy when the BLOB owns its whole buffer. A + * typed-array BLOB may be a slice of a larger (pooled) buffer: only its own + * byteOffset/byteLength are read, and the slice is copied so a cached vector never keeps the + * whole pooled buffer alive (the cache bound counts the vector's bytes). */ private blobToEmbedding(blob: any): Float32Array { if (isArrayBuffer(blob)) { @@ -258,8 +260,8 @@ class RagDatabase { } if (isArrayBuffer(blob?.buffer) && typeof blob.byteOffset === 'number') { const length = Math.floor(blob.byteLength / 4); - if (blob.byteOffset % 4 === 0) { - return new Float32Array(blob.buffer, blob.byteOffset, length); + if (blob.byteOffset === 0 && blob.buffer.byteLength === length * 4) { + return new Float32Array(blob.buffer, 0, length); } return new Float32Array( blob.buffer.slice(blob.byteOffset, blob.byteOffset + length * 4), From 0db0f8a512bcc2676a078b5d4f38f26bfbbbe002 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 15:31:27 +0000 Subject: [PATCH 4/4] fix(sync): drop a failed-reconnect message once the device connects Pins mobile-pro with the fix and covers it in the Known Devices integration test. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016JzAmMBhYRz9FHP1r3mEVr --- .../KnownDevicesSection.integration.test.tsx | 32 +++++++++++++++++++ pro | 2 +- 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/__tests__/pro/sync/KnownDevicesSection.integration.test.tsx b/__tests__/pro/sync/KnownDevicesSection.integration.test.tsx index ab29945d2..8e32361aa 100644 --- a/__tests__/pro/sync/KnownDevicesSection.integration.test.tsx +++ b/__tests__/pro/sync/KnownDevicesSection.integration.test.tsx @@ -108,4 +108,36 @@ describe(' action ownership', () => { ); expect(ui.getByText(/Offline/)).toBeTruthy(); }); + + it('drops a failed-reconnect message once the device connects', async () => { + const offline = savedDevice('mac-debug', 'OGAD: Mac (Debug)'); + const section = (device: SyncManagedDevice) => ( + undefined} + onRepair={async () => 'reconnected'} + onDisconnect={async () => undefined} + onReconnect={async () => { + throw new Error('connection timed out after 5000 ms'); + }} + onSetManualEndpoint={() => undefined} + manualEndpointDeviceIds={[]} + onSendModel={() => undefined} + onForget={async () => undefined} + reachabilityErrors={{}} + /> + ); + const ui = render(section(offline)); + + fireEvent.press(ui.getByTestId('sync-reconnect-mac-debug')); + expect( + await ui.findByText(/Could not reach OGAD: Mac \(Debug\)/), + ).toBeTruthy(); + + // The peer comes back on its own (it was busy, not gone): the row now reads Connected. + ui.rerender(section({ ...offline, state: 'connected', onNetwork: true })); + + expect(ui.queryByText(/Could not reach/)).toBeNull(); + }); }); diff --git a/pro b/pro index 661315f13..28b7ef69e 160000 --- a/pro +++ b/pro @@ -1 +1 @@ -Subproject commit 661315f13088ce9146486d40ab77793da1f780c2 +Subproject commit 28b7ef69ecc21b1c85a4d0b92fde91bcd84b9ae1