From d6f929bfbd603ea975daf20031db25bf9337d34e Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Sat, 1 Aug 2026 05:40:14 +0800 Subject: [PATCH] fix(embed): stop one bad chunk darkening a page, and exit non-zero on failures (#3037) (#3532) Co-Authored-By: Garry Tan --- docs/architecture/KEY_FILES.md | 2 +- src/cli.ts | 9 +- src/commands/embed.ts | 253 ++++++++++++++--- test/embed-exit-code-3037.serial.test.ts | 132 +++++++++ .../embed-partial-failure-3037.serial.test.ts | 266 ++++++++++++++++++ 5 files changed, 624 insertions(+), 38 deletions(-) create mode 100644 test/embed-exit-code-3037.serial.test.ts create mode 100644 test/embed-partial-failure-3037.serial.test.ts diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 6a465c00e..d7cffd6cb 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -192,7 +192,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/extraction-review.ts` — Extraction quarantine lane markers (issue #160), sibling of `src/core/quarantine.ts` / `embed-skip.ts` (frontmatter-key pattern, no schema migration). Auto-extracted stubs from untrusted input carry the PAIR `provenance: 'auto-extracted'` + `status: 'unverified'` (both required — user pages with their own `status`/`provenance` never match). Exports `quarantineMarkers()`, `isUnverifiedExtraction()` (JS predicate) and `unverifiedExtractionFragment(alias)` — the single SQL source of truth consumed by `buildSourceFactorCase` (namespace source-boost guard), both engines' `getUnverifiedExtractionPageIds`, the `extraction_pending` op, and the `unverified_extractions` doctor check, so filter and marker keys can never drift. Consequences: unverified stubs are excluded from the compiled-truth fusion boost + the `people/`/`companies/` source-boost (rank as ordinary content), stamped `unverified: true` in search results (`stampUnverifiedExtractions`, hybrid.ts), listed by `extraction_pending`, promoted (status → `verified`, provenance kept for audit) or rejected (soft-delete) by the owner-only `extraction_review` op. Pinned by `test/extraction-review.test.ts` (PGLite) + `test/e2e/extraction-review-postgres.test.ts` (live Postgres parity). - `src/commands/enrich.ts` + `src/core/enrich/thin.ts` + `src/core/cycle/enrich-thin.ts` — `gbrain enrich --thin`: batch-develops stub (thin) pages via **brain-internal grounded synthesis**. gbrain's model tooling sees only brain-internal context (search / get_page / facts / backlinks), not the web, so enrich consolidates what the brain ALREADY knows about an entity (scattered across meetings, other pages, deals, facts) into one cited page via ONE `gateway.chat` call per page; web research stays the agent-driven `enrich` SKILL's job. `runEnrichCore(engine, opts, signal)` (strict per-source; multi-source iteration is the caller's job) drives `enrichOne` per candidate: `withRefreshingLock('enrich::')` → `getPage` → deterministic retrieve (hybridSearch + getBacklinks + facts + raw_data, source-scoped, sanitized via `INJECTION_PATTERNS`) → `assessGrounding` gate (skip < `MIN_CONTEXT_CHARS`, no LLM) → `buildEnrichPrompt` (grounded dossier, `[Source: slug]` citations, SKIP sentinel) → synth → `put_page` handler (`remote:false`, auto-link + write-through) stamping `enriched_at` + `enriched_by:'cli:enrich'`. Candidate selection is the SQL-native `engine.listEnrichCandidates(opts)` (`src/core/engine.ts` interface + `EnrichCandidate`/`EnrichCandidatesOpts`/`ENRICH_ORDER_SQL` in `src/core/types.ts` + pg/pglite impls): thin-filter + per-page source-correct inbound count (`to_page_id = p.id`, `mentions` excluded) + `enriched_at` recency guard + whitelisted ORDER BY + LIMIT, lightweight projection (NO bodies). Resume via `src/core/op-checkpoint.ts` (local `enrichFingerprint`); budget via `BudgetTracker` + `withBudgetTracker` (best-effort under `--workers > 1` — `runSlidingPool` aborts new claims on `BUDGET_EXHAUSTED` but does NOT cancel in-flight `gateway.chat`; pin `--workers 1` for a hard ceiling). `sanitizeContext` (thin.ts) neutralizes the `…` data-envelope delimiters (injection escape, mirrors the `` convention); the `--background` multi-source fan-out idempotency key carries the run fingerprint via exported `backgroundIdempotencyKey(sid, args)` (a bare `enrich:${sid}` would return stale completed jobs); `runEnrichCore` flags `budget_exhausted` post-hoc when `tracker.totalSpent > tracker.cap` even when the gateway swallowed the final-call throw (via read-only `BudgetTracker.cap` getter); `body()` flushes the checkpoint on `BudgetExhausted` before it propagates so resume doesn't re-charge. The opt-in `enrich_thin` cycle phase (default OFF via `cycle.enrich_thin.enabled`) trickles `max_pages_per_tick` (default 3) per source with per-source cost cap enforced as `min(per_source_cap, brain_wide_remaining)` + brain-wide total + walltime caps. Wired into `cycle.ts` (`CyclePhase`/`ALL_PHASES` between `conversation_facts_backfill` and `skillopt`/`embed`; `PHASE_SCOPE='source'`; `NEEDS_LOCK`; dispatch), `cli.ts` (`CLI_ONLY` + `CLI_ONLY_SELF_HELP` + `THIN_CLIENT_REFUSED_COMMANDS` + dispatch), `jobs.ts` (Minion `enrich` handler, strict per-source, NOT in `PROTECTED_JOB_NAMES`). DI seam `opts.synthesizeFn` keeps tests hermetic (no API key, no mock.module). Pinned by `test/enrich/thin.test.ts`, `test/enrich/idempotency.test.ts`, `test/enrich-cycle-phase.test.ts`, `test/e2e/enrich-pglite.test.ts` (grew-cited, skip, ordering, multi-source, recency, resume, budget abort + checkpoint flush, final-call overage, lock-skip, provenance), `test/e2e/engine-parity.test.ts` (`listEnrichCandidates` pg↔pglite parity). - `src/core/data-research.ts` — Recipe validation, field extraction (MRR/ARR regex), dedup, tracker parsing, HTML stripping. -- `src/commands/embed.ts` — `gbrain embed [--stale|--all] [--slugs ...]`. `--stale` starts with `engine.countStaleChunks()` (single SELECT count(*) WHERE embedding IS NULL, ~50 bytes wire) so a fully-embedded brain short-circuits with no further reads. When stale chunks exist, `engine.listStaleChunks()` returns just the chunks needing embeddings (slug + chunk_index + chunk_text + metadata, no `vector(1536)` payload); caller groups by slug, embeds, re-upserts via `upsertChunks`. All `console.log`/`console.error` call sites use `slog`/`serr` from `src/core/console-prefix.ts` so when `runEmbedCore` runs inside a per-source `withSourcePrefix` scope (installed by the `gbrain sync --all` worker pool) every line carries the `[] ` prefix; standalone callers see identical output because slog/serr fall through to bare console fns outside the wrap. Every embed-write path stamps `pages.embedding_signature` via `engine.setPageEmbeddingSignature(slug, {sourceId, signature: currentEmbeddingSignature()})` so a later model/dims swap is detectable as stale. The per-slug path (`embedPage`, used by `gbrain embed ` AND sync's post-import embed step) and the full-re-embed path (`embedAll`) stamp unconditionally per page. The stale path (`embedAllStale`) first calls `invalidateStaleSignatureEmbeddings` on a live run so signature-drifted pages flow through the NULL cursor, then stamps each page — but ONLY when EVERY chunk was stale this pass (a partially-stale page keeps preserved chunks of unknown provenance, so it stays unstamped rather than falsely marked current; `embed --all` fully re-embeds + stamps those). dry-run never mutates: it counts signature-drift via the widened `countStaleChunks({signature})` predicate without NULLing anything. `--include-null-signature` (#3391) lifts the NULL-signature grandfather clause: threads `includeNullSignature: true` into the invalidation + counts so pages that predate the v108 stamp re-embed too after a model swap (both engines' `countStaleChunks`/`sumStaleChunkChars`/`invalidateStaleSignatureEmbeddings` accept the flag; predicate becomes `sig IS NULL OR sig <> current`). Without the flag, a live stale run that just invalidated drifted rows probes for left-behind NULL-signature chunks and emits a loud stderr warning naming the count + the fix — mixed embedding spaces in one index are never silent. Pinned by `test/embedding-migration.test.ts` + `test/e2e/migrate-embeddings-postgres.test.ts`. +- `src/commands/embed.ts` — `gbrain embed [--stale|--all] [--slugs ...]`. `--stale` starts with `engine.countStaleChunks()` (single SELECT count(*) WHERE embedding IS NULL, ~50 bytes wire) so a fully-embedded brain short-circuits with no further reads. When stale chunks exist, `engine.listStaleChunks()` returns just the chunks needing embeddings (slug + chunk_index + chunk_text + metadata, no `vector(1536)` payload); caller groups by slug, embeds, re-upserts via `upsertChunks`. All `console.log`/`console.error` call sites use `slog`/`serr` from `src/core/console-prefix.ts` so when `runEmbedCore` runs inside a per-source `withSourcePrefix` scope (installed by the `gbrain sync --all` worker pool) every line carries the `[] ` prefix; standalone callers see identical output because slog/serr fall through to bare console fns outside the wrap. Every embed-write path stamps `pages.embedding_signature` via `engine.setPageEmbeddingSignature(slug, {sourceId, signature: currentEmbeddingSignature()})` so a later model/dims swap is detectable as stale. The per-slug path (`embedPage`, used by `gbrain embed ` AND sync's post-import embed step) and the full-re-embed path (`embedAll`) stamp per page when every chunk embedded cleanly. The stale path (`embedAllStale`) first calls `invalidateStaleSignatureEmbeddings` on a live run so signature-drifted pages flow through the NULL cursor, then stamps each page — but ONLY when EVERY chunk was stale this pass (a partially-stale page keeps preserved chunks of unknown provenance, so it stays unstamped rather than falsely marked current; `embed --all` fully re-embeds + stamps those). dry-run never mutates: it counts signature-drift via the widened `countStaleChunks({signature})` predicate without NULLing anything. `--include-null-signature` (#3391) lifts the NULL-signature grandfather clause: threads `includeNullSignature: true` into the invalidation + counts so pages that predate the v108 stamp re-embed too after a model swap (both engines' `countStaleChunks`/`sumStaleChunkChars`/`invalidateStaleSignatureEmbeddings` accept the flag; predicate becomes `sig IS NULL OR sig <> current`). Without the flag, a live stale run that just invalidated drifted rows probes for left-behind NULL-signature chunks and emits a loud stderr warning naming the count + the fix — mixed embedding spaces in one index are never silent. Pinned by `test/embedding-migration.test.ts` + `test/e2e/migrate-embeddings-postgres.test.ts`. Embed failures are never silent (#3037): all three page paths embed via `embedPageTexts`, which tries the page's chunks in one batch and, on a PERMANENT request-shaped failure (non-429, non-`AITransientError`, non-auth), retries once per chunk so one bad chunk costs one chunk instead of darkening the whole page (failed chunks stay `embedding IS NULL` for the next `--stale` pass; a partially-failed page is never signature-stamped). Rate-limit/outage/auth failures do NOT fan out (cost bounding — `embedBatchWithBackoff` already owns 429 backoff). Failed chunk counts land on `EmbedResult.failures` + capped `failure_samples`, and `src/cli.ts`'s embed case sets a non-zero exit verdict on `failures > 0` (mirror of the `import` errors>0 guard). Pinned by `test/embed-partial-failure-3037.serial.test.ts` + `test/embed-exit-code-3037.serial.test.ts` (real spawned CLI). - `src/core/retrieval-upgrade-planner.ts` — `runSchemaTransition(engine, targetDim)` (exported) is the ONE atomic dimension-transition path, shared by `ze-switch` and `gbrain migrate embeddings`. In a single transaction it rebuilds ALL THREE dim-pinned text-embedding-space columns at `targetDim` — `content_chunks.embedding`, `query_cache.embedding`, `facts.embedding` — preserving each column's declared type (`vector` vs `halfvec`, probed from `information_schema`) and recreating its HNSW index with the matching opclass, gated on `hnswIndexExpected` (above the per-type dim ceiling pgvector refuses the index and exact scans remain the path). query_cache + facts are created at brain-birth width by `migrate.ts` and NO migration ever ALTERs them, so omitting either leaves it silently broken: a narrow `query_cache.embedding` makes every `store()`/`lookup()` fail inside the cache's own error-swallowing (permanent 0% hit rate), and a narrow `facts.embedding` fails every per-fact embed write (the doctor check that would warn is skipped on PGLite, the default engine). `content_chunks.embedding_image` / `embedding_multimodal` are the deliberate exception — separate multimodal models, dimensions independent of the text model. Pinned by `test/embedding-migration.test.ts` (all three widths + a real INSERT at the new width into each) and `test/e2e/migrate-embeddings-postgres.test.ts`. - `src/core/embedding-migration.ts` — provider-agnostic embedding migration core (#3390): `planEmbeddingMigration` (workload counts via the widened stale predicates with the TARGET signature + `includeNullSignature`, so a mid-migration re-plan counts only what remains; cost via `embedding-pricing.ts`; `null_signature_chunks` split out for #3391 visibility; reranker-on-outgoing-provider warning), `applyEmbeddingMigration` (env-override gate BEFORE any mutation → in-flight state marker `embedding_migration.state` → `runSchemaTransition` when the ACTUAL column width differs from target → DB-plane `embedding_model`/`embedding_dimensions` → `persistConfig` callback for the file plane → `invalidateStaleSignatureEmbeddings({includeNullSignature: true})` → `SemanticQueryCache.clear()`), `completeEmbeddingMigration` (clears the marker + stamps `embedding_migration.completed`; call only at zero backlog), `resolveMigrationTarget` (validates `provider:model` via `resolveRecipe`, dims via `embeddingDimsForModel` or explicit `--dim`), `migrationSignature` (matches `currentEmbeddingSignature()` shape). Engine-pure; every step idempotent under crash + re-run — the NULL-embedding column is the checkpoint. Reuses `runSchemaTransition` (now exported from `retrieval-upgrade-planner.ts`) so ze-switch and the migration share ONE dimension-transition path. `reconcilePageSignatures(engine, plan)` runs after the re-embed drain and BEFORE the completion probe: it stamps the target signature on every page that has zero NULL-embedding chunks, covering pages whose chunks straddle a `listStaleChunks` batch boundary (the embed loop only stamps when `stale.length === existing.length`, so a split page is embedded correctly but never stamped — without the reconcile a >1-batch brain reports "incomplete" and the re-run re-invalidates and re-pays for those pages). Sound only because apply() invalidated everything not already in the target space; pages with a remaining NULL chunk stay unstamped so a real embed failure still surfaces. Invalidation is ordered BEFORE the config writes so a crash on a same-dim swap leaves rows merely stale (empty results) rather than new-space queries scored against old-space vectors (silently wrong). Pinned by `test/embedding-migration.test.ts` (PGLite) + `test/e2e/migrate-embeddings-postgres.test.ts` (real pgvector). - `src/commands/migrate-embeddings.ts` — `gbrain migrate embeddings --to [--dim N] [--dry-run] [--yes] [--json] [--no-embed] [--pace[=mode]] [--ignore-env-override]` (alias: `gbrain retrieval-upgrade`, the command README/doctor promised since v0.36). Flow: plan → render (stderr when `--json` so stdout stays JSON-clean) → consent gate (TTY y/N prompt or `--yes`; non-TTY without `--yes` refuses exit 2, mirroring the reindex-code cost gate) → live probe (one embed against the TARGET model/dims BEFORE any mutation — bad key/model/dim fails with nothing changed) → `applyEmbeddingMigration` with `persistEmbeddingFileConfig` (writes `~/.gbrain/config.json` + reconfigures the in-process gateway — the gateway reads file/env, NOT the DB plane) → `runEmbedCore({stale, catchUp, singleFlight, includeNullSignature, pace})` → drain check → `completeEmbeddingMigration` or exit 1 with the resume hint (re-run the same command). Also surfaced as the `migrate_embeddings` op (scope admin, localOnly, hidden cliHints; handler hard-refuses `ctx.remote !== false` and returns `needs_confirmation` + plan without `yes: true`). Pinned by `test/migrate-embeddings-flow.serial.test.ts` (full lifecycle incl. interrupted-run resume on PGLite). diff --git a/src/cli.ts b/src/cli.ts index ba62f4147..502cb0d05 100755 --- a/src/cli.ts +++ b/src/cli.ts @@ -1832,7 +1832,14 @@ async function handleCliOnly(command: string, args: string[]) { } case 'embed': { const { runEmbed } = await import('./commands/embed.ts'); - await runEmbed(engine, args); + // #3037: mirror the `import` case above — the CLI was discarding the + // result, so a run where every chunk failed to embed still exited 0 + // and cron/CI/health gates read total silence as success. Surface + // non-zero on failures > 0. (undefined = backgrounded via --background.) + const embedResult = await runEmbed(engine, args); + if (embedResult && embedResult.failures > 0) { + setCliExitVerdict(1); + } break; } case 'serve': { diff --git a/src/commands/embed.ts b/src/commands/embed.ts index 5760155f4..cb017aa43 100644 --- a/src/commands/embed.ts +++ b/src/commands/embed.ts @@ -19,10 +19,26 @@ import { } from '../core/pace-mode.ts'; import { tryAcquireDbLock, type DbLockHandle } from '../core/db-lock.ts'; import { embedBackfillLockId } from '../core/embed-backfill-lock.ts'; +import { AITransientError } from '../core/ai/errors.ts'; import { wrapChunkTextsForStoredMode } from '../core/embedding-context.ts'; import { titleTierCorpusGeneration } from '../core/contextual-retrieval-service.ts'; import type { Page } from '../core/types.ts'; +/** #3037: cap failure samples so a corpus-wide outage doesn't bloat --json. */ +const FAILURE_SAMPLE_CAP = 10; + +/** + * #3037: record embed failures on the run result. `chunkCount` is the number + * of chunks left un-embedded by this failure (1 for page-level errors where + * the chunk count isn't known at the catch site). + */ +function recordFailure(result: EmbedResult, chunkCount: number, slug: string, e: unknown): void { + result.failures += chunkCount; + if (result.failure_samples.length < FAILURE_SAMPLE_CAP) { + result.failure_samples.push(`${slug}: ${e instanceof Error ? e.message : String(e)}`); + } +} + /** * #3507 — after a plain re-embed fully re-embedded a `per_chunk_synopsis` * page at the title-only tier (see wrapChunkTextsForStoredMode), restamp the @@ -166,6 +182,20 @@ export interface EmbedResult { total_chunks: number; /** Number of pages processed (whether or not they had stale chunks). */ pages_processed: number; + /** + * #3037: chunks that FAILED to embed this run (batch failures + per-chunk + * isolation failures). Callers must not read total silence as success: + * `src/cli.ts` turns `failures > 0` into a non-zero exit verdict (mirrors + * the `import` errors>0 guard), and structured consumers (--json, minion + * handlers) can surface it. 0 on a clean run. Additive field. + */ + failures: number; + /** + * #3037: up to 10 `slug: error-message` samples of what failed, so the + * operator gets a diagnosis without scrolling stderr. Capped so a + * corpus-wide outage doesn't bloat structured output. Additive field. + */ + failure_samples: string[]; /** True if this run was a dry-run. */ dryRun: boolean; /** @@ -284,6 +314,8 @@ export async function runEmbedCore(engine: BrainEngine, opts: EmbedOpts): Promis would_embed: 0, total_chunks: 0, pages_processed: 0, + failures: 0, + failure_samples: [], dryRun: !!opts.dryRun, }; @@ -293,6 +325,11 @@ export async function runEmbedCore(engine: BrainEngine, opts: EmbedOpts): Promis try { await embedPage(engine, s, !!opts.dryRun, result, opts.sourceId, opts.signal, opts.quiet); } catch (e: unknown) { + if (isAborted(opts.signal)) break; // shutdown, not a failure + // #3037: a page-level error (not found, DB write) must not exit 0. + // Chunk-level embed failures are counted inside embedPage; this + // counts the page itself (chunk count unknown at this site). + recordFailure(result, 1, s, e); serr(` Error embedding ${s}: ${e instanceof Error ? e.message : e}`); } } @@ -535,6 +572,12 @@ export async function runEmbed(engine: BrainEngine, args: string[]): Promise0 into a + // non-zero exit verdict. + if (result.failures > 0) { + serr(`[embed] ${result.failures} chunk(s) failed to embed. First error: ${result.failure_samples[0] ?? 'unknown'}`); + } return result; } catch (e) { if (progressStarted) progress.finish(); @@ -623,10 +666,32 @@ async function embedPage( // contextual prefix when the page was embedded wrapped), not raw // chunk_text — otherwise a re-embed silently strips the contextual // prefixes the sync path applied. fenced_code chunks stay unwrapped. - const embeddings = await embedBatch(wrapChunkTextsForStoredMode(page, toEmbed), { abortSignal: signal }); + // #3037: per-chunk failure isolation — one bad chunk must not leave the + // page's sibling chunks NULL. The wrapped texts (computed once) feed the + // fan-out too, so an isolation retry never strips the prefixes. Total + // embed failure is recorded here (where the chunk count is known) and + // swallowed: the page stays NULL exactly as before, but the run now + // reports it (result.failures → non-zero exit) instead of pretending + // success. Abort (shutdown) still propagates. + let embeddings: (Float32Array | null)[]; + let failed = 0; + let firstError: unknown; + try { + ({ embeddings, failed, firstError } = await embedPageTexts( + wrapChunkTextsForStoredMode(page, toEmbed), + signal ? { abortSignal: signal } : {}, + )); + } catch (e: unknown) { + if (isAborted(signal)) throw e; + recordFailure(result, toEmbed.length, slug, e); + result.pages_processed++; + serr(` Error embedding ${slug}: ${e instanceof Error ? e.message : e}`); + return; + } const embeddingMap = new Map(); for (let j = 0; j < toEmbed.length; j++) { - embeddingMap.set(toEmbed[j].chunk_index, embeddings[j]); + const emb = embeddings[j]; + if (emb) embeddingMap.set(toEmbed[j].chunk_index, emb); } const updated: ChunkInput[] = chunks.map(c => preserveCodeMetadata(c, { chunk_index: c.chunk_index, @@ -643,16 +708,21 @@ async function embedPage( // Guard: only stamp when EVERY chunk was (re)embedded this pass. If some // chunks were preserved from a prior embed (unknown/old provenance), the // page is mixed — don't claim it's current. `embed --all` fully re-embeds - // such a page and then stamps it. - if (toEmbed.length === chunks.length) { + // such a page and then stamps it. #3037: a partial failure leaves failed + // chunks NULL, so don't stamp then either. + if (failed === 0 && toEmbed.length === chunks.length) { await engine.setPageEmbeddingSignature(slug, { sourceId, signature: currentEmbeddingSignature() }); // #3507: a fully re-embedded per_chunk_synopsis page landed at the // title tier — keep the stamped mode honest. await restampIfDemotedToTitleTier(engine, page, slug, page.source_id); } - result.embedded += toEmbed.length; + result.embedded += toEmbed.length - failed; + if (failed > 0) { + recordFailure(result, failed, slug, firstError); + serr(` ${slug}: ${failed} chunk(s) failed to embed; embedded the other ${toEmbed.length - failed}`); + } result.pages_processed++; - if (!quiet) slog(`${slug}: embedded ${toEmbed.length} chunks`); + if (!quiet) slog(`${slug}: embedded ${toEmbed.length - failed} chunks`); } /** @@ -791,11 +861,18 @@ async function embedAll( try { // #3507: reproduce the page's stored wrapping convention (see embedPage). - const embeddings = await embedBatch(wrapChunkTextsForStoredMode(page, toEmbed)); + // #3037: per-chunk failure isolation — one bad chunk costs one chunk, + // not the whole page's siblings. The wrapped texts feed the fan-out + // too, so an isolation retry never strips the contextual prefixes. + const { embeddings, failed, firstError } = await embedPageTexts( + wrapChunkTextsForStoredMode(page, toEmbed), + signal ? { abortSignal: signal } : {}, + ); // Build a map of new embeddings by chunk_index const embeddingMap = new Map(); for (let j = 0; j < toEmbed.length; j++) { - embeddingMap.set(toEmbed[j].chunk_index, embeddings[j]); + const emb = embeddings[j]; + if (emb) embeddingMap.set(toEmbed[j].chunk_index, emb); } // Preserve ALL chunks, only update embeddings for stale ones. // preserveCodeMetadata threads code-chunk metadata (#769) so re-embed @@ -809,17 +886,30 @@ async function embedAll( })); await observed(pacer, () => engine.upsertChunks(page.slug, updated, pageOpts)); // v0.41.31: stamp embedding provenance so a later model swap is - // detectable as stale. - await observed(pacer, () => - engine.setPageEmbeddingSignature(page.slug, { sourceId: pageSourceId, signature }), - ); - // #3507: --all fully re-embeds; a per_chunk_synopsis page landed at - // the title tier — keep the stamped mode honest. - await observed(pacer, () => - restampIfDemotedToTitleTier(engine, page, page.slug, pageSourceId), - ); - result.embedded += toEmbed.length; + // detectable as stale. #3037: not on partial failure — failed chunks + // stay NULL under unknown provenance. + if (failed === 0) { + await observed(pacer, () => + engine.setPageEmbeddingSignature(page.slug, { sourceId: pageSourceId, signature }), + ); + // #3507: --all fully re-embeds; a per_chunk_synopsis page landed at + // the title tier — keep the stamped mode honest. #3037: gated on + // failed === 0 — a partially-failed page was NOT fully re-embedded, + // so restamping would make contextual_retrieval_mode lie again + // (the exact #3461 bug). + await observed(pacer, () => + restampIfDemotedToTitleTier(engine, page, page.slug, pageSourceId), + ); + } + result.embedded += toEmbed.length - failed; + if (failed > 0) { + recordFailure(result, failed, page.slug, firstError); + serr(`\n ${page.slug}: ${failed} chunk(s) failed to embed; embedded the other ${toEmbed.length - failed}`); + } } catch (e: unknown) { + // #3037: count the darkened page so the run can't exit 0 (abort is a + // shutdown, not a failure). + if (!isAborted(signal)) recordFailure(result, toEmbed.length, page.slug, e); serr(`\n Error embedding ${page.slug}: ${e instanceof Error ? e.message : e}`); } @@ -1037,10 +1127,8 @@ async function embedAllStale( let afterUpdatedAt: string | null = null; let totalChunksLoaded = 0; let budgetExitNotified = false; - // #1946 (OV2a): track chunks that errored out so a catch-up pass that finishes - // with stale chunks still remaining (un-embeddable for a non-transient reason) - // surfaces that loudly instead of looking like a clean run. - let embedFailures = 0; + // #1946 (OV2a) + #3037: embed failures are tracked on result.failures so + // the catch-up warning below AND the CLI exit verdict both see them. // E-3 (paced-backfill): bounded end-of-run re-entry. A longer paced run gives // a live writer (sync / put_page) more time to insert NEW stale rows BEHIND @@ -1137,12 +1225,19 @@ async function embedAllStale( // NORMAL post-model-migration path, so raw-text embedding here // quietly converted whole corpora to the unwrapped convention. const pageRow = await observed(pacer, () => engine.getPage(slug, { sourceId: keySourceId })); - const embeddings = await embedBatchWithBackoff(wrapChunkTextsForStoredMode(pageRow, stale), { abortSignal: effectiveSignal }); + // #3037: per-chunk failure isolation — one bad chunk costs one + // chunk, not the whole page's siblings. The wrapped texts feed the + // fan-out too, so an isolation retry never strips the prefixes. + const { embeddings, failed, firstError } = await embedPageTexts( + wrapChunkTextsForStoredMode(pageRow, stale), + { abortSignal: effectiveSignal }, + ); // Re-fetch existing chunks and merge to avoid deleting non-stale chunks. const existing = await observed(pacer, () => engine.getChunks(slug, { sourceId: keySourceId })); const staleIdxToEmbedding = new Map(); for (let j = 0; j < stale.length; j++) { - staleIdxToEmbedding.set(stale[j].chunk_index, embeddings[j]); + const emb = embeddings[j]; + if (emb) staleIdxToEmbedding.set(stale[j].chunk_index, emb); } // preserveCodeMetadata threads code-chunk metadata (#769) so the // autopilot --stale path doesn't clobber language/symbol_name/etc @@ -1160,7 +1255,8 @@ async function embedAllStale( // A partially-stale page keeps preserved chunks of unknown/old // provenance, so don't claim it's current. (After invalidate, a // signature-drifted page IS fully stale → this stamps it.) - if (signature && stale.length === existing.length) { + // #3037: not on partial failure — failed chunks stay NULL. + if (signature && failed === 0 && stale.length === existing.length) { await observed(pacer, () => engine.setPageEmbeddingSignature(slug, { sourceId: keySourceId, signature }), ); @@ -1168,17 +1264,24 @@ async function embedAllStale( // #3507: a FULLY re-embedded per_chunk_synopsis page landed at the // title tier — keep the stamped mode honest. Partially-stale pages // stay stamped as-is (mixed provenance; reindex sweeps fix them). - if (stale.length === existing.length) { + // #3037: `failed === 0` is part of "fully re-embedded" — if the + // per-chunk isolation left some chunks NULL, restamping would make + // contextual_retrieval_mode lie again (the exact #3461 bug). + if (failed === 0 && stale.length === existing.length) { await observed(pacer, () => restampIfDemotedToTitleTier(engine, pageRow, slug, keySourceId), ); } - result.embedded += stale.length; + result.embedded += stale.length - failed; + if (failed > 0) { + recordFailure(result, failed, slug, firstError); + serr(`\n ${slug}: ${failed} chunk(s) failed to embed; embedded the other ${stale.length - failed}`); + } } catch (e: unknown) { // Budget/abort-fired cancellations are expected on the way out; don't // spam per-page "Error embedding" lines when we're shutting down. if (effectiveSignal.aborted) return; - embedFailures++; + recordFailure(result, stale.length, slug, e); serr(`\n Error embedding ${slug}: ${e instanceof Error ? e.message : e}`); } totalProcessedPages++; @@ -1231,14 +1334,14 @@ async function embedAllStale( // chunks unembedded means those chunks are stuck (a non-transient embed // failure), not that we ran out of time. Surface it loudly so it doesn't read // as a clean run — re-running won't help until the underlying failure is fixed. - if (staleOpts?.catchUp && !effectiveSignal.aborted && embedFailures > 0) { + if (staleOpts?.catchUp && !effectiveSignal.aborted && result.failures > 0) { const remaining = await engine.countStaleChunks( signature ? { signature, ...(sourceId ? { sourceId } : {}), ...(includeNullSig && { includeNullSignature: true }) } : (sourceId ? { sourceId } : undefined), ); if (remaining > 0) { - serr(`\n [embed] catch-up finished but ${remaining} chunk(s) remain stale after ${embedFailures} embed failure(s). These are not embeddable as-is; re-running won't clear them until the underlying error is resolved.`); + serr(`\n [embed] catch-up finished but ${remaining} chunk(s) remain stale after ${result.failures} embed failure(s). These are not embeddable as-is; re-running won't clear them until the underlying error is resolved.`); } } } @@ -1358,12 +1461,7 @@ export async function embedBatchWithBackoff( // If the budget fired we may have been aborted mid-fetch; bubble out. if (signal?.aborted) throw e; const msg = e instanceof Error ? e.message : String(e); - // D4: structured detection first (handles gateway-wrapped errors via - // cause chain); message-match as fallback for providers whose wrappers - // strip `cause.status`. - const isRateLimit = detect429FromCause(e) - || /rate.?limit|429/i.test(msg); - if (!isRateLimit || attempt === MAX_RATE_LIMIT_RETRIES) throw e; + if (!isRateLimitError(e) || attempt === MAX_RATE_LIMIT_RETRIES) throw e; const delayMs = parseRetryDelayMs(msg); serr(` [rate-limit] attempt ${attempt + 1}/${MAX_RATE_LIMIT_RETRIES}, waiting ${delayMs}ms...`); @@ -1373,3 +1471,86 @@ export async function embedBatchWithBackoff( // Unreachable, but TypeScript needs it. return embedBatch(texts); } + +/** + * 429 judgment shared by embedBatchWithBackoff (retry decision) and + * embedPageTexts (fan-out decision). D4: structured detection first + * (gateway-wrapped errors via cause chain); message-match as fallback for + * providers whose wrappers strip `cause.status`. + */ +function isRateLimitError(e: unknown): boolean { + const msg = e instanceof Error ? e.message : String(e); + return detect429FromCause(e) || /rate.?limit|429/i.test(msg); +} + +/** Walk the cause chain (like detect429FromCause) for the first HTTP status. */ +function statusFromCause(e: unknown): number | undefined { + let cur: unknown = e; + for (let depth = 0; depth < 5 && cur !== undefined && cur !== null; depth++) { + const obj = cur as { status?: unknown; statusCode?: unknown; cause?: unknown }; + if (typeof obj.status === 'number') return obj.status; + if (typeof obj.statusCode === 'number') return obj.statusCode; + cur = obj.cause; + } + return undefined; +} + +/** + * #3037: embed one page's chunk texts with per-chunk failure isolation. + * + * All three embed paths used to send a page's chunks in ONE + * embedBatch call, so one bad chunk (e.g. an oversized chunk the provider + * 400s) left EVERY sibling chunk NULL — an ~8.6x blast radius. This wrapper + * tries the batch first (the cheap, common path), and only on a + * PERMANENT-looking batch failure retries once per chunk so one bad chunk + * costs one chunk. + * + * Cost bounding — when we do NOT fan out (rethrow instead): + * - 429 / rate limit: embedBatchWithBackoff already retried with backoff; + * fanning out N single-chunk calls would hammer the same limiter N-fold. + * - AITransientError (5xx / network / unknown, per normalizeAIError): the + * batch CONTENT isn't the problem, so isolation can't help — during an + * outage it would just multiply failing calls per page. + * - 401/403 (auth): nothing chunk-specific; every call would fail. + * When we DO fan out (permanent request-shaped 4xx like 400/413/422), the + * per-chunk pass happens at most ONCE per page per run and re-spends roughly + * the same tokens the failed batch would have — bounded, no recursion. A + * fresh 429 arising DURING the fan-out still gets the normal backoff (each + * single-chunk call goes through embedBatchWithBackoff). + * + * Throws when nothing could be embedded (total failure — same contract as + * the pre-#3037 single batch call). Returns `null` at the index of each + * failed chunk otherwise. + */ +async function embedPageTexts( + texts: string[], + opts: EmbedBatchWithBackoffOpts = {}, +): Promise<{ embeddings: (Float32Array | null)[]; failed: number; firstError?: unknown }> { + try { + return { embeddings: await embedBatchWithBackoff(texts, opts), failed: 0 }; + } catch (e: unknown) { + if (opts.abortSignal?.aborted) throw e; // shutdown, not a chunk problem + if (texts.length <= 1) throw e; // nothing to isolate + if (isRateLimitError(e) || e instanceof AITransientError) throw e; + const status = statusFromCause(e); + if (status === 401 || status === 403) throw e; + + const embeddings: (Float32Array | null)[] = []; + let failed = 0; + let firstError: unknown; + for (const t of texts) { + try { + const single = await embedBatchWithBackoff([t], opts); + embeddings.push(single[0] ?? null); + if (single[0] === undefined) { failed++; firstError ??= e; } + } catch (chunkErr: unknown) { + if (opts.abortSignal?.aborted) throw chunkErr; + embeddings.push(null); + failed++; + firstError ??= chunkErr; + } + } + if (failed === texts.length) throw firstError ?? e; // total failure: pre-#3037 contract + return { embeddings, failed, firstError }; + } +} diff --git a/test/embed-exit-code-3037.serial.test.ts b/test/embed-exit-code-3037.serial.test.ts new file mode 100644 index 000000000..1e74a7a79 --- /dev/null +++ b/test/embed-exit-code-3037.serial.test.ts @@ -0,0 +1,132 @@ +/** + * #3037 — `gbrain embed` must exit non-zero when chunks failed to embed. + * + * Pre-fix, src/cli.ts discarded runEmbed's result entirely, so a run where + * EVERY chunk failed to embed still exited 0 — cron, CI and health gates read + * total failure as success. The fix mirrors the `import` case's + * `errors > 0 → setCliExitVerdict(1)` guard. + * + * Real spawned CLI against a tmpdir PGLite brain, with the embedding + * provider pointed at a local mock llama-server (OpenAI-compatible, no auth) + * that can be flipped between failing and healthy. Single test, single + * brain: every spawn pays a cold transpile cost (see + * apply-migrations-pglite-spawn.serial.test.ts for the rationale). + * + * Serial: spawns subprocesses + binds a local port + writes tmpdirs. + */ +import { describe, test, expect } from 'bun:test'; +import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'fs'; +import { join } from 'path'; +import { tmpdir } from 'os'; + +const REPO = new URL('..', import.meta.url).pathname.replace(/\/$/, ''); +const DIMS = 16; + +async function runCli( + args: string[], + env: Record, + timeoutMs: number, +): Promise<{ exitCode: number; stdout: string; stderr: string }> { + const proc = Bun.spawn(['bun', 'run', `${REPO}/src/cli.ts`, ...args], { + cwd: REPO, + env: { ...process.env, ...env }, + stdout: 'pipe', + stderr: 'pipe', + }); + const killer = setTimeout(() => { + try { proc.kill('SIGKILL'); } catch { /* already dead */ } + }, timeoutMs); + try { + const [stdout, stderr, exitCode] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + proc.exited, + ]); + return { exitCode, stdout, stderr }; + } finally { + clearTimeout(killer); + } +} + +describe('gbrain embed exit code on failures (#3037)', () => { + test('embed --stale exits non-zero when embedding fails, 0 once it succeeds', async () => { + // Mock OpenAI-compatible embeddings endpoint, flippable between modes. + let mode: 'fail' | 'ok' = 'fail'; + const server = Bun.serve({ + port: 0, + async fetch(req) { + const url = new URL(req.url); + if (url.pathname.endsWith('/embeddings')) { + if (mode === 'fail') { + return new Response(JSON.stringify({ error: { message: 'mock provider exploded' } }), { + status: 500, headers: { 'Content-Type': 'application/json' }, + }); + } + const body = await req.json() as { input: string | string[] }; + const inputs = Array.isArray(body.input) ? body.input : [body.input]; + const vec = Array.from({ length: DIMS }, () => 0.1); + return new Response(JSON.stringify({ + data: inputs.map((_, i) => ({ object: 'embedding', index: i, embedding: vec })), + usage: { prompt_tokens: 3, total_tokens: 3 }, + }), { status: 200, headers: { 'Content-Type': 'application/json' } }); + } + // /v1/models probe shape. + return new Response(JSON.stringify({ data: [{ id: 'test-model' }] }), { + status: 200, headers: { 'Content-Type': 'application/json' }, + }); + }, + }); + + const home = mkdtempSync(join(tmpdir(), 'gbrain-3037-exit-')); + const notes = mkdtempSync(join(tmpdir(), 'gbrain-3037-notes-')); + try { + mkdirSync(join(home, '.gbrain'), { recursive: true }); + writeFileSync( + join(home, '.gbrain', 'config.json'), + JSON.stringify({ + engine: 'pglite', + database_path: join(home, '.gbrain', 'brain.pglite'), + embedding_model: 'llama-server:test-model', + embedding_dimensions: DIMS, + }) + '\n', + ); + writeFileSync(join(notes, 'note.md'), '# A note\n\nSome content to embed.\n'); + const env = { + HOME: home, + GBRAIN_HOME: home, + LLAMA_SERVER_BASE_URL: `http://127.0.0.1:${server.port}/v1`, + }; + + const init = await runCli(['init', '--migrate-only'], env, 120_000); + expect(init.exitCode).toBe(0); + + const imp = await runCli(['import', notes, '--no-embed'], env, 90_000); + expect(imp.exitCode).toBe(0); + + // THE #3037 PIN: provider fails every embed call → the run must exit + // non-zero. Pre-fix this exited 0 (result discarded by cli.ts). + const failing = await runCli(['embed', '--stale'], env, 90_000); + if (failing.exitCode === 0) { + console.error('--- failing-embed stdout ---\n' + failing.stdout); + console.error('--- failing-embed stderr ---\n' + failing.stderr); + } + expect(failing.exitCode).not.toBe(0); + expect(failing.stderr).toMatch(/failed to embed/i); + + // Same brain, healthy provider: converges and exits 0 (failure exit is + // not sticky; the failed chunks stayed NULL so --stale picks them up). + mode = 'ok'; + const healthy = await runCli(['embed', '--stale'], env, 90_000); + if (healthy.exitCode !== 0) { + console.error('--- healthy-embed stdout ---\n' + healthy.stdout); + console.error('--- healthy-embed stderr ---\n' + healthy.stderr); + } + expect(healthy.exitCode).toBe(0); + expect(healthy.stdout + healthy.stderr).toMatch(/Embedded [1-9]\d* chunks/); + } finally { + server.stop(true); + try { rmSync(home, { recursive: true, force: true }); } catch { /* best effort */ } + try { rmSync(notes, { recursive: true, force: true }); } catch { /* best effort */ } + } + }, 480_000); +}); diff --git a/test/embed-partial-failure-3037.serial.test.ts b/test/embed-partial-failure-3037.serial.test.ts new file mode 100644 index 000000000..4cd0d41ce --- /dev/null +++ b/test/embed-partial-failure-3037.serial.test.ts @@ -0,0 +1,266 @@ +/** + * #3037 — one oversized/bad chunk must not darken its ENTIRE page, and embed + * failures must be visible on the run result. + * + * Pre-fix, all three embed paths sent a page's chunks in ONE embedBatch call + * inside a try whose catch only logged to stderr: when the batch threw, + * upsertChunks never ran, so EVERY sibling chunk stayed NULL (~8.6x blast + * radius from a single bad chunk), and EmbedResult had no failure field — + * `gbrain embed` exited 0 on a total no-op. + * + * Pinned here: + * 1. --stale and --all: a page with ONE bad chunk still embeds its other + * chunks (per-chunk isolation via embedPageTexts), failures are counted + * on result.failures, and the embedding signature is NOT stamped for a + * partially-failed page. + * 2. Cost bounding: a 429 (rate limit) does NOT fan out into N + * single-chunk calls, and neither does an AITransientError (outage) — + * isolation only fires for permanent request-shaped failures. + * + * Serial: uses mock.module (leaks across files sharing a bun process). + * The CLI exit-code half of #3037 is pinned by + * test/embed-exit-code-3037.serial.test.ts (real spawned CLI). + */ +import { describe, test, expect, mock, beforeEach, afterEach } from 'bun:test'; +import type { BrainEngine } from '../src/core/engine.ts'; +import { AITransientError } from '../src/core/ai/errors.ts'; + +// Track every embedBatch call's shape so tests can assert batch-vs-single +// fan-out behavior. +let embedCalls: string[][] = []; +let embedBatchBehavior: ((texts: string[], opts?: unknown) => Promise) | null = null; + +mock.module('../src/core/embedding.ts', () => ({ + embedBatch: async (texts: string[], opts?: unknown) => { + embedCalls.push([...texts]); + if (embedBatchBehavior) return embedBatchBehavior(texts, opts); + return texts.map(() => new Float32Array(1536)); + }, + currentEmbeddingSignature: () => 'test:model:1536', +})); + +// Import AFTER mocking. +const { runEmbedCore } = await import('../src/commands/embed.ts'); + +// Preflight seam (same as test/embed.serial.test.ts): make +// diagnoseEmbedding's fast-path pass without real env vars. +const { __setEmbedTransportForTests } = await import('../src/core/ai/gateway.ts'); +__setEmbedTransportForTests(async () => ({ embeddings: [], usage: { tokens: 0 } } as any)); + +function mockEngine(overrides: Partial> = {}): BrainEngine { + const calls: { method: string; args: any[] }[] = []; + const track = (method: string) => (...args: any[]) => { + calls.push({ method, args }); + if (overrides[method]) return overrides[method](...args); + return Promise.resolve(null); + }; + return new Proxy({} as any, { + get(_, prop: string) { + if (prop === '_calls') return calls; + if (overrides[prop]) return overrides[prop]; + return track(prop); + }, + }); +} + +/** Permanent 400-shaped batch failure (e.g. one oversized chunk). */ +function permanentBatchError(): Error { + const err = new Error('batch contains an invalid input'); + (err as any).cause = { status: 400 }; + return err; +} + +beforeEach(() => { + embedCalls = []; + embedBatchBehavior = null; + process.env.GBRAIN_EMBED_CONCURRENCY = '1'; +}); + +afterEach(() => { + delete process.env.GBRAIN_EMBED_CONCURRENCY; +}); + +// Behavior: the whole-page batch 400s; retried per-chunk, only 'BAD' fails. +function oneBadChunkBehavior() { + embedBatchBehavior = async (texts: string[]) => { + if (texts.length > 1) throw permanentBatchError(); + if (texts[0] === 'BAD') throw permanentBatchError(); + return texts.map(() => new Float32Array(1536)); + }; +} + +const THREE_CHUNKS = [ + { chunk_index: 0, chunk_text: 'good-a', chunk_source: 'compiled_truth' as const, embedded_at: null, token_count: 1 }, + { chunk_index: 1, chunk_text: 'BAD', chunk_source: 'compiled_truth' as const, embedded_at: null, token_count: 1 }, + { chunk_index: 2, chunk_text: 'good-b', chunk_source: 'compiled_truth' as const, embedded_at: null, token_count: 1 }, +]; + +describe('#3037 — one bad chunk no longer darkens its page', () => { + test('--stale: siblings of one bad chunk get embedded; failure counted; signature not stamped', async () => { + oneBadChunkBehavior(); + const stale = THREE_CHUNKS.map(c => ({ + slug: 'poisoned-page', chunk_index: c.chunk_index, chunk_text: c.chunk_text, + chunk_source: c.chunk_source, model: null, token_count: 1, source_id: 'default', page_id: 1, + })); + const upsertCalls: Array<{ slug: string; chunks: any[] }> = []; + const engine = mockEngine({ + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, + getChunks: async () => THREE_CHUNKS, + upsertChunks: async (slug: string, chunks: any[]) => { upsertCalls.push({ slug, chunks }); }, + }); + + const result = await runEmbedCore(engine, { stale: true }); + + // Pre-fix: the batch threw, upsertChunks never ran, embedded stayed 0. + expect(upsertCalls).toHaveLength(1); + const byIdx = new Map(upsertCalls[0].chunks.map((c: any) => [c.chunk_index, c])); + expect(byIdx.get(0)!.embedding).toBeInstanceOf(Float32Array); + expect(byIdx.get(2)!.embedding).toBeInstanceOf(Float32Array); + expect(byIdx.get(1)!.embedding).toBeUndefined(); // bad chunk stays NULL (re-run picks it up) + expect(result.embedded).toBe(2); + expect(result.failures).toBe(1); + expect(result.failure_samples).toHaveLength(1); + expect(result.failure_samples[0]).toContain('poisoned-page'); + // Partially-failed page must NOT be stamped as current provenance. + const stamps = (engine as any)._calls.filter((c: any) => c.method === 'setPageEmbeddingSignature'); + expect(stamps).toHaveLength(0); + }); + + test('--all: same isolation on the listPages path', async () => { + oneBadChunkBehavior(); + const upsertCalls: Array<{ slug: string; chunks: any[] }> = []; + const engine = mockEngine({ + listPages: async () => [{ slug: 'poisoned-page', source_id: 'default' }], + getChunks: async () => THREE_CHUNKS, + upsertChunks: async (slug: string, chunks: any[]) => { upsertCalls.push({ slug, chunks }); }, + }); + + const result = await runEmbedCore(engine, { all: true }); + + expect(upsertCalls).toHaveLength(1); + const byIdx = new Map(upsertCalls[0].chunks.map((c: any) => [c.chunk_index, c])); + expect(byIdx.get(0)!.embedding).toBeInstanceOf(Float32Array); + expect(byIdx.get(1)!.embedding).toBeUndefined(); + expect(result.embedded).toBe(2); + expect(result.failures).toBe(1); + const stamps = (engine as any)._calls.filter((c: any) => c.method === 'setPageEmbeddingSignature'); + expect(stamps).toHaveLength(0); + }); + + test('--stale x #3507: fan-out retries the WRAPPED texts and a partially-failed page is not restamped', async () => { + // Composition pin for the #3037 + #3538 merge: the per-chunk isolation + // retry must re-send the contextually WRAPPED text (raw chunk_text here + // would silently strip prefixes on exactly the pages that hit an error), + // and restampIfDemotedToTitleTier must NOT fire when isolation left + // chunks NULL (the page was not fully re-embedded — restamping would + // make contextual_retrieval_mode lie again, the exact #3461 bug). + embedBatchBehavior = async (texts: string[]) => { + if (texts.length > 1) throw permanentBatchError(); + if (texts[0].includes('BAD')) throw permanentBatchError(); + return texts.map(() => new Float32Array(1536)); + }; + const stale = THREE_CHUNKS.map(c => ({ + slug: 'wrapped-page', chunk_index: c.chunk_index, chunk_text: c.chunk_text, + chunk_source: c.chunk_source, model: null, token_count: 1, source_id: 'default', page_id: 1, + })); + const engine = mockEngine({ + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, + getPage: async () => ({ + slug: 'wrapped-page', title: 'My Title', compiled_truth: 'x', timeline: '', + source_id: 'default', contextual_retrieval_mode: 'per_chunk_synopsis', + }), + getChunks: async () => THREE_CHUNKS, + upsertChunks: async () => {}, + }); + + const result = await runEmbedCore(engine, { stale: true }); + + // Every embed call — the failed batch AND each single-chunk retry — + // carries the stored-mode contextual prefix (fenced_code exemption is + // pinned upstream in test/embedding-context.test.ts). + expect(embedCalls.length).toBeGreaterThan(1); + for (const call of embedCalls) { + for (const text of call) expect(text).toStartWith('My Title\n\n'); + } + expect(result.embedded).toBe(2); + expect(result.failures).toBe(1); + // Partially-failed page: neither signature-stamped nor CR-restamped. + const calls = (engine as any)._calls as Array<{ method: string }>; + expect(calls.filter(c => c.method === 'setPageEmbeddingSignature')).toHaveLength(0); + expect(calls.filter(c => c.method === 'updatePageContextualRetrievalState')).toHaveLength(0); + }); + + test('--stale: a fully-failed page is counted on result.failures (no more silent no-op)', async () => { + embedBatchBehavior = async () => { throw permanentBatchError(); }; + const stale = THREE_CHUNKS.map(c => ({ + slug: 'dark-page', chunk_index: c.chunk_index, chunk_text: c.chunk_text, + chunk_source: c.chunk_source, model: null, token_count: 1, source_id: 'default', page_id: 1, + })); + const engine = mockEngine({ + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, + getChunks: async () => THREE_CHUNKS, + upsertChunks: async () => {}, + }); + + const result = await runEmbedCore(engine, { stale: true }); + + expect(result.embedded).toBe(0); + expect(result.failures).toBe(3); + expect(result.failure_samples[0]).toContain('dark-page'); + }); +}); + +describe('#3037 — cost bounding: no per-chunk fan-out on transient failures', () => { + test('sustained 429 does not fan out into single-chunk calls', async () => { + embedBatchBehavior = async () => { + const err = new Error('Rate limit reached. Please try again in 0ms.'); + (err as any).cause = { status: 429 }; + throw err; + }; + const stale = THREE_CHUNKS.map(c => ({ + slug: 'rate-limited-page', chunk_index: c.chunk_index, chunk_text: c.chunk_text, + chunk_source: c.chunk_source, model: null, token_count: 1, source_id: 'default', page_id: 1, + })); + const engine = mockEngine({ + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, + getChunks: async () => THREE_CHUNKS, + upsertChunks: async () => {}, + }); + + const result = await runEmbedCore(engine, { stale: true }); + + // Every call must be the full 3-text batch: embedBatchWithBackoff's own + // retries (initial + MAX_RATE_LIMIT_RETRIES), never a 1-text isolation call + // hammering the limiter. + expect(embedCalls.length).toBeGreaterThan(1); + for (const call of embedCalls) expect(call).toHaveLength(3); + expect(result.embedded).toBe(0); + expect(result.failures).toBe(3); + }, 30_000); + + test('AITransientError (outage/network) does not fan out', async () => { + embedBatchBehavior = async () => { throw new AITransientError('upstream 502', { status: 502 }); }; + const stale = THREE_CHUNKS.map(c => ({ + slug: 'outage-page', chunk_index: c.chunk_index, chunk_text: c.chunk_text, + chunk_source: c.chunk_source, model: null, token_count: 1, source_id: 'default', page_id: 1, + })); + const engine = mockEngine({ + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, + getChunks: async () => THREE_CHUNKS, + upsertChunks: async () => {}, + }); + + const result = await runEmbedCore(engine, { stale: true }); + + // Non-429 → no backoff retries; transient → no isolation. Exactly 1 call. + expect(embedCalls).toHaveLength(1); + expect(embedCalls[0]).toHaveLength(3); + expect(result.embedded).toBe(0); + expect(result.failures).toBe(3); + }); +});