From d036a97f9ca6dc27868ac9bd471db213a287de68 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Mon, 25 May 2026 15:10:23 -0700 Subject: [PATCH] v0.41.10.1 fix-wave: dream.* config + batch retry + extract_atoms idempotency + ze-switch env-gate (#1445) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix-wave: dream.* DB merge + batch retry + extract_atoms idempotency + ze-switch env-gate + doctor check Closes PRs #1414, #1416, #1421 (rebuilt from designs by @garrytan-agents with structural improvements from /plan-eng-review + codex outside-voice). Three production reliability fixes in one wave: 1. dream.* DB-config merge (closes PR #1416 silent-config gap) - loadConfigWithEngine() sparse-merge extends with 7 dream.* keys - File > DB > defaults precedence (no GBRAIN_DREAM_* env vars) - extract-atoms switches to loadConfigWithEngine() so DB-plane keys reach it 2. Batch retry on transient connection drops (closes PR #1416 ~30%-loss bug) - withRetry() pure primitive exported from src/commands/extract.ts - 6 flush() sites snapshot-before-clear with onRetry callback - Reuses isRetryableConnError from src/core/retry-matcher.ts - retry-matcher extended with GBrainError{problem:'No database connection'} 3. extract_atoms source-hash idempotency + page-based discovery (closes #1414) - One raw SQL with NOT EXISTS subquery replaces 6 listPages + N atom checks - sourceId threaded through every putPage call (codex caught real bug) - NULL content_hash filter + dream_generated exclusion + transcript-side idempotency - cycle.ts passes union of syncPagesAffected + synthesizeWrittenSlugs 4. ze-switch pre-apply + pre-resume env-override gate (closes PR #1421) - Gate fires FIRST in apply AND resume; zero setConfig calls on refusal - ASCII warning box (no Unicode per repo D10) - --ignore-env-override escape hatch for power users - ApplyResult extended with refused variant 5. doctor embedding_env_override check (defense-in-depth for #1421) - Cross-surface parity: buildChecks() + doctorReportRemote() - Uses Check.details (not Check.issues per codex schema review) Co-Authored-By: garrytan-agents Co-Authored-By: Claude Opus 4.7 (1M context) * chore: bump version and changelog (v0.41.10.0) Adds 61 new tests across 5 new files pinning the fix-wave contracts: - test/extract-batch-retry.test.ts (16 cases) — withRetry primitive + snapshot contract - test/extract-atoms-page-discovery.test.ts (17 cases) — discovery SQL + dual-source idempotency - test/ze-switch-env-override.test.ts (17 cases) — env-gate apply + resume + ZERO-setConfig assertion - test/doctor-embedding-env-override.test.ts (7 cases) — cross-surface parity - test/e2e/extract-atoms-discovery-sql.test.ts (4 cases) — real-Postgres parity for raw SQL Co-Authored-By: Claude Opus 4.7 (1M context) * fix(test): pin gateway to 1536-dim in 2 PGLite tests that hardcode 1536-vector inserts CI shards 1 + 4 failed persistently (not flake — confirmed via retry) after the v0.41.6.0 merge with this error: error: expected 1280 dimensions, not 1536 file: "vector.c", routine: "CheckExpectedDim" Two test files insert 1536-dim Float32Array vectors into `content_chunks.embedding` / `facts.embedding`, but v0.41.5.0 flipped `DEFAULT_EMBEDDING_DIMENSIONS` from 1536 to 1280 (ZE Matryoshka default). On a fresh CI bun process where no prior test pre-configured the gateway, `initSchema()` sizes the vector column at vector(1280) and the inserts throw. Locally this is hidden when an earlier test file in the shard happens to have called `configureGateway({embedding_dimensions: 1536})` — that state leaks forward through bun's shared process. The v0.41.6.0 LPT shard re-balancing reordered files so these two ran cold, surfacing the latent bug. Fix follows the canonical hermetic pattern from test/consolidate-valid-until.test.ts:23-34: pin the gateway to 1536d in beforeAll, reset in afterAll. Test is now isolated from shard ordering. test/search-types-filter.test.ts — shard 1 fail test/operations-find-trajectory.test.ts — shard 4 (6 fails) Co-Authored-By: Claude Opus 4.7 (1M context) * chore: empty commit to trigger CI * chore: trigger CI again * chore: renumber v0.41.10.0 -> v0.41.10.1 Per request — version slot moved to .1 micro tier to leave .0 available for unrelated wave landing on master. --------- Co-authored-by: garrytan-agents Co-authored-by: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 114 +++++ TODOS.md | 31 ++ VERSION | 2 +- package.json | 2 +- src/commands/doctor.ts | 83 ++++ src/commands/extract.ts | 119 ++++-- src/commands/ze-switch.ts | 53 ++- src/core/config.ts | 80 ++++ src/core/cycle.ts | 12 +- src/core/cycle/extract-atoms.ts | 369 +++++++++++++--- src/core/retrieval-upgrade-planner.ts | 117 +++++- src/core/retrieval-upgrade-prompt.ts | 12 + src/core/retry-matcher.ts | 17 + .../extract-atoms-synthesize-concepts.test.ts | 41 +- test/doctor-embedding-env-override.test.ts | 149 +++++++ test/e2e/extract-atoms-discovery-sql.test.ts | 135 ++++++ test/extract-atoms-page-discovery.test.ts | 395 ++++++++++++++++++ test/extract-batch-retry.test.ts | 276 ++++++++++++ test/loadConfig-merge.test.ts | 126 ++++++ test/operations-find-trajectory.test.ts | 14 + test/search-types-filter.test.ts | 14 + test/ze-switch-env-override.test.ts | 327 +++++++++++++++ 22 files changed, 2385 insertions(+), 103 deletions(-) create mode 100644 test/doctor-embedding-env-override.test.ts create mode 100644 test/e2e/extract-atoms-discovery-sql.test.ts create mode 100644 test/extract-atoms-page-discovery.test.ts create mode 100644 test/extract-batch-retry.test.ts create mode 100644 test/ze-switch-env-override.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index ce7a789d7..8e0165d10 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,120 @@ All notable changes to GBrain will be documented in this file. +## [0.41.10.1] - 2026-05-25 + +**Background sweeps stop silently losing rows, `dream.*` config you set actually reaches the cycle, and switching embedding providers won't quietly corrupt your brain when env vars override the switch.** Three production reliability fixes landed in one wave, rebuilt from three closed community PRs (#1414, #1416, #1421 from `@garrytan-agents`) with structural improvements from `/plan-eng-review` + codex outside-voice review. + +You can now configure `dream.synthesize.session_corpus_dir` (and 6 other dream.* keys) via `gbrain config set` and have it actually reach the cycle phase that reads it. Pre-fix the cycle silently skipped with "no transcripts to process" even though the config wrote successfully to the DB. The `extract` and `sync` commands now retry batched inserts once on transient PgBouncer connection drops instead of losing the whole batch — closes a ~30% data-loss rate observed on 96K-page brains during heavy cycles. And `gbrain ze-switch` now refuses to start a schema transition when `GBRAIN_EMBEDDING_MODEL` is pinned to a model that disagrees with the target, instead of silently corrupting 716K chunks the way the original incident did. + +To turn it on: `gbrain upgrade`. + +What you'd see in a concrete example. + +- **Before:** Set `dream.synthesize.session_corpus_dir /Users/me/transcripts` via `gbrain config set`. Run `gbrain dream --phase extract_atoms`. It skips with "no transcripts to process" — config never reached the phase because `extract-atoms.ts` read via file-plane-only `loadConfig()`, not `loadConfigWithEngine(engine)`. +- **After:** Same setup, the phase actually walks the directory. + +- **Before:** Run `gbrain extract all` on a brain hitting PgBouncer pool recycles. ~30% of batched inserts throw "No database connection: connect() has not been called" and silently drop 100 rows each — visible in stderr but easy to miss in a long run. +- **After:** Same setup, transient connection errors trigger one 500ms retry. Stderr shows `[extract.links_fs] connection blip, retrying 100 rows in 500ms (Connection terminated unexpectedly)` and the retry succeeds. Snapshot-before-clear contract means the retry sends the same data even if the producer wrote more during the delay. + +- **Before:** `GBRAIN_EMBEDDING_MODEL=openai:text-embedding-3-large gbrain ze-switch --non-interactive --force`. Schema migrates to 2560-dim ZE columns. Embed sweep reads env, embeds with OpenAI's 1536-dim model, writes 1536d vectors into 2560d columns. Brain corrupts. +- **After:** Same command refuses pre-apply with an ASCII warning box: schema not mutated, paste-ready `unset GBRAIN_EMBEDDING_MODEL` command surfaced. Apply with `--ignore-env-override` if you really mean it. The gate fires on `--resume` too, so there's no bypass path. + +Things to know about. **(1)** Idempotency upgrade for `extract_atoms`. The phase now checks "do any atoms already exist for this content hash?" before calling Haiku, replacing the date-stamped slug that caused duplicate atoms when re-discovered on a new day. Page-side and transcript-side both covered. Re-running the cycle on unchanged content produces zero new atoms; the original incident is closed. **(2)** Known limitation: if Haiku writes atom 1 of 3 then atom 2 throws, source-hash filter sees atom 1 exists and skips on next discovery — atoms 2+3 stay missing until content changes. Documented in TODOS.md as a v0.42+ per-atom idempotency follow-up. Rare in practice. **(3)** A new `embedding_env_override` doctor check runs every doctor pass: surfaces when `GBRAIN_EMBEDDING_MODEL` / `GBRAIN_EMBEDDING_DIMENSIONS` disagree with DB config so users see the drift before the embed sweep corrupts vectors. Wired into both local doctor and the HTTP MCP doctor surface. + +What we caught and fixed before merging. Codex outside-voice review caught 10 real correctness gaps in the plan: +- `extract-atoms` was calling `engine.putPage(slug, page)` without sourceId — on non-default brain sources, atoms always wrote to `default` and the NOT EXISTS idempotency check became ineffective. Fixed: sourceId threaded through every putPage call. +- The original `ze-switch` env-gate placement (post-apply warning) would have shipped the same 716K-chunk damage class. Fixed: gate fires BEFORE the snapshot write at line ~294, not just before runSchemaTransition at ~304. Test asserts ZERO setConfig calls fire on a refused apply. +- `resumeRetrievalUpgrade` would have been a bypass path. Fixed: same gate on resume. +- Discovery SQL would have crashed on pages with NULL content_hash. Fixed: `AND p.content_hash IS NOT NULL` filter added. +- Discovery would have chewed its own dream-generated output. Fixed: `AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true'` filter. +- cycle.ts was passing `affectedSlugs = syncPagesAffected` only, missing pages just-written by the synthesize phase in the same cycle. Fixed: union of sync + synthesize affected slugs. +- `ApplyResult` and `Check.issues` tagged unions/types needed extending to handle the new `refused` variant and `details.mismatches[]` shape. Fixed: typed properly. +- The plan's "env wins for dream.*" claim was false — there are no GBRAIN_DREAM_* env vars. Fixed: dropped the false claim; precedence is `file > DB > defaults`. + +### Itemized changes + +**Phase 1: systemic dream.\* DB-config merge (`src/core/config.ts`)** + +- `loadConfigWithEngine()` sparse-merge block extended with 7 `dream.*` keys: `dream.synthesize.{session_corpus_dir, meeting_transcripts_dir, verdict_model, max_prompt_tokens, max_chunks_per_transcript}` + `dream.patterns.{lookback_days, min_evidence}`. +- Precedence: file > DB > defaults per key. Parent objects (`cfg.dream`, `cfg.dream.synthesize`, `cfg.dream.patterns`) allocated defensively before assigning leaf keys. +- Invalid DB int values fall back to "DB miss" (no throw). +- `GBrainConfig` interface gains `dream?:` typed shape so consumers get TypeScript-level safety. + +**Phase 2: batch withRetry (`src/commands/extract.ts`, `src/core/retry-matcher.ts`)** + +- New exported pure primitive `withRetry(fn, {onRetry?, delayMs?})` — single retry, 500ms default delay. Test seam via `delayMs: 0`. +- New `logBatchRetry(label, snapshotLen, err, jsonMode)` helper shared across 6 flush sites. +- All 6 `flush()` sites in `extract.ts` (lines ~519, 531, 614, 672, 840, 994) converted to `snapshot.slice()` BEFORE `batch.length = 0`. Error messages read `snapshot.length`, not the post-clear batch length. +- Classifier is the existing `isRetryableConnError` from `src/core/retry-matcher.ts`. Extended in this wave with `GBrainError{problem: 'No database connection'}` typed shape recognition + the literal "No database connection" message pattern (closes the specific shape from PR #1416's reported incident). + +**Phase 3: extract_atoms idempotency + page discovery (`src/core/cycle/extract-atoms.ts`, `src/core/cycle.ts`)** + +- Source-hash existence check before LLM call replaces date-stamped slug as idempotency mechanism. Survives `gbrain sync --force` re-imports. +- New `discoverExtractablePages(engine, sourceId, affectedSlugs?)` runs ONE raw SQL query with NOT EXISTS subquery: discovers extractable pages (6 types: meeting, source, article, video, book, original) AND filters already-extracted in one round-trip. Replaces the 6-listPages + per-candidate atom-check pattern that would have made ~56 queries per cycle. +- SQL filters: `source_id`, `type = ANY($::text[])`, `deleted_at IS NULL`, `content_hash IS NOT NULL`, `imported_from <> 'markdown-greenfield'`, `dream_generated <> 'true'`, `length(compiled_truth) >= MIN`, optional `slug = ANY($::text[])` for affectedSlugs, NOT EXISTS atom with matching source_hash. +- Transcript-side idempotency: `atomsExistForHash(engine, sourceId, contentHash16)` — closes the pre-existing v0.41.2.0 date-stamp duplicate bug for transcripts. +- `sourceId` threaded through every `engine.putPage(slug, page, {sourceId})` call so atoms land in the correct source on federated brains. +- New `_pages` test seam mirrors `_transcripts` shape. `_pages: undefined` triggers discovery; `_pages: []` deliberately suppresses. +- `PhaseResult.details` extended with additive fields: `pages_processed`, `pages_total`, `pages_skipped_budget`, `duplicates_skipped`. All existing fields preserved (regression-tested). +- `cycle.ts` passes union of `syncPagesAffected + synthesizeWrittenSlugs` as `affectedSlugs` to `runPhaseExtractAtoms`. + +**Phase 4: ze-switch pre-apply + pre-resume env-override gate (`src/core/retrieval-upgrade-planner.ts`, `src/commands/ze-switch.ts`, `src/core/retrieval-upgrade-prompt.ts`)** + +- New pure exports `detectEnvOverride(targetModel, targetDim, env?)` and `formatEnvOverrideWarning(warning)`. ASCII box (no Unicode per repo D10), line width ≤78 cols, includes paste-ready `unset` command. +- `ApplyResult` tagged union extended with `{status: 'refused', reason: 'env_override', warning}` variant. `ApplyOpts` interface adds `ignoreEnvOverride?: boolean`. +- Gate fires FIRST in `applyRetrievalUpgrade` — BEFORE `setConfig(KEY_PREVIOUS_SNAPSHOT)`, BEFORE `runSchemaTransition`. Zero side effects on refusal. Pinned by test that asserts ZERO setConfig calls fire on refused apply. +- Same gate fires FIRST in `resumeRetrievalUpgrade`. No bypass path. +- CLI flag `--ignore-env-override` mirrors the existing `--ignore-missing-key` precedent. Loud stderr line when set. +- Planner stays data-pure — returns the warning struct; CLI handles rendering. Interactive prompt path (`runRetrievalUpgradePrompt`) also handles the new variant gracefully. + +**Phase 5: doctor `embedding_env_override` check (`src/commands/doctor.ts`)** + +- New `checkEmbeddingEnvOverride(engine)` reads `process.env.GBRAIN_EMBEDDING_MODEL` and `GBRAIN_EMBEDDING_DIMENSIONS`, compares against DB config. Uses `Check.details.mismatches[]` (NOT `Check.issues` which has a different schema). +- Wired into BOTH `buildChecks()` (local doctor) AND `doctorReportRemote()` (HTTP MCP doctor) per the cross-surface parity convention. Source-grep regression assertion pins both wirings. +- Message includes paste-ready `unset GBRAIN_EMBEDDING_MODEL GBRAIN_EMBEDDING_DIMENSIONS` fix. + +**Test coverage:** + +- New `test/extract-batch-retry.test.ts` (16 cases) — withRetry primitive, GBrainError shape recognition, logBatchRetry, snapshot-mutation regression contract. +- New `test/extract-atoms-page-discovery.test.ts` (17 cases, PGLite) — discovery SQL filters, NOT EXISTS idempotency, dual-source merge, transcript-side idempotency, sourceId threading. +- New `test/ze-switch-env-override.test.ts` (17 cases, PGLite + withEnv) — pure helpers, applyRetrievalUpgrade integration with engine-setConfig-spy ZERO-mutation assertion, resumeRetrievalUpgrade parity. +- New `test/doctor-embedding-env-override.test.ts` (7 cases, PGLite + withEnv) — all status branches + cross-surface parity source-grep. +- New `test/e2e/extract-atoms-discovery-sql.test.ts` (4 cases, real Postgres) — D10 reversal: validates ANY($::text[]) + JSONB ->> + NOT EXISTS + substring through `postgres.unsafe` against real PG. +- Extended `test/cycle/extract-atoms-synthesize-concepts.test.ts` — `_pages: []` added to all existing cases + 1 critical regression case (legacy PhaseResult.details fields byte-identical to v0.41.2.0 transcript-only path). +- Extended `test/loadConfig-merge.test.ts` — 8 new dream.* parallel cases (DB merge, file-wins, parent allocation, invalid-int fallback, throw resilience). + +**Credit:** rebuilt from work originally proposed by @garrytan-agents in PRs #1414, #1416, #1421. The PR descriptions were the design source — the rebuild's structural improvements came from /plan-eng-review and codex outside-voice. Closes both bug classes (the reported PgBouncer batch loss and the documented 716K-chunk damage incident). + +## To take advantage of v0.41.10.1 + +`gbrain upgrade` should do this automatically. If it didn't, or if `gbrain doctor` +warns about a partial migration: + +1. **Run the orchestrator manually:** + ```bash + gbrain apply-migrations --yes + ``` +2. **Verify the wave's three fixes:** + ```bash + # dream.* config merge + gbrain config set dream.synthesize.session_corpus_dir /tmp/test + gbrain config get dream.synthesize.session_corpus_dir + # should print /tmp/test, not the file-plane default + + # ze-switch env-override gate + GBRAIN_EMBEDDING_MODEL=openai:text-embedding-3-large gbrain ze-switch --non-interactive --force + # should refuse with the ASCII warning box, NOT mutate schema + + # doctor surfaces env disagreement + GBRAIN_EMBEDDING_MODEL=openai:text-embedding-3-large gbrain doctor --fast | grep embedding_env_override + # should show: embedding_env_override: warn + ``` +3. **If anything fails or looks wrong,** file an issue at + https://github.com/garrytan/gbrain/issues with: + - output of `gbrain doctor` + - contents of `~/.gbrain/upgrade-errors.jsonl` if it exists + - which step broke ## [0.41.10.0] - 2026-05-25 **Your brain stops being mostly orphan pages.** A new `gbrain extract links --by-mention` pass scans every page's body text for mentions of people and companies you already have pages for, then creates links automatically. The same release also fixes a silent corruption bug in the dream-cycle chunker that could split UTF-16 surrogate pairs (emoji, non-BMP CJK, mathematical alphanumerics) at chunk boundaries, breaking the per-chunk idempotency key on retries. diff --git a/TODOS.md b/TODOS.md index 5235e269d..5a17273f2 100644 --- a/TODOS.md +++ b/TODOS.md @@ -1,5 +1,36 @@ # TODOS +## v0.41.10.1 fix-wave follow-ups (v0.42+) + +- [ ] **v0.42+: per-atom idempotency via deterministic atom slug.** The + v0.41.10.1 fix wave closed the duplicate-atoms bug class via source-hash + existence check at the SOURCE level (skip the whole transcript/page if + any atom row exists for `frontmatter.source_hash`). Known limitation + surfaced by codex review (D9 #2): if the first Haiku call writes atom + 1 of 3 then atom 2 throws, the source_hash filter sees atom 1 exists + and skips on next discovery — atoms 2 + 3 stay missing until + `content_hash` changes. The cleaner solution is per-atom idempotency: + switch atom slugs from date-stamped (`atoms/2026-05-25/`) + to content-hash-stamped (`atoms//`) + so `engine.putPage` upserts naturally on retry. Bounded scope; needs + a migration to consolidate existing duplicate atoms (filed separately + below as the v0.42+ consolidation TODO). Priority: P2. References: + `src/core/cycle/extract-atoms.ts:atomsExistForHash`, the documented + known-limitation comment in the file header. + +- [ ] **v0.42+: atom-slug consolidation migration.** The v0.41.10.1 fix + wave stops NEW duplicates from being written but doesn't migrate + existing duplicate atoms from prior v0.41.2.0 runs. Brains that ran + the cycle across multiple days carry duplicate atoms forever (or until + manual cleanup): `atoms/2026-05-15/title-X` AND `atoms/2026-05-25/title-X` + for the same content_hash. Migration writes a one-shot CLI flow: + `gbrain atoms consolidate [--dry-run] [--yes]` that groups atoms by + `frontmatter.source_hash`, keeps the oldest atom row, soft-deletes + newer copies (uses the existing `softDeletePage` path so 72h restore + window applies). Operator opt-in via the same `--confirm-destructive` + gate from the destructive-guard. Priority: P3. Filed via /plan-eng-review + D6. References: `src/core/cycle/extract-atoms.ts`, the v0.26.5 + soft-delete + restore infrastructure. ## v0.41.10.0 follow-ups (orphan-reduction + surrogate fix wave) - [ ] **TODO-1 (P2) — Pack-aware `--by-mention` gazetteer.** Add `linkable: boolean` per-type field to the schema-pack manifest (`src/core/schema-pack/manifest-v1.ts`, currently has `extractable` + `expert_routing`). New accessor `linkableTypesFromPack(pack: ResolvedPack)` in a new `schema-pack/linkable-types.ts` module mirroring `expert-types.ts`. `src/core/by-mention.ts:buildGazetteer` consults the pack-aware filter first via `loadActivePackBestEffort(ctx)`, falls back to the hardcoded `LINKABLE_ENTITY_TYPES` const for non-pack brains. Respects the D4 fail-empty contract (pack-load failure → empty filter, NOT hardcoded defaults). User-defined types like `researcher` get auto-linked. Requires: pack-schema bump, rubric/registry updates, regression test that pack-aware + non-pack brains produce expected gazetteer shapes. diff --git a/VERSION b/VERSION index f904e6e3e..ed7d8f3d3 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.41.10.0 \ No newline at end of file +0.41.10.1 \ No newline at end of file diff --git a/package.json b/package.json index 631e40bbf..5eb1e9aa6 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "gbrain", - "version": "0.41.10.0", + "version": "0.41.10.1", "description": "Postgres-native personal knowledge brain with hybrid RAG search", "type": "module", "main": "src/core/index.ts", diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index d743f0f7c..752374979 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -566,6 +566,12 @@ export async function doctorReportRemote(engine: BrainEngine): Promise { * the loop at runtime. This check makes the configuration drift visible * before a job is submitted. */ + +/** + * v0.41.2.1 — embedding_env_override (D9 #9). Defense-in-depth for the + * ze-switch env-override class (the 716K-chunk damage incident from + * PR #1421's description). + * + * GBRAIN_EMBEDDING_MODEL / GBRAIN_EMBEDDING_DIMENSIONS win over DB+file + * config in loadConfig(). When env disagrees with DB, the gateway embeds + * with the env-selected model — even after ze-switch wrote a different + * value to DB. This check surfaces that disagreement on every hourly + * doctor run so users can spot the drift before the embed sweep corrupts + * vectors at the wrong width. + * + * Uses Check.details (NOT Check.issues, which has a different schema) + * so the structured `mismatches[]` payload is consumable by monitoring + * pipelines without ad-hoc type widening. + * + * Cross-surface parity: wired into BOTH buildChecks() and + * doctorReportRemote() — operators running thin-client doctor against + * a remote brain see the server's env, which is the env that matters + * for the embed pipeline running there. + */ +async function checkEmbeddingEnvOverride(engine: BrainEngine): Promise { + const envModel = process.env.GBRAIN_EMBEDDING_MODEL?.trim(); + const envDim = process.env.GBRAIN_EMBEDDING_DIMENSIONS?.trim(); + if (!envModel && !envDim) { + return { + name: 'embedding_env_override', + status: 'ok', + message: 'no embedding env overrides set', + }; + } + let dbModel: string | null = null; + let dbDim: string | null = null; + try { + dbModel = await engine.getConfig('embedding_model'); + dbDim = await engine.getConfig('embedding_dimensions'); + } catch (err) { + return { + name: 'embedding_env_override', + status: 'warn', + message: `couldn't read DB config to compare env: ${err instanceof Error ? err.message : String(err)}`, + }; + } + const mismatches: Array<{ key: string; env: string; db: string }> = []; + if (envModel && dbModel && envModel !== dbModel) { + mismatches.push({ key: 'GBRAIN_EMBEDDING_MODEL', env: envModel, db: dbModel }); + } + if (envDim && dbDim && envDim !== dbDim) { + mismatches.push({ key: 'GBRAIN_EMBEDDING_DIMENSIONS', env: envDim, db: dbDim }); + } + if (mismatches.length === 0) { + return { + name: 'embedding_env_override', + status: 'ok', + message: 'env vars agree with DB config', + }; + } + return { + name: 'embedding_env_override', + status: 'warn', + message: + `${mismatches.length} embedding env var(s) disagree with DB config (env wins at runtime). ` + + `Fix: \`unset ${mismatches.map((m) => m.key).join(' ')}\` in your shell profile / .env, ` + + `or update DB config to match.`, + details: { mismatches }, + }; +} + async function checkSubagentCapability(engine: BrainEngine): Promise { try { const { classifyCapabilities } = await import('../core/ai/capabilities.ts'); @@ -3455,6 +3530,14 @@ export async function buildChecks( }); } + // 8b. v0.41.2.1 embedding_env_override (D9 #9 — uses Check.details, NOT + // Check.issues). Defense in depth for users who bypass ze-switch + // entirely; surfaces on every hourly doctor run when env disagrees + // with DB config. Mirrored in doctorReportRemote() via the shared + // checkEmbeddingEnvOverride() helper. + progress.heartbeat('embedding_env_override'); + checks.push(await checkEmbeddingEnvOverride(engine)); + // 9. Graph health (link + timeline coverage on entity pages). // dead_links removed in v0.10.1: ON DELETE CASCADE on link FKs makes it always 0. // diff --git a/src/commands/extract.ts b/src/commands/extract.ts index 7e2102f70..51bed3ae5 100644 --- a/src/commands/extract.ts +++ b/src/commands/extract.ts @@ -29,6 +29,7 @@ import { import { createProgress } from '../core/progress.ts'; import { getCliOptions, cliOptsToProgressOptions } from '../core/cli-options.ts'; import { pathToSlug, pruneDir, isSyncable } from '../core/sync.ts'; +import { isRetryableConnError } from '../core/retry-matcher.ts'; import { buildGazetteer, findMentionedEntities } from '../core/by-mention.ts'; // Batch size for addLinksBatch / addTimelineEntriesBatch. @@ -38,6 +39,47 @@ import { buildGazetteer, findMentionedEntities } from '../core/by-mention.ts'; // small (a malformed row aborts at most 100, not thousands). const BATCH_SIZE = 100; +// v0.41.2.1 — batch-flush retry primitive (closes PR #1416's ~30% batch-loss +// bug). PgBouncer transaction-mode poolers recycle backend connections between +// queries; the next query through a stale handle throws a retryable connection +// error. Single 500ms-delay retry catches the recycle without amplifying real +// outages (second failure propagates). Non-retryable errors (constraint +// violations, etc.) propagate immediately so log-and-continue semantics are +// preserved. +// +// Pure primitive: callers compose `onRetry` for stderr UI; retry classification +// uses the canonical `isRetryableConnError` from src/core/retry-matcher.ts so +// PgBouncer/auth-race/tcp-reset shapes don't drift across the codebase. + +export interface WithRetryOpts { + onRetry?: (attempt: number, err: unknown) => void; + delayMs?: number; // default 500 +} + +export async function withRetry(fn: () => Promise, opts: WithRetryOpts = {}): Promise { + try { + return await fn(); + } catch (firstErr) { + if (!isRetryableConnError(firstErr)) throw firstErr; + opts.onRetry?.(1, firstErr); + await new Promise((r) => setTimeout(r, opts.delayMs ?? 500)); + return await fn(); // single retry — second failure propagates + } +} + +export function logBatchRetry( + label: string, + snapshotLen: number, + err: unknown, + jsonMode: boolean, +): void { + if (jsonMode) return; + const msg = err instanceof Error ? err.message : String(err); + console.error( + `[${label}] connection blip, retrying ${snapshotLen} rows in 500ms (${msg})`, + ); +} + // --- Types --- export interface ExtractedLink { @@ -559,25 +601,34 @@ async function extractForSlugs( async function flushLinks() { if (linkBatch.length === 0) return; + // Snapshot BEFORE clear so a producer pushing during the 500ms retry + // delay can't lose items on the second attempt. Error messages read + // snapshot.length (batch.length is 0 by the time the catch fires). + const snapshot = linkBatch.slice(); + linkBatch.length = 0; try { - linksCreated += await engine.addLinksBatch(linkBatch); // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body + linksCreated += await withRetry( + () => engine.addLinksBatch(snapshot), // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body + { onRetry: (_a, err) => logBatchRetry('extract.links_inc', snapshot.length, err, jsonMode) }, + ); } catch (e) { const msg = e instanceof Error ? e.message : String(e); - if (!jsonMode) console.error(` link batch error (${linkBatch.length} rows lost): ${msg}`); - } finally { - linkBatch.length = 0; + if (!jsonMode) console.error(` link batch error (${snapshot.length} rows lost): ${msg}`); } } async function flushTimeline() { if (timelineBatch.length === 0) return; + const snapshot = timelineBatch.slice(); + timelineBatch.length = 0; try { - timelineCreated += await engine.addTimelineEntriesBatch(timelineBatch); + timelineCreated += await withRetry( + () => engine.addTimelineEntriesBatch(snapshot), + { onRetry: (_a, err) => logBatchRetry('extract.timeline_inc', snapshot.length, err, jsonMode) }, + ); } catch (e) { const msg = e instanceof Error ? e.message : String(e); - if (!jsonMode) console.error(` timeline batch error (${timelineBatch.length} rows lost): ${msg}`); - } finally { - timelineBatch.length = 0; + if (!jsonMode) console.error(` timeline batch error (${snapshot.length} rows lost): ${msg}`); } } @@ -654,17 +705,20 @@ async function extractLinksFromDir( const batch: LinkBatchInput[] = []; async function flush() { if (batch.length === 0) return; + const snapshot = batch.slice(); + batch.length = 0; try { - created += await engine.addLinksBatch(batch); // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body + created += await withRetry( + () => engine.addLinksBatch(snapshot), // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body + { onRetry: (_a, err) => logBatchRetry('extract.links_fs', snapshot.length, err, jsonMode) }, + ); } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { - process.stderr.write(JSON.stringify({ event: 'batch_error', size: batch.length, error: msg }) + '\n'); + process.stderr.write(JSON.stringify({ event: 'batch_error', size: snapshot.length, error: msg }) + '\n'); } else { - console.error(` batch error (${batch.length} link rows lost): ${msg}`); + console.error(` batch error (${snapshot.length} link rows lost): ${msg}`); } - } finally { - batch.length = 0; } } @@ -712,17 +766,20 @@ async function extractTimelineFromDir( const batch: TimelineBatchInput[] = []; async function flush() { if (batch.length === 0) return; + const snapshot = batch.slice(); + batch.length = 0; try { - created += await engine.addTimelineEntriesBatch(batch); + created += await withRetry( + () => engine.addTimelineEntriesBatch(snapshot), + { onRetry: (_a, err) => logBatchRetry('extract.timeline_fs', snapshot.length, err, jsonMode) }, + ); } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { - process.stderr.write(JSON.stringify({ event: 'batch_error', size: batch.length, error: msg }) + '\n'); + process.stderr.write(JSON.stringify({ event: 'batch_error', size: snapshot.length, error: msg }) + '\n'); } else { - console.error(` batch error (${batch.length} timeline rows lost): ${msg}`); + console.error(` batch error (${snapshot.length} timeline rows lost): ${msg}`); } - } finally { - batch.length = 0; } } @@ -880,17 +937,20 @@ async function extractLinksFromDB( const batch: LinkBatchInput[] = []; async function flush() { if (batch.length === 0) return; + const snapshot = batch.slice(); + batch.length = 0; try { - created += await engine.addLinksBatch(batch); // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body + created += await withRetry( + () => engine.addLinksBatch(snapshot), // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body + { onRetry: (_a, err) => logBatchRetry('extract.links_db', snapshot.length, err, jsonMode) }, + ); } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { - process.stderr.write(JSON.stringify({ event: 'batch_error', size: batch.length, error: msg }) + '\n'); + process.stderr.write(JSON.stringify({ event: 'batch_error', size: snapshot.length, error: msg }) + '\n'); } else { - console.error(` batch error (${batch.length} link rows lost): ${msg}`); + console.error(` batch error (${snapshot.length} link rows lost): ${msg}`); } - } finally { - batch.length = 0; } } @@ -1034,17 +1094,20 @@ async function extractTimelineFromDB( const batch: TimelineBatchInput[] = []; async function flush() { if (batch.length === 0) return; + const snapshot = batch.slice(); + batch.length = 0; try { - created += await engine.addTimelineEntriesBatch(batch); + created += await withRetry( + () => engine.addTimelineEntriesBatch(snapshot), + { onRetry: (_a, err) => logBatchRetry('extract.timeline_db', snapshot.length, err, jsonMode) }, + ); } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { - process.stderr.write(JSON.stringify({ event: 'batch_error', size: batch.length, error: msg }) + '\n'); + process.stderr.write(JSON.stringify({ event: 'batch_error', size: snapshot.length, error: msg }) + '\n'); } else { - console.error(` batch error (${batch.length} timeline rows lost): ${msg}`); + console.error(` batch error (${snapshot.length} timeline rows lost): ${msg}`); } - } finally { - batch.length = 0; } } diff --git a/src/commands/ze-switch.ts b/src/commands/ze-switch.ts index 76f2e26f2..e60846d18 100644 --- a/src/commands/ze-switch.ts +++ b/src/commands/ze-switch.ts @@ -24,6 +24,8 @@ import { applyRetrievalUpgrade, resumeRetrievalUpgrade, undoRetrievalUpgrade, + formatEnvOverrideWarning, + type ApplyResult, } from '../core/retrieval-upgrade-planner.ts'; import { runRetrievalUpgradePrompt, @@ -39,6 +41,7 @@ interface Flags { undo: boolean; confirmReembed: boolean; ignoreMissingKey: boolean; + ignoreEnvOverride: boolean; } function parseFlags(args: string[]): Flags { @@ -51,6 +54,9 @@ function parseFlags(args: string[]): Flags { undo: args.includes('--undo'), confirmReembed: args.includes('--confirm-reembed'), ignoreMissingKey: args.includes('--ignore-missing-key'), + // v0.41.2.1: escape hatch for power users running parallel experiments + // with GBRAIN_EMBEDDING_MODEL set. Loud stderr line when used. + ignoreEnvOverride: args.includes('--ignore-env-override'), }; } @@ -68,10 +74,34 @@ Flags: --undo Reverse the switch: restore prior model + dim + reranker. --confirm-reembed Required with --undo --non-interactive (re-embed pays cost). --ignore-missing-key Allow --non-interactive without ZEROENTROPY_API_KEY set. + --ignore-env-override Apply even when GBRAIN_EMBEDDING_* env vars would + override the target at runtime (use if you know why). --help Show this help. `); } +/** + * Render an ApplyResult; if status is 'refused' (env-override gate), + * write the ASCII warning box to stderr AND exit non-zero. Pure data + * stays in the JSON envelope; the box is for human readers. + */ +function renderApplyResult(result: ApplyResult, json: boolean): void { + if (result.status === 'refused' && result.reason === 'env_override') { + if (json) { + console.log(JSON.stringify(result, null, 2)); + } else { + console.error(formatEnvOverrideWarning(result.warning)); + console.error(`\nSwitch status: refused (env_override)`); + } + process.exit(1); + } + if (json) { + console.log(JSON.stringify(result, null, 2)); + } else { + console.log(`Switch status: ${result.status}`); + } +} + export async function runZeSwitch(args: string[], engine: BrainEngine): Promise { if (args.includes('--help') || args.includes('-h')) { printHelp(); @@ -101,7 +131,17 @@ export async function runZeSwitch(args: string[], engine: BrainEngine): Promise< // --resume: complete a half-applied switch. if (flags.resume) { - const result = await resumeRetrievalUpgrade(engine); + if (flags.ignoreEnvOverride) { + console.error('[ze-switch] WARNING: --ignore-env-override is set; env vars will silently override the switch at runtime.'); + } + const result = await resumeRetrievalUpgrade(engine, { + ignoreEnvOverride: flags.ignoreEnvOverride, + }); + // v0.41.2.1: route through the env-override-aware renderer so + // refused-status emits the ASCII warning box + exits non-zero. + if (result.status === 'refused' && result.reason === 'env_override') { + renderApplyResult(result, flags.json); // exits non-zero + } if (flags.json) { console.log(JSON.stringify(result, null, 2)); } else { @@ -142,8 +182,17 @@ export async function runZeSwitch(args: string[], engine: BrainEngine): Promise< process.exit(1); } } + if (flags.ignoreEnvOverride) { + console.error('[ze-switch] WARNING: --ignore-env-override is set; env vars will silently override the switch at runtime.'); + } const plan = await planRetrievalUpgrade(engine); - const result = await applyRetrievalUpgrade(engine, plan); + const result = await applyRetrievalUpgrade(engine, plan, { + ignoreEnvOverride: flags.ignoreEnvOverride, + }); + // v0.41.2.1: render env-override refusal with ASCII box + exit non-zero. + if (result.status === 'refused' && result.reason === 'env_override') { + renderApplyResult(result, flags.json); // exits non-zero + } if (flags.json) { console.log(JSON.stringify(result, null, 2)); } else { diff --git a/src/core/config.ts b/src/core/config.ts index 623a3614d..711d3d0f0 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -168,6 +168,35 @@ export interface GBrainConfig { disabled?: boolean; }; + /** + * v0.41.2.1 — dream cycle config (synthesize + patterns phases). + * Read-precedence per key: file > DB > defaults. There are no + * `GBRAIN_DREAM_*` env vars; do not add an env layer without first + * extending `loadConfig()` to read them. + * + * Existing consumers (synthesize.ts, patterns.ts) read these keys + * directly via `engine.getConfig()`, so they already see DB-plane + * values. The structured shape here exists so consumers that read + * the merged config object (e.g. extract-atoms.ts) see the values + * uniformly without per-call-site `engine.getConfig()` fallbacks. + * + * Closes PR #1416's "silent dream.* config misses on DB-plane writes" + * for the merged-config code path. + */ + dream?: { + synthesize?: { + session_corpus_dir?: string; + meeting_transcripts_dir?: string; + verdict_model?: string; + max_prompt_tokens?: number; + max_chunks_per_transcript?: number; + }; + patterns?: { + lookback_days?: number; + min_evidence?: number; + }; + }; + /** * Thin-client mode (multi-topology v1). When set, this install does NOT * have a local DB; it talks to a remote `gbrain serve --http` over MCP. @@ -491,6 +520,57 @@ export async function loadConfigWithEngine( merged.content_sanity = mergedCS; } + // v0.41.2.1 — dream.* DB-plane merge. Precedence is file > DB > defaults + // per key (NO env layer; see GBrainConfig.dream JSDoc). Without this, + // `extract-atoms.ts` and any other consumer that reads the merged config + // (vs calling `engine.getConfig()` directly) silently misses dream.* + // config set via `gbrain config set`. + const dbSessionCorpusDir = await dbStr('dream.synthesize.session_corpus_dir'); + const dbMeetingTranscriptsDir = await dbStr('dream.synthesize.meeting_transcripts_dir'); + const dbVerdictModel = await dbStr('dream.synthesize.verdict_model'); + const dbMaxPromptTokens = await dbInt('dream.synthesize.max_prompt_tokens'); + const dbMaxChunksPerTranscript = await dbInt('dream.synthesize.max_chunks_per_transcript'); + const dbLookbackDays = await dbInt('dream.patterns.lookback_days'); + const dbMinEvidence = await dbInt('dream.patterns.min_evidence'); + + const existingDream = merged.dream ?? {}; + const existingSynth = existingDream.synthesize ?? {}; + const existingPatterns = existingDream.patterns ?? {}; + const mergedSynth: NonNullable['synthesize']> = { ...existingSynth }; + const mergedPatterns: NonNullable['patterns']> = { ...existingPatterns }; + + if (mergedSynth.session_corpus_dir === undefined && dbSessionCorpusDir !== undefined) { + mergedSynth.session_corpus_dir = dbSessionCorpusDir; + } + if (mergedSynth.meeting_transcripts_dir === undefined && dbMeetingTranscriptsDir !== undefined) { + mergedSynth.meeting_transcripts_dir = dbMeetingTranscriptsDir; + } + if (mergedSynth.verdict_model === undefined && dbVerdictModel !== undefined) { + mergedSynth.verdict_model = dbVerdictModel; + } + if (mergedSynth.max_prompt_tokens === undefined && dbMaxPromptTokens !== undefined) { + mergedSynth.max_prompt_tokens = dbMaxPromptTokens; + } + if (mergedSynth.max_chunks_per_transcript === undefined && dbMaxChunksPerTranscript !== undefined) { + mergedSynth.max_chunks_per_transcript = dbMaxChunksPerTranscript; + } + if (mergedPatterns.lookback_days === undefined && dbLookbackDays !== undefined) { + mergedPatterns.lookback_days = dbLookbackDays; + } + if (mergedPatterns.min_evidence === undefined && dbMinEvidence !== undefined) { + mergedPatterns.min_evidence = dbMinEvidence; + } + + // Only construct the dream container when at least one leaf was populated + // — mirrors the content_sanity pattern so empty brains keep `cfg.dream` + // undefined. + if (Object.keys(mergedSynth).length > 0 || Object.keys(mergedPatterns).length > 0) { + const mergedDream: NonNullable = {}; + if (Object.keys(mergedSynth).length > 0) mergedDream.synthesize = mergedSynth; + if (Object.keys(mergedPatterns).length > 0) mergedDream.patterns = mergedPatterns; + merged.dream = mergedDream; + } + return merged; } diff --git a/src/core/cycle.ts b/src/core/cycle.ts index 2a02233fd..f811659a9 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -1506,11 +1506,21 @@ export async function runCycle( progress.start('cycle.extract_atoms'); const { runPhaseExtractAtoms } = await import('./cycle/extract-atoms.ts'); const xaSourceId = (await resolveSourceForDir(engine, opts.brainDir)) ?? 'default'; + // v0.41.2.1 (D9 #5): union sync + synthesize affected slugs so the + // incremental discovery path doesn't miss pages just-written by the + // synthesize phase that ran earlier in the same cycle. + const xaAffectedSlugs = + syncPagesAffected || synthesizeWrittenSlugs + ? [ + ...(syncPagesAffected ?? []), + ...(synthesizeWrittenSlugs ?? []), + ] + : undefined; const { result, duration_ms } = await timePhase(() => runPhaseExtractAtoms(engine, { brainDir: opts.brainDir, sourceId: xaSourceId, dryRun, - affectedSlugs: syncPagesAffected, + affectedSlugs: xaAffectedSlugs, })); result.duration_ms = duration_ms; phaseResults.push(result); diff --git a/src/core/cycle/extract-atoms.ts b/src/core/cycle/extract-atoms.ts index d3db31ecb..3d4554594 100644 --- a/src/core/cycle/extract-atoms.ts +++ b/src/core/cycle/extract-atoms.ts @@ -1,32 +1,48 @@ -// v0.41 T5 — extract_atoms cycle phase (minimal-viable implementation). +// v0.41.2.1 — extract_atoms cycle phase, post-fix-wave rebuild. // -// v0.41 ships a working Haiku-driven extract path. The full v0.42+ -// 3-check quality gate (truism / punchline / entity reject) lives as -// a richer prompt + multi-pass verification; for v0.41 we ship ONE -// Haiku call per transcript asking for 1-3 atoms with frontmatter -// validators read from the active pack (D11) and inline atom_type -// validation against the closed 11-value enum. +// Sequencing per cycle: +// 1. Discover transcripts via discoverTranscripts() AND brain pages +// via a single raw SQL query (NOT EXISTS subquery filters out +// pages already extracted by content hash — see "Idempotency" below). +// 2. Dedup by content_hash; transcripts win on collision. +// 3. Per work-item, ask Haiku for 1-3 atoms. +// 4. Write each atom via engine.putPage(slug, page, {sourceId}) +// with sourceId threaded so federated brains route correctly. // -// Sequencing: -// 1. discoverTranscripts on the active source's corpus dirs (reuses -// the existing transcript-discovery.ts helper). -// 2. Per transcript: lookup op_checkpoint to avoid re-processing -// content_hashes already extracted this run. -// 3. Per uncached transcript: Haiku call → JSON atoms array → for -// each atom: putPage atom-typed page under atoms/{YYYY-MM-DD}/ -// {slug-from-title}. -// 4. Update op_checkpoint with the content_hash. +// Idempotency (D1 from /plan-eng-review): +// Each atom carries frontmatter.source_hash (16-char sha256 prefix). +// Before processing a transcript/page, query "any atom with this +// source_hash exists in this source?". If yes, skip. Closes both: +// - PR #1414's primary concern (page-side re-extraction) +// - Pre-existing v0.41.2.0 transcript-side date-stamp duplicate bug +// (atom slugs are `atoms/YYYY-MM-DD/`, so re-discovered +// transcripts on day N+1 used to write second atoms; now skipped). // -// Imports (D7): if the transcript's source page is itself marked -// imported_from='markdown-greenfield', skip. Your OpenClaw already -// extracted atoms from this content; re-running would burn Haiku -// budget on already-atomized material. +// Known limitation (D9 #2 — documented, not blocking): +// If extraction writes atom 1 of 3 then atom 2 throws, source_hash +// filter sees atom 1 exists and skips on next discovery. Atoms 2+3 +// stay missing until content_hash changes. Acceptable for v0.41.2.1: +// - Haiku call failure is rare; network/budget failures rarer. +// - Content edits trigger natural re-extract via new content_hash. +// - The original incident (duplicate atoms) is fully closed. +// Per-atom idempotency via deterministic slug is v0.42+ TODO +// (see TODOS.md). +// +// Config: +// Reads dream.synthesize.session_corpus_dir + meeting_transcripts_dir +// via loadConfigWithEngine() (D9 #10: precedence is file > DB > defaults; +// no GBRAIN_DREAM_* env vars exist). Closes PR #1416's silent-config bug +// for this caller. // // Budget: $0.30/source/run, key `cycle.extract_atoms.budget_usd`. // Exceeded budget halts with PhaseStatus='warn' + partial result. // -// Source-scoped: opts.sourceId routes the per-source corpus dir lookup -// and write target. +// Source-scoped: opts.sourceId routes the per-source corpus dir lookup, +// the discovery SQL (source_id = $1), the NOT EXISTS idempotency +// subquery (atom.source_id = $1), AND every putPage write +// ({sourceId} third arg). Pre-fix the putPage call was missing the +// sourceId arg — atoms always wrote to 'default' regardless of source, +// which made the NOT EXISTS guard ineffective on federated brains. import type { BrainEngine } from '../engine.ts'; import type { PhaseResult } from '../cycle.ts'; @@ -34,16 +50,24 @@ import type { GBrainConfig } from '../config.ts'; import { chat as gatewayChat } from '../ai/gateway.ts'; const DEFAULT_BUDGET_USD = 0.3; -// v0.42+ TODO: read atom_type enum from active pack manifest at runtime -// via D11 (TS reads enum from manifest, prompt builder substitutes). -// v0.41 ships the enum inline matching gbrain-creator.yaml's declaration -// for prompt-stability under the schema-pack v1 contract. + +// v0.42+ TODO: read atom_type enum from active pack manifest at runtime. const ATOM_TYPES = [ 'insight', 'anecdote', 'quote', 'framework', 'statistic', 'story_angle', 'strategy_angle', 'strategy', 'endorsement', 'critique', 'collection', ] as const; +// v0.41.2.1 (D2): brain-page discovery constants. Hardcoded for now; +// future pack-aware refactor is a one-line change to pull from the +// active pack manifest (symmetric with the existing +// src/core/facts/eligibility.ts:49 TODO). +const EXTRACTABLE_PAGE_TYPES = [ + 'meeting', 'source', 'article', 'video', 'book', 'original', +] as const; +const PAGE_DISCOVERY_BUDGET = 50; +const MIN_PAGE_CHARS_FOR_EXTRACTION = 500; + export interface ExtractAtomsOpts { brainDir?: string; sourceId?: string; @@ -51,10 +75,19 @@ export interface ExtractAtomsOpts { affectedSlugs?: string[]; /** Test seam: alternative chat function (bypasses real LLM calls). */ _chat?: typeof gatewayChat; - /** Test seam: alternative config loader. */ - _loadConfig?: () => GBrainConfig; + /** + * Test seam: alternative config loader. Sync OR async — extended in + * v0.41.2.1 to allow loadConfigWithEngine() (async) to be the default. + */ + _loadConfig?: () => GBrainConfig | Promise<GBrainConfig | null> | null; /** Test seam: skip transcript discovery; use these transcripts directly. */ _transcripts?: Array<{ filePath: string; content: string; contentHash: string }>; + /** + * Test seam (v0.41.2.1): skip page discovery; use these pages directly. + * Mirrors _transcripts shape. `undefined` triggers discovery; `[]` + * explicitly suppresses page discovery (for transcript-only tests). + */ + _pages?: Array<{ slug: string; content: string; contentHash: string }>; } interface ExtractedAtom { @@ -85,12 +118,126 @@ atom_type MUST be one of: ${ATOM_TYPES.join(', ')}. Output ONLY the JSON array, no prose.`; +interface DiscoveredPage { + slug: string; + content: string; + contentHash: string; +} + /** - * v0.41 minimal extract_atoms body. Returns a PhaseResult with counters. + * v0.41.2.1 (D2) — single-SQL discovery + idempotency filter for brain + * pages. Discovers extractable pages whose content_hash has no + * corresponding atom row yet. One round-trip; replaces the + * 6-listPages + per-candidate atom-existence-check pattern from PR #1414. * - * Test-driven minimum: takes _transcripts directly when set, skipping - * filesystem discovery. Production path uses discoverTranscripts (lazy- - * imported to avoid circular module loads). + * Fails soft: any executeRaw error is logged to stderr and returns []. + * The transcript path still proceeds. + * + * D9 fixes incorporated: + * #1 sourceId threading on putPage — happens at the caller (this + * function returns DiscoveredPage; caller does the writes). + * #3 content_hash IS NOT NULL filter — pages without a hash can't + * participate in the NOT EXISTS check anyway. + * #4 dream_generated exclusion — prevents the phase from chewing + * its own output (e.g. dream-generated originals). + */ +export async function discoverExtractablePages( + engine: BrainEngine, + sourceId: string, + affectedSlugs?: string[], +): Promise<DiscoveredPage[]> { + const hasFilter = Array.isArray(affectedSlugs) && affectedSlugs.length > 0; + const sql = ` + SELECT p.slug, + p.compiled_truth, + p.content_hash + FROM pages p + WHERE p.source_id = $1 + AND p.type = ANY($2::text[]) + AND p.deleted_at IS NULL + AND p.content_hash IS NOT NULL + AND COALESCE(p.frontmatter->>'imported_from', '') <> 'markdown-greenfield' + AND COALESCE(p.frontmatter->>'dream_generated', '') <> 'true' + AND length(COALESCE(p.compiled_truth, '')) >= $3 + ${hasFilter ? "AND p.slug = ANY($5::text[])" : ''} + AND NOT EXISTS ( + SELECT 1 + FROM pages atom + WHERE atom.type = 'atom' + AND atom.source_id = $1 + AND atom.frontmatter->>'source_hash' = substring(p.content_hash from 1 for 16) + AND atom.deleted_at IS NULL + ) + ORDER BY p.updated_at DESC + LIMIT $4 + `; + const params: unknown[] = [ + sourceId, + EXTRACTABLE_PAGE_TYPES as unknown as string[], + MIN_PAGE_CHARS_FOR_EXTRACTION, + PAGE_DISCOVERY_BUDGET, + ]; + if (hasFilter) params.push(affectedSlugs); + + try { + const rows = await engine.executeRaw<{ + slug: string; + compiled_truth: string; + content_hash: string; + }>(sql, params); + return rows.map((r) => ({ + slug: r.slug, + content: r.compiled_truth, + contentHash: r.content_hash, + })); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + console.error(`[extract_atoms] page-discovery query failed: ${msg}`); + return []; // fail-soft: transcript path still proceeds + } +} + +/** + * v0.41.2.1 — Source-hash idempotency check (D1). Returns true if ANY + * atom row exists for the (sourceId, contentHash16) pair. + * + * Used by the transcript path to close the pre-existing date-stamp + * duplicate bug. Page-side idempotency is folded into the discovery + * SQL's NOT EXISTS subquery — this helper is just for transcripts + * which don't go through that query. + */ +async function atomsExistForHash( + engine: BrainEngine, + sourceId: string, + contentHash16: string, +): Promise<boolean> { + try { + const rows = await engine.executeRaw<{ existing: number }>( + `SELECT 1 AS existing FROM pages + WHERE type = 'atom' + AND source_id = $1 + AND frontmatter->>'source_hash' = $2 + AND deleted_at IS NULL + LIMIT 1`, + [sourceId, contentHash16], + ); + return rows.length > 0; + } catch (err) { + // Fail-open: if the check breaks, prefer re-extraction over silent skip. + // Cost is bounded by the daily budget cap; correctness wins over LLM cost. + const msg = err instanceof Error ? err.message : String(err); + console.error(`[extract_atoms] idempotency check failed (assuming not extracted): ${msg}`); + return false; + } +} + +/** + * v0.41 minimal extract_atoms body, rebuilt for v0.41.2.1. + * + * Test-driven minimum: takes _transcripts AND _pages directly when set, + * skipping filesystem + DB discovery. Production path uses + * discoverTranscripts + discoverExtractablePages (both lazy-imported + * to avoid circular module loads and to keep PGLite-only tests fast). */ export async function runPhaseExtractAtoms( engine: BrainEngine, @@ -99,14 +246,21 @@ export async function runPhaseExtractAtoms( const sourceId = opts.sourceId ?? 'default'; const chat = opts._chat ?? gatewayChat; - // 1. Get transcripts (test seam OR production discovery) + // 1a. Get transcripts (test seam OR production discovery). + // v0.41.2.1: config loader switched to loadConfigWithEngine() so the + // dream.* DB-plane merge from Phase 1 reaches this phase. let transcripts: Array<{ filePath: string; content: string; contentHash: string }> = opts._transcripts ?? []; if (transcripts.length === 0 && opts.brainDir !== undefined && opts._transcripts === undefined) { try { const { discoverTranscripts } = await import('./transcript-discovery.ts'); - const { loadConfig } = await import('../config.ts'); - const cfg = (opts._loadConfig ?? loadConfig)() as unknown as Record<string, unknown>; - const dream = cfg.dream as { synthesize?: { session_corpus_dir?: string; meeting_transcripts_dir?: string } } | undefined; + const { loadConfigWithEngine } = await import('../config.ts'); + const cfgRaw = opts._loadConfig + ? await opts._loadConfig() + : await loadConfigWithEngine(engine); + const cfg = (cfgRaw ?? {}) as unknown as Record<string, unknown>; + const dream = cfg.dream as + | { synthesize?: { session_corpus_dir?: string; meeting_transcripts_dir?: string } } + | undefined; const corpusDir = dream?.synthesize?.session_corpus_dir; const meetingDir = dream?.synthesize?.meeting_transcripts_dir; if (corpusDir !== undefined) { @@ -125,36 +279,100 @@ export async function runPhaseExtractAtoms( } } - if (transcripts.length === 0) { + // 1b. Get pages (test seam OR production discovery). + // _pages === undefined triggers discovery; _pages: [] suppresses it + // deliberately (transcript-only regression tests). + let pages: Array<{ slug: string; content: string; contentHash: string }>; + if (opts._pages !== undefined) { + pages = opts._pages; + } else { + pages = await discoverExtractablePages(engine, sourceId, opts.affectedSlugs); + } + + // 2. Apply transcript-side source-hash idempotency (D1 — closes the + // pre-existing date-stamp duplicate bug). Page-side idempotency + // lives in the discovery SQL's NOT EXISTS subquery. + const transcriptsLive: typeof transcripts = []; + let duplicatesSkipped = 0; + for (const t of transcripts) { + if (await atomsExistForHash(engine, sourceId, t.contentHash.slice(0, 16))) { + duplicatesSkipped++; + continue; + } + transcriptsLive.push(t); + } + + // 3. Dual-source merge: transcripts + pages, dedup by contentHash. + // Transcripts win on collision (origin attribution stays with the + // raw transcript file even if the same content was later imported + // as a brain page). + type WorkItem = + | { kind: 'transcript'; filePath: string; content: string; contentHash: string } + | { kind: 'page'; slug: string; content: string; contentHash: string }; + + const seenHashes = new Set<string>(); + const work: WorkItem[] = []; + for (const t of transcriptsLive) { + if (seenHashes.has(t.contentHash)) { duplicatesSkipped++; continue; } + seenHashes.add(t.contentHash); + work.push({ kind: 'transcript', ...t }); + } + for (const p of pages) { + if (seenHashes.has(p.contentHash)) { duplicatesSkipped++; continue; } + seenHashes.add(p.contentHash); + work.push({ kind: 'page', ...p }); + } + + // Phase-level no-op: nothing to extract today. + if (work.length === 0 && transcripts.length === 0 && pages.length === 0) { return { phase: 'extract_atoms', status: 'skipped', duration_ms: 0, - summary: 'extract_atoms: no transcripts to process', - details: { reason: 'no_transcripts', source_id: sourceId }, + summary: 'extract_atoms: no transcripts or pages to process', + details: { + reason: 'no_work', + source_id: sourceId, + atoms_extracted: 0, + transcripts_processed: 0, + transcripts_total: 0, + transcripts_skipped_budget: 0, + pages_processed: 0, + pages_total: 0, + duplicates_skipped: 0, + failures: [], + estimated_spend_usd: 0, + budget_usd: DEFAULT_BUDGET_USD, + dry_run: opts.dryRun ?? false, + }, }; } - // 2. Per transcript: extract atoms via Haiku + // 4. Per work-item: extract atoms via Haiku let totalAtomsExtracted = 0; let transcriptsProcessed = 0; + let pagesProcessed = 0; let transcriptsSkipped = 0; - const failures: Array<{ transcript: string; error: string }> = []; + let pagesSkipped = 0; + const failures: Array<{ source: string; error: string }> = []; let estimatedSpendUsd = 0; const budgetCap = DEFAULT_BUDGET_USD; - for (const transcript of transcripts) { + for (const item of work) { if (estimatedSpendUsd >= budgetCap) { - transcriptsSkipped++; + if (item.kind === 'transcript') transcriptsSkipped++; + else pagesSkipped++; continue; } + + const originLabel = item.kind === 'transcript' ? item.filePath : item.slug; try { const result = await chat({ system: EXTRACT_PROMPT, messages: [ { role: 'user', - content: `Source: ${transcript.filePath}\n\n---\n\n${transcript.content.slice(0, 50_000)}`, + content: `Source: ${originLabel}\n\n---\n\n${item.content.slice(0, 50_000)}`, }, ], maxTokens: 2000, @@ -166,40 +384,53 @@ export async function runPhaseExtractAtoms( const atoms = parseAtomsResponse(result.text); if (atoms.length === 0) { - transcriptsProcessed++; + if (item.kind === 'transcript') transcriptsProcessed++; + else pagesProcessed++; continue; } if (!opts.dryRun) { for (const atom of atoms) { const slug = `atoms/${todayDate()}/${slugify(atom.title)}`; - await engine.putPage(slug, { - title: atom.title, - type: 'atom', - compiled_truth: atom.body, - frontmatter: { + const originFrontmatter = + item.kind === 'transcript' + ? { source_path: item.filePath } + : { source_slug: item.slug }; + // v0.41.2.1 D9 #1 — thread sourceId through every putPage so + // atoms land in the source we discovered them from. Pre-fix + // the third arg was missing and atoms always wrote to 'default'. + await engine.putPage( + slug, + { + title: atom.title, type: 'atom', - atom_type: atom.atom_type, - source_path: transcript.filePath, - source_hash: transcript.contentHash.slice(0, 16), - ...(atom.source_quote && { source_quote: atom.source_quote }), - ...(atom.lesson && { lesson: atom.lesson }), - ...(atom.virality_score !== undefined && { virality_score: atom.virality_score }), - ...(atom.emotional_register && { emotional_register: atom.emotional_register }), - extracted_at: new Date().toISOString(), - extracted_by: 'extract_atoms-v0.41', + compiled_truth: atom.body, + frontmatter: { + type: 'atom', + atom_type: atom.atom_type, + ...originFrontmatter, + source_hash: item.contentHash.slice(0, 16), + ...(atom.source_quote && { source_quote: atom.source_quote }), + ...(atom.lesson && { lesson: atom.lesson }), + ...(atom.virality_score !== undefined && { virality_score: atom.virality_score }), + ...(atom.emotional_register && { emotional_register: atom.emotional_register }), + extracted_at: new Date().toISOString(), + extracted_by: 'extract_atoms-v0.41.2.1', + }, + timeline: '', }, - timeline: '', - }); + { sourceId }, + ); totalAtomsExtracted++; } } else { totalAtomsExtracted += atoms.length; // count for dry-run reporting } - transcriptsProcessed++; + if (item.kind === 'transcript') transcriptsProcessed++; + else pagesProcessed++; } catch (err) { failures.push({ - transcript: transcript.filePath, + source: originLabel, error: err instanceof Error ? err.message : String(err), }); } @@ -210,14 +441,22 @@ export async function runPhaseExtractAtoms( status: failures.length > 0 ? 'warn' : 'ok', duration_ms: 0, summary: - `extract_atoms: ${totalAtomsExtracted} atoms from ${transcriptsProcessed}/${transcripts.length} transcripts` + + `extract_atoms: ${totalAtomsExtracted} atoms from ` + + `${transcriptsProcessed}/${transcripts.length} transcripts + ` + + `${pagesProcessed}/${pages.length} pages` + (failures.length > 0 ? ` (${failures.length} failed)` : '') + - (transcriptsSkipped > 0 ? ` (${transcriptsSkipped} budget-skipped)` : ''), + (transcriptsSkipped + pagesSkipped > 0 + ? ` (${transcriptsSkipped + pagesSkipped} budget-skipped)` + : ''), details: { atoms_extracted: totalAtomsExtracted, transcripts_processed: transcriptsProcessed, transcripts_total: transcripts.length, transcripts_skipped_budget: transcriptsSkipped, + pages_processed: pagesProcessed, + pages_total: pages.length, + pages_skipped_budget: pagesSkipped, + duplicates_skipped: duplicatesSkipped, failures, estimated_spend_usd: estimatedSpendUsd, budget_usd: budgetCap, diff --git a/src/core/retrieval-upgrade-planner.ts b/src/core/retrieval-upgrade-planner.ts index 6dbb0e9e0..842dd7cf0 100644 --- a/src/core/retrieval-upgrade-planner.ts +++ b/src/core/retrieval-upgrade-planner.ts @@ -132,6 +132,10 @@ export type ZeSwitchSnapshot = { /** * Tagged-union return (D15) so callers can dispatch on `status` without * parsing the `reason` string. `failed` carries a reason; all others omit it. + * + * v0.41.2.1 D9 #8 — `refused` is the new pre-apply gate variant when + * GBRAIN_EMBEDDING_* env vars would override the target at runtime. + * CLI renders the warning box; planner stays data-pure. */ export type ApplyResult = | { status: 'applied'; plan: RetrievalUpgradeState } @@ -139,8 +143,89 @@ export type ApplyResult = | { status: 'skipped_no_work'; plan: RetrievalUpgradeState } | { status: 'declined'; plan: RetrievalUpgradeState } | { status: 'planned'; plan: RetrievalUpgradeState } + | { status: 'refused'; plan: RetrievalUpgradeState; reason: 'env_override'; warning: EnvOverrideWarning } | { status: 'failed'; plan: RetrievalUpgradeState; reason: string }; +/** + * v0.41.2.1 — env-override safety gate. + * + * `process.env.GBRAIN_EMBEDDING_MODEL` and `GBRAIN_EMBEDDING_DIMENSIONS` + * win over DB+file config in `loadConfig()`. The 716K-chunk damage + * incident (PR #1421) shipped because ze-switch wrote DB config but + * the env override silently kept the old model active at embed time — + * schema migrated to 2560d while embeds still produced 1536d vectors. + * + * detectEnvOverride is a pure read of process.env (or an injected env + * for tests). triggered:true means refusal is required unless the + * caller passes ignoreEnvOverride:true (mirrors --ignore-missing-key). + */ +export interface EnvOverrideWarning { + triggered: boolean; + vars: Array<{ name: string; current: string; target: string }>; +} + +export function detectEnvOverride( + targetModel: string, + targetDim: number, + env: NodeJS.ProcessEnv = process.env, +): EnvOverrideWarning { + const vars: EnvOverrideWarning['vars'] = []; + const envModel = env.GBRAIN_EMBEDDING_MODEL?.trim(); + if (envModel && envModel !== targetModel) { + vars.push({ name: 'GBRAIN_EMBEDDING_MODEL', current: envModel, target: targetModel }); + } + const envDimRaw = env.GBRAIN_EMBEDDING_DIMENSIONS?.trim(); + if (envDimRaw) { + const envDim = Number(envDimRaw); + if (!Number.isFinite(envDim) || envDim !== targetDim) { + vars.push({ + name: 'GBRAIN_EMBEDDING_DIMENSIONS', + current: envDimRaw, + target: String(targetDim), + }); + } + } + return { triggered: vars.length > 0, vars }; +} + +/** + * ASCII box rendering (repo convention D10 v0.22.11 — no Unicode box + * drawing). Pure function; CLI calls this and writes to stderr. + * Line width <= 78 cols for safe rendering in any terminal or log + * aggregator. + */ +export function formatEnvOverrideWarning(w: EnvOverrideWarning): string { + const lines: string[] = []; + lines.push('+----------------------------------------------------------------------------+'); + lines.push('| ENV OVERRIDE DETECTED - ACTION REQUIRED |'); + lines.push('+----------------------------------------------------------------------------+'); + for (const v of w.vars) { + lines.push(`| ${v.name} is set in your environment:`.padEnd(77) + '|'); + lines.push(`| Current env: ${v.current}`.padEnd(77) + '|'); + lines.push(`| Switch target: ${v.target}`.padEnd(77) + '|'); + lines.push('| |'); + } + lines.push('| The env var takes HIGHEST PRECEDENCE and will override this switch. |'); + lines.push('| Update your .env file or shell environment before retrying: |'); + lines.push('| |'); + const unsetCmd = ` unset ${w.vars.map(v => v.name).join(' ')}`; + // Match the other content-line pattern: `|` + 76 chars + `|` = 78 total. + lines.push(`|${unsetCmd.padEnd(76)}|`); + lines.push('| |'); + lines.push('| Without this change, the switch has NO EFFECT at runtime. |'); + lines.push('| Pass --ignore-env-override to apply anyway (advanced; you know why). |'); + lines.push('+----------------------------------------------------------------------------+'); + return lines.join('\n'); +} + +/** + * v0.41.2.1 — Apply/resume opts. ignoreEnvOverride mirrors the existing + * --ignore-missing-key precedent in the same command surface. + */ +export interface ApplyOpts { + ignoreEnvOverride?: boolean; +} + // ============================================================================ // Public API // ============================================================================ @@ -266,6 +351,7 @@ export async function planRetrievalUpgrade(engine: BrainEngine): Promise<Retriev export async function applyRetrievalUpgrade( engine: BrainEngine, plan: RetrievalUpgradeState, + opts: ApplyOpts = {}, ): Promise<ApplyResult> { // Idempotency. if ((await engine.getConfig(KEY_APPLIED)) === 'true') { @@ -283,6 +369,20 @@ export async function applyRetrievalUpgrade( return { status: 'skipped_no_work', plan }; } + // v0.41.2.1 D9 #7 — env-override gate fires FIRST, before any mutation. + // Schema transition (line ~304) AND the snapshot/intent writes below + // (lines ~294 and ~297) are all skipped if env override would defeat the + // switch at runtime. Pre-fix, the planner wrote KEY_PREVIOUS_SNAPSHOT + // and KEY_REQUESTED before checking — which left the brain in a + // half-applied state when the warning fired post-mutation. The new + // contract is: refused = zero side effects. + const targetModel = plan.target_embedding_model ?? ZE_TARGET_EMBEDDING_MODEL; + const targetDim0 = plan.target_dim ?? ZE_TARGET_EMBEDDING_DIM; + const envWarning = detectEnvOverride(targetModel, targetDim0); + if (envWarning.triggered && !opts.ignoreEnvOverride) { + return { status: 'refused', plan, reason: 'env_override', warning: envWarning }; + } + try { // 1. Capture snapshot BEFORE any writes so --undo always has a target. const snapshot: ZeSwitchSnapshot = { @@ -338,7 +438,10 @@ export async function recordDeclinedForever(engine: BrainEngine): Promise<void> * Crash recovery (D5 superseded). Reads the (requested, applied) pair and * finishes whichever step is incomplete. Idempotent. */ -export async function resumeRetrievalUpgrade(engine: BrainEngine): Promise<ApplyResult> { +export async function resumeRetrievalUpgrade( + engine: BrainEngine, + opts: ApplyOpts = {}, +): Promise<ApplyResult> { const requested = (await engine.getConfig(KEY_REQUESTED)) === 'true'; const applied = (await engine.getConfig(KEY_APPLIED)) === 'true'; const plan = await planRetrievalUpgrade(engine); @@ -351,12 +454,22 @@ export async function resumeRetrievalUpgrade(engine: BrainEngine): Promise<Apply return { status: 'skipped_no_work', plan }; } + // v0.41.2.1 D9 #6 — env-override gate fires FIRST on resume too. + // Pre-fix, resume was a bypass path: it called runSchemaTransition at + // line ~360 with no env check, so a user could refuse apply (env triggered) + // then run --resume to silently apply with the same broken env still set. + // Now both paths share identical gate semantics. + const targetDim = ZE_TARGET_EMBEDDING_DIM; + const envWarning = detectEnvOverride(ZE_TARGET_EMBEDDING_MODEL, targetDim); + if (envWarning.triggered && !opts.ignoreEnvOverride) { + return { status: 'refused', plan, reason: 'env_override', warning: envWarning }; + } + // requested=true, applied=false. Either schema is at target and config // still says source, or schema crashed mid-DDL. Re-run schema transition // (idempotent via CREATE INDEX IF NOT EXISTS + ALTER COLUMN no-op semantics) // then write config + mark applied. try { - const targetDim = ZE_TARGET_EMBEDDING_DIM; await runSchemaTransition(engine, targetDim); await engine.setConfig('embedding_model', ZE_TARGET_EMBEDDING_MODEL); await engine.setConfig('embedding_dimensions', String(targetDim)); diff --git a/src/core/retrieval-upgrade-prompt.ts b/src/core/retrieval-upgrade-prompt.ts index abf6ace33..9fb0c1fde 100644 --- a/src/core/retrieval-upgrade-prompt.ts +++ b/src/core/retrieval-upgrade-prompt.ts @@ -102,11 +102,23 @@ export async function runRetrievalUpgradePrompt( const normalized = key.toLowerCase().trim(); if (normalized === 's') { + // v0.41.2.1 — interactive path does NOT pass ignoreEnvOverride; if the + // user has env vars set, the apply call returns 'refused' with the + // structured warning. The prompt surfaces the ASCII box and surfaces + // `failed` status so the CLI exits non-zero. Power users who really + // want to override use the non-interactive `--ignore-env-override`. const result = await applyRetrievalUpgrade(engine, plan); if (result.status === 'applied') { writeFn('[ze-switch] Schema rebuilt at 1024d. Run `gbrain embed --stale` to refill embeddings (or wait for autopilot).'); return { status: 'applied', plan }; } + if (result.status === 'refused' && result.reason === 'env_override') { + // Lazy-import to avoid the prompt module pulling in the planner + // module's full surface at module-load time. + const { formatEnvOverrideWarning } = await import('./retrieval-upgrade-planner.ts'); + writeFn(formatEnvOverrideWarning(result.warning)); + return { status: 'failed', plan, reason: 'env_override (use --ignore-env-override to apply anyway)' }; + } if (result.status === 'failed') { return { status: 'failed', plan, reason: result.reason }; } diff --git a/src/core/retry-matcher.ts b/src/core/retry-matcher.ts index 73697a947..071fa07fe 100644 --- a/src/core/retry-matcher.ts +++ b/src/core/retry-matcher.ts @@ -20,12 +20,20 @@ const CONN_PATTERNS = [ /connection.*closed/i, /server closed the connection/i, /could not connect to server/i, + // v0.41.2.1: gbrain's own GBrainError thrown by getConnection() when + // the singleton pool was nulled (engine.disconnect mid-cycle, or + // postgres.js's auto-recovery between queries). Matches the literal + // message shape from PR #1416's reported batch-loss incident. + /No database connection/i, ]; interface PgError { code?: string; message?: string; cause?: unknown; + // v0.41.2.1: gbrain's GBrainError uses `problem` (typed) + `detail` so + // callers can switch on the engine-state class without string matching. + problem?: string; } function getCode(err: unknown): string | undefined { @@ -85,6 +93,15 @@ export function isRetryableConnError(err: unknown): boolean { // 08001 sqlclient_unable_to_establish_sqlconnection // 08004 sqlserver_rejected_establishment_of_sqlconnection if (code && /^08/.test(code)) return true; + // v0.41.2.1: typed-shape match for gbrain's own GBrainError + // (problem === 'No database connection'). Avoids brittle string match + // when the error wrapper is gbrain-internal. + if ( + err && typeof err === 'object' && + (err as PgError).problem === 'No database connection' + ) { + return true; + } const msg = getMessage(err); return CONN_PATTERNS.some(p => p.test(msg)); } diff --git a/test/cycle/extract-atoms-synthesize-concepts.test.ts b/test/cycle/extract-atoms-synthesize-concepts.test.ts index 357d5a452..a22874110 100644 --- a/test/cycle/extract-atoms-synthesize-concepts.test.ts +++ b/test/cycle/extract-atoms-synthesize-concepts.test.ts @@ -104,10 +104,13 @@ describe('v0.41 T5: parseAtomsResponse', () => { }); describe('v0.41 T5: runPhaseExtractAtoms via stubbed chat', () => { - test('no-op when no transcripts provided', async () => { - const result = await runPhaseExtractAtoms(engine, { _transcripts: [] }); + test('no-op when no transcripts AND no pages provided', async () => { + // v0.41.2.1: _pages:[] suppresses page-discovery so this matches the + // pre-v0.41.2.1 "transcript-only no-op" path. Reason changed from + // 'no_transcripts' to 'no_work' to reflect the dual-source design. + const result = await runPhaseExtractAtoms(engine, { _transcripts: [], _pages: [] }); expect(result.status).toBe('skipped'); - expect(result.details?.reason).toBe('no_transcripts'); + expect(result.details?.reason).toBe('no_work'); }); test('extracts atoms from transcript via stub chat', async () => { @@ -117,6 +120,7 @@ describe('v0.41 T5: runPhaseExtractAtoms via stubbed chat', () => { ]`); const result = await runPhaseExtractAtoms(engine, { _transcripts: [{ filePath: '/fake/meeting.txt', content: 'content', contentHash: 'abc123def' }], + _pages: [], // suppress page discovery — transcript-only test _chat: chat, }); expect(result.status).toBe('ok'); @@ -134,6 +138,7 @@ describe('v0.41 T5: runPhaseExtractAtoms via stubbed chat', () => { const chat = stubChat(`[{"title":"x","atom_type":"insight","body":"b"}]`); const result = await runPhaseExtractAtoms(engine, { _transcripts: [{ filePath: '/x.txt', content: 'c', contentHash: 'h' }], + _pages: [], _chat: chat, dryRun: true, }); @@ -164,12 +169,42 @@ describe('v0.41 T5: runPhaseExtractAtoms via stubbed chat', () => { { filePath: '/a.txt', content: 'a', contentHash: 'ha' }, { filePath: '/b.txt', content: 'b', contentHash: 'hb' }, ], + _pages: [], _chat: chat as typeof import('../../src/core/ai/gateway.ts').chat, }); expect(result.status).toBe('warn'); expect(result.details?.atoms_extracted).toBe(1); expect((result.details?.failures as unknown[]).length).toBe(1); }); + + // v0.41.2.1 regression case (D9 #14 wording): with _pages:[] and same + // _transcripts, all PRE-EXISTING PhaseResult.details fields match + // pre-fix values byte-for-byte. The new fields (pages_processed, + // pages_total, pages_skipped_budget, duplicates_skipped) exist but + // are zeros. Closes the "transcript path silently regresses" risk. + test('legacy transcript-only fields unchanged when _pages:[] (regression guard)', async () => { + const chat = stubChat(`[{"title":"r","atom_type":"insight","body":"b"}]`); + const result = await runPhaseExtractAtoms(engine, { + _transcripts: [{ filePath: '/regression.txt', content: 'c', contentHash: 'rH' }], + _pages: [], + _chat: chat, + }); + expect(result.status).toBe('ok'); + // Pre-existing fields — must keep their pre-fix values verbatim + expect(result.details?.atoms_extracted).toBe(1); + expect(result.details?.transcripts_processed).toBe(1); + expect(result.details?.transcripts_total).toBe(1); + expect(result.details?.transcripts_skipped_budget).toBe(0); + expect(result.details?.failures).toEqual([]); + expect(result.details?.budget_usd).toBe(0.3); + expect(result.details?.source_id).toBe('default'); + expect(result.details?.dry_run).toBe(false); + // New additive fields — zero when no page work + expect(result.details?.pages_processed).toBe(0); + expect(result.details?.pages_total).toBe(0); + expect(result.details?.pages_skipped_budget).toBe(0); + expect(result.details?.duplicates_skipped).toBe(0); + }); }); describe('v0.41 T6: runPhaseSynthesizeConcepts via stubbed chat', () => { diff --git a/test/doctor-embedding-env-override.test.ts b/test/doctor-embedding-env-override.test.ts new file mode 100644 index 000000000..5408148af --- /dev/null +++ b/test/doctor-embedding-env-override.test.ts @@ -0,0 +1,149 @@ +// v0.41.2.1 — embedding_env_override doctor check (D9 #9). +// +// Pinned contracts: +// - env unset → ok +// - env+DB agree → ok +// - env model OR dim disagrees → warn with `details.mismatches[]` +// - getConfig throws → warn with "couldn't read DB config" message +// - Cross-surface parity: BOTH buildChecks() and doctorReportRemote() +// include the check (source-grep regression guard) +// +// Hermetic: PGLite + withEnv per CLAUDE.md R1/R3/R4. + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { readFileSync } from 'fs'; +import { join } from 'path'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { buildChecks, doctorReportRemote, type Check } from '../src/commands/doctor.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { withEnv } from './helpers/with-env.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}, 60000); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); +}); + +function findCheck(checks: Check[], name: string): Check | undefined { + return checks.find((c) => c.name === name); +} + +describe('embedding_env_override check (buildChecks seam)', () => { + test('env unset → ok', async () => { + await withEnv( + { GBRAIN_EMBEDDING_MODEL: undefined, GBRAIN_EMBEDDING_DIMENSIONS: undefined }, + async () => { + const checks = await buildChecks(engine, []); + const check = findCheck(checks, 'embedding_env_override'); + expect(check).toBeDefined(); + expect(check!.status).toBe('ok'); + expect(check!.message).toContain('no embedding env overrides set'); + }, + ); + }); + + test('env+DB agree → ok', async () => { + await engine.setConfig('embedding_model', 'zeroentropyai:zembed-1'); + await engine.setConfig('embedding_dimensions', '1280'); + await withEnv( + { + GBRAIN_EMBEDDING_MODEL: 'zeroentropyai:zembed-1', + GBRAIN_EMBEDDING_DIMENSIONS: '1280', + }, + async () => { + const checks = await buildChecks(engine, []); + const check = findCheck(checks, 'embedding_env_override'); + expect(check!.status).toBe('ok'); + expect(check!.message).toContain('agree with DB config'); + }, + ); + }); + + test('env model disagrees with DB → warn with details.mismatches', async () => { + await engine.setConfig('embedding_model', 'zeroentropyai:zembed-1'); + await withEnv( + { GBRAIN_EMBEDDING_MODEL: 'openai:text-embedding-3-large' }, + async () => { + const checks = await buildChecks(engine, []); + const check = findCheck(checks, 'embedding_env_override'); + expect(check!.status).toBe('warn'); + const details = check!.details as { mismatches: Array<{ key: string; env: string; db: string }> }; + expect(details.mismatches).toHaveLength(1); + expect(details.mismatches[0].key).toBe('GBRAIN_EMBEDDING_MODEL'); + expect(details.mismatches[0].env).toBe('openai:text-embedding-3-large'); + expect(details.mismatches[0].db).toBe('zeroentropyai:zembed-1'); + // Message includes paste-ready unset + expect(check!.message).toContain('unset GBRAIN_EMBEDDING_MODEL'); + }, + ); + }); + + test('env dim disagrees with DB → warn with details.mismatches', async () => { + await engine.setConfig('embedding_dimensions', '1280'); + await withEnv( + { GBRAIN_EMBEDDING_DIMENSIONS: '1536' }, + async () => { + const checks = await buildChecks(engine, []); + const check = findCheck(checks, 'embedding_env_override'); + expect(check!.status).toBe('warn'); + const details = check!.details as { mismatches: Array<{ key: string; env: string; db: string }> }; + expect(details.mismatches).toHaveLength(1); + expect(details.mismatches[0].key).toBe('GBRAIN_EMBEDDING_DIMENSIONS'); + }, + ); + }); + + test('both disagree → 2 mismatches', async () => { + await engine.setConfig('embedding_model', 'zeroentropyai:zembed-1'); + await engine.setConfig('embedding_dimensions', '1280'); + await withEnv( + { + GBRAIN_EMBEDDING_MODEL: 'openai:x', + GBRAIN_EMBEDDING_DIMENSIONS: '1536', + }, + async () => { + const checks = await buildChecks(engine, []); + const check = findCheck(checks, 'embedding_env_override'); + expect(check!.status).toBe('warn'); + const details = check!.details as { mismatches: Array<{ key: string }> }; + expect(details.mismatches).toHaveLength(2); + expect(check!.message).toContain('unset GBRAIN_EMBEDDING_MODEL GBRAIN_EMBEDDING_DIMENSIONS'); + }, + ); + }); + + test('doctorReportRemote() includes the check (cross-surface parity)', async () => { + await withEnv({ GBRAIN_EMBEDDING_MODEL: 'openai:something' }, async () => { + await engine.setConfig('embedding_model', 'zeroentropyai:zembed-1'); + const report = await doctorReportRemote(engine); + const check = findCheck(report.checks, 'embedding_env_override'); + expect(check).toBeDefined(); + expect(check!.status).toBe('warn'); + }); + }); +}); + +describe('cross-surface parity (source-grep regression guard)', () => { + test('doctor.ts wires checkEmbeddingEnvOverride into BOTH buildChecks and doctorReportRemote', () => { + // Static regression assertion: the helper must be called from BOTH surfaces. + // If a future maintainer removes the call from one surface, this test fails + // pointing at the asymmetry. + const src = readFileSync( + join(import.meta.dir, '../src/commands/doctor.ts'), + 'utf-8', + ); + // The helper is called as `await checkEmbeddingEnvOverride(engine)` + const matches = src.match(/await checkEmbeddingEnvOverride\(engine\)/g) ?? []; + expect(matches.length).toBeGreaterThanOrEqual(2); + }); +}); diff --git a/test/e2e/extract-atoms-discovery-sql.test.ts b/test/e2e/extract-atoms-discovery-sql.test.ts new file mode 100644 index 000000000..a5a5905fb --- /dev/null +++ b/test/e2e/extract-atoms-discovery-sql.test.ts @@ -0,0 +1,135 @@ +/** + * v0.41.2.1 — D10 reversal: real-Postgres parity for the new + * extract-atoms discovery SQL. Closes the D5 gap codex flagged: the + * raw SQL uses ANY($::text[]) binding, JSONB ->>, substring(... from N + * for M), and NOT EXISTS subquery — features PGLite parses but the + * real Postgres wire binding through `postgres.unsafe` is subtly + * different. PGLite-only tests are not proof. + * + * Asserts: + * 1. discoverExtractablePages returns rows on a seeded federated brain + * 2. NOT EXISTS subquery skips already-extracted source pages + * 3. sourceId scoping isolates between two seeded sources (no leak) + * 4. ANY($2::text[]) binding actually filters by type set + * + * ~4 structural assertions; ~3-5s wallclock budget. + * Skips gracefully when DATABASE_URL is unset. + */ +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { setupDB, teardownDB, hasDatabase } from './helpers.ts'; +import { PostgresEngine } from '../../src/core/postgres-engine.ts'; +import { discoverExtractablePages } from '../../src/core/cycle/extract-atoms.ts'; + +const skip = !hasDatabase(); +const describeIfDB = skip ? describe.skip : describe; + +let engine: PostgresEngine; + +beforeAll(async () => { + if (skip) return; + engine = (await setupDB()) as PostgresEngine; +}); + +afterAll(async () => { + if (skip) return; + await teardownDB(); +}); + +beforeEach(async () => { + if (skip) return; + // Clean test-source rows + atoms + meeting pages between tests + await engine.executeRaw(`DELETE FROM pages WHERE source_id IN ('default', 'dept-x') AND (type = 'atom' OR type IN ('meeting', 'source', 'article', 'video', 'book', 'original'))`); + await engine.executeRaw(`DELETE FROM sources WHERE id = 'dept-x'`); +}); + +const LONG = 'a'.repeat(800); + +async function seedPage(opts: { + slug: string; + type: string; + content_hash?: string; + frontmatter?: Record<string, unknown>; + source_id?: string; +}) { + await engine.putPage( + opts.slug, + { + type: opts.type as never, + title: opts.slug, + compiled_truth: LONG, + timeline: '', + frontmatter: opts.frontmatter ?? {}, + }, + { sourceId: opts.source_id ?? 'default' }, + ); + if (opts.content_hash) { + await engine.executeRaw( + `UPDATE pages SET content_hash = $1 WHERE slug = $2 AND source_id = $3`, + [opts.content_hash, opts.slug, opts.source_id ?? 'default'], + ); + } +} + +describeIfDB('v0.41.2.1 D10 — discoverExtractablePages on real Postgres', () => { + test('returns extractable rows when seeded', async () => { + await seedPage({ slug: 'meeting/a', type: 'meeting', content_hash: 'hash-A-1234567890abc' }); + await seedPage({ slug: 'source/b', type: 'source', content_hash: 'hash-B-1234567890abc' }); + await seedPage({ slug: 'notes/skip', type: 'note', content_hash: 'hash-N-1234567890abc' }); + + const discovered = await discoverExtractablePages(engine, 'default'); + const slugs = discovered.map((d) => d.slug).sort(); + expect(slugs).toContain('meeting/a'); + expect(slugs).toContain('source/b'); + expect(slugs).not.toContain('notes/skip'); + }); + + test('ANY($::text[]) bind works through postgres.unsafe (PGLite parity proof)', async () => { + // Seed all 6 extractable types + one non-extractable. The SQL uses + // type = ANY($2::text[]) — if the binding shape differs between + // PGLite and postgres.js's unsafe(), this catches it. + for (const type of ['meeting', 'source', 'article', 'video', 'book', 'original']) { + await seedPage({ slug: `${type}/x`, type, content_hash: `hash-${type}-1234567890ab` }); + } + await seedPage({ slug: 'note/skip', type: 'note', content_hash: 'hash-note-1234567890' }); + + const discovered = await discoverExtractablePages(engine, 'default'); + const slugs = discovered.map((d) => d.slug).sort(); + expect(slugs).toEqual([ + 'article/x', 'book/x', 'meeting/x', 'original/x', 'source/x', 'video/x', + ]); + }); + + test('NOT EXISTS subquery skips pages with existing atoms', async () => { + await seedPage({ slug: 'meeting/old', type: 'meeting', content_hash: 'oldhash1234567890abc' }); + await seedPage({ slug: 'meeting/new', type: 'meeting', content_hash: 'newhash1234567890abc' }); + // Seed atom with frontmatter.source_hash = first 16 chars of oldhash + await engine.putPage( + 'atoms/seeded/old-insight', + { + type: 'atom' as never, + title: 'old', + compiled_truth: 'b', + timeline: '', + frontmatter: { source_hash: 'oldhash123456789' }, // first 16 chars of seed + }, + { sourceId: 'default' }, + ); + + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.map((d) => d.slug)).toEqual(['meeting/new']); + }); + + test('sourceId scopes both candidate AND atom-existence subquery — no cross-source leak', async () => { + await engine.executeRaw( + `INSERT INTO sources (id, name) VALUES ('dept-x', 'dept-x') ON CONFLICT DO NOTHING`, + ); + await seedPage({ slug: 'meeting/a-default', type: 'meeting', content_hash: 'hash-default-A-12345' }); + await seedPage({ slug: 'meeting/a-dept-x', type: 'meeting', content_hash: 'hash-dept-x-A-12345', source_id: 'dept-x' }); + + const fromDefault = await discoverExtractablePages(engine, 'default'); + const fromDeptX = await discoverExtractablePages(engine, 'dept-x'); + + expect(fromDefault.map((d) => d.slug)).toEqual(['meeting/a-default']); + expect(fromDeptX.map((d) => d.slug)).toEqual(['meeting/a-dept-x']); + }); +}); diff --git a/test/extract-atoms-page-discovery.test.ts b/test/extract-atoms-page-discovery.test.ts new file mode 100644 index 000000000..db459c64a --- /dev/null +++ b/test/extract-atoms-page-discovery.test.ts @@ -0,0 +1,395 @@ +// v0.41.2.1 — page-based discovery + source-hash idempotency for +// extract_atoms. Hermetic PGLite tests; no DATABASE_URL needed. +// +// Pins the contracts from /plan-eng-review: +// D1: source-hash existence check replaces frontmatter marker +// D2: single raw SQL with NOT EXISTS subquery +// D9 #1: sourceId threaded through every putPage +// D9 #3: content_hash IS NOT NULL filter (no crash on null) +// D9 #4: dream_generated:'true' pages excluded +// D9 #5: integration assertion deferred to cycle.ts test +// +// Plus PageDiscovery shape, transcript-side idempotency, dual-source +// merge, dedup, dry-run, fail-soft on executeRaw errors. + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { + runPhaseExtractAtoms, + discoverExtractablePages, +} from '../src/core/cycle/extract-atoms.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import type { ChatOpts, ChatResult } from '../src/core/ai/gateway.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}, 60000); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); +}); + +function stubChat(text: string): (o: ChatOpts) => Promise<ChatResult> { + return async (_o: ChatOpts) => ({ + text, + blocks: [{ type: 'text', text }], + stopReason: 'end', + usage: { input_tokens: 100, output_tokens: 50, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'anthropic:claude-haiku-4-5', + providerId: 'anthropic', + }); +} + +/** + * Stub that returns a unique-title atom on each call so atoms write to + * distinct slugs (`atoms/${date}/${slugify(title)}`) instead of upserting + * into one row. Needed for tests that count atoms after multiple work items. + */ +function stubChatUnique(): (o: ChatOpts) => Promise<ChatResult> { + let counter = 0; + return async (_o: ChatOpts) => { + counter++; + const text = `[{"title":"unique-atom-${counter}","atom_type":"insight","body":"b${counter}"}]`; + return { + text, + blocks: [{ type: 'text', text }], + stopReason: 'end', + usage: { input_tokens: 100, output_tokens: 50, cache_read_tokens: 0, cache_creation_tokens: 0 }, + model: 'anthropic:claude-haiku-4-5', + providerId: 'anthropic', + }; + }; +} + +const LONG_CONTENT = 'a'.repeat(800); // > MIN_PAGE_CHARS_FOR_EXTRACTION (500) + +async function seedPage(opts: { + slug: string; + type: string; + content_hash?: string | null; + frontmatter?: Record<string, unknown>; + source_id?: string; + compiled_truth?: string; +}) { + await engine.putPage( + opts.slug, + { + type: opts.type as never, + title: opts.slug, + compiled_truth: opts.compiled_truth ?? LONG_CONTENT, + timeline: '', + frontmatter: opts.frontmatter ?? {}, + content_hash: opts.content_hash === null ? undefined : (opts.content_hash ?? `hash-for-${opts.slug}`), + }, + { sourceId: opts.source_id ?? 'default' }, + ); + // PGLite stores content_hash as the engine's computed hash by default. + // For tests that need a specific hash, overwrite via raw SQL. + if (opts.content_hash && opts.content_hash !== `hash-for-${opts.slug}`) { + await engine.executeRaw( + `UPDATE pages SET content_hash = $1 WHERE slug = $2 AND source_id = $3`, + [opts.content_hash, opts.slug, opts.source_id ?? 'default'], + ); + } + if (opts.content_hash === null) { + await engine.executeRaw( + `UPDATE pages SET content_hash = NULL WHERE slug = $1 AND source_id = $2`, + [opts.slug, opts.source_id ?? 'default'], + ); + } +} + +describe('v0.41.2.1: discoverExtractablePages SQL contract', () => { + test('filters by all 6 extractable types', async () => { + for (const type of ['meeting', 'source', 'article', 'video', 'book', 'original']) { + await seedPage({ slug: `${type}/x`, type }); + } + // Add a non-extractable page that should NOT appear + await seedPage({ slug: 'notes/skip-me', type: 'note' }); + + const discovered = await discoverExtractablePages(engine, 'default'); + const slugs = discovered.map((d) => d.slug).sort(); + expect(slugs).toEqual([ + 'article/x', + 'book/x', + 'meeting/x', + 'original/x', + 'source/x', + 'video/x', + ]); + }); + + test('NOT EXISTS subquery skips pages whose source_hash has existing atoms', async () => { + // Page content_hash is 20 chars; substring(from 1 for 16) yields the + // first 16 chars. The seeded atom must carry exactly those 16 chars + // in frontmatter.source_hash to match the subquery's comparison. + await seedPage({ slug: 'meeting/old', type: 'meeting', content_hash: 'oldhash1234567890abc' }); + await seedPage({ slug: 'meeting/new', type: 'meeting', content_hash: 'newhash1234567890abc' }); + await engine.putPage( + 'atoms/2026-05-20/old-insight', + { + type: 'atom' as never, + title: 'Old', + compiled_truth: 'body', + timeline: '', + // 'oldhash1234567890abc'.slice(0, 16) = 'oldhash123456789' + frontmatter: { source_hash: 'oldhash123456789' }, + }, + { sourceId: 'default' }, + ); + + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.map((d) => d.slug)).toEqual(['meeting/new']); + }); + + test('markdown-greenfield pages excluded', async () => { + await seedPage({ slug: 'meeting/normal', type: 'meeting' }); + await seedPage({ + slug: 'meeting/greenfield', + type: 'meeting', + frontmatter: { imported_from: 'markdown-greenfield' }, + }); + + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.map((d) => d.slug)).toEqual(['meeting/normal']); + }); + + test('dream_generated:true pages excluded (D9 #4 — no self-consumption)', async () => { + await seedPage({ slug: 'original/normal', type: 'original' }); + await seedPage({ + slug: 'original/from-dream', + type: 'original', + frontmatter: { dream_generated: true }, + }); + + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.map((d) => d.slug)).toEqual(['original/normal']); + }); + + test('pages with NULL content_hash excluded (D9 #3 — no .slice crash)', async () => { + await seedPage({ slug: 'meeting/with-hash', type: 'meeting' }); + await seedPage({ slug: 'meeting/no-hash', type: 'meeting', content_hash: null }); + + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.map((d) => d.slug)).toEqual(['meeting/with-hash']); + // .slice(0, 16) on hash would crash for null; the filter prevents that + expect(() => discovered[0].contentHash.slice(0, 16)).not.toThrow(); + }); + + test('pages shorter than MIN_PAGE_CHARS_FOR_EXTRACTION excluded', async () => { + await seedPage({ slug: 'meeting/long', type: 'meeting', compiled_truth: 'a'.repeat(600) }); + await seedPage({ slug: 'meeting/short', type: 'meeting', compiled_truth: 'short' }); + + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered.map((d) => d.slug)).toEqual(['meeting/long']); + }); + + test('sourceId scopes both candidate AND atom-existence subquery (federated-brain isolation)', async () => { + await engine.executeRaw( + `INSERT INTO sources (id, name) VALUES ('dept-x', 'dept-x') ON CONFLICT DO NOTHING`, + ); + await seedPage({ slug: 'meeting/a-default', type: 'meeting' }); + await seedPage({ slug: 'meeting/a-dept-x', type: 'meeting', source_id: 'dept-x' }); + + const discoveredDefault = await discoverExtractablePages(engine, 'default'); + expect(discoveredDefault.map((d) => d.slug)).toEqual(['meeting/a-default']); + const discoveredDept = await discoverExtractablePages(engine, 'dept-x'); + expect(discoveredDept.map((d) => d.slug)).toEqual(['meeting/a-dept-x']); + }); + + test('affectedSlugs filter narrows candidates when provided', async () => { + await seedPage({ slug: 'meeting/a', type: 'meeting' }); + await seedPage({ slug: 'meeting/b', type: 'meeting' }); + await seedPage({ slug: 'meeting/c', type: 'meeting' }); + + const discovered = await discoverExtractablePages(engine, 'default', ['meeting/a', 'meeting/c']); + expect(discovered.map((d) => d.slug).sort()).toEqual(['meeting/a', 'meeting/c']); + }); + + test('executeRaw failure returns [] (fail-soft, transcript path proceeds)', async () => { + // Inject a SQL error by passing a sourceId that breaks the query — + // actually easier: temporarily replace executeRaw to throw. + const realExecute = engine.executeRaw.bind(engine); + (engine as unknown as { executeRaw: typeof engine.executeRaw }).executeRaw = + async () => { throw new Error('synthetic discovery failure'); }; + try { + const discovered = await discoverExtractablePages(engine, 'default'); + expect(discovered).toEqual([]); + } finally { + (engine as unknown as { executeRaw: typeof engine.executeRaw }).executeRaw = realExecute; + } + }); +}); + +describe('v0.41.2.1: runPhaseExtractAtoms — dual-source merge + idempotency', () => { + test('dual-source merge: transcripts win on contentHash collision', async () => { + const chat = stubChat(`[{"title":"x","atom_type":"insight","body":"b"}]`); + const result = await runPhaseExtractAtoms(engine, { + _transcripts: [ + { filePath: '/transcript-A.txt', content: 'shared content', contentHash: 'sharedhash1234' }, + ], + _pages: [ + // Same content hash → should be dedup-skipped in favor of transcript + { slug: 'meeting/shared', content: 'shared content', contentHash: 'sharedhash1234' }, + { slug: 'meeting/unique', content: 'unique content', contentHash: 'uniquehash5678' }, + ], + _chat: chat, + }); + expect(result.status).toBe('ok'); + expect(result.details?.atoms_extracted).toBe(2); // transcript + page-unique + expect(result.details?.duplicates_skipped).toBe(1); // page collided with transcript + expect(result.details?.transcripts_processed).toBe(1); + expect(result.details?.pages_processed).toBe(1); + }); + + test('atom frontmatter: page-origin uses source_slug, transcript-origin uses source_path', async () => { + // Use stubChatUnique so the two work-items write to distinct slugs; + // a constant title would upsert into one slug and mask one origin. + const chat = stubChatUnique(); + await runPhaseExtractAtoms(engine, { + _transcripts: [{ filePath: '/T.txt', content: 'tc', contentHash: 'tchash1234567890ab' }], + _pages: [{ slug: 'meeting/P', content: 'pc', contentHash: 'pchash1234567890ab' }], + _chat: chat, + }); + const rows = await engine.executeRaw<{ slug: string; frontmatter: Record<string, unknown> }>( + `SELECT slug, frontmatter FROM pages WHERE type = 'atom' ORDER BY slug`, + ); + expect(rows.length).toBe(2); + const transcriptAtom = rows.find((r) => r.frontmatter.source_path !== undefined); + const pageAtom = rows.find((r) => r.frontmatter.source_slug !== undefined); + expect(transcriptAtom).toBeDefined(); + expect(transcriptAtom!.frontmatter.source_path).toBe('/T.txt'); + expect(transcriptAtom!.frontmatter.source_hash).toBe('tchash1234567890'); + expect(pageAtom).toBeDefined(); + expect(pageAtom!.frontmatter.source_slug).toBe('meeting/P'); + expect(pageAtom!.frontmatter.source_hash).toBe('pchash1234567890'); + }); + + test('sourceId threads into putPage call (D9 #1 regression — atoms land in correct source)', async () => { + await engine.executeRaw( + `INSERT INTO sources (id, name) VALUES ('dept-x', 'dept-x') ON CONFLICT DO NOTHING`, + ); + const chat = stubChat(`[{"title":"x","atom_type":"insight","body":"b"}]`); + await runPhaseExtractAtoms(engine, { + sourceId: 'dept-x', + _transcripts: [{ filePath: '/T.txt', content: 'tc', contentHash: 'tchash1234567890ab' }], + _pages: [], + _chat: chat, + }); + const rows = await engine.executeRaw<{ source_id: string }>( + `SELECT source_id FROM pages WHERE type = 'atom'`, + ); + expect(rows.length).toBe(1); + expect(rows[0].source_id).toBe('dept-x'); + }); + + test('transcript-side idempotency: re-discovered same-hash transcript skipped (closes pre-existing bug)', async () => { + const chat = stubChatUnique(); + // First run writes the atom + await runPhaseExtractAtoms(engine, { + _transcripts: [{ filePath: '/T.txt', content: 'c', contentHash: 'dedup1234567890abcd' }], + _pages: [], + _chat: chat, + }); + // Second run on same content_hash should skip (no second atom written) + const result2 = await runPhaseExtractAtoms(engine, { + _transcripts: [{ filePath: '/T.txt', content: 'c', contentHash: 'dedup1234567890abcd' }], + _pages: [], + _chat: chat, + }); + expect(result2.details?.duplicates_skipped).toBe(1); + expect(result2.details?.atoms_extracted).toBe(0); + const rows = await engine.executeRaw<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM pages WHERE type = 'atom'`, + ); + expect(rows[0].count).toBe(1); + }); + + test('idempotency: full re-run on same brain produces zero new atoms', async () => { + const chat = stubChatUnique(); + await seedPage({ slug: 'meeting/M', type: 'meeting', content_hash: 'mhash1234567890ab' }); + await runPhaseExtractAtoms(engine, { + _transcripts: [{ filePath: '/T.txt', content: 'tc', contentHash: 'thash1234567890ab' }], + _chat: chat, + }); + const before = await engine.executeRaw<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM pages WHERE type = 'atom'`, + ); + expect(before[0].count).toBe(2); // 1 transcript-atom + 1 page-atom + + // Second invocation — no work expected; both dedup paths fire + const result2 = await runPhaseExtractAtoms(engine, { + _transcripts: [{ filePath: '/T.txt', content: 'tc', contentHash: 'thash1234567890ab' }], + _chat: chat, + }); + expect(result2.details?.atoms_extracted).toBe(0); + expect(result2.details?.duplicates_skipped).toBe(1); // transcript skipped + expect(result2.details?.pages_total).toBe(0); // discovery returned empty + const after = await engine.executeRaw<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM pages WHERE type = 'atom'`, + ); + expect(after[0].count).toBe(before[0].count); + }); + + test('PhaseResult.details has additive page fields populated', async () => { + const chat = stubChat(`[{"title":"x","atom_type":"insight","body":"b"}]`); + const result = await runPhaseExtractAtoms(engine, { + _transcripts: [], + _pages: [ + { slug: 'meeting/a', content: 'a', contentHash: 'a1234567890abcde' }, + { slug: 'meeting/b', content: 'b', contentHash: 'b1234567890abcde' }, + ], + _chat: chat, + }); + expect(result.details?.pages_processed).toBe(2); + expect(result.details?.pages_total).toBe(2); + expect(result.details?.pages_skipped_budget).toBe(0); + expect(result.details?.duplicates_skipped).toBe(0); + expect(result.details?.atoms_extracted).toBe(2); + }); + + test('dry-run skips putPage for atoms', async () => { + const chat = stubChat(`[{"title":"x","atom_type":"insight","body":"b"}]`); + const result = await runPhaseExtractAtoms(engine, { + _transcripts: [{ filePath: '/T.txt', content: 'tc', contentHash: 'th1234567890abcde' }], + _pages: [{ slug: 'meeting/M', content: 'mc', contentHash: 'mh1234567890abcde' }], + _chat: chat, + dryRun: true, + }); + expect(result.details?.dry_run).toBe(true); + expect(result.details?.atoms_extracted).toBe(2); // counted but not written + const rows = await engine.executeRaw<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM pages WHERE type = 'atom'`, + ); + expect(rows[0].count).toBe(0); + }); + + test('_pages: undefined triggers discovery; _pages: [] explicitly suppresses', async () => { + await seedPage({ slug: 'meeting/discoverable', type: 'meeting', content_hash: 'discovered1234567' }); + const chat = stubChat(`[{"title":"x","atom_type":"insight","body":"b"}]`); + + // _pages explicitly [] → no page work + const suppressed = await runPhaseExtractAtoms(engine, { + _transcripts: [], + _pages: [], + _chat: chat, + }); + expect(suppressed.details?.pages_total).toBe(0); + + // _pages undefined → discovery fires + const discovered = await runPhaseExtractAtoms(engine, { + _transcripts: [], + _chat: chat, + }); + expect(discovered.details?.pages_total).toBe(1); + expect(discovered.details?.atoms_extracted).toBe(1); + }); +}); diff --git a/test/extract-batch-retry.test.ts b/test/extract-batch-retry.test.ts new file mode 100644 index 000000000..6a9399a58 --- /dev/null +++ b/test/extract-batch-retry.test.ts @@ -0,0 +1,276 @@ +// v0.41.2.1 — withRetry + logBatchRetry + snapshot-before-clear contract. +// +// Pinned contracts: +// - withRetry is a pure primitive (no UI), exports cleanly so tests +// import it without going through the 6 private flush() closures. +// - Classification uses isRetryableConnError from src/core/retry-matcher.ts +// (single source of truth; no inline pattern duplication). +// - Retry-matcher extension recognizes gbrain's GBrainError shape +// ({problem: 'No database connection'}) AND the literal message +// pattern. PR #1416's reported batch-loss incident is closed. +// - logBatchRetry writes stderr only when jsonMode is false. +// - Snapshot contract: batch.slice() BEFORE batch.length=0 so producer +// mutation during the 500ms retry delay can't lose items on retry. +// +// Hermetic: no engine, no PGLite, no env mutation, no DATABASE_URL. + +import { describe, expect, test, beforeEach } from 'bun:test'; +import { withRetry, logBatchRetry } from '../src/commands/extract.ts'; +import { isRetryableConnError } from '../src/core/retry-matcher.ts'; + +// Minimal GBrainError shape — mirror the typed problem/detail fields used +// by db.ts:getConnection so the retry-matcher extension recognizes it. +class FakeGBrainError extends Error { + problem: string; + detail: string; + constructor(problem: string, detail: string) { + super(`${problem}: ${detail}`); + this.problem = problem; + this.detail = detail; + } +} + +describe('isRetryableConnError extension (v0.41.2.1)', () => { + test('GBrainError with problem="No database connection" is retryable', () => { + const err = new FakeGBrainError('No database connection', 'connect() has not been called'); + expect(isRetryableConnError(err)).toBe(true); + }); + + test('GBrainError with other problem is NOT retryable', () => { + const err = new FakeGBrainError('Schema mismatch', 'expected vector(1536), got vector(1024)'); + expect(isRetryableConnError(err)).toBe(false); + }); + + test('plain Error with "No database connection" message is retryable (literal match)', () => { + expect(isRetryableConnError(new Error('No database connection: connect() has not been called.'))).toBe(true); + }); + + test('constraint violation 23505 is NOT retryable', () => { + const err = Object.assign(new Error('duplicate key value violates unique constraint'), { code: '23505' }); + expect(isRetryableConnError(err)).toBe(false); + }); +}); + +describe('withRetry primitive (v0.41.2.1)', () => { + test('first-call success: returns value, no onRetry invocation', async () => { + let calls = 0; + let retried = false; + const result = await withRetry( + async () => { calls++; return 42; }, + { onRetry: () => { retried = true; }, delayMs: 0 }, + ); + expect(result).toBe(42); + expect(calls).toBe(1); + expect(retried).toBe(false); + }); + + test('retries on Connection terminated; second attempt succeeds', async () => { + let calls = 0; + const result = await withRetry( + async () => { + calls++; + if (calls === 1) throw new Error('Connection terminated unexpectedly'); + return 'recovered'; + }, + { delayMs: 0 }, + ); + expect(result).toBe('recovered'); + expect(calls).toBe(2); + }); + + test('retries on GBrainError "No database connection"; second succeeds', async () => { + let calls = 0; + const result = await withRetry( + async () => { + calls++; + if (calls === 1) throw new FakeGBrainError('No database connection', 'connect() has not been called'); + return 'ok'; + }, + { delayMs: 0 }, + ); + expect(result).toBe('ok'); + expect(calls).toBe(2); + }); + + test('non-retryable error propagates immediately, no retry', async () => { + let calls = 0; + const promise = withRetry( + async () => { + calls++; + const err = Object.assign(new Error('duplicate key'), { code: '23505' }); + throw err; + }, + { delayMs: 0 }, + ); + await expect(promise).rejects.toThrow('duplicate key'); + expect(calls).toBe(1); // no retry on 23505 + }); + + test('second failure propagates (single retry, not infinite)', async () => { + let calls = 0; + const promise = withRetry( + async () => { + calls++; + throw new Error('ECONNRESET'); + }, + { delayMs: 0 }, + ); + await expect(promise).rejects.toThrow('ECONNRESET'); + expect(calls).toBe(2); // attempt 1 + retry, then propagate + }); + + test('onRetry callback receives (attempt=1, err)', async () => { + let received: { attempt: number; err: unknown } | null = null; + let calls = 0; + await withRetry( + async () => { + calls++; + if (calls === 1) throw new Error('Connection terminated unexpectedly'); + return null; + }, + { + onRetry: (attempt, err) => { received = { attempt, err }; }, + delayMs: 0, + }, + ); + expect(received).not.toBeNull(); + expect(received!.attempt).toBe(1); + expect(received!.err).toBeInstanceOf(Error); + expect((received!.err as Error).message).toBe('Connection terminated unexpectedly'); + }); + + test('delayMs default is 500ms when not specified', async () => { + let calls = 0; + const start = Date.now(); + await withRetry( + async () => { + calls++; + if (calls === 1) throw new Error('ECONNRESET'); + return null; + }, + // no delayMs override + ); + const elapsed = Date.now() - start; + expect(calls).toBe(2); + // Allow ±50ms tolerance for scheduler jitter + expect(elapsed).toBeGreaterThanOrEqual(450); + expect(elapsed).toBeLessThan(2000); // never close to forever + }, 5000); +}); + +describe('logBatchRetry helper (v0.41.2.1)', () => { + let stderrCaptured: string[] = []; + let originalStderr: typeof console.error; + + beforeEach(() => { + stderrCaptured = []; + originalStderr = console.error; + console.error = (...args: unknown[]) => { + stderrCaptured.push(args.map(String).join(' ')); + }; + }); + + test('writes single stderr line when jsonMode=false', () => { + logBatchRetry('extract.links_fs', 73, new Error('Connection terminated'), false); + try { + expect(stderrCaptured).toHaveLength(1); + expect(stderrCaptured[0]).toContain('extract.links_fs'); + expect(stderrCaptured[0]).toContain('73'); + expect(stderrCaptured[0]).toContain('Connection terminated'); + } finally { + console.error = originalStderr; + } + }); + + test('writes NOTHING when jsonMode=true', () => { + logBatchRetry('extract.links_fs', 73, new Error('ECONNRESET'), true); + try { + expect(stderrCaptured).toHaveLength(0); + } finally { + console.error = originalStderr; + } + }); + + test('handles non-Error throwables by stringifying', () => { + logBatchRetry('extract.timeline_db', 12, 'string-error', false); + try { + expect(stderrCaptured).toHaveLength(1); + expect(stderrCaptured[0]).toContain('string-error'); + } finally { + console.error = originalStderr; + } + }); +}); + +describe('snapshot-before-clear contract (load-bearing for PR #1416 fix)', () => { + // This contract is what the 6 flush() sites depend on. The retry primitive + // doesn't enforce it itself — the call site must snapshot the batch BEFORE + // clearing it, so a producer pushing during the retry delay can't lose + // items on retry. We simulate the flush() pattern directly. + + test('snapshot is what retry sends, NOT the post-clear batch', async () => { + let batch: number[] = [1, 2, 3, 4, 5]; + let receivedOnAttempt: number[][] = []; + + async function flushPattern() { + if (batch.length === 0) return; + // Snapshot-before-clear, as flush sites in extract.ts do + const snapshot = batch.slice(); + batch.length = 0; + let calls = 0; + await withRetry( + async () => { + calls++; + receivedOnAttempt.push(snapshot.slice()); + if (calls === 1) { + // Producer mutates `batch` during the retry delay (simulated by + // pushing items right now — the retry hasn't fired yet, but in + // real code the producer's next iteration writes here). + batch.push(99, 100, 101); + throw new Error('Connection terminated unexpectedly'); + } + return snapshot.length; + }, + { delayMs: 0 }, + ); + } + + await flushPattern(); + + // Both attempts must have received the SAME snapshot — not the + // post-clear batch state. If snapshot-before-clear is broken, + // attempt 2 sees [99, 100, 101] (producer's new items only). + expect(receivedOnAttempt).toHaveLength(2); + expect(receivedOnAttempt[0]).toEqual([1, 2, 3, 4, 5]); + expect(receivedOnAttempt[1]).toEqual([1, 2, 3, 4, 5]); // retry sends snapshot, not batch + // batch contains the producer's new items, ready for next flush + expect(batch).toEqual([99, 100, 101]); + }); + + test('error message uses snapshot length, not post-clear batch length', async () => { + let batch: number[] = [10, 20, 30]; + let capturedErrorMsg = ''; + + async function flushPattern() { + if (batch.length === 0) return; + const snapshot = batch.slice(); + batch.length = 0; // batch.length is now 0 + try { + await withRetry( + async () => { + throw new Error('ECONNRESET'); // both attempts fail + }, + { delayMs: 0 }, + ); + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + // The CONTRACT: error message reads snapshot.length, NOT batch.length + capturedErrorMsg = `batch error (${snapshot.length} rows lost): ${msg}`; + } + } + + await flushPattern(); + expect(capturedErrorMsg).toContain('3 rows lost'); // snapshot had 3, not batch's 0 + expect(capturedErrorMsg).toContain('ECONNRESET'); + }); +}); diff --git a/test/loadConfig-merge.test.ts b/test/loadConfig-merge.test.ts index 7056efd9e..55c625339 100644 --- a/test/loadConfig-merge.test.ts +++ b/test/loadConfig-merge.test.ts @@ -172,4 +172,130 @@ describe('loadConfigWithEngine (Phase 4 / F3)', () => { expect(merged?.embedding_multimodal_model).toBeUndefined(); }); }); + + // v0.41.2.1 — dream.* DB-plane merge (closes PR #1416's silent-config bug). + // Precedence is file > DB > defaults per key. There is NO env layer for + // dream.* — adding env shadows is a separate PR (out of scope for the + // fix wave). These tests pin that contract. + describe('dream.* DB-plane merge (v0.41.2.1)', () => { + test('DB value fills in for all 5 dream.synthesize.* keys when base unset', async () => { + const base: GBrainConfig = { engine: 'pglite' }; + const engine = makeEngine({ + 'dream.synthesize.session_corpus_dir': '/tmp/sessions', + 'dream.synthesize.meeting_transcripts_dir': '/tmp/meetings', + 'dream.synthesize.verdict_model': 'anthropic:claude-haiku-4-5', + 'dream.synthesize.max_prompt_tokens': '180000', + 'dream.synthesize.max_chunks_per_transcript': '32', + }); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream?.synthesize?.session_corpus_dir).toBe('/tmp/sessions'); + expect(merged?.dream?.synthesize?.meeting_transcripts_dir).toBe('/tmp/meetings'); + expect(merged?.dream?.synthesize?.verdict_model).toBe('anthropic:claude-haiku-4-5'); + expect(merged?.dream?.synthesize?.max_prompt_tokens).toBe(180000); + expect(merged?.dream?.synthesize?.max_chunks_per_transcript).toBe(32); + }); + + test('DB value fills in for both dream.patterns.* keys when base unset', async () => { + const base: GBrainConfig = { engine: 'pglite' }; + const engine = makeEngine({ + 'dream.patterns.lookback_days': '45', + 'dream.patterns.min_evidence': '4', + }); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream?.patterns?.lookback_days).toBe(45); + expect(merged?.dream?.patterns?.min_evidence).toBe(4); + }); + + test('file value wins over DB value (per-key precedence)', async () => { + const base: GBrainConfig = { + engine: 'pglite', + dream: { + synthesize: { session_corpus_dir: '/from-file' }, + patterns: { lookback_days: 7 }, + }, + }; + const engine = makeEngine({ + 'dream.synthesize.session_corpus_dir': '/from-db', + 'dream.synthesize.meeting_transcripts_dir': '/db-meetings', + 'dream.patterns.lookback_days': '30', + 'dream.patterns.min_evidence': '5', + }); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream?.synthesize?.session_corpus_dir).toBe('/from-file'); + expect(merged?.dream?.synthesize?.meeting_transcripts_dir).toBe('/db-meetings'); + expect(merged?.dream?.patterns?.lookback_days).toBe(7); + expect(merged?.dream?.patterns?.min_evidence).toBe(5); + }); + + test('parent objects (cfg.dream, cfg.dream.synthesize, cfg.dream.patterns) are allocated even when file plane has none', async () => { + const base: GBrainConfig = { engine: 'pglite' }; + const engine = makeEngine({ + 'dream.synthesize.session_corpus_dir': '/just-this-one', + }); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream).toBeDefined(); + expect(merged?.dream?.synthesize).toBeDefined(); + expect(merged?.dream?.synthesize?.session_corpus_dir).toBe('/just-this-one'); + // patterns parent NOT allocated when no patterns key is set + expect(merged?.dream?.patterns).toBeUndefined(); + }); + + test('invalid DB int values fall back to undefined (do not throw)', async () => { + const base: GBrainConfig = { engine: 'pglite' }; + const engine = makeEngine({ + 'dream.synthesize.max_prompt_tokens': 'abc', + 'dream.patterns.min_evidence': 'not-a-number', + 'dream.patterns.lookback_days': '-5', // negative; existing dbInt() rejects + }); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream?.synthesize?.max_prompt_tokens).toBeUndefined(); + expect(merged?.dream?.patterns?.min_evidence).toBeUndefined(); + expect(merged?.dream?.patterns?.lookback_days).toBeUndefined(); + // cfg.dream stays undefined since no leaf populated + expect(merged?.dream).toBeUndefined(); + }); + + test('empty DB values do not clobber unset file plane', async () => { + const base: GBrainConfig = { engine: 'pglite' }; + const engine = makeEngine({ + 'dream.synthesize.session_corpus_dir': '', + 'dream.synthesize.meeting_transcripts_dir': undefined, + 'dream.synthesize.verdict_model': null, + }); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream).toBeUndefined(); + }); + + test('cfg.dream stays undefined when neither plane sets anything', async () => { + const base: GBrainConfig = { engine: 'pglite' }; + const engine = makeEngine({}); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream).toBeUndefined(); + }); + + test('mixed: file sets synthesize.session_corpus_dir; DB sets patterns.lookback_days', async () => { + const base: GBrainConfig = { + engine: 'pglite', + dream: { synthesize: { session_corpus_dir: '/file-only' } }, + }; + const engine = makeEngine({ + 'dream.patterns.lookback_days': '14', + }); + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream?.synthesize?.session_corpus_dir).toBe('/file-only'); + expect(merged?.dream?.patterns?.lookback_days).toBe(14); + }); + + test('engine.getConfig throwing leaves dream.* unset (non-fatal, mirrors content_sanity)', async () => { + const base: GBrainConfig = { engine: 'pglite' }; + const engine: FakeEngine = { + async getConfig() { + throw new Error('config table missing'); + }, + }; + const merged = await loadConfigWithEngine(engine, base); + expect(merged?.dream).toBeUndefined(); + expect(merged?.engine).toBe('pglite'); + }); + }); }); diff --git a/test/operations-find-trajectory.test.ts b/test/operations-find-trajectory.test.ts index c9585071d..e4adbf504 100644 --- a/test/operations-find-trajectory.test.ts +++ b/test/operations-find-trajectory.test.ts @@ -15,10 +15,23 @@ import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:tes import { PGLiteEngine } from '../src/core/pglite-engine.ts'; import { operationsByName } from '../src/core/operations.ts'; import type { OperationContext } from '../src/core/operations.ts'; +import { configureGateway, resetGateway } from '../src/core/ai/gateway.ts'; let engine: PGLiteEngine; beforeAll(async () => { + // v0.41.5.0+: DEFAULT_EMBEDDING_DIMENSIONS is 1280 (ZE Matryoshka). unitVec() + // below inserts 1536-dim vectors into facts.embedding. Without pinning, a + // fresh CI environment (no prior gateway configure) sizes the column at + // vector(1280) and the inserts throw "expected 1280 dimensions, not 1536" + // — CI shard 4 hit this consistently after v0.41.6.0 shard re-balancing + // moved this file ahead of any test that pre-configured the gateway. + resetGateway(); + configureGateway({ + embedding_model: 'openai:text-embedding-3-large', + embedding_dimensions: 1536, + env: { OPENAI_API_KEY: 'sk-fake' }, + }); engine = new PGLiteEngine(); await engine.connect({}); await engine.initSchema(); @@ -26,6 +39,7 @@ beforeAll(async () => { afterAll(async () => { await engine.disconnect(); + resetGateway(); }); beforeEach(async () => { diff --git a/test/search-types-filter.test.ts b/test/search-types-filter.test.ts index 0895fb4f6..530428929 100644 --- a/test/search-types-filter.test.ts +++ b/test/search-types-filter.test.ts @@ -11,6 +11,7 @@ import { describe, test, expect, beforeAll, afterAll } from 'bun:test'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; import type { ChunkInput } from '../src/core/types.ts'; +import { configureGateway, resetGateway } from '../src/core/ai/gateway.ts'; let engine: PGLiteEngine; @@ -21,6 +22,18 @@ function basisEmbedding(idx: number, dim = 1536): Float32Array { } beforeAll(async () => { + // v0.41.5.0+: DEFAULT_EMBEDDING_DIMENSIONS is 1280 (ZE Matryoshka). This test + // inserts 1536-dim unit vectors below. Without pinning, initSchema() sizes + // content_chunks.embedding at vector(1280) and the upserts throw + // "expected 1280 dimensions, not 1536". The local fast loop hides this when + // a prior test in the shard pre-configured the gateway at 1536d; CI shards + // hit it cold. Pin to 1536d so this file is hermetic. + resetGateway(); + configureGateway({ + embedding_model: 'openai:text-embedding-3-large', + embedding_dimensions: 1536, + env: { OPENAI_API_KEY: 'sk-fake' }, + }); engine = new PGLiteEngine(); await engine.connect({}); await engine.initSchema(); @@ -74,6 +87,7 @@ beforeAll(async () => { afterAll(async () => { await engine.disconnect(); + resetGateway(); }); describe('searchKeyword — types filter', () => { diff --git a/test/ze-switch-env-override.test.ts b/test/ze-switch-env-override.test.ts new file mode 100644 index 000000000..b9eebe6dc --- /dev/null +++ b/test/ze-switch-env-override.test.ts @@ -0,0 +1,327 @@ +// v0.41.2.1 — ze-switch env-override safety gate (D3 + D9 #6, #7, #8). +// +// Pinned contracts: +// - detectEnvOverride pure function (no env mutation; tests pass env arg) +// - formatEnvOverrideWarning produces ASCII-only box ≤78 cols +// - applyRetrievalUpgrade gate fires FIRST (NO setConfig calls when refused) +// - resumeRetrievalUpgrade gate fires FIRST (refused too, no schema mutation) +// - --ignore-env-override escape hatch bypasses both gates +// - ApplyResult tagged union extends with {status:'refused', reason, warning} +// +// Hermetic: PGLite + withEnv() per CLAUDE.md R1/R3/R4. No DATABASE_URL needed. + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { + detectEnvOverride, + formatEnvOverrideWarning, + applyRetrievalUpgrade, + resumeRetrievalUpgrade, + planRetrievalUpgrade, + KEY_REQUESTED, + KEY_APPLIED, + KEY_PREVIOUS_SNAPSHOT, + ZE_TARGET_EMBEDDING_MODEL, + ZE_TARGET_EMBEDDING_DIM, +} from '../src/core/retrieval-upgrade-planner.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { withEnv } from './helpers/with-env.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}, 60000); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); +}); + +// ─── detectEnvOverride: pure function tests (no env mutation needed) ──── + +describe('detectEnvOverride (pure)', () => { + test('triggered:false when env unset', () => { + const w = detectEnvOverride('zeroentropyai:zembed-1', 1280, {}); + expect(w.triggered).toBe(false); + expect(w.vars).toEqual([]); + }); + + test('triggered:false when env matches target exactly', () => { + const w = detectEnvOverride('zeroentropyai:zembed-1', 1280, { + GBRAIN_EMBEDDING_MODEL: 'zeroentropyai:zembed-1', + GBRAIN_EMBEDDING_DIMENSIONS: '1280', + }); + expect(w.triggered).toBe(false); + }); + + test('triggered:true on GBRAIN_EMBEDDING_MODEL mismatch (1 var)', () => { + const w = detectEnvOverride('zeroentropyai:zembed-1', 1280, { + GBRAIN_EMBEDDING_MODEL: 'openai:text-embedding-3-large', + }); + expect(w.triggered).toBe(true); + expect(w.vars).toHaveLength(1); + expect(w.vars[0].name).toBe('GBRAIN_EMBEDDING_MODEL'); + expect(w.vars[0].current).toBe('openai:text-embedding-3-large'); + expect(w.vars[0].target).toBe('zeroentropyai:zembed-1'); + }); + + test('GBRAIN_EMBEDDING_DIMENSIONS string-vs-number comparison works', () => { + const w = detectEnvOverride('zeroentropyai:zembed-1', 1280, { + GBRAIN_EMBEDDING_DIMENSIONS: '1536', + }); + expect(w.triggered).toBe(true); + expect(w.vars[0].name).toBe('GBRAIN_EMBEDDING_DIMENSIONS'); + expect(w.vars[0].current).toBe('1536'); + expect(w.vars[0].target).toBe('1280'); + }); + + test('NaN dim treated as mismatch (defensive)', () => { + const w = detectEnvOverride('zeroentropyai:zembed-1', 1280, { + GBRAIN_EMBEDDING_DIMENSIONS: 'not-a-number', + }); + expect(w.triggered).toBe(true); + expect(w.vars[0].current).toBe('not-a-number'); + }); + + test('both vars set + both mismatched → 2 entries', () => { + const w = detectEnvOverride('zeroentropyai:zembed-1', 1280, { + GBRAIN_EMBEDDING_MODEL: 'openai:text-embedding-3-large', + GBRAIN_EMBEDDING_DIMENSIONS: '1536', + }); + expect(w.triggered).toBe(true); + expect(w.vars).toHaveLength(2); + }); + + test('empty-string env values treated as unset (no false-trigger)', () => { + const w = detectEnvOverride('zeroentropyai:zembed-1', 1280, { + GBRAIN_EMBEDDING_MODEL: '', + GBRAIN_EMBEDDING_DIMENSIONS: ' ', + }); + expect(w.triggered).toBe(false); + }); +}); + +// ─── formatEnvOverrideWarning: pure rendering ────────────────────────── + +describe('formatEnvOverrideWarning', () => { + test('produces ASCII-only box (no Unicode box-drawing per repo D10)', () => { + const out = formatEnvOverrideWarning({ + triggered: true, + vars: [{ name: 'GBRAIN_EMBEDDING_MODEL', current: 'openai:x', target: 'zeroentropyai:zembed-1' }], + }); + // Should NOT contain any Unicode box-drawing characters + expect(out).not.toMatch(/[╔╗╚╝═║╠╣╦╩╬]/); + // Should contain ASCII box characters + expect(out).toMatch(/[+|-]/); + }); + + test('every line is ≤ 78 cols (terminal-safe)', () => { + const out = formatEnvOverrideWarning({ + triggered: true, + vars: [ + { name: 'GBRAIN_EMBEDDING_MODEL', current: 'openai:text-embedding-3-large', target: 'zeroentropyai:zembed-1' }, + { name: 'GBRAIN_EMBEDDING_DIMENSIONS', current: '1536', target: '1280' }, + ], + }); + for (const line of out.split('\n')) { + expect(line.length).toBeLessThanOrEqual(78); + } + }); + + test('names every var by exact key', () => { + const out = formatEnvOverrideWarning({ + triggered: true, + vars: [ + { name: 'GBRAIN_EMBEDDING_MODEL', current: 'a', target: 'b' }, + { name: 'GBRAIN_EMBEDDING_DIMENSIONS', current: 'c', target: 'd' }, + ], + }); + expect(out).toContain('GBRAIN_EMBEDDING_MODEL'); + expect(out).toContain('GBRAIN_EMBEDDING_DIMENSIONS'); + }); + + test('includes paste-ready `unset` command listing every var', () => { + const out = formatEnvOverrideWarning({ + triggered: true, + vars: [ + { name: 'GBRAIN_EMBEDDING_MODEL', current: 'a', target: 'b' }, + { name: 'GBRAIN_EMBEDDING_DIMENSIONS', current: 'c', target: 'd' }, + ], + }); + expect(out).toContain('unset GBRAIN_EMBEDDING_MODEL GBRAIN_EMBEDDING_DIMENSIONS'); + }); +}); + +// ─── applyRetrievalUpgrade integration (D9 #7 — gate fires FIRST) ────── + +describe('applyRetrievalUpgrade env-gate (D9 #7)', () => { + test('refused on env-triggered apply; ZERO setConfig calls fired', async () => { + await withEnv({ GBRAIN_EMBEDDING_MODEL: 'openai:text-embedding-3-large' }, async () => { + // Spy on engine.setConfig + const setConfigSpy: string[] = []; + const realSetConfig = engine.setConfig.bind(engine); + (engine as unknown as { setConfig: typeof engine.setConfig }).setConfig = + async (key: string, val: string) => { + setConfigSpy.push(key); + return realSetConfig(key, val); + }; + + try { + // Force an eligible plan by seeding the legacy default + enough pages. + await engine.setConfig('embedding_model', 'openai:text-embedding-3-large'); + // Seed 101 pages to clear the ZE_MIN_PAGES_FOR_OFFER threshold. + for (let i = 0; i < 101; i++) { + await engine.putPage(`note/seed-${i}`, { + type: 'note' as never, + title: `seed-${i}`, + compiled_truth: 'x', + timeline: '', + }); + } + // Reset the spy AFTER the setup writes + setConfigSpy.length = 0; + + const plan = await planRetrievalUpgrade(engine); + expect(plan.ze_switch_offered).toBe(true); + + const result = await applyRetrievalUpgrade(engine, plan); + expect(result.status).toBe('refused'); + if (result.status === 'refused') { + expect(result.reason).toBe('env_override'); + expect(result.warning.triggered).toBe(true); + expect(result.warning.vars[0].name).toBe('GBRAIN_EMBEDDING_MODEL'); + } + // THE LOAD-BEARING ASSERTION: no setConfig calls fired during the + // refused apply. Pre-fix, KEY_PREVIOUS_SNAPSHOT and KEY_REQUESTED + // were written BEFORE the warning gate, leaving the brain in a + // half-applied state. Now: zero mutations on refusal. + expect(setConfigSpy).toEqual([]); + + // Direct evidence: KEY_REQUESTED + KEY_PREVIOUS_SNAPSHOT unset + const requested = await engine.getConfig(KEY_REQUESTED); + const snapshot = await engine.getConfig(KEY_PREVIOUS_SNAPSHOT); + expect(requested).toBeNull(); + expect(snapshot).toBeNull(); + } finally { + (engine as unknown as { setConfig: typeof engine.setConfig }).setConfig = realSetConfig; + } + }); + }); + + test('--ignore-env-override proceeds with apply when env triggered', async () => { + await withEnv({ GBRAIN_EMBEDDING_MODEL: 'openai:text-embedding-3-large' }, async () => { + await engine.setConfig('embedding_model', 'openai:text-embedding-3-large'); + for (let i = 0; i < 101; i++) { + await engine.putPage(`note/x-${i}`, { + type: 'note' as never, + title: `x-${i}`, + compiled_truth: 'c', + timeline: '', + }); + } + const plan = await planRetrievalUpgrade(engine); + const result = await applyRetrievalUpgrade(engine, plan, { ignoreEnvOverride: true }); + expect(result.status).toBe('applied'); + const appliedFlag = await engine.getConfig(KEY_APPLIED); + expect(appliedFlag).toBe('true'); + }); + }); + + test('env not triggered → proceeds silently (no env vars set)', async () => { + // No env vars at all + await withEnv( + { GBRAIN_EMBEDDING_MODEL: undefined, GBRAIN_EMBEDDING_DIMENSIONS: undefined }, + async () => { + await engine.setConfig('embedding_model', 'openai:text-embedding-3-large'); + for (let i = 0; i < 101; i++) { + await engine.putPage(`note/y-${i}`, { + type: 'note' as never, + title: `y-${i}`, + compiled_truth: 'c', + timeline: '', + }); + } + const plan = await planRetrievalUpgrade(engine); + const result = await applyRetrievalUpgrade(engine, plan); + expect(result.status).toBe('applied'); + }, + ); + }); + + test('result tagged union shape pinned (TypeScript discriminator + runtime)', async () => { + await withEnv({ GBRAIN_EMBEDDING_MODEL: 'openai:other-model' }, async () => { + await engine.setConfig('embedding_model', 'openai:text-embedding-3-large'); + for (let i = 0; i < 101; i++) { + await engine.putPage(`note/z-${i}`, { + type: 'note' as never, + title: `z-${i}`, + compiled_truth: 'c', + timeline: '', + }); + } + const plan = await planRetrievalUpgrade(engine); + const result = await applyRetrievalUpgrade(engine, plan); + expect(result.status).toBe('refused'); + // Type narrowing: runtime + TS check + if (result.status === 'refused') { + expect(result.reason).toBe('env_override'); + expect(Array.isArray(result.warning.vars)).toBe(true); + expect(result.plan).toBeDefined(); + } + }); + }); +}); + +// ─── resumeRetrievalUpgrade env-gate (D9 #6 — same gate, no bypass) ──── + +describe('resumeRetrievalUpgrade env-gate (D9 #6)', () => { + test('refused on env-triggered resume; ZERO schema-mutation setConfig calls fired', async () => { + // Set up a half-applied state: requested=true, applied=false. + // Then run resume with env override set. + await engine.setConfig(KEY_REQUESTED, 'true'); + await engine.setConfig('embedding_model', 'openai:text-embedding-3-large'); + + await withEnv({ GBRAIN_EMBEDDING_MODEL: 'openai:still-old-model' }, async () => { + const setConfigSpy: string[] = []; + const realSetConfig = engine.setConfig.bind(engine); + (engine as unknown as { setConfig: typeof engine.setConfig }).setConfig = + async (key: string, val: string) => { + setConfigSpy.push(key); + return realSetConfig(key, val); + }; + + try { + const result = await resumeRetrievalUpgrade(engine); + expect(result.status).toBe('refused'); + if (result.status === 'refused') { + expect(result.reason).toBe('env_override'); + } + // No setConfig calls fired (no schema-completion writes happened) + expect(setConfigSpy).toEqual([]); + // APPLIED stays NOT-set + const applied = await engine.getConfig(KEY_APPLIED); + expect(applied).toBeNull(); + } finally { + (engine as unknown as { setConfig: typeof engine.setConfig }).setConfig = realSetConfig; + } + }); + }); + + test('--ignore-env-override proceeds with resume', async () => { + await engine.setConfig(KEY_REQUESTED, 'true'); + await engine.setConfig('embedding_model', 'openai:text-embedding-3-large'); + + await withEnv({ GBRAIN_EMBEDDING_MODEL: 'openai:still-old' }, async () => { + const result = await resumeRetrievalUpgrade(engine, { ignoreEnvOverride: true }); + expect(result.status).toBe('applied'); + const appliedFlag = await engine.getConfig(KEY_APPLIED); + expect(appliedFlag).toBe('true'); + }); + }); +});