Revert "perf(contextual-retrieval): bound per-chunk synopsis concurrency (#2628)"

This reverts commit 3454dca0b4.

src/core/contextual-retrieval-service.ts

@@ -45,6 +45,7 @@ import { embedBatch } from './embedding.ts';
 import { resolveContextualRetrievalMode } from './contextual-retrieval-resolver.ts';
 import {
   buildContextualPrefix,
+  extractFirstTwoSentences,
   modeRequiresHaiku,
   modeRequiresWrapper,
   sanitizeTitle,
@@ -56,8 +57,10 @@ import {
   SYNOPSIS_DOC_MAX_CHARS,
   type GeneratePerChunkSynopsisResult,
 } from './page-summary.ts';
-import type { SynopsisFailureKind } from './audit-synopsis.ts';
-import { runSlidingPool } from './worker-pool.ts';
+import {
+  logSynopsisFailure,
+  type SynopsisFailureKind,
+} from './audit-synopsis.ts';
 import type { BrainEngine } from './engine.ts';
 import type { ChunkInput, CRMode, Page } from './types.ts';
 import type { SourceRow } from './sources-ops.ts';
@@ -70,24 +73,6 @@ import type { SourceRow } from './sources-ops.ts';
  * corpus_generation hash.
  */
 export const TITLE_WRAPPER_VERSION = 1;
-const DEFAULT_HAIKU_MODEL = 'anthropic:claude-haiku-4-5-20251001';
-export const DEFAULT_CONTEXTUAL_CHUNK_CONCURRENCY = 4;
-export const MAX_CONTEXTUAL_CHUNK_CONCURRENCY = 16;
-
-export function resolveContextualChunkConcurrency(
-  env: Record<string, string | undefined> = process.env,
-): number {
-  const raw = env.GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY;
-  if (raw === undefined || raw.trim() === '') return DEFAULT_CONTEXTUAL_CHUNK_CONCURRENCY;
-  const n = Number(raw);
-  if (!Number.isFinite(n)) return DEFAULT_CONTEXTUAL_CHUNK_CONCURRENCY;
-  return clampContextualChunkConcurrency(n);
-}
-
-function clampContextualChunkConcurrency(n: number): number {
-  if (!Number.isFinite(n)) return DEFAULT_CONTEXTUAL_CHUNK_CONCURRENCY;
-  return Math.max(1, Math.min(MAX_CONTEXTUAL_CHUNK_CONCURRENCY, Math.trunc(n)));
-}
 
 /**
  * Embedding model placeholder. The actual model name lands here from
@@ -223,16 +208,12 @@ export interface ReembedPageArgs {
    * src/core/minions/rate-leases.ts here; inline callers (import-file,
    * reindex command) pass undefined and rely on gateway-level retry.
    */
-  acquireSynopsisLease?: () => Promise<unknown>;
-  releaseSynopsisLease?: (lease?: unknown) => Promise<void>;
-  /**
-   * Intra-page per-chunk synopsis concurrency. 1 preserves the legacy
-   * sequential loop exactly; higher values only parallelize Haiku synopsis
-   * calls. Embedding remains one batch after all synopses succeed.
-   */
-  chunkConcurrency?: number;
+  acquireSynopsisLease?: () => Promise<void>;
+  releaseSynopsisLease?: () => Promise<void>;
 }
 
+const DEFAULT_HAIKU_MODEL = 'anthropic:claude-haiku-4-5-20251001';
+
 /**
  * Re-embed one page through the active CR mode. Implements the D26 P0-2
  * two-phase build pattern.
@@ -451,41 +432,82 @@ async function tryBuildPhase1(opts: {
   }
 
   // per_chunk_synopsis path. Read source text via fallback chain,
-  // generate synopsis per chunk through a bounded sliding pool, then
+  // generate synopsis per chunk sequentially within this page (D10),
   // batch embed at the end (D27 P2-2).
   const sourceText = readSourceTextWithFallback(page, chunks);
-  const wrappedTexts: string[] = new Array(chunks.length);
-  const chunkConcurrency = clampContextualChunkConcurrency(
-    args.chunkConcurrency ?? resolveContextualChunkConcurrency(),
-  );
+  const wrappedTexts: string[] = [];
 
-  const poolResult = await runSlidingPool({
-    items: chunks,
-    workers: chunkConcurrency,
-    signal: args.abortSignal,
-    onError: 'abort',
-    failureLabel: (c) => String(c.chunk_index),
-    onItem: async (c, i) => {
-      wrappedTexts[i] = await buildWrappedChunkText({
-        chunk: c,
-        sourceText,
-        safeTitle,
-        page,
-        args,
-        haikuModel,
-      });
-    },
-  });
+  for (let i = 0; i < chunks.length; i++) {
+    const c = chunks[i];
 
-  if (poolResult.failures.length > 0) {
-    const failure = [...poolResult.failures].sort((a, b) => a.idx - b.idx)[0].error;
-    if (failure instanceof ChunkSynopsisPhase1Error) {
-      return failure.result;
+    // Code chunks always bypass the wrapper (D20-T4) — pass through.
+    if (c.chunk_source === 'fenced_code') {
+      wrappedTexts.push(c.chunk_text);
+      continue;
     }
-    throw failure;
-  }
-  if (poolResult.aborted || args.abortSignal?.aborted) {
-    return { kind: 'transient', cause: 'timeout', detail: 'aborted' };
+
+    // Acquire rate-lease per chunk (D26 P0-3). Inline callers pass no
+    // hooks; only the Minion handler wires through rate-leases.ts.
+    if (args.acquireSynopsisLease) {
+      await args.acquireSynopsisLease();
+    }
+
+    let synopsisResult: GeneratePerChunkSynopsisResult;
+    try {
+      synopsisResult = await generatePerChunkSynopsis({
+        documentText: sourceText,
+        chunkText: c.chunk_text,
+        pageTitle: page.title,
+        pageSlug: args.pageSlug,
+        sourceId: args.sourceId,
+        chunkIndex: c.chunk_index,
+        model: haikuModel,
+        abortSignal: args.abortSignal,
+      });
+    } finally {
+      if (args.releaseSynopsisLease) {
+        try {
+          await args.releaseSynopsisLease();
+        } catch {
+          // Lease release failure shouldn't abort the page; surfacing it
+          // would race with the synopsis result. Audit-only.
+        }
+      }
+    }
+
+    if (synopsisResult.kind === 'success') {
+      const prefix = buildContextualPrefix(safeTitle, synopsisResult.synopsis);
+      wrappedTexts.push(
+        wrapChunkForEmbedding(c.chunk_text, prefix, c.chunk_source),
+      );
+      continue;
+    }
+
+    // Failure classification per D27 P1-2:
+    //   refusal | empty | malformed → page-level fall-back to title-only
+    //   auth_failure → permanent (won't fix with retry)
+    //   rate_limit | timeout | network | provider_5xx → transient
+    //   source_missing → walked into fallback already; would be 'malformed'
+    //     from generatePerChunkSynopsis if we ever propagated it here
+    if (
+      synopsisResult.kind === 'refusal' ||
+      synopsisResult.kind === 'empty' ||
+      synopsisResult.kind === 'malformed'
+    ) {
+      return { kind: 'page_level_fallback_requested', cause: synopsisResult.kind };
+    }
+    if (synopsisResult.kind === 'auth_failure') {
+      return {
+        kind: 'permanent',
+        cause: synopsisResult.kind,
+        detail: synopsisResult.detail ?? 'auth failure',
+      };
+    }
+    return {
+      kind: 'transient',
+      cause: synopsisResult.kind,
+      detail: synopsisResult.detail ?? 'transient',
+    };
   }
 
   // All chunks synthesized successfully. Single batch embed (D27 P2-2).
@@ -506,113 +528,6 @@ async function tryBuildPhase1(opts: {
   }
 }
 
-class ChunkSynopsisPhase1Error extends Error {
-  constructor(readonly result: Exclude<Phase1Result, Phase1Success>) {
-    super(`chunk synopsis failed: ${result.kind}`);
-    this.name = 'ChunkSynopsisPhase1Error';
-  }
-}
-
-async function buildWrappedChunkText(opts: {
-  chunk: ChunkInput;
-  sourceText: string;
-  safeTitle: string;
-  page: Page;
-  args: ReembedPageArgs;
-  haikuModel: string;
-}): Promise<string> {
-  const { chunk: c, sourceText, safeTitle, page, args, haikuModel } = opts;
-
-  // Code chunks always bypass the wrapper (D20-T4) — pass through.
-  if (c.chunk_source === 'fenced_code') {
-    return c.chunk_text;
-  }
-
-  // Acquire rate-lease per chunk (D26 P0-3). Inline callers pass no
-  // hooks; only the Minion handler wires through rate-leases.ts.
-  let lease: unknown;
-  let leaseAcquired = false;
-  let synopsisResult: GeneratePerChunkSynopsisResult;
-  try {
-    if (args.acquireSynopsisLease) {
-      try {
-        lease = await args.acquireSynopsisLease();
-      } catch (err) {
-        if (args.abortSignal?.aborted || isAbortError(err)) {
-          throw new ChunkSynopsisPhase1Error({
-            kind: 'transient',
-            cause: 'timeout',
-            detail: 'aborted',
-          });
-        }
-        throw err;
-      }
-      leaseAcquired = true;
-    }
-    synopsisResult = await generatePerChunkSynopsis({
-      documentText: sourceText,
-      chunkText: c.chunk_text,
-      pageTitle: page.title,
-      pageSlug: args.pageSlug,
-      sourceId: args.sourceId,
-      chunkIndex: c.chunk_index,
-      model: haikuModel,
-      abortSignal: args.abortSignal,
-    });
-  } finally {
-    if (leaseAcquired && args.releaseSynopsisLease) {
-      try {
-        await args.releaseSynopsisLease(lease);
-      } catch {
-        // Lease release failure shouldn't abort the page; surfacing it
-        // would race with the synopsis result. Audit-only.
-      }
-    }
-  }
-
-  if (synopsisResult.kind === 'success') {
-    const prefix = buildContextualPrefix(safeTitle, synopsisResult.synopsis);
-    return wrapChunkForEmbedding(c.chunk_text, prefix, c.chunk_source);
-  }
-
-  // Failure classification per D27 P1-2:
-  //   refusal | empty | malformed → page-level fall-back to title-only
-  //   auth_failure → permanent (won't fix with retry)
-  //   rate_limit | timeout | network | provider_5xx → transient
-  //   source_missing → walked into fallback already; would be 'malformed'
-  //     from generatePerChunkSynopsis if we ever propagated it here
-  if (
-    synopsisResult.kind === 'refusal' ||
-    synopsisResult.kind === 'empty' ||
-    synopsisResult.kind === 'malformed'
-  ) {
-    throw new ChunkSynopsisPhase1Error({
-      kind: 'page_level_fallback_requested',
-      cause: synopsisResult.kind,
-    });
-  }
-  if (synopsisResult.kind === 'auth_failure') {
-    throw new ChunkSynopsisPhase1Error({
-      kind: 'permanent',
-      cause: synopsisResult.kind,
-      detail: synopsisResult.detail ?? 'auth failure',
-    });
-  }
-  throw new ChunkSynopsisPhase1Error({
-    kind: 'transient',
-    cause: synopsisResult.kind,
-    detail: synopsisResult.detail ?? 'transient',
-  });
-}
-
-function isAbortError(err: unknown): boolean {
-  return (
-    typeof err === 'object' &&
-    err !== null &&
-    (err as { name?: unknown }).name === 'AbortError'
-  );
-}
-
 /**
  * Source-text fallback chain per D11:
  *   1. read page.source_path from disk (truest "document")

src/core/minions/handlers/contextual-reindex-per-chunk.ts

@@ -40,7 +40,6 @@ import { UnrecoverableError } from '../types.ts';
 import type { BrainEngine } from '../../engine.ts';
 import {
   reembedPageWithContextualRetrieval,
-  resolveContextualChunkConcurrency,
   type ReembedPageResult,
 } from '../../contextual-retrieval-service.ts';
 import {
@@ -133,7 +132,7 @@ export function makeContextualReindexHandler(opts: MakeContextualReindexHandlerO
     // call inside the service acquires/releases a lease against the
     // shared key across all worker processes.
     const maxConcurrent = resolveMaxConcurrent();
-    const chunkConcurrency = resolveContextualChunkConcurrency();
+    let currentLeaseId: number | null = null;
 
     const result: ReembedPageResult = await reembedPageWithContextualRetrieval({
       engine,
@@ -142,32 +141,32 @@ export function makeContextualReindexHandler(opts: MakeContextualReindexHandlerO
       globalMode,
       killSwitchDisabled,
       abortSignal: ctx.signal,
-      chunkConcurrency,
       acquireSynopsisLease: async () => {
         // Poll-acquire with brief backoff. The service's per-chunk loop
-        // is bounded within a page; this guards against the cross-worker
-        // pile-up and remains the global rate governor.
+        // is sequential within a page; this guards against the cross-
+        // worker pile-up.
         let attempts = 0;
         const maxAttempts = 60; // ~1 min max wait per chunk before giving up
         while (attempts < maxAttempts) {
-          if (ctx.signal.aborted) throw abortError();
           const res = await acquireLease(engine, RATE_LEASE_KEY, ctx.id, maxConcurrent, {
             ttlMs: 60_000,
           });
           if (res.acquired && res.leaseId != null) {
-            return res.leaseId;
+            currentLeaseId = res.leaseId;
+            return;
           }
           attempts++;
-          await sleepWithAbort(1000, ctx.signal);
+          await new Promise((r) => setTimeout(r, 1000));
         }
         throw new Error(
           `Failed to acquire ${RATE_LEASE_KEY} lease after ${maxAttempts} attempts; ` +
             `Haiku rate limit pile-up too deep.`,
         );
       },
-      releaseSynopsisLease: async (lease) => {
-        if (typeof lease === 'number') {
-          await releaseLease(engine, lease);
+      releaseSynopsisLease: async () => {
+        if (currentLeaseId != null) {
+          await releaseLease(engine, currentLeaseId);
+          currentLeaseId = null;
         }
       },
     });
@@ -219,26 +218,6 @@ async function tryLoadPageAcrossSources(
   return null;
 }
 
-function sleepWithAbort(ms: number, signal: AbortSignal): Promise<void> {
-  return new Promise((resolve, reject) => {
-    if (signal.aborted) {
-      reject(abortError());
-      return;
-    }
-    const timer = setTimeout(resolve, ms);
-    signal.addEventListener('abort', () => {
-      clearTimeout(timer);
-      reject(abortError());
-    }, { once: true });
-  });
-}
-
-function abortError(): Error {
-  const err = new Error('aborted');
-  err.name = 'AbortError';
-  return err;
-}
-
 function classifyResult(
   pageSlug: string,
   result: ReembedPageResult,

test/contextual-retrieval-service-pure.test.ts

@@ -1,38 +1,20 @@
 /**
  * Pure-function tests for src/core/contextual-retrieval-service.ts.
  *
- * This file pins the service's pure helpers plus hermetic service behavior
- * driven through fake engine + gateway seams. Full PGLite coverage lives in
- * test/e2e/contextual-retrieval-pglite.test.ts.
+ * The full service test (PHASE 1 + PHASE 2 happy path, refusal restart,
+ * transient error propagation) needs a real PGLite + gateway stub seam.
+ * That lands in test/e2e/contextual-retrieval.test.ts. This file pins
+ * the service's pure helpers: corpus_generation hash composition + the
+ * expectedMode helper used by the T9 reindex sweep predicate.
  */
 
-import { afterEach, describe, test, expect } from 'bun:test';
+import { describe, test, expect } from 'bun:test';
 import {
   computeCorpusGeneration,
   computeSourceTextHash,
   expectedModeForPageSourceOnly,
-  reembedPageWithContextualRetrieval,
-  resolveContextualChunkConcurrency,
   TITLE_WRAPPER_VERSION,
 } from '../src/core/contextual-retrieval-service.ts';
-import {
-  __setChatTransportForTests,
-  __setEmbedTransportForTests,
-  configureGateway,
-  resetGateway,
-  type ChatOpts,
-  type ChatResult,
-} from '../src/core/ai/gateway.ts';
-import type { ChunkInput } from '../src/core/types.ts';
-import { withEnv } from './helpers/with-env.ts';
-
-const TEST_DIMS = 1536;
-
-afterEach(() => {
-  __setChatTransportForTests(null);
-  __setEmbedTransportForTests(null);
-  resetGateway();
-});
 
 describe('computeCorpusGeneration', () => {
   test('returns 16-char hex hash', () => {
@@ -156,317 +138,3 @@ describe('expectedModeForPageSourceOnly (T9 reindex sweep helper)', () => {
     }
   });
 });
-
-describe('resolveContextualChunkConcurrency', () => {
-  test('defaults to 4 and reads the process env', async () => {
-    await withEnv({ GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY: undefined }, async () => {
-      expect(resolveContextualChunkConcurrency()).toBe(4);
-    });
-    await withEnv({ GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY: '7' }, async () => {
-      expect(resolveContextualChunkConcurrency()).toBe(7);
-    });
-  });
-
-  test('clamps to [1, 16] and ignores invalid values', () => {
-    expect(resolveContextualChunkConcurrency({
-      GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY: '0',
-    })).toBe(1);
-    expect(resolveContextualChunkConcurrency({
-      GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY: '-3',
-    })).toBe(1);
-    expect(resolveContextualChunkConcurrency({
-      GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY: '99',
-    })).toBe(16);
-    expect(resolveContextualChunkConcurrency({
-      GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY: '1.9',
-    })).toBe(1);
-    expect(resolveContextualChunkConcurrency({
-      GBRAIN_CONTEXTUAL_CHUNK_CONCURRENCY: 'not-a-number',
-    })).toBe(4);
-  });
-});
-
-describe('per-chunk synopsis concurrency', () => {
-  test('concurrency > 1 preserves chunk-order embed input', async () => {
-    const chunks = makeChunks(['alpha', 'beta', 'gamma', 'delta']);
-    const delays: Record<string, number> = { alpha: 30, beta: 5, gamma: 20, delta: 1 };
-    const sequential = await runWithChatStub({
-      chunks,
-      concurrency: 1,
-      delayForChunk: (chunk) => delays[chunk] ?? 1,
-    });
-    const parallel = await runWithChatStub({
-      chunks,
-      concurrency: 4,
-      delayForChunk: (chunk) => delays[chunk] ?? 1,
-    });
-
-    expect(parallel.result.kind).toBe('success');
-    expect(parallel.embedInputs).toEqual(sequential.embedInputs);
-    expect(parallel.embeddedChunks.map((c) => c.chunk_text)).toEqual(
-      chunks.map((c) => c.chunk_text),
-    );
-  });
-
-  test('concurrency is bounded', async () => {
-    let active = 0;
-    let maxActive = 0;
-    let leaseActive = 0;
-    let maxLeaseActive = 0;
-    let acquired = 0;
-    let released = 0;
-    const chunks = makeChunks(Array.from({ length: 8 }, (_, i) => `chunk-${i}`));
-    const out = await runWithChatStub({
-      chunks,
-      concurrency: 3,
-      acquireSynopsisLease: async () => {
-        acquired++;
-        leaseActive++;
-        maxLeaseActive = Math.max(maxLeaseActive, leaseActive);
-        return acquired;
-      },
-      releaseSynopsisLease: async () => {
-        released++;
-        leaseActive--;
-      },
-      chat: async (opts) => {
-        active++;
-        maxActive = Math.max(maxActive, active);
-        try {
-          await delay(20, opts.abortSignal);
-          return chatSuccess(`Synopsis for ${extractChunk(opts)}`);
-        } finally {
-          active--;
-        }
-      },
-    });
-
-    expect(out.result.kind).toBe('success');
-    expect(maxActive).toBeGreaterThan(1);
-    expect(maxActive).toBeLessThanOrEqual(3);
-    expect(maxLeaseActive).toBeLessThanOrEqual(3);
-    expect(acquired).toBe(8);
-    expect(released).toBe(8);
-    expect(leaseActive).toBe(0);
-  });
-
-  test('one chunk failure aborts queued work and falls back at page level', async () => {
-    let started = 0;
-    const chunks = makeChunks(Array.from({ length: 9 }, (_, i) => `chunk-${i}`));
-    const out = await runWithChatStub({
-      chunks,
-      concurrency: 3,
-      chat: async (opts) => {
-        started++;
-        const chunk = extractChunk(opts);
-        if (chunk === 'chunk-0') return chatSuccess('');
-        await delay(30, opts.abortSignal);
-        return chatSuccess(`Synopsis for ${chunk}`);
-      },
-    });
-
-    expect(out.result.kind).toBe('page_fallback');
-    expect(started).toBeLessThanOrEqual(3);
-  });
-
-  test('fenced code chunks bypass synopsis calls and leases', async () => {
-    let chatCalls = 0;
-    let leaseCalls = 0;
-    const chunks: ChunkInput[] = [
-      { chunk_index: 0, chunk_text: 'intro', chunk_source: 'compiled_truth' },
-      { chunk_index: 1, chunk_text: 'const x = 1;', chunk_source: 'fenced_code' },
-      { chunk_index: 2, chunk_text: 'outro', chunk_source: 'compiled_truth' },
-    ];
-
-    const out = await runWithChatStub({
-      chunks,
-      concurrency: 3,
-      acquireSynopsisLease: async () => {
-        leaseCalls++;
-      },
-      releaseSynopsisLease: async () => {},
-      chat: async (opts) => {
-        chatCalls++;
-        return chatSuccess(`Synopsis for ${extractChunk(opts)}`);
-      },
-    });
-
-    expect(out.result.kind).toBe('success');
-    expect(chatCalls).toBe(2);
-    expect(leaseCalls).toBe(2);
-    expect(out.embedInputs[1]).toBe('const x = 1;');
-  });
-
-  test('abortSignal cancels in-flight and queued synopsis work promptly', async () => {
-    const controller = new AbortController();
-    let started = 0;
-    const chunks = makeChunks(Array.from({ length: 20 }, (_, i) => `chunk-${i}`));
-    const startedAt = Date.now();
-    const promise = runWithChatStub({
-      chunks,
-      concurrency: 4,
-      abortSignal: controller.signal,
-      chat: async (opts) => {
-        started++;
-        await delay(1000, opts.abortSignal);
-        return chatSuccess(`Synopsis for ${extractChunk(opts)}`);
-      },
-    });
-    setTimeout(() => controller.abort(), 20);
-
-    const out = await promise;
-    expect(out.result.kind).toBe('transient_error');
-    if (out.result.kind === 'transient_error') {
-      expect(out.result.cause).toBe('timeout');
-    }
-    expect(started).toBeLessThanOrEqual(4);
-    expect(Date.now() - startedAt).toBeLessThan(300);
-  });
-});
-
-function makeChunks(texts: string[]): ChunkInput[] {
-  return texts.map((text, i) => ({
-    chunk_index: i,
-    chunk_text: text,
-    chunk_source: 'compiled_truth',
-  }));
-}
-
-async function runWithChatStub(opts: {
-  chunks: ChunkInput[];
-  concurrency: number;
-  abortSignal?: AbortSignal;
-  delayForChunk?: (chunk: string) => number;
-  chat?: (opts: ChatOpts) => Promise<ChatResult>;
-  acquireSynopsisLease?: () => Promise<unknown>;
-  releaseSynopsisLease?: (lease?: unknown) => Promise<void>;
-}) {
-  configureGateway({
-    embedding_model: 'openai:text-embedding-3-large',
-    embedding_dimensions: TEST_DIMS,
-    env: { OPENAI_API_KEY: 'sk-test' },
-  });
-
-  const embedInputs: string[][] = [];
-  __setEmbedTransportForTests(async ({ values }: any) => {
-    embedInputs.push([...values]);
-    return {
-      embeddings: values.map((_: string, i: number) =>
-        Array.from({ length: TEST_DIMS }, () => 0.001 + i * 0.001),
-      ),
-      usage: { tokens: 0 },
-    } as any;
-  });
-
-  __setChatTransportForTests(opts.chat ?? (async (chatOpts) => {
-    const chunk = extractChunk(chatOpts);
-    await delay(opts.delayForChunk?.(chunk) ?? 1, chatOpts.abortSignal);
-    return chatSuccess(`Synopsis for ${chunk}`);
-  }));
-
-  const engine = makeServiceEngine(opts.chunks);
-  const result = await reembedPageWithContextualRetrieval({
-    engine,
-    pageSlug: 'wiki/concepts/concurrency-test',
-    sourceId: 'default',
-    globalMode: 'per_chunk_synopsis',
-    chunkConcurrency: opts.concurrency,
-    abortSignal: opts.abortSignal,
-    ...(opts.acquireSynopsisLease && { acquireSynopsisLease: opts.acquireSynopsisLease }),
-    ...(opts.releaseSynopsisLease && { releaseSynopsisLease: opts.releaseSynopsisLease }),
-  });
-
-  return {
-    result,
-    embedInputs: embedInputs.flat(),
-    embeddedChunks: engine.embeddedChunks as ChunkInput[],
-  };
-}
-
-function makeServiceEngine(chunks: ChunkInput[]) {
-  const engine: any = {
-    embeddedChunks: [] as ChunkInput[],
-    async getPage() {
-      return {
-        id: 1,
-        slug: 'wiki/concepts/concurrency-test',
-        source_id: 'default',
-        type: 'concept',
-        title: 'Concurrency Test',
-        compiled_truth: chunks.map((c) => c.chunk_text).join('\n\n'),
-        timeline: '',
-        frontmatter: {},
-        created_at: new Date('2026-01-01T00:00:00Z'),
-        updated_at: new Date('2026-01-01T00:00:00Z'),
-        deleted_at: null,
-      };
-    },
-    async executeRaw() {
-      return [{
-        id: 'default',
-        name: 'Default',
-        local_path: null,
-        last_commit: null,
-        last_sync_at: null,
-        config: {},
-        created_at: new Date('2026-01-01T00:00:00Z'),
-        contextual_retrieval_mode: null,
-        trust_frontmatter_overrides: false,
-      }];
-    },
-    async getChunks() {
-      return chunks;
-    },
-    async transaction(fn: (tx: any) => Promise<void>) {
-      await fn({
-        upsertChunks: async (_slug: string, embedded: ChunkInput[]) => {
-          engine.embeddedChunks = embedded;
-        },
-        updatePageContextualRetrievalState: async () => {},
-      });
-    },
-    async updatePageContextualRetrievalState() {},
-  };
-  return engine;
-}
-
-function extractChunk(opts: ChatOpts): string {
-  const content = String(opts.messages[0]?.content ?? '');
-  return content.match(/<chunk>\n([\s\S]*?)\n<\/chunk>/)?.[1] ?? '';
-}
-
-function chatSuccess(text: string): ChatResult {
-  return {
-    text,
-    blocks: [],
-    stopReason: 'end',
-    usage: {
-      input_tokens: 0,
-      output_tokens: 0,
-      cache_read_tokens: 0,
-      cache_creation_tokens: 0,
-    },
-    model: 'stub:chat',
-    providerId: 'stub',
-  };
-}
-
-function delay(ms: number, signal?: AbortSignal): Promise<void> {
-  return new Promise((resolve, reject) => {
-    if (signal?.aborted) {
-      reject(abortError());
-      return;
-    }
-    const timer = setTimeout(resolve, ms);
-    signal?.addEventListener('abort', () => {
-      clearTimeout(timer);
-      reject(abortError());
-    }, { once: true });
-  });
-}
-
-function abortError(): Error {
-  const err = new Error('aborted');
-  err.name = 'AbortError';
-  return err;
-}