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
9 changed files with 65 additions and 159 deletions
+34 -29
View File
@@ -1,5 +1,5 @@
import type { BrainEngine } from '../core/engine.ts';
import { embedBatch, currentEmbeddingSignature, resolveEmbeddingModelLabel } from '../core/embedding.ts';
import { embedBatch, currentEmbeddingSignature } from '../core/embedding.ts';
import type { ChunkInput } from '../core/types.ts';
import { chunkText } from '../core/chunkers/recursive.ts';
import { createProgress, type ProgressReporter } from '../core/progress.ts';
@@ -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;
}
@@ -581,16 +591,11 @@ async function embedPage(
for (let j = 0; j < toEmbed.length; j++) {
embeddingMap.set(toEmbed[j].chunk_index, embeddings[j]);
}
// #1717: label each (re)embedded chunk with the model that actually
// produced its vector. Preserved chunks (not re-embedded this pass) keep
// their existing model so a mixed-model page isn't relabeled wholesale.
const embedModelLabel = resolveEmbeddingModelLabel();
const updated: ChunkInput[] = chunks.map(c => ({
chunk_index: c.chunk_index,
chunk_text: c.chunk_text,
chunk_source: c.chunk_source,
embedding: embeddingMap.get(c.chunk_index),
model: embeddingMap.has(c.chunk_index) && embedModelLabel ? embedModelLabel : c.model,
token_count: c.token_count || Math.ceil(c.chunk_text.length / 4),
}));
@@ -607,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(
@@ -625,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,
) {
@@ -722,16 +729,12 @@ async function embedAll(
for (let j = 0; j < toEmbed.length; j++) {
embeddingMap.set(toEmbed[j].chunk_index, embeddings[j]);
}
// #1717: stamp the resolved embedding model on (re)embedded chunks;
// preserve the existing model on chunks left untouched.
const embedModelLabel = resolveEmbeddingModelLabel();
// Preserve ALL chunks, only update embeddings for stale ones
const updated: ChunkInput[] = chunks.map(c => ({
chunk_index: c.chunk_index,
chunk_text: c.chunk_text,
chunk_source: c.chunk_source,
embedding: embeddingMap.get(c.chunk_index) ?? undefined,
model: embeddingMap.has(c.chunk_index) && embedModelLabel ? embedModelLabel : c.model,
token_count: c.token_count || Math.ceil(c.chunk_text.length / 4),
}));
await observed(pacer, () => engine.upsertChunks(page.slug, updated, pageOpts));
@@ -772,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`);
}
}
}
@@ -811,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,
@@ -828,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`);
}
}
@@ -839,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;
}
@@ -851,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;
}
@@ -1021,15 +1030,11 @@ async function embedAllStale(
for (let j = 0; j < stale.length; j++) {
staleIdxToEmbedding.set(stale[j].chunk_index, embeddings[j]);
}
// #1717: label the re-embedded (stale) chunks with the resolved
// model; preserve the existing model on the non-stale chunks.
const embedModelLabel = resolveEmbeddingModelLabel();
const merged: ChunkInput[] = existing.map(c => ({
chunk_index: c.chunk_index,
chunk_text: c.chunk_text,
chunk_source: c.chunk_source,
embedding: staleIdxToEmbedding.get(c.chunk_index) ?? undefined,
model: staleIdxToEmbedding.has(c.chunk_index) && embedModelLabel ? embedModelLabel : c.model,
token_count: c.token_count || Math.ceil(c.chunk_text.length / 4),
}));
await observed(pacer, () => engine.upsertChunks(slug, merged, { sourceId: keySourceId }));
@@ -1095,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',
-7
View File
@@ -20,7 +20,6 @@
import type { BrainEngine } from './engine.ts';
import type { ChunkInput } from './types.ts';
import { embedBatchWithBackoff } from '../commands/embed.ts';
import { resolveEmbeddingModelLabel } from './embedding.ts';
import { type DbPacer, createNoopPacer, observed } from './db-pacer.ts';
import { AbortError } from './abort-check.ts';
@@ -201,17 +200,11 @@ export async function embedStaleForSource(
for (let j = 0; j < stale.length; j++) {
staleIdxToEmbedding.set(stale[j].chunk_index, embeddings[j]);
}
// #1717: label re-embedded chunks with the model that produced the
// vector; preserved chunks keep their existing model. Without this,
// upsertChunks falls back to DEFAULT_EMBEDDING_MODEL for every chunk
// (the same mislabel the embed.ts paths fixed).
const embedModelLabel = resolveEmbeddingModelLabel();
const merged: ChunkInput[] = existing.map((c) => ({
chunk_index: c.chunk_index,
chunk_text: c.chunk_text,
chunk_source: c.chunk_source,
embedding: staleIdxToEmbedding.get(c.chunk_index) ?? undefined,
model: staleIdxToEmbedding.has(c.chunk_index) && embedModelLabel ? embedModelLabel : c.model,
token_count: c.token_count || Math.ceil(c.chunk_text.length / 4),
// Carry through per-chunk metadata. upsertChunks writes these as
// EXCLUDED.<col> (not COALESCE), so omitting them here resets image
-15
View File
@@ -113,21 +113,6 @@ export async function embedBatch(
return results;
}
/**
* Resolve the embedding model label (`provider:model`) to stamp onto
* `content_chunks.model`, so each chunk records the model that actually
* produced its vector instead of the engine's hardcoded default (#1717).
* Returns undefined if the gateway is unconfigured; callers then fall back
* to the chunk's existing model rather than mislabeling it.
*/
export function resolveEmbeddingModelLabel(): string | undefined {
try {
return gatewayGetModel();
} catch {
return undefined;
}
}
/** Currently-configured embedding model (short form without provider prefix). */
export function getEmbeddingModelName(): string {
return gatewayGetModel().split(':').slice(1).join(':') || 'text-embedding-3-large';
+1 -11
View File
@@ -8,7 +8,7 @@ import { chunkText } from './chunkers/recursive.ts';
import { chunkCodeText, chunkCodeTextFull, detectCodeLanguage, CHUNKER_VERSION } from './chunkers/code.ts';
import { findChunkForOffset } from './chunkers/edge-extractor.ts';
import { extractCodeRefs, imageOfCandidates } from './link-extraction.ts';
import { embedBatch, embedMultimodal, currentEmbeddingSignature, resolveEmbeddingModelLabel } from './embedding.ts';
import { embedBatch, embedMultimodal, currentEmbeddingSignature } from './embedding.ts';
import { slugifyPath, slugifyCodePath, isCodeFilePath } from './sync.ts';
import type { ChunkInput, PageInput, PageType } from './types.ts';
import { computeEffectiveDate } from './effective-date.ts';
@@ -716,12 +716,8 @@ export async function importFromContent(
? chunks.map((c) => wrapChunkForEmbedding(c.chunk_text, prefix, c.chunk_source))
: chunks.map((c) => c.chunk_text);
const embeddings = await embedBatch(wrappedTexts);
// #1717: label each chunk with the model that actually produced its
// vector, not the engine's hardcoded default.
const embedModelLabel = resolveEmbeddingModelLabel();
for (let i = 0; i < chunks.length; i++) {
chunks[i].embedding = embeddings[i];
if (embedModelLabel) chunks[i].model = embedModelLabel;
// token_count tracks the wrapped string length so cost reporting
// reflects what we actually sent to the embedder.
chunks[i].token_count = Math.ceil(wrappedTexts[i].length / 4);
@@ -1145,10 +1141,7 @@ export async function importCodeFile(
const matched = existingByKey.get(key);
if (matched && matched.embedding) {
// Reuse the existing embedding verbatim. No API call, no cost.
// #1717: carry the existing model label along with the reused vector
// so the upsert doesn't relabel it with the engine default.
chunks[i]!.embedding = matched.embedding as Float32Array;
chunks[i]!.model = matched.model ?? undefined;
chunks[i]!.token_count = matched.token_count ?? undefined;
} else {
needsEmbedIndexes.push(i);
@@ -1160,12 +1153,9 @@ export async function importCodeFile(
try {
const textsToEmbed = needsEmbedIndexes.map((i) => chunks[i]!.chunk_text);
const embeddings = await embedBatch(textsToEmbed);
// #1717: stamp the model that produced these vectors.
const embedModelLabel = resolveEmbeddingModelLabel();
for (let j = 0; j < needsEmbedIndexes.length; j++) {
const i = needsEmbedIndexes[j]!;
chunks[i]!.embedding = embeddings[j]!;
if (embedModelLabel) chunks[i]!.model = embedModelLabel;
chunks[i]!.token_count = Math.ceil(chunks[i]!.chunk_text.length / 4);
}
} catch (e: unknown) {
+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)); });
-46
View File
@@ -15,7 +15,6 @@ import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:tes
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
import { resetPgliteState } from './helpers/reset-pglite.ts';
import { embedStaleForSource } from '../src/core/embed-stale.ts';
import { configureGateway, resetGateway } from '../src/core/ai/gateway.ts';
import type { ChunkInput } from '../src/core/types.ts';
let engine: PGLiteEngine;
@@ -277,49 +276,4 @@ describe('embedStaleForSource', () => {
// The stale text row actually got its embedding.
expect(txtRow.embedded_at).not.toBeNull();
});
// #1717: the backfill path must label re-embedded chunks with the model
// that produced the vector, and preserve the existing label on chunks it
// did not touch (before the fix, both were reset to the engine default).
test('labels re-embedded chunks with the gateway model, preserves untouched labels (#1717)', async () => {
configureGateway({
embedding_model: 'openai:text-embedding-3-large',
env: { OPENAI_API_KEY: 'sk-test-embed-stale-1717' },
});
try {
await engine.putPage('notes/model-label', {
type: 'note',
title: 'model-label',
compiled_truth: '# model-label\n\nseeded',
});
await engine.upsertChunks('notes/model-label', [
{
chunk_index: 0,
chunk_text: 'already embedded elsewhere',
chunk_source: 'compiled_truth',
embedding: new Float32Array(1536).fill(0.01),
model: 'voyage:voyage-3',
token_count: 4,
},
{
chunk_index: 1,
chunk_text: 'stale chunk needing embed',
chunk_source: 'compiled_truth',
token_count: 5,
embedding: undefined, // stale
},
]);
const result = await embedStaleForSource(engine, 'default', { embedFn: fakeEmbedFn });
expect(result.embedded).toBe(1);
const after = await engine.getChunks('notes/model-label');
const preserved = after.find((c) => c.chunk_index === 0)!;
const reembedded = after.find((c) => c.chunk_index === 1)!;
expect(reembedded.model).toBe('openai:text-embedding-3-large');
expect(preserved.model).toBe('voyage:voyage-3');
} finally {
resetGateway();
}
});
});
-33
View File
@@ -37,8 +37,6 @@ mock.module('../src/core/embedding.ts', () => ({
// setPageEmbeddingSignature / invalidateStaleSignatureEmbeddings resolve to
// null via the Proxy default, so the signature value is inert here.
currentEmbeddingSignature: () => 'test:model:1536',
// #1717: embed paths stamp this label on (re)embedded chunks.
resolveEmbeddingModelLabel: () => 'openai:text-embedding-3-large',
}));
// Import AFTER mocking.
@@ -805,34 +803,3 @@ describe('embedAllStale --source threading (D7)', () => {
expect((firstCallOpts as { sourceId?: string }).sourceId).toBe('media-corpus');
});
});
// #1717: content_chunks.model must record the model that actually produced
// each vector, not the gateway/engine default.
describe('content_chunks.model labeling (#1717)', () => {
test('stamps the resolved embedding model on re-embedded chunks, preserves it on untouched chunks', async () => {
let upserted: any[] | undefined;
// Chunk 0 is stale (no embedded_at) → gets re-embedded this pass.
// Chunk 1 is already embedded with a DIFFERENT model → must be preserved,
// not relabeled to the current model.
const chunks = [
{ chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', embedded_at: null, model: 'zeroentropyai:zembed-1', token_count: 1 },
{ chunk_index: 1, chunk_text: 'b', chunk_source: 'compiled_truth', embedded_at: '2026-01-01', embedding: new Float32Array(1536), model: 'voyage:voyage-3', token_count: 1 },
];
const engine = mockEngine({
getPage: async () => ({ slug: 'notes/x', compiled_truth: 'a', timeline: '', source_id: 'default' }),
getChunks: async () => chunks,
upsertChunks: async (_slug: string, c: any[]) => { upserted = c; },
setPageEmbeddingSignature: async () => null,
});
await runEmbedCore(engine, { slugs: ['notes/x'] });
expect(upserted).toBeDefined();
const byIdx = Object.fromEntries(upserted!.map(c => [c.chunk_index, c]));
// Re-embedded chunk carries the model that produced its vector (was
// mislabeled with the default before the fix).
expect(byIdx[0].model).toBe('openai:text-embedding-3-large');
// Untouched chunk keeps its original model — no wholesale relabel.
expect(byIdx[1].model).toBe('voyage:voyage-3');
});
});
@@ -73,21 +73,4 @@ describe('importFromContent embedding_signature stamping (F1)', () => {
await importFromContent(engine, 'concepts/unstamped', '# Unstamped\n\nbody content.', { noEmbed: true });
expect(await signatureOf('concepts/unstamped')).toBeNull();
});
// #1717: content_chunks.model must record the model that produced the
// vector (the configured gateway model), not the engine's hardcoded
// default. The gateway here is configured to openai:text-embedding-3-large,
// which differs from DEFAULT_EMBEDDING_MODEL — so this fails without the
// import-path model stamping.
test('inline embed labels content_chunks.model with the configured model (#1717)', async () => {
await importFromContent(engine, 'concepts/labeled', '# Labeled\n\nsome body content to chunk and embed.', {});
const rows = await engine.executeRaw<{ model: string }>(
`SELECT cc.model FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE p.slug = $1 AND p.source_id = 'default'`,
['concepts/labeled'],
);
expect(rows.length).toBeGreaterThan(0);
for (const r of rows) expect(r.model).toBe('openai:text-embedding-3-large');
});
});