fix(doctor): clear extract_atoms raw source-holder backlog + normalize pooler direct-URL overrides

Takeover of #2242, split to the two concerns that survive review:

- extract_atoms: exclude source pages whose frontmatter declares a raw
  payload pointer from discovery AND the doctor backlog count (shared SQL
  fragment so they can't drift). Extraction on these yields zero atoms, so
  no atom row is ever written and they re-enter the backlog every cycle —
  a permanent no-progress doctor blocker.
- connection-manager: a direct-URL override (opts/env) that still points at
  the Supavisor TRANSACTION pooler (port 6543, usually a copy-paste of the
  primary URL) is normalized to the real direct host via deriveDirectUrl.
  Session-mode pooler overrides (port 5432) pass through — they are a
  legitimate direct-ish target, which the original PR would have nulled out.

Dropped from the original PR: orphan-reporting atom exclusions (master
already excludes atoms/ and raw/ first segments plus /raw/ segments in
src/commands/orphans.ts) and the drain dry-run status tweak.

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

src/core/connection-manager.ts

@@ -167,6 +167,33 @@ export function deriveDirectUrl(url: string): string | null {
   }
 }
 
+/** True when the URL targets Supavisor transaction mode (port 6543). */
+function isTransactionPoolerUrl(url: string): boolean {
+  try {
+    return new URL(url.replace(/^postgres(ql)?:\/\//, 'http://')).port === '6543';
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * Resolve the direct-pool URL from an explicit/env override + the primary URL.
+ *
+ * A direct override still pointing at the TRANSACTION-mode pooler (port 6543,
+ * usually a copy-paste of the primary URL) is a misconfiguration: the manager
+ * would believe DDL/bulk work runs on a long-timeout direct connection while
+ * still routing through Supavisor transaction mode (short timeouts, no
+ * prepared statements). Normalize it via deriveDirectUrl. Session-mode pooler
+ * URLs (pooler host, port 5432) pass through — they are a legitimate
+ * direct-ish target when the db.<ref> host is unreachable.
+ */
+export function normalizeDirectUrl(primaryUrl: string, override?: string | null): string | null {
+  const candidate = override ?? deriveDirectUrl(primaryUrl);
+  if (!candidate) return null;
+  if (!isTransactionPoolerUrl(candidate)) return candidate;
+  return deriveDirectUrl(candidate) ?? deriveDirectUrl(primaryUrl);
+}
+
 /**
  * Read kill-switch state from env. Subordinate to parent manager's state
  * when present (A2 inheritance).
@@ -213,9 +240,10 @@ export class ConnectionManager {
     } else {
       this._killSwitch = readKillSwitchEnv();
       this._isSupabase = isSupabasePoolerUrl(opts.url);
-      // Direct URL: explicit override > env > derive > null
+      // Direct URL: explicit override > env > derive > null. Pooler-shaped
+      // overrides are normalized to a real direct host (or dropped).
       const envOverride = process.env.GBRAIN_DIRECT_DATABASE_URL;
-      this._directUrl = opts.directUrl ?? envOverride ?? deriveDirectUrl(opts.url);
+      this._directUrl = normalizeDirectUrl(opts.url, opts.directUrl ?? envOverride);
     }
   }
 

src/core/cycle/extract-atoms.ts

@@ -83,6 +83,14 @@ const SYNTHESIS_OUTPUT_TYPES = new Set<string>(['atom', 'concept']);
 
 const PAGE_DISCOVERY_BUDGET = 50;
 const MIN_PAGE_CHARS_FOR_EXTRACTION = 500;
+// Source pages whose frontmatter declares a `raw` payload pointer hold raw
+// import data, not extractable prose. Extraction on them yields zero atoms,
+// so no atom row is ever written and they re-enter discovery + the doctor
+// backlog count on every cycle — a permanent no-progress loop. Shared by
+// discoverExtractablePages and countExtractAtomsBacklog so the phase and the
+// doctor check can't drift.
+const RAW_SOURCE_HOLDER_EXCLUSION_SQL =
+  `AND NOT (p.type = 'source' AND COALESCE(p.frontmatter ? 'raw', false))`;
 
 /**
  * Pure allowlist policy: the legacy floor UNION the pack's `extractable: true`
@@ -207,6 +215,10 @@ interface DiscoveredPage {
  *      participate in the NOT EXISTS check anyway.
  *   #4 dream_generated exclusion — prevents the phase from chewing
  *      its own output (e.g. dream-generated originals).
+ *   #5 raw source-holder exclusion — source pages that only point at a raw
+ *      import payload are not extractable prose; counting them creates a
+ *      permanent backlog/no-progress loop (see
+ *      RAW_SOURCE_HOLDER_EXCLUSION_SQL).
  */
 export async function discoverExtractablePages(
   engine: BrainEngine,
@@ -225,6 +237,7 @@ export async function discoverExtractablePages(
       AND p.content_hash IS NOT NULL
       AND COALESCE(p.frontmatter->>'imported_from',   '') <> 'markdown-greenfield'
       AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true'
+      ${RAW_SOURCE_HOLDER_EXCLUSION_SQL}
       AND length(COALESCE(p.compiled_truth, '')) >= $3
       ${hasFilter ? "AND p.slug = ANY($5::text[])" : ''}
       AND NOT EXISTS (
@@ -297,6 +310,7 @@ export async function countExtractAtomsBacklog(
            AND p.content_hash IS NOT NULL
            AND COALESCE(p.frontmatter->>'imported_from',   '') <> 'markdown-greenfield'
            AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true'
+           ${RAW_SOURCE_HOLDER_EXCLUSION_SQL}
            AND length(COALESCE(p.compiled_truth, '')) >= $3
            AND NOT EXISTS (
              SELECT 1 FROM pages atom
@@ -310,6 +324,7 @@ export async function countExtractAtomsBacklog(
            AND p.content_hash IS NOT NULL
            AND COALESCE(p.frontmatter->>'imported_from',   '') <> 'markdown-greenfield'
            AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true'
+           ${RAW_SOURCE_HOLDER_EXCLUSION_SQL}
            AND length(COALESCE(p.compiled_truth, '')) >= $2
            AND NOT EXISTS (
              SELECT 1 FROM pages atom

test/connection-manager.serial.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, test, beforeEach, afterEach } from 'bun:test';
 import {
   isSupabasePoolerUrl,
   deriveDirectUrl,
+  normalizeDirectUrl,
   readKillSwitchEnv,
   resolveDirectPoolSize,
   ConnectionManager,
@@ -78,6 +79,44 @@ describe('deriveDirectUrl', () => {
   });
 });
 
+describe('normalizeDirectUrl', () => {
+  test('normalizes a transaction-pooler (6543) override to the real direct host', () => {
+    const direct = normalizeDirectUrl(
+      'postgresql://postgres.abcxyz:p@aws-0-us-west-2.pooler.supabase.com:6543/postgres',
+      'postgresql://postgres.abcxyz:p@aws-0-us-west-2.pooler.supabase.com:6543/postgres',
+    );
+    expect(direct).toBe('postgresql://postgres:p@db.abcxyz.supabase.co:5432/postgres');
+  });
+
+  test('keeps a non-pooler direct override', () => {
+    const direct = normalizeDirectUrl(
+      'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
+      'postgresql://u:p@custom-direct.example.com:5432/db',
+    );
+    expect(direct).toBe('postgresql://u:p@custom-direct.example.com:5432/db');
+  });
+
+  test('keeps a session-mode pooler override (pooler host, port 5432)', () => {
+    const sessionUrl = 'postgresql://postgres.abc:p@aws.pooler.supabase.com:5432/db';
+    const direct = normalizeDirectUrl(
+      'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
+      sessionUrl,
+    );
+    expect(direct).toBe(sessionUrl);
+  });
+
+  test('no override: derives from the primary as before', () => {
+    const direct = normalizeDirectUrl(
+      'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
+    );
+    expect(direct).toContain('db.abc.supabase.co:5432');
+  });
+
+  test('no override, non-Supabase primary: null', () => {
+    expect(normalizeDirectUrl('postgresql://u:p@localhost:5432/db')).toBeNull();
+  });
+});
+
 describe('readKillSwitchEnv', () => {
   let original: string | undefined;
   beforeEach(() => { original = process.env.GBRAIN_DISABLE_DIRECT_POOL; });
@@ -145,13 +184,18 @@ describe('resolveDirectPoolSize', () => {
 
 describe('ConnectionManager — describeMode + dual-pool routing', () => {
   let originalKillSwitch: string | undefined;
+  let originalDirectUrl: string | undefined;
   beforeEach(() => {
     originalKillSwitch = process.env.GBRAIN_DISABLE_DIRECT_POOL;
+    originalDirectUrl = process.env.GBRAIN_DIRECT_DATABASE_URL;
     delete process.env.GBRAIN_DISABLE_DIRECT_POOL;
+    delete process.env.GBRAIN_DIRECT_DATABASE_URL;
   });
   afterEach(() => {
     if (originalKillSwitch === undefined) delete process.env.GBRAIN_DISABLE_DIRECT_POOL;
     else process.env.GBRAIN_DISABLE_DIRECT_POOL = originalKillSwitch;
+    if (originalDirectUrl === undefined) delete process.env.GBRAIN_DIRECT_DATABASE_URL;
+    else process.env.GBRAIN_DIRECT_DATABASE_URL = originalDirectUrl;
   });
 
   test('non-Supabase URL → single mode', () => {
@@ -190,6 +234,25 @@ describe('ConnectionManager — describeMode + dual-pool routing', () => {
     expect(cm.resolveDirectUrl()).toContain('custom-direct.example.com');
   });
 
+  test('explicit transaction-pooler directUrl override is normalized to direct host', () => {
+    const cm = new ConnectionManager({
+      url: 'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
+      directUrl: 'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
+    });
+    expect(cm.resolveDirectUrl()).toContain('db.abc.supabase.co:5432');
+    expect(cm.resolveDirectUrl()).not.toContain(':6543');
+  });
+
+  test('env transaction-pooler directUrl override is normalized to direct host', () => {
+    process.env.GBRAIN_DIRECT_DATABASE_URL =
+      'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db';
+    const cm = new ConnectionManager({
+      url: 'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
+    });
+    expect(cm.resolveDirectUrl()).toContain('db.abc.supabase.co:5432');
+    expect(cm.resolveDirectUrl()).not.toContain(':6543');
+  });
+
   test('host string contains creds neither in describeMode nor resolveDirectUrl logging', () => {
     const cm = new ConnectionManager({
       url: 'postgresql://postgres.abc:secret@aws.pooler.supabase.com:6543/db',

test/doctor-extract-atoms-backlog.test.ts

@@ -76,6 +76,17 @@ describe('countExtractAtomsBacklog (issue #1678)', () => {
     });
     expect(await countExtractAtomsBacklog(engine)).toBe(0);
   });
+
+  it('ignores raw source-holder pages (permanent no-progress backlog otherwise)', async () => {
+    await engine.putPage('wiki/raw-email-source', {
+      type: 'source',
+      title: 'Raw email source',
+      compiled_truth: BODY,
+      frontmatter: { raw: 'raw/email/example.md' },
+    });
+    expect(await countExtractAtomsBacklog(engine)).toBe(0);
+    expect(await countExtractAtomsBacklog(engine, 'default')).toBe(0);
+  });
 });
 
 describe('computeExtractAtomsBacklogCheck (issue #1678)', () => {

test/extract-atoms-page-discovery.test.ts

@@ -178,6 +178,18 @@ describe('v0.41.2.1: discoverExtractablePages SQL contract', () => {
     expect(discovered.map((d) => d.slug)).toEqual(['original/normal']);
   });
 
+  test('raw source-holder pages excluded (#5 — no permanent no-progress backlog)', async () => {
+    await seedPage({ slug: 'source/normal', type: 'source' });
+    await seedPage({
+      slug: 'wiki/raw-email-source',
+      type: 'source',
+      frontmatter: { raw: 'raw/email/example.md' },
+    });
+
+    const discovered = await discoverExtractablePages(engine, 'default');
+    expect(discovered.map((d) => d.slug)).toEqual(['source/normal']);
+  });
+
   test('pages with NULL content_hash excluded (D9 #3 — no .slice crash)', async () => {
     await seedPage({ slug: 'meeting/with-hash', type: 'meeting' });
     await seedPage({ slug: 'meeting/no-hash', type: 'meeting', content_hash: null });