feat(lbug): runtime-guard streamQuery against the WAL-checkpoint driver (#2264)

streamQuery is deliberately not wrapped in withConnLock (its per-row callback can
re-enter the adapter), so its unlocked per-row reads could race a CHECKPOINT on the
shared connection — the corruption window the lock serializes everything else
against. That invariant was comment-only, safe today only because the serve/read
path forks analyze workers. Make it enforced:

- lbug-adapter: a walDriverActive flag + markWalDriverActive(bool); streamQuery
  throws an actionable error when the driver is active.
- wal-checkpoint-driver: arm the flag on start, disarm in stop() AFTER the in-flight
  CHECKPOINT drains (clearing earlier would briefly allow a race).

A future in-process analyze overlapping a stream now fails loud instead of
corrupting native state. (reentrancy test's lbug-adapter mock gains the new export.)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JBJomjoTdBV2eveDVq4JMm

gitnexus/src/core/lbug/lbug-adapter.ts

@@ -198,6 +198,20 @@ let conn: lbug.Connection | null = null;
 // reassigned only at open/close, never mid-load).
 const isSharedSingletonConn = (c: lbug.Connection): boolean => c === conn;
 
+// True while the manual WAL-checkpoint driver is running (toggled by the driver's
+// start/stop). `streamQuery` is deliberately NOT wrapped in withConnLock (its
+// per-row callback can re-enter the adapter), so it must NOT run while a CHECKPOINT
+// can fire on the unlocked read connection — the exact overlap the lock serializes
+// everything else against (#2264). The serve/read path never starts the driver, so
+// this stays false there; an in-process analyze overlapping a stream would trip the
+// guard in streamQuery instead of silently corrupting native state.
+let walDriverActive = false;
+
+/** Toggled by the WAL-checkpoint driver's start (true) / stop (false). @see streamQuery */
+export const markWalDriverActive = (active: boolean): void => {
+  walDriverActive = active;
+};
+
 let currentDbPath: string | null = null;
 let currentDbReadOnly = false;
 let ftsLoaded = false;
@@ -1529,6 +1543,18 @@ export const streamQuery = async (
   cypher: string,
   onRow: (row: any) => void | Promise<void>,
 ): Promise<number> => {
+  if (walDriverActive) {
+    // streamQuery reads rows on the singleton connection WITHOUT withConnLock; if
+    // the WAL-checkpoint driver is live, those reads could race a CHECKPOINT — the
+    // #2264 corruption window. Today the serve/read path never runs the driver
+    // (analyze runs in a forked worker), so this fails loud only if a future
+    // in-process analyze overlaps a stream. Run analysis in a worker, or stop the
+    // driver before streaming. See conn-lock.ts.
+    throw new Error(
+      'streamQuery cannot run while the WAL-checkpoint driver is active (it would ' +
+        'race a CHECKPOINT on the unlocked read connection — #2264).',
+    );
+  }
   if (!conn) {
     throw new Error('LadybugDB not initialized. Call initLbug first.');
   }

gitnexus/src/core/lbug/wal-checkpoint-driver.ts

@@ -33,7 +33,7 @@
  */
 
 import { logger } from '../logger.js';
-import { tryFlushWAL } from './lbug-adapter.js';
+import { tryFlushWAL, markWalDriverActive } from './lbug-adapter.js';
 import { isLbugCheckpointIoError } from './lbug-config.js';
 
 /**
@@ -162,6 +162,10 @@ export const startWalCheckpointDriver = (
   let stopped = false;
   let inflight: Promise<void> | null = null;
 
+  // Arm the streamQuery guard: while this driver runs, an unlocked streamQuery on
+  // the singleton connection could race a CHECKPOINT (#2264). Cleared in stop().
+  markWalDriverActive(true);
+
   const tick = async (): Promise<void> => {
     if (stopped) return;
     // Reentrancy guard: setInterval keeps firing on its fixed cadence even when
@@ -218,6 +222,9 @@ export const startWalCheckpointDriver = (
           /* swallowed in tick() — surface path is the surrounding write */
         }
       }
+      // Disarm AFTER the in-flight CHECKPOINT drains — clearing it earlier would
+      // briefly let a streamQuery race the still-finishing CHECKPOINT (#2264).
+      markWalDriverActive(false);
     },
   };
 };

gitnexus/test/unit/stream-query-driver-guard.test.ts

@@ -0,0 +1,52 @@
+/**
+ * Unit tests for the streamQuery WAL-driver guard (#2264). streamQuery is
+ * deliberately NOT wrapped in withConnLock (its per-row callback re-enters the
+ * adapter), so it must refuse to run while the WAL-checkpoint driver is live —
+ * otherwise its unlocked per-row reads could race a CHECKPOINT on the shared
+ * connection (the corruption window the lock serializes everything else against).
+ * Today the serve/read path never starts the driver; this guard fails loud if a
+ * future in-process analyze ever overlaps a stream.
+ */
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
+import { streamQuery, markWalDriverActive } from '../../src/core/lbug/lbug-adapter.js';
+import { startWalCheckpointDriver } from '../../src/core/lbug/wal-checkpoint-driver.js';
+
+describe('streamQuery WAL-driver guard (#2264)', () => {
+  beforeEach(() => {
+    // Manual checkpoint defaults on; pin it so the driver path is deterministic.
+    vi.stubEnv('GITNEXUS_WAL_MANUAL_CHECKPOINT', '1');
+  });
+
+  afterEach(() => {
+    markWalDriverActive(false);
+    vi.unstubAllEnvs();
+  });
+
+  it('throws when the WAL-checkpoint driver is active', async () => {
+    markWalDriverActive(true);
+    await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow(
+      /WAL-checkpoint driver is active/,
+    );
+  });
+
+  it('passes the guard when inactive (reaching the not-initialized check)', async () => {
+    markWalDriverActive(false);
+    await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow(
+      /not initialized/,
+    );
+  });
+
+  it('startWalCheckpointDriver arms the guard; stop() disarms it', async () => {
+    const driver = startWalCheckpointDriver({ periodMs: 1_000_000 });
+    try {
+      await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow(
+        /WAL-checkpoint driver is active/,
+      );
+    } finally {
+      await driver.stop();
+    }
+    await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow(
+      /not initialized/,
+    );
+  });
+});

gitnexus/test/unit/wal-checkpoint-driver-reentrancy.test.ts

@@ -14,6 +14,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
 vi.mock('../../src/core/lbug/lbug-adapter.js', () => ({
   tryFlushWAL: vi.fn(),
+  // The driver now toggles the streamQuery guard on start/stop (#2264).
+  markWalDriverActive: vi.fn(),
 }));
 
 import { startWalCheckpointDriver } from '../../src/core/lbug/wal-checkpoint-driver.js';