perf: reduce hashing and query overhead in history processing (#277)

* perf: batch semantic hashing and stream history revisions

* perf: reuse prepared queries for streaming turn lookups

* perf: retain owned history across the durable store boundary

Author: SaladDay

packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts

@@ -564,7 +564,14 @@ function decodeRecords(records: readonly CanonicalHistoryEnvelope[]): CanonicalH
 }
 
 function revisionOfNormalized(records: readonly CanonicalHistoryEnvelope[]): string {
-  return `sha256:${createHash('sha256').update(canonicalJson(records), 'utf8').digest('hex')}`;
+  // Preserve the canonical JSON array bytes without building a sorted copy and
+  // serialized string of the entire history at once.
+  const hash = createHash('sha256').update('[');
+  for (let index = 0; index < records.length; index += 1) {
+    if (index > 0) hash.update(',');
+    hash.update(canonicalJson(records[index]), 'utf8');
+  }
+  return `sha256:${hash.update(']').digest('hex')}`;
 }
 
 function decodeMessage(value: unknown): CanonicalHistoryMessage {

packages/local-runtime-v2/src/service/session-system/query-collapse-state.ts

@@ -217,19 +217,30 @@ function readByCurrentTurn(db: AppDb, sessionId: string, currentTurnId: string)
     .get();
 }
 
+const processingQueries = new WeakMap<AppDb, ReturnType<typeof prepareProcessingQuery>>();
+
 function readProcessingByCurrentTurn(db: AppDb, sessionId: string, currentTurnId: string) {
+  let query = processingQueries.get(db);
+  if (!query) {
+    query = prepareProcessingQuery(db);
+    processingQueries.set(db, query);
+  }
+  return query.get({ sessionId, currentTurnId });
+}
+
+function prepareProcessingQuery(db: AppDb) {
   return db
     .select()
     .from(queryCollapseViewStates)
     .where(
       and(
-        eq(queryCollapseViewStates.sessionId, sessionId),
-        eq(queryCollapseViewStates.currentTurnId, currentTurnId),
+        eq(queryCollapseViewStates.sessionId, sql.placeholder('sessionId')),
+        eq(queryCollapseViewStates.currentTurnId, sql.placeholder('currentTurnId')),
         isNull(queryCollapseViewStates.processingFinishedAtMs),
       ),
     )
     .orderBy(desc(queryCollapseViewStates.updatedAtMs), desc(queryCollapseViewStates.queryKey))
-    .get();
+    .prepare();
 }
 
 function readByKey(db: AppDb, sessionId: string, queryKey: string) {

packages/local-runtime-v2/src/service/turn-system/agent-host/history/durable-canonical-history-store.ts

@@ -146,11 +146,12 @@ export class DurableCanonicalHistoryStore implements CanonicalHistoryStore {
     const detached = captureSemanticSnapshot(snapshot).value;
     validateCanonicalHistorySnapshot(detached);
     assertCanonicalIdentityVector(detached);
-    return Object.freeze({
+    return captureSemanticSnapshot({
       revision: detached.revision.trim(),
-      messages: Object.freeze([...detached.messages]),
-      identityVector: Object.freeze([...detached.identityVector]),
-    });
+      // Keep the separately owned arrays reusable at the next snapshot boundary.
+      messages: captureSemanticSnapshot([...detached.messages]).value,
+      identityVector: captureSemanticSnapshot([...detached.identityVector]).value,
+    }).value;
   }
 }
 

packages/local-runtime-v2/src/service/turn-system/agent-host/history/incremental-sha256.ts

@@ -2,26 +2,42 @@ import { createHash } from 'node:crypto';
 
 const UTF8_CHUNK_CODE_UNITS = 8 * 1_024;
 
-/** Native SHA-256 with bounded UTF-8 chunks and the existing update boundaries. */
+/** Native SHA-256 with bounded batching and independent UTF-8 update boundaries. */
 export class IncrementalSha256 {
   private readonly hash = createHash('sha256');
   private finalized = false;
+  private pending = '';
 
   update(value: string): void {
     if (this.finalized) throw new Error('SHA-256 digest is already finalized.');
     let offset = 0;
     while (offset < value.length) {
       const end = safeUtf8ChunkEnd(value, offset);
-      this.hash.update(value.slice(offset, end), 'utf8');
+      let chunk = value.slice(offset, end);
+      // Each update encodes separately. A trailing high surrogate must not pair
+      // with a low surrogate supplied by a later update when we batch strings.
+      if (isHighSurrogate(chunk.charCodeAt(chunk.length - 1))) {
+        chunk = chunk.slice(0, -1) + '\ufffd';
+      }
+      if (this.pending.length + chunk.length > UTF8_CHUNK_CODE_UNITS) this.flush();
+      this.pending += chunk;
+      if (this.pending.length === UTF8_CHUNK_CODE_UNITS) this.flush();
       offset = end;
     }
   }
 
   digestHex(): string {
     if (this.finalized) throw new Error('SHA-256 digest is already finalized.');
+    this.flush();
     this.finalized = true;
     return this.hash.digest('hex');
   }
+
+  private flush(): void {
+    if (this.pending.length === 0) return;
+    this.hash.update(this.pending, 'utf8');
+    this.pending = '';
+  }
 }
 
 function safeUtf8ChunkEnd(value: string, offset: number): number {

packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from 'node:crypto';
 
-import { and, desc, eq, gt, inArray } from 'drizzle-orm';
+import { and, desc, eq, gt, inArray, sql } from 'drizzle-orm';
 
 import {
   sessionLocks,
@@ -491,11 +491,29 @@ async function completeTurnSessionDeletion(
   );
 }
 
+const receiptQueries = new WeakMap<
+  TurnRepositoryOptions['db'],
+  ReturnType<typeof prepareReceiptQuery>
+>();
+
+function prepareReceiptQuery(db: TurnRepositoryOptions['db']) {
+  return db
+    .select()
+    .from(turnIngress)
+    .where(eq(turnIngress.turnId, sql.placeholder('turnId')))
+    .prepare();
+}
+
 async function findTurnReceipt(
   options: TurnRepositoryOptions,
   turnId: string,
 ): ReturnType<TurnRepository['findReceipt']> {
-  const receipt = options.db.select().from(turnIngress).where(eq(turnIngress.turnId, turnId)).get();
+  let query = receiptQueries.get(options.db);
+  if (!query) {
+    query = prepareReceiptQuery(options.db);
+    receiptQueries.set(options.db, query);
+  }
+  const receipt = query.get({ turnId });
   if (!receipt) return undefined;
   if (
     !storedBusyReason(receipt.busyReason) ||

test/history-processing.test.ts

@@ -1,9 +1,23 @@
 import { createHash } from 'node:crypto';
-import { CanonicalHistoryJsonlDataSource } from '../packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.js';
+import {
+  canonicalActiveHistoryRevision,
+  canonicalHistoryRevision,
+  CanonicalHistoryJsonlDataSource,
+  decodeCanonicalHistoryEnvelope,
+} from '../packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.js';
+import { canonicalJson } from '../packages/local-runtime-v2/src/infra/file/canonical-history-json-value.js';
 import { mkdtemp, rm, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { describe, expect, it, vi } from 'vitest';
+import { eq } from 'drizzle-orm';
+import { DatabaseClient } from '../packages/local-runtime-v2/src/infra/db/client.js';
+import { initializeDatabase } from '../packages/local-runtime-v2/src/infra/db/initialize.js';
+import { queryCollapseViewStates } from '../packages/local-runtime-v2/src/infra/db/schema/query-collapse.js';
+import { turnIngress } from '../packages/local-runtime-v2/src/infra/db/schema/turn.js';
+import { createQueryCollapseState } from '../packages/local-runtime-v2/src/service/session-system/query-collapse-state.js';
+import { createQueueTurnAdmissionPriorityFence } from '../packages/local-runtime-v2/src/service/session-system/index.js';
+import { createTurnRepository } from '../packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.js';
 import { IncrementalSha256 } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/incremental-sha256.js';
 import { captureSemanticSnapshot } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.js';
 import { DurableCanonicalHistoryStore } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/durable-canonical-history-store.js';
@@ -18,6 +32,106 @@ import {
 } from '../packages/local-runtime-v2/src/service/session-system/messages/history/session-history-paths.js';
 import type { SessionRecord } from '../packages/local-runtime-v2/src/service/session-system/sessions/repo/contract.js';
 
+describe('prepared runtime reads', () => {
+  async function withDatabase(
+    run: (client: DatabaseClient, writer: DatabaseClient) => Promise<void>,
+  ) {
+    const dataDir = await mkdtemp(join(tmpdir(), 'mcode-prepared-reads-'));
+    const client = new DatabaseClient({ dataDir });
+    const writer = new DatabaseClient({ dataDir });
+    try {
+      await initializeDatabase({ database: client, dataDir });
+      await run(client, writer);
+    } finally {
+      writer.close();
+      client.close();
+      await rm(dataDir, { recursive: true, force: true });
+    }
+  }
+
+  it('keeps processing reads fresh across sessions, completion and another connection', async () => {
+    await withDatabase(async (client, writer) => {
+      const state = createQueryCollapseState({ db: client.db, nowMs: () => 1 });
+      expect(await state.findProcessingByCurrentTurn('s1', 't1')).toBeUndefined();
+      await state.start({ sessionId: 's1', currentTurnId: 't1', queryKey: 'a' });
+      await state.start({ sessionId: 's1', currentTurnId: 't1', queryKey: 'b' });
+      await state.start({ sessionId: 's2', currentTurnId: 't1', queryKey: 'other' });
+      expect((await state.findProcessingByCurrentTurn('s1', 't1'))?.queryKey).toBe('b');
+      expect((await state.findProcessingByCurrentTurn('s2', 't1'))?.queryKey).toBe('other');
+      expect(await state.findProcessingByCurrentTurn('s1', 'missing')).toBeUndefined();
+      await state.finish({
+        sessionId: 's1',
+        currentTurnId: 't1',
+        queryKey: 'b',
+        forceExpanded: false,
+      });
+      expect((await state.findProcessingByCurrentTurn('s1', 't1'))?.queryKey).toBe('a');
+      writer.db
+        .update(queryCollapseViewStates)
+        .set({ processingFinishedAtMs: 2 })
+        .where(eq(queryCollapseViewStates.sessionId, 's1'))
+        .run();
+      expect(await state.findProcessingByCurrentTurn('s1', 't1')).toBeUndefined();
+      expect((await state.findProcessingByCurrentTurn('s2', 't1'))?.queryKey).toBe('other');
+      client.close();
+      const reopened = createQueryCollapseState({ db: client.db });
+      expect((await reopened.findProcessingByCurrentTurn('s2', 't1'))?.queryKey).toBe('other');
+    });
+  });
+
+  it('rereads receipts and validates corruption after a previous successful lookup', async () => {
+    await withDatabase(async (client, writer) => {
+      const repository = createTurnRepository({
+        db: client.db,
+        priorityFence: createQueueTurnAdmissionPriorityFence(),
+        sessionAdmission: { rejectionInTransaction: () => undefined },
+      });
+      expect(await repository.findReceipt('turn-1')).toBeUndefined();
+      for (let i = 1; i <= 2; i += 1) {
+        writer.db
+          .insert(turnIngress)
+          .values({
+            turnId: `turn-${i}`,
+            sessionId: `s${i}`,
+            busyReason: 'turn',
+            inputJson: '{}',
+            status: 'accepted',
+            acceptedAtMs: 1,
+            acceptedSequence: i,
+            inputDigest: `digest-${i}`,
+          })
+          .run();
+      }
+      expect(await repository.findReceipt('turn-1')).toMatchObject({
+        sessionId: 's1',
+        acceptedSequence: 1,
+      });
+      expect(await repository.findReceipt('turn-2')).toMatchObject({
+        sessionId: 's2',
+        acceptedSequence: 2,
+      });
+      writer.db
+        .update(turnIngress)
+        .set({ inputDigest: '' })
+        .where(eq(turnIngress.turnId, 'turn-1'))
+        .run();
+      await expect(repository.findReceipt('turn-1')).rejects.toThrow('malformed');
+      writer.db
+        .update(turnIngress)
+        .set({ inputDigest: 'edited', acceptedSequence: 3 })
+        .where(eq(turnIngress.turnId, 'turn-1'))
+        .run();
+      expect(await repository.findReceipt('turn-1')).toMatchObject({
+        inputDigest: 'edited',
+        acceptedSequence: 3,
+      });
+      writer.db.delete(turnIngress).where(eq(turnIngress.turnId, 'turn-1')).run();
+      expect(await repository.findReceipt('turn-1')).toBeUndefined();
+      expect(await repository.findReceipt("turn-2' OR 1=1 --")).toBeUndefined();
+    });
+  });
+});
+
 describe('native incremental semantic hashing', () => {
   const inputs = ['', 'abc', '中文🙂', '\ud800', '\udc00', 'a'.repeat(8191) + '🙂tail'];
   for (const length of [55, 56, 63, 64, 65, 8191, 8192, 8193, 32769]) {
@@ -38,6 +152,63 @@ describe('native incremental semantic hashing', () => {
       createHash('sha256').update('\ud83d').update('\ude42').digest('hex'),
     );
   });
+  it('matches native updates across repeated batches and split surrogate pairs', () => {
+    const actual = new IncrementalSha256();
+    const expected = createHash('sha256');
+    const parts = ['key', ':', '', '\ud83d', '\ude42', '中文🙂', 'x'.repeat(8191), '🙂tail'];
+    for (let index = 0; index < 257; index += 1) {
+      for (const part of parts) {
+        actual.update(part);
+        expected.update(part, 'utf8');
+      }
+    }
+    expect(actual.digestHex()).toBe(expected.digest('hex'));
+  });
+});
+
+describe('streamed canonical history revisions', () => {
+  it.each([0, 1, 100])('preserves the canonical JSON digest for %i records', (length) => {
+    const records = Array.from({ length }, (_, index) => ({
+      message_id: `msg-${index}`,
+      turn_id: `turn-${index}`,
+      message: {
+        role: 'user',
+        timestamp: index,
+        content: '中文🙂\ud800'.repeat(2048),
+        metadata: { z: [null, true, -0], '10': 'ten', '2': 'two', a: { b: '"\\\n' } },
+      },
+    }));
+    const expected = `sha256:${createHash('sha256')
+      .update(canonicalJson(records.map(decodeCanonicalHistoryEnvelope)), 'utf8')
+      .digest('hex')}`;
+    expect(canonicalHistoryRevision(records)).toBe(expected);
+    expect(canonicalActiveHistoryRevision(records)).toBe(expected);
+    if (records.length > 0) {
+      records[0]!.message.content = 'edited';
+      expect(canonicalHistoryRevision(records)).not.toBe(expected);
+    }
+  });
+  it('keeps active and settled sequence validation distinct', () => {
+    const pending = [
+      {
+        message_id: 'msg-assistant',
+        turn_id: 'turn-1',
+        message: {
+          role: 'assistant',
+          timestamp: 1,
+          content: [
+            { type: 'toolCall', id: 'call-1', name: 'bash', arguments: { command: 'pwd' } },
+          ],
+        },
+      },
+    ];
+    const expected = `sha256:${createHash('sha256')
+      .update(canonicalJson(pending.map(decodeCanonicalHistoryEnvelope)), 'utf8')
+      .digest('hex')}`;
+    expect(canonicalActiveHistoryRevision(pending)).toBe(expected);
+    expect(() => canonicalHistoryRevision(pending)).toThrow('tool results');
+    expect(() => canonicalActiveHistoryRevision([pending[0]!, pending[0]!])).toThrow();
+  });
 });
 
 describe('semantic snapshots', () => {
@@ -486,6 +657,49 @@ describe('committed history read reuse', () => {
     expect(readActive).not.toHaveBeenCalled();
     expect((await store.read('synthetic-session')).revision).toBe('r2');
   });
+  it('preserves owned history through the store and the next commit snapshot', async () => {
+    const original = {
+      revision: ' r1 ',
+      messages: [{ role: 'user', timestamp: 1, content: 'original'.repeat(4096) }],
+      identityVector: ['msg-1'],
+    };
+    const store = new DurableCanonicalHistoryStore({
+      read: async () => original,
+      readActive: async () => original,
+      append: async () => original,
+      replace: async () => original,
+    });
+    const committed = await store.append(change);
+    const expectedContent = original.messages[0]!.content;
+    original.messages[0]!.content = 'changed';
+    original.identityVector[0] = 'changed';
+    expect(committed.revision).toBe('r1');
+    expect(committed.messages).toEqual([{ role: 'user', timestamp: 1, content: expectedContent }]);
+    expect(committed.identityVector).toEqual(['msg-1']);
+    expect(Object.isFrozen(committed.messages[0])).toBe(true);
+    expect(Object.isFrozen(committed.messages)).toBe(true);
+    expect(Object.isFrozen(committed.identityVector)).toBe(true);
+    const clone = vi.spyOn(globalThis, 'structuredClone');
+    try {
+      expect(captureSemanticSnapshot(committed).value).toBe(committed);
+      expect(
+        captureSemanticSnapshot({ committedMessages: committed.messages }).value.committedMessages,
+      ).toBe(committed.messages);
+      expect(clone).not.toHaveBeenCalled();
+    } finally {
+      clone.mockRestore();
+    }
+    const sharedEmpty: never[] = [];
+    const emptyStore = new DurableCanonicalHistoryStore({
+      read: async () => ({ revision: 'r2', messages: sharedEmpty, identityVector: sharedEmpty }),
+      readActive: async () => original,
+      append: async () => original,
+      replace: async () => original,
+    });
+    const empty = await emptyStore.read('synthetic-session');
+    expect(empty.messages).not.toBe(empty.identityVector);
+    expect(empty.messages).toEqual([]);
+  });
   it('retains legacy rereads and rejects invalid commits or write failures', async () => {
     const readActive = vi.fn(async () => ({
       revision: 'r1',