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
7 changed files with 67 additions and 281 deletions
+2 -11
View File
@@ -808,20 +808,12 @@ async function makeContext(engine: BrainEngine, params: Record<string, unknown>)
// 'default'. Wrapped in try/catch so a doctor / single-source brain that
// never set up sources still returns 'default' silently.
let sourceId: string | undefined;
// #2561: when the source resolved via a NON-explicit tier (path-match /
// brain default / sole-non-default / seed default), unqualified search-shaped
// reads span every `config.federated = true` source. Computed here (the
// trusted local boundary) and consumed by federatedSearchScope in
// operations.ts, which additionally gates on ctx.remote === false.
let localFederated: string[] | undefined;
try {
const { resolveSourceWithTier, localFederatedSourceIds } = await import('./core/source-resolver.ts');
const { resolveSourceId } = await import('./core/source-resolver.ts');
// params.source is set when a CLI flag was parsed for the op (rare; most
// CLI ops don't take --source). Falls through to env/dotfile/path-match.
const explicit = (params.source as string | undefined) ?? null;
const resolved = await resolveSourceWithTier(engine, explicit);
sourceId = resolved.source_id;
localFederated = await localFederatedSourceIds(engine, resolved.source_id, resolved.tier);
sourceId = await resolveSourceId(engine, explicit);
} catch {
// Source resolution failed (e.g. sources table doesn't exist on a fresh
// pre-init brain). Leave sourceId unset; engine read methods fall through
@@ -842,7 +834,6 @@ async function makeContext(engine: BrainEngine, params: Record<string, unknown>)
// table). Matches dispatch.ts's auto-fill so the contract holds across
// every transport.
sourceId: sourceId ?? 'default',
...(localFederated ? { localFederatedSourceIds: localFederated } : {}),
};
}
+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
+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',
+2 -61
View File
@@ -424,23 +424,6 @@ export interface OperationContext {
* satisfied even on single-source brains.
*/
sourceId: string;
/**
* #2561 federated read scope for UNQUALIFIED local CLI reads.
*
* Set ONLY by the local CLI's context builder (src/cli.ts makeContext), and
* only when the source resolved via a non-explicit tier (local_path /
* brain_default / sole_non_default / seed_default NOT --source, NOT
* GBRAIN_SOURCE, NOT a .gbrain-source dotfile). Contains the resolved
* source first, then every other `config.federated = true` source, so an
* unqualified `gbrain search "X"` spans federated sources as
* docs/guides/multi-source-brains.md promises.
*
* Consumed exclusively by `federatedSearchScope` and ONLY when
* `ctx.remote === false` a remote caller's scope stays governed by
* `ctx.auth.allowedSources` / scalar `ctx.sourceId` (source-isolation
* invariant, fail-closed).
*/
localFederatedSourceIds?: string[];
}
/**
@@ -556,45 +539,6 @@ export function resolveRequestedScope(
return sourceScopeOpts(ctx);
}
/**
* #2561 source scope for the search-shaped read ops (`search`, `query`).
*
* Delegates to `resolveRequestedScope` (the single trust+grant resolver), then
* widens an UNQUALIFIED trusted-local scalar scope to the CLI-computed
* federated set (`ctx.localFederatedSourceIds`, resolved source first). This is
* what makes `sources add --federated` mean something for local search: a
* federated source participates in unqualified `gbrain search "X"` results.
*
* The expansion NEVER applies when:
* - the caller is not strictly trusted-local (`ctx.remote !== false`)
* remote scope stays grant-governed (fail-closed source isolation);
* - a per-call `source_id` was passed (explicit wins, including `__all__`);
* - the resolver already produced a federated array (OAuth grant);
* - the CLI resolved the source from an explicit signal (--source / env /
* dotfile) makeContext leaves `localFederatedSourceIds` unset then.
*
* Deliberately NOT inside `sourceScopeOpts`: code-intel ops collapse a
* multi-element scope to an error (`resolveCodeIntelScope`), and non-search
* reads (get_page, get_links, ) keep their long-standing scalar behavior.
*/
export function federatedSearchScope(
ctx: OperationContext,
sourceIdParam?: string,
): { sourceId?: string; sourceIds?: string[] } {
const scope = resolveRequestedScope(ctx, sourceIdParam);
if (
ctx.remote === false &&
sourceIdParam === undefined &&
scope.sourceId !== undefined &&
scope.sourceIds === undefined &&
ctx.localFederatedSourceIds !== undefined &&
ctx.localFederatedSourceIds.length > 1
) {
return { sourceIds: ctx.localFederatedSourceIds };
}
return scope;
}
/**
* Code-intel adapter for `resolveRequestedScope`. Graph traversal
* (code_callers/code_callees/code_blast/code_flow) is single-source by design
@@ -1504,8 +1448,7 @@ const search: Operation = {
const queryText = p.query as string;
const limit = (p.limit as number) || 20;
const offset = (p.offset as number) || 0;
// #2561: unqualified trusted-local search spans federated sources.
const scope = federatedSearchScope(ctx);
const scope = sourceScopeOpts(ctx);
// T4/D5 — per-call mode honored ONLY for trusted/local callers so a remote
// OAuth client can't escalate to the costly tokenmax bundle. Local + unknown
@@ -1667,9 +1610,7 @@ const query: Operation = {
// is spread into BOTH the image-similarity searchVector path and the text
// hybridSearch path below, so both honor the same grant.
const sourceIdParam = typeof p.source_id === 'string' ? p.source_id : undefined;
// #2561: unqualified trusted-local query spans federated sources (per-call
// source_id / remote grants still resolve through resolveRequestedScope).
const querySourceScope = federatedSearchScope(ctx, sourceIdParam);
const querySourceScope = resolveRequestedScope(ctx, sourceIdParam);
// v0.27.1: image-similarity branch. Bypasses hybridSearch (which is
// text-only); embeds the image via embedMultimodal and runs a direct
-39
View File
@@ -353,45 +353,6 @@ export async function resolveSourceWithTier(
return { source_id: 'default', tier: 'seed_default' };
}
/**
* #2561 compute the federated read scope for an UNQUALIFIED local CLI call.
*
* `sources add --federated` promises that a `config.federated = true` source
* "participates in unqualified `gbrain search` results"
* (docs/guides/multi-source-brains.md). This helper turns that promise into a
* scope: given the resolved source and WHICH tier resolved it, return
* `[resolvedSource, ...other federated source ids]` or `undefined` when the
* expansion must not apply:
*
* - explicit tiers (`flag` / `env` / `dotfile`): the user named a source;
* scalar scope stands (that IS the qualified case);
* - no other federated source exists: keep the scalar fast path unchanged.
*
* Archived sources are excluded (same rationale as pickSoleNonDefaultSource);
* the archived column is v34+, so fall back to the un-archived query on older
* brains. Callers put the result on `OperationContext.localFederatedSourceIds`
* consumed only by `federatedSearchScope` and only when `remote === false`.
*/
export async function localFederatedSourceIds(
engine: BrainEngine,
sourceId: string,
tier: SourceTier,
): Promise<string[] | undefined> {
if (tier === 'flag' || tier === 'env' || tier === 'dotfile') return undefined;
let rows: Array<{ id: string }>;
try {
rows = await engine.executeRaw<{ id: string }>(
`SELECT id FROM sources WHERE config->>'federated' = 'true' AND archived = false ORDER BY id`,
);
} catch {
rows = await engine.executeRaw<{ id: string }>(
`SELECT id FROM sources WHERE config->>'federated' = 'true' ORDER BY id`,
);
}
const ids = [sourceId, ...rows.map((r) => r.id).filter((id) => id !== sourceId)];
return ids.length > 1 ? ids : undefined;
}
/** Exposed for tests. */
export const __testing = {
readDotfileWalk,
+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)); });
-154
View File
@@ -1,154 +0,0 @@
/**
* #2561 sources.config.federated participates in UNQUALIFIED local CLI
* search/query.
*
* Pre-fix: the local CLI always emitted a scalar `{sourceId}` scope (required
* field, auto-filled 'default'), so a source registered with
* `gbrain sources add --federated` was invisible to an unqualified
* `gbrain search "X"` contradicting docs/guides/multi-source-brains.md
* ("Source participates in unqualified `gbrain search` results").
*
* Fix: the CLI context builder computes `ctx.localFederatedSourceIds`
* (resolved source + every other federated source) whenever the source
* resolved via a NON-explicit tier; `federatedSearchScope` widens the scalar
* scope to that set for the `search` / `query` ops trusted-local only
* (`ctx.remote === false`), never for remote callers, never when a per-call
* `source_id` or an explicit --source/env/dotfile was given.
*/
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
import { localFederatedSourceIds } from '../src/core/source-resolver.ts';
import {
federatedSearchScope,
operations,
type OperationContext,
} from '../src/core/operations.ts';
let engine: PGLiteEngine;
const search = operations.find((o) => o.name === 'search')!;
function ctxOf(overrides: Partial<OperationContext> = {}): OperationContext {
return {
engine: engine as any,
config: {} as any,
logger: console as any,
dryRun: false,
remote: false,
sourceId: 'default',
...overrides,
};
}
beforeAll(async () => {
engine = new PGLiteEngine();
await engine.connect({});
await engine.initSchema();
// Seeded 'default' source is federated=true. Add:
// wiki — federated (must join unqualified search)
// private — NOT federated (must stay invisible unless explicitly named)
// oldnews — federated but archived (must stay excluded)
await engine.executeRaw(
`INSERT INTO sources (id, name, local_path, config) VALUES ('wiki', 'wiki', '/tmp/wiki', '{"federated": true}'::jsonb)`,
);
await engine.executeRaw(
`INSERT INTO sources (id, name, local_path, config) VALUES ('private', 'private', '/tmp/private', '{}'::jsonb)`,
);
await engine.executeRaw(
`INSERT INTO sources (id, name, local_path, config, archived) VALUES ('oldnews', 'oldnews', '/tmp/oldnews', '{"federated": true}'::jsonb, true)`,
);
const pages: Array<[slug: string, sourceId: string, where: string]> = [
['notes/home', 'default', 'default'],
['wiki/topic', 'wiki', 'wiki'],
['private/topic', 'private', 'private'],
['old/topic', 'oldnews', 'oldnews'],
];
for (const [slug, sourceId, where] of pages) {
await engine.putPage(slug, {
type: 'note', title: `Topic in ${where}`, compiled_truth: `the zebra telescope in ${where}`, frontmatter: {},
}, { sourceId });
await engine.upsertChunks(slug, [
{ chunk_index: 0, chunk_text: `the zebra telescope in ${where}`, chunk_source: 'compiled_truth' },
], { sourceId });
}
// Keyword-only search path: no embedding provider needed in tests.
await engine.setConfig('search.mcp_keyword_only', 'true');
}, 60_000);
afterAll(async () => {
if (engine) await engine.disconnect();
}, 60_000);
describe('localFederatedSourceIds — CLI-side scope computation', () => {
test('non-explicit tier: resolved source first, then other federated, archived excluded', async () => {
expect(await localFederatedSourceIds(engine, 'default', 'seed_default')).toEqual(['default', 'wiki']);
});
test('non-federated resolved source still joins its own scope', async () => {
expect(await localFederatedSourceIds(engine, 'private', 'brain_default')).toEqual(['private', 'default', 'wiki']);
});
test('explicit tiers (--source / env / dotfile) never expand', async () => {
expect(await localFederatedSourceIds(engine, 'default', 'flag')).toBeUndefined();
expect(await localFederatedSourceIds(engine, 'default', 'env')).toBeUndefined();
expect(await localFederatedSourceIds(engine, 'default', 'dotfile')).toBeUndefined();
});
test('single federated source (the resolved one) keeps the scalar fast path', async () => {
const solo = { executeRaw: async () => [{ id: 'default' }] } as any;
expect(await localFederatedSourceIds(solo, 'default', 'seed_default')).toBeUndefined();
});
});
describe('federatedSearchScope — trust + explicitness matrix', () => {
test('trusted local + unqualified widens to the federated set', () => {
const ctx = ctxOf({ localFederatedSourceIds: ['default', 'wiki'] });
expect(federatedSearchScope(ctx)).toEqual({ sourceIds: ['default', 'wiki'] });
});
test('remote caller NEVER widens (fail-closed), even if the field is set', () => {
const ctx = ctxOf({ remote: true, localFederatedSourceIds: ['default', 'wiki'] });
expect(federatedSearchScope(ctx)).toEqual({ sourceId: 'default' });
});
test('per-call source_id wins over the federated set', () => {
const ctx = ctxOf({ localFederatedSourceIds: ['default', 'wiki'] });
expect(federatedSearchScope(ctx, 'wiki')).toEqual({ sourceId: 'wiki' });
});
test('per-call __all__ keeps the whole-brain semantics for trusted local', () => {
const ctx = ctxOf({ localFederatedSourceIds: ['default', 'wiki'] });
expect(federatedSearchScope(ctx, '__all__')).toEqual({});
});
test('a federated OAuth grant wins over the local set', () => {
const ctx = ctxOf({
localFederatedSourceIds: ['default', 'wiki'],
auth: { allowedSources: ['a', 'b'] } as OperationContext['auth'],
});
expect(federatedSearchScope(ctx)).toEqual({ sourceIds: ['a', 'b'] });
});
test('no local federated set → unchanged scalar scope', () => {
expect(federatedSearchScope(ctxOf())).toEqual({ sourceId: 'default' });
});
});
describe('search op — unqualified local search spans federated sources', () => {
test('federated source results appear; non-federated + archived stay invisible', async () => {
const ctx = ctxOf({
localFederatedSourceIds: await localFederatedSourceIds(engine, 'default', 'seed_default'),
});
const results = (await search.handler(ctx, { query: 'zebra telescope' })) as Array<{ slug: string }>;
const slugs = results.map((r) => r.slug);
expect(slugs).toContain('notes/home');
expect(slugs).toContain('wiki/topic'); // pre-#2561 this was missing
expect(slugs).not.toContain('private/topic');
expect(slugs).not.toContain('old/topic');
});
test('explicit source resolution (no federated set on ctx) stays single-source', async () => {
const results = (await search.handler(ctxOf(), { query: 'zebra telescope' })) as Array<{ slug: string }>;
const slugs = results.map((r) => r.slug);
expect(slugs).toEqual(['notes/home']);
});
});