Compare commits

..
Author SHA1 Message Date
ddb000b0e2 fix(dream): keep dream --dry-run --json stdout clean of embed summaries (#394)
The cycle's embed phase called runEmbedCore with no output suppression, so
the '[dry-run] Would embed ...' / 'Embedded N chunks ...' slog summaries
landed on stdout ahead of the JSON CycleReport, breaking the documented
stdout-clean-for-JSON contract (docs/progress-events.md).

Adds EmbedOpts.quiet gating the human stdout summary slog sites in
embed.ts (embedPage, embedAll, embedAllStale); the cycle's runPhaseEmbed
sets quiet: true since it reports counts via its own PhaseResult. Errors
and warnings still go to stderr regardless.

Takeover of #854 (same approach, reimplemented on current master — the
original patch predates the slog migration and the widened
embedAll/embedAllStale signatures). Regression test ported from #854.

Co-authored-by: Kage18 <Kage18@users.noreply.github.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-21 14:31:13 -07:00
8 changed files with 75 additions and 149 deletions
+5 -8
View File
@@ -21,17 +21,14 @@ GBrain is tuned for the Supabase **Transaction pooler** (port 6543): it
auto-disables prepared statements there and routes `engine.transaction()`
(migrations, DDL, sync imports) to a derived **direct** connection
(`db.<ref>.supabase.co:5432`). That direct host is IPv6-only, so on an
IPv4-only host it is unreachable. When that happens gbrain now falls back to
the pooler automatically (one stderr warning, then single-pool mode for the
rest of the process) — but the pooler's ~2-min statement timeout can truncate
very long migrations or bulk imports.
IPv4-only host, reads work but sync **silently skips most pages**. This is the
number one cause of "sync ran but nothing happened."
Fix: make the direct connection reachable over IPv4. Either set
`GBRAIN_DIRECT_DATABASE_URL` to the **Session pooler** string (port 5432 on the
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on.
`GBRAIN_DISABLE_DIRECT_POOL=1` skips the direct pool (and the fallback warning)
entirely. Verify by running `gbrain sync` and checking that the page count in
`gbrain stats` matches the syncable file count in the repo.
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on. Verify by
running `gbrain sync` and checking that the page count in `gbrain stats` matches
the syncable file count in the repo.
### The Primitives
+5 -8
View File
@@ -2720,17 +2720,14 @@ GBrain is tuned for the Supabase **Transaction pooler** (port 6543): it
auto-disables prepared statements there and routes `engine.transaction()`
(migrations, DDL, sync imports) to a derived **direct** connection
(`db.<ref>.supabase.co:5432`). That direct host is IPv6-only, so on an
IPv4-only host it is unreachable. When that happens gbrain now falls back to
the pooler automatically (one stderr warning, then single-pool mode for the
rest of the process) — but the pooler's ~2-min statement timeout can truncate
very long migrations or bulk imports.
IPv4-only host, reads work but sync **silently skips most pages**. This is the
number one cause of "sync ran but nothing happened."
Fix: make the direct connection reachable over IPv4. Either set
`GBRAIN_DIRECT_DATABASE_URL` to the **Session pooler** string (port 5432 on the
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on.
`GBRAIN_DISABLE_DIRECT_POOL=1` skips the direct pool (and the fallback warning)
entirely. Verify by running `gbrain sync` and checking that the page count in
`gbrain stats` matches the syncable file count in the repo.
`pooler.supabase.com` host, IPv4), or enable Supabase's IPv4 add-on. Verify by
running `gbrain sync` and checking that the page count in `gbrain stats` matches
the syncable file count in the repo.
### The Primitives
+33 -15
View File
@@ -107,6 +107,14 @@ export interface EmbedOpts {
* runs lock every source in sorted order. dryRun skips it.
*/
singleFlight?: boolean;
/**
* #394: suppress human stdout summaries (the `[dry-run] Would embed ...` /
* `Embedded N chunks ...` slog lines). Set by structured-output callers —
* the cycle's embed phase (dream --json must keep stdout JSON-clean per
* docs/progress-events.md) reports counts via its own PhaseResult instead.
* Errors/warnings still go to stderr regardless.
*/
quiet?: boolean;
}
/**
@@ -253,7 +261,7 @@ export async function runEmbedCore(engine: BrainEngine, opts: EmbedOpts): Promis
for (const s of opts.slugs) {
if (isAborted(opts.signal)) break; // #1737: stop the per-slug loop on abort
try {
await embedPage(engine, s, !!opts.dryRun, result, opts.sourceId, opts.signal);
await embedPage(engine, s, !!opts.dryRun, result, opts.sourceId, opts.signal, opts.quiet);
} catch (e: unknown) {
serr(` Error embedding ${s}: ${e instanceof Error ? e.message : e}`);
}
@@ -347,6 +355,7 @@ export async function runEmbedCore(engine: BrainEngine, opts: EmbedOpts): Promis
catchUp: opts.catchUp,
pacer,
paceMaxConcurrency,
quiet: opts.quiet,
}, opts.signal);
} finally {
// E1: surface pacing telemetry (human + structured) when pacing was on.
@@ -376,7 +385,7 @@ export async function runEmbedCore(engine: BrainEngine, opts: EmbedOpts): Promis
return result;
}
if (opts.slug) {
await embedPage(engine, opts.slug, !!opts.dryRun, result, opts.sourceId, opts.signal);
await embedPage(engine, opts.slug, !!opts.dryRun, result, opts.sourceId, opts.signal, opts.quiet);
return result;
}
throw new Error('No embed target specified. Pass { slug }, { slugs }, { all }, or { stale }.');
@@ -521,6 +530,7 @@ async function embedPage(
result: EmbedResult,
sourceId?: string,
signal?: AbortSignal,
quiet?: boolean,
) {
const opts = sourceId ? { sourceId } : undefined;
const page = await engine.getPage(slug, opts);
@@ -565,7 +575,7 @@ async function embedPage(
result.skipped += chunks.length - toEmbed.length;
if (toEmbed.length === 0) {
slog(`${slug}: all ${chunks.length} chunks already embedded`);
if (!quiet) slog(`${slug}: all ${chunks.length} chunks already embedded`);
result.pages_processed++;
return;
}
@@ -602,7 +612,7 @@ async function embedPage(
}
result.embedded += toEmbed.length;
result.pages_processed++;
slog(`${slug}: embedded ${toEmbed.length} chunks`);
if (!quiet) slog(`${slug}: embedded ${toEmbed.length} chunks`);
}
async function embedAll(
@@ -620,6 +630,8 @@ async function embedAll(
pacer?: DbPacer;
/** Resolved concurrency cap (E-1: the worker count, no separate permit). */
paceMaxConcurrency?: number;
/** #394: suppress human stdout summaries (structured-output callers). */
quiet?: boolean;
},
signal?: AbortSignal,
) {
@@ -763,10 +775,12 @@ async function embedAll(
});
// Stdout summary preserved for scripts/tests that grep for counts.
if (dryRun) {
slog(`[dry-run] Would embed ${result.would_embed} chunks across ${pages.length} pages`);
} else {
slog(`Embedded ${result.embedded} chunks across ${pages.length} pages`);
if (!staleOpts?.quiet) {
if (dryRun) {
slog(`[dry-run] Would embed ${result.would_embed} chunks across ${pages.length} pages`);
} else {
slog(`Embedded ${result.embedded} chunks across ${pages.length} pages`);
}
}
}
@@ -802,6 +816,8 @@ async function embedAllStale(
pacer?: DbPacer;
/** Resolved concurrency cap (E-1: the worker count, no separate permit). */
paceMaxConcurrency?: number;
/** #394: suppress human stdout summaries (structured-output callers). */
quiet?: boolean;
},
signature?: string,
externalSignal?: AbortSignal,
@@ -819,7 +835,7 @@ async function embedAllStale(
signature,
...(sourceId && { sourceId }),
});
if (invalidated > 0) {
if (invalidated > 0 && !staleOpts?.quiet) {
slog(`[embed] invalidated ${invalidated} chunk(s) embedded under a prior model signature`);
}
}
@@ -830,10 +846,12 @@ async function embedAllStale(
dryRun && signature ? { ...sourceOpt, signature } : sourceOpt,
);
if (staleCount === 0) {
if (dryRun) {
slog('[dry-run] Would embed 0 chunks (0 stale found)');
} else {
slog('Embedded 0 chunks (0 stale found)');
if (!staleOpts?.quiet) {
if (dryRun) {
slog('[dry-run] Would embed 0 chunks (0 stale found)');
} else {
slog('Embedded 0 chunks (0 stale found)');
}
}
return;
}
@@ -842,7 +860,7 @@ async function embedAllStale(
result.would_embed += staleCount;
result.total_chunks += staleCount;
if (onProgress) onProgress(1, 1, 0);
slog(`[dry-run] Would embed ${staleCount} stale chunks`);
if (!staleOpts?.quiet) slog(`[dry-run] Would embed ${staleCount} stale chunks`);
return;
}
@@ -1082,7 +1100,7 @@ async function embedAllStale(
if (budgetTimer) clearTimeout(budgetTimer);
}
slog(`Embedded ${result.embedded} chunks across ${totalProcessedPages} pages`);
if (!staleOpts?.quiet) slog(`Embedded ${result.embedded} chunks across ${totalProcessedPages} pages`);
// #1946 (OV2a): a catch-up pass that completed without being aborted but left
// chunks unembedded means those chunks are stuck (a non-transient embed
-6
View File
@@ -1078,9 +1078,6 @@ async function initPostgres(opts: {
console.warn(' Direct connections are IPv6 only and fail in many environments.');
console.warn(' Use the Transaction pooler connection string instead (port 6543):');
console.warn(' Supabase Dashboard > Connect (top bar) > Connection String > Transaction pooler');
console.warn(' (With a pooler URL, gbrain derives a direct connection for DDL and falls back');
console.warn(' to the pooler automatically if that host is unreachable. Power users:');
console.warn(' GBRAIN_DIRECT_DATABASE_URL overrides the derived URL; GBRAIN_DISABLE_DIRECT_POOL=1 disables it.)');
console.warn('');
}
@@ -1094,9 +1091,6 @@ async function initPostgres(opts: {
if (databaseUrl.includes('supabase.co') && (msg.includes('ECONNREFUSED') || msg.includes('ETIMEDOUT'))) {
console.error('Connection failed. Supabase direct connections (db.*.supabase.co:5432) are IPv6 only.');
console.error('Use the Transaction pooler connection string instead (port 6543).');
console.error('(gbrain derives its own direct connection from pooler URLs for DDL; if that host is');
console.error('unreachable it falls back to the pooler. GBRAIN_DIRECT_DATABASE_URL overrides the');
console.error('derived URL; GBRAIN_DISABLE_DIRECT_POOL=1 disables the direct pool entirely.)');
}
throw e;
}
+2 -48
View File
@@ -167,25 +167,6 @@ export function deriveDirectUrl(url: string): string | null {
}
}
/**
* Error codes that mean "the direct host is unreachable from this network"
* (#1641). The auto-derived db.<ref>.supabase.co host is IPv6-only without
* the paid IPv4 add-on, so ENOTFOUND/ECONNREFUSED here is expected on
* IPv4-only networks — we fall back to the pooler instead of failing init.
*/
const NETWORK_UNREACHABLE_CODES = [
'ENOTFOUND', 'ECONNREFUSED', 'ENETUNREACH', 'EHOSTUNREACH',
'ETIMEDOUT', 'CONNECT_TIMEOUT',
];
/** True when err looks like a network-unreachable failure (not auth/SQL). */
export function isNetworkUnreachableError(err: unknown): boolean {
const code = (err as { code?: unknown } | null)?.code;
if (typeof code === 'string' && NETWORK_UNREACHABLE_CODES.includes(code)) return true;
const msg = err instanceof Error ? err.message : String(err);
return NETWORK_UNREACHABLE_CODES.some(c => msg.includes(c));
}
/**
* Read kill-switch state from env. Subordinate to parent manager's state
* when present (A2 inheritance).
@@ -338,30 +319,7 @@ export class ConnectionManager {
throw err;
});
}
let pool: Sql | null;
try {
pool = await this._directInit;
} catch (err) {
// #1641: the derived direct host (db.<ref>.supabase.co) is IPv6-only
// without Supabase's IPv4 add-on. On IPv4-only networks the direct
// pool can never connect — permanently fall back to the read pool
// (self-activating kill-switch) instead of failing init/migrations.
// Non-network errors (auth, SQL) still throw: they mean misconfig,
// not unreachability.
if (isNetworkUnreachableError(err)) {
const alreadyWarned = this._killSwitch;
this._killSwitch = true;
const msg = err instanceof Error ? err.message : String(err);
if (!alreadyWarned) console.error(
`gbrain: direct connection to ${this._directUrl ? this.hostOnly(this._directUrl) : 'unknown host'} unreachable (${msg}); ` +
'falling back to the pooler for DDL/bulk (long migrations may hit the pooler statement timeout). ' +
'Set GBRAIN_DIRECT_DATABASE_URL to a reachable direct URL (e.g. the Session pooler, port 5432) or enable the Supabase IPv4 add-on; ' +
'GBRAIN_DISABLE_DIRECT_POOL=1 silences this.',
);
return this.getReadPool();
}
throw err;
}
const pool = await this._directInit;
if (!pool) {
// Defensive — initDirectPool should have thrown.
throw new Error('connection-manager: direct pool init returned null');
@@ -392,9 +350,8 @@ export class ConnectionManager {
},
};
const t0 = Date.now();
let pool: Sql | null = null;
try {
pool = postgres(this._directUrl, opts);
const pool = postgres(this._directUrl, opts);
// Probe to validate connectivity early.
await pool`SELECT 1`;
logConnectionEvent({
@@ -405,9 +362,6 @@ export class ConnectionManager {
});
return pool;
} catch (err) {
// Don't leak the failed pool's sockets/timers (#1641 fallback keeps
// the process running afterward).
if (pool) await endPoolBounded(pool);
logConnectionEvent({
pool: 'ddl',
op: 'error',
+3 -1
View File
@@ -1214,7 +1214,9 @@ async function runPhaseEmbed(engine: BrainEngine, dryRun: boolean, signal?: Abor
// 10-15 min one) bails within a batch instead of running to completion
// after the job was killed — which left gbrain_cycle_locks held and
// wedged every subsequent autopilot cycle.
const result = await runEmbedCore(engine, { stale: true, dryRun, signal });
// #394: quiet — the cycle reports embed counts via its own PhaseResult;
// raw `[dry-run] Would embed ...` stdout lines would corrupt `dream --json`.
const result = await runEmbedCore(engine, { stale: true, dryRun, signal, quiet: true });
const embeddedCount = dryRun ? result.would_embed : result.embedded;
return {
phase: 'embed',
-63
View File
@@ -3,7 +3,6 @@ import {
isSupabasePoolerUrl,
deriveDirectUrl,
readKillSwitchEnv,
isNetworkUnreachableError,
resolveDirectPoolSize,
ConnectionManager,
DEFAULT_DIRECT_POOL_SIZE,
@@ -239,65 +238,3 @@ describe('ConnectionManager — parent inheritance (A2)', () => {
}
});
});
describe('isNetworkUnreachableError (#1641)', () => {
test('classifies network codes as unreachable', () => {
for (const code of ['ENOTFOUND', 'ECONNREFUSED', 'ENETUNREACH', 'EHOSTUNREACH', 'ETIMEDOUT', 'CONNECT_TIMEOUT']) {
const err = Object.assign(new Error('connect failed'), { code });
expect(isNetworkUnreachableError(err)).toBe(true);
}
});
test('classifies by message when code absent', () => {
expect(isNetworkUnreachableError(new Error('getaddrinfo ENOTFOUND db.abc.supabase.co'))).toBe(true);
});
test('auth/SQL errors are NOT unreachable', () => {
expect(isNetworkUnreachableError(new Error('password authentication failed for user "postgres"'))).toBe(false);
expect(isNetworkUnreachableError(new Error('syntax error at or near "SELEC"'))).toBe(false);
expect(isNetworkUnreachableError(null)).toBe(false);
});
});
describe('ConnectionManager — direct-pool fallback on unreachable host (#1641)', () => {
let originalKillSwitch: string | undefined;
let originalError: typeof console.error;
let errLines: string[];
beforeEach(() => {
originalKillSwitch = process.env.GBRAIN_DISABLE_DIRECT_POOL;
delete process.env.GBRAIN_DISABLE_DIRECT_POOL;
originalError = console.error;
errLines = [];
console.error = (...args: unknown[]) => { errLines.push(args.join(' ')); };
});
afterEach(() => {
console.error = originalError;
if (originalKillSwitch === undefined) delete process.env.GBRAIN_DISABLE_DIRECT_POOL;
else process.env.GBRAIN_DISABLE_DIRECT_POOL = originalKillSwitch;
});
test('ddl() falls back to the read pool when the direct host is unreachable', async () => {
const cm = new ConnectionManager({
url: 'postgresql://postgres.abc:p@aws.pooler.supabase.com:6543/db',
// 127.0.0.1:9 (discard) → instant ECONNREFUSED, the IPv4-only-network shape.
directUrl: 'postgresql://postgres:p@127.0.0.1:9/db',
});
const fakeReadPool = {} as ReturnType<typeof ConnectionManager.prototype.read>;
cm.setReadPool(fakeReadPool);
expect(cm.isDualPoolActive()).toBe(true);
const pool = await cm.ddl(); // without the fix this throws ECONNREFUSED
expect(pool).toBe(fakeReadPool);
// Self-activating kill-switch: subsequent calls skip the direct pool.
expect(cm.isKillSwitchActive()).toBe(true);
expect(cm.isDualPoolActive()).toBe(false);
expect(cm.describeMode().mode).toBe('single (kill-switch)');
// One stderr line mentioning the power-user override.
const warning = errLines.filter(l => l.includes('GBRAIN_DIRECT_DATABASE_URL'));
expect(warning.length).toBe(1);
const again = await cm.ddl();
expect(again).toBe(fakeReadPool);
expect(errLines.filter(l => l.includes('GBRAIN_DIRECT_DATABASE_URL')).length).toBe(1);
}, 20000);
});
+27
View File
@@ -292,6 +292,33 @@ describe('runDream — output format', () => {
expect(parsed).toHaveProperty('totals');
});
// #394 / takeover of #854: the embed phase's `[dry-run] Would embed ...`
// summary must not leak onto stdout ahead of the JSON CycleReport.
test('--dry-run --json emits only JSON even when embed has stale chunks', async () => {
await engine.putPage('concepts/testing', {
type: 'concept',
title: 'Testing',
compiled_truth: 'Testing keeps JSON contracts honest.',
timeline: '',
});
await engine.upsertChunks('concepts/testing', [
{ chunk_index: 0, chunk_text: 'Testing keeps JSON contracts honest.', chunk_source: 'compiled_truth' },
]);
const lines: string[] = [];
const logSpy = spyOn(console, 'log').mockImplementation((msg: string) => { lines.push(String(msg)); });
await runDream(engine, ['--dir', repo, '--phase', 'embed', '--dry-run', '--json']);
logSpy.mockRestore();
const output = lines.join('\n');
expect(output.trimStart().startsWith('{')).toBe(true);
const parsed = JSON.parse(output);
expect(parsed.schema_version).toBe('1');
expect(parsed.phases[0].phase).toBe('embed');
// The stale chunk was still counted in the structured report.
expect(parsed.phases[0].details.would_embed).toBe(1);
});
test('human output for clean status mentions "Brain is healthy"', async () => {
const lines: string[] = [];
const logSpy = spyOn(console, 'log').mockImplementation((msg: string) => { lines.push(String(msg)); });