perf(mobile): keep chat history safe on failed loads, index and cache RAG vectors

__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 };
       });

__tests__/pro/sync/KnownDevicesSection.integration.test.tsx

@@ -108,4 +108,36 @@ describe('<KnownDevicesSection/> 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) => (
+      <KnownDevicesSection
+        devices={[device]}
+        pairingDeviceId={null}
+        onPair={() => 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();
+  });
 });

__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<string, unknown>) {
+      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<string, unknown> = {},
+) => ({
+  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<typeof chatList>['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: [],
+    });
+  });
+});

__tests__/unit/services/rag/database.test.ts

@@ -150,8 +150,40 @@ 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);
     });
+
+    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]);
+        // 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,
+        );
+      },
+    );
   });
 
   describe('hasEmbeddingsForDocument', () => {