fix(dream): stamp incremental extraction watermark (#2636)

The Dream cycle disables sync's inline extraction and routes changed
slugs through extractForSlugs, which flushed link/timeline batches but
never stamped links_extracted_at — so incrementally extracted pages
stayed permanently visible to `extract --stale` / doctor.

Collect processedRefs per successfully processed page and stamp them
via stampExtracted (best-effort) after both batch flushes, non-dry-run
mode 'all' only. Source-id threading from the original PR #2637 already
landed on master via #1503/#1747, so this rebase carries only the
missing watermark stamp plus regression tests.

Takeover of #2637.

Co-authored-by: JavanC <JavanC@users.noreply.github.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/commands/extract.ts

@@ -1025,6 +1025,10 @@ async function extractForSlugs(
   let linksCreated = 0;
   let timelineCreated = 0;
   let pagesProcessed = 0;
+  // #2636: successfully processed pages get their extraction watermark
+  // stamped after the final flush (mode 'all' only — a partial-mode run
+  // hasn't done the full extraction the watermark asserts).
+  const processedRefs: Array<{ slug: string; source_id: string }> = [];
 
   // Issue #972: read the basename flag once per extract run.
   const globalBasename = await isGlobalBasenameEnabled(engine);
@@ -1113,6 +1117,7 @@ async function extractForSlugs(
         }
 
         pagesProcessed++;
+        if (!dryRun) processedRefs.push({ slug, source_id: sourceId ?? 'default' });
       } catch { /* skip unreadable */ }
       progress.tick(1);
     },
@@ -1120,6 +1125,13 @@ async function extractForSlugs(
 
   await flushLinks();
   await flushTimeline();
+  // #2636: the Dream cycle disables sync's inline extraction and routes
+  // changed slugs through this incremental path — without a stamp here,
+  // those pages never get links_extracted_at and stay permanently visible
+  // to `extract --stale` / doctor. Stamp only after BOTH batches flushed.
+  if (!dryRun && mode === 'all') {
+    await stampExtracted(engine, processedRefs);
+  }
   progress.finish();
 
   if (!jsonMode) {

src/core/ai/recipes/litellm-proxy.ts

@@ -56,36 +56,6 @@ export const litellmProxy: Recipe = {
       cost_per_1m_output_usd: undefined,
       price_last_verified: '2026-06-14',
     },
-    // LiteLLM normalizes Cohere / Voyage / Jina / etc. rerank backends to the
-    // same wire shape gbrain's gateway.rerank() already speaks (the
-    // ZeroEntropy/llama.cpp contract):
-    //   { model, query, documents, top_n } → { results: [{ index, relevance_score }] }
-    // So any rerank model the user registers in their LiteLLM config is
-    // reachable via `gbrain config set search.reranker.model litellm:<model>`
-    // with no request/response adapter — same as embeddings ride the proxy.
-    reranker: {
-      models: [], // user-provided; whatever rerank models the proxy serves
-      // No canonical default — the proxy defines its own model ids. The user
-      // sets search.reranker.model explicitly (mirrors the embedding
-      // touchpoint's user_provided_models contract).
-      default_model: '',
-      // The proxied backend bills (Cohere/Voyage/…); pricing-unknown is the
-      // honest state — same stance as this recipe's embedding/chat
-      // touchpoints and budget-tracker's deliberate litellm exclusion from
-      // the free-provider sets.
-      cost_per_1m_tokens_usd: undefined,
-      price_last_verified: '2026-06-27',
-      max_payload_bytes: 5_000_000,
-      // LEAF path only (matches llama-server-reranker's convention). LiteLLM
-      // serves both `/rerank` and `/v1/rerank`, and LITELLM_BASE_URL may be
-      // set with or without the `/v1` suffix (the setup_hint allows both), so
-      // the leaf form yields a valid route either way:
-      //   http://localhost:4000    + /rerank → /rerank        ✓
-      //   http://localhost:4000/v1 + /rerank → /v1/rerank     ✓
-      // Pinning '/v1/rerank' here would double to /v1/v1/rerank → 404 on
-      // /v1-suffixed bases.
-      path: '/rerank',
-    },
   },
-  setup_hint: 'Run LiteLLM (https://docs.litellm.ai) in front of any provider; set LITELLM_BASE_URL (include the /v1 suffix if your proxy serves the OpenAI route there, e.g. http://localhost:4000/v1) + pass --embedding-model litellm:<model> and --embedding-dimensions <N>. For rerank: register a rerank model in LiteLLM and set search.reranker.model litellm:<model-name>.',
+  setup_hint: 'Run LiteLLM (https://docs.litellm.ai) in front of any provider; set LITELLM_BASE_URL (include the /v1 suffix if your proxy serves the OpenAI route there, e.g. http://localhost:4000/v1) + pass --embedding-model litellm:<model> and --embedding-dimensions <N>.',
 };

test/ai/recipe-litellm-reranker.test.ts

@@ -1,88 +0,0 @@
-/**
- * litellm-proxy reranker touchpoint smoke.
- *
- * Sibling of recipe-llama-server-reranker.test.ts. Pins the reranker
- * touchpoint on the LiteLLM proxy recipe so:
- *  - the touchpoint exists with the LEAF '/rerank' path (LiteLLM serves both
- *    /rerank and /v1/rerank, so the leaf form is valid whether or not the
- *    user's LITELLM_BASE_URL carries the /v1 suffix the setup_hint allows)
- *  - a /v1-suffixed base URL does NOT produce /v1/v1/rerank (the original
- *    community PR pinned '/v1/rerank' which 404s on /v1-suffixed bases)
- *  - models: [] (user-provided; proxy defines the model ids)
- *  - pricing stays undefined (proxy can front a paid provider — same honest
- *    pricing-unknown stance as the embedding/chat touchpoints)
- *
- * The gateway.rerank() URL tests drive the real URL builder via the stubbed
- * transport (same seam as test/ai/rerank.test.ts).
- */
-
-import { describe, expect, test, afterEach } from 'bun:test';
-import { getRecipe } from '../../src/core/ai/recipes/index.ts';
-import {
-  configureGateway,
-  resetGateway,
-  rerank,
-  __setRerankTransportForTests,
-} from '../../src/core/ai/gateway.ts';
-
-afterEach(() => {
-  __setRerankTransportForTests(null);
-  resetGateway();
-});
-
-describe('recipe: litellm reranker touchpoint', () => {
-  test('declares reranker touchpoint with leaf /rerank path', () => {
-    const r = getRecipe('litellm')!;
-    const tp = r.touchpoints.reranker;
-    expect(tp).toBeDefined();
-    expect(tp!.path).toBe('/rerank');
-    expect(tp!.max_payload_bytes).toBe(5_000_000);
-  });
-
-  test('reranker touchpoint uses empty models[] for user-provided model ids', () => {
-    const r = getRecipe('litellm')!;
-    expect(r.touchpoints.reranker!.models).toEqual([]);
-  });
-
-  test('pricing stays undefined — proxy can front a paid provider', () => {
-    const r = getRecipe('litellm')!;
-    expect(r.touchpoints.reranker!.cost_per_1m_tokens_usd).toBeUndefined();
-  });
-
-  test('setup_hint keeps the /v1-suffix guidance AND mentions rerank', () => {
-    const r = getRecipe('litellm')!;
-    expect(r.setup_hint).toMatch(/\/v1 suffix/);
-    expect(r.setup_hint).toMatch(/search\.reranker\.model litellm:/);
-  });
-});
-
-describe('gateway.rerank() URL via litellm recipe', () => {
-  async function capturedRerankUrl(baseUrl?: string): Promise<string> {
-    configureGateway({
-      reranker_model: 'litellm:my-reranker',
-      env: {},
-      ...(baseUrl ? { base_urls: { litellm: baseUrl } } : {}),
-    });
-    let capturedUrl = '';
-    __setRerankTransportForTests(async (url) => {
-      capturedUrl = url;
-      return new Response(
-        JSON.stringify({ results: [{ index: 0, relevance_score: 0.9 }] }),
-        { status: 200, headers: { 'content-type': 'application/json' } },
-      );
-    });
-    await rerank({ query: 'q', documents: ['d'] });
-    return capturedUrl;
-  }
-
-  test('default base (no /v1 suffix) → /rerank', async () => {
-    const url = await capturedRerankUrl();
-    expect(url).toBe('http://localhost:4000/rerank');
-  });
-
-  test('/v1-suffixed base → /v1/rerank, NOT /v1/v1/rerank', async () => {
-    const url = await capturedRerankUrl('http://localhost:4000/v1');
-    expect(url).toBe('http://localhost:4000/v1/rerank');
-    expect(url).not.toContain('/v1/v1/');
-  });
-});

test/extract-incremental.test.ts

@@ -60,6 +60,51 @@ async function seedPage(slug: string, body: string): Promise<void> {
 }
 
 describe('runExtractCore — incremental cycle path (#417)', () => {
+  test('Dream incremental all-mode stamps the source-scoped extraction watermark (#2636)', async () => {
+    await engine.executeRaw(
+      `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3)`,
+      ['repo-a', 'repo-a', tempDir],
+    );
+    await engine.putPage('people/alice-example', {
+      type: 'person',
+      title: 'alice-example',
+      compiled_truth: '# alice',
+      timeline: '',
+      frontmatter: {},
+      content_hash: 'h',
+    }, { sourceId: 'repo-a' });
+    writeFileSync(join(tempDir, 'people/alice-example.md'), '# alice');
+
+    await runExtractCore(engine as unknown as BrainEngine, {
+      mode: 'all',
+      dir: tempDir,
+      slugs: ['people/alice-example'],
+      sourceId: 'repo-a',
+    });
+
+    const rows = await engine.executeRaw<{ links_extracted_at: string | null }>(
+      `SELECT links_extracted_at FROM pages WHERE slug = $1 AND source_id = $2`,
+      ['people/alice-example', 'repo-a'],
+    );
+    expect(rows[0]?.links_extracted_at).not.toBeNull();
+    expect(await engine.countStalePagesForExtraction({ sourceId: 'repo-a' })).toBe(0);
+  });
+
+  test('Dream incremental dry-run does NOT stamp the watermark', async () => {
+    await seedPage('people/alice-example', '# alice');
+    await runExtractCore(engine as unknown as BrainEngine, {
+      mode: 'all',
+      dir: tempDir,
+      slugs: ['people/alice-example'],
+      dryRun: true,
+    });
+    const rows = await engine.executeRaw<{ links_extracted_at: string | null }>(
+      `SELECT links_extracted_at FROM pages WHERE slug = $1`,
+      ['people/alice-example'],
+    );
+    expect(rows[0]?.links_extracted_at ?? null).toBeNull();
+  });
+
   test('1. slugs: [] returns immediately with zero counts (early-return path)', async () => {
     await seedPage('people/alice-example', '# alice');
     const result = await runExtractCore(engine as unknown as BrainEngine, {