From 84fed4194a0596dce05052934079c656fd305d16 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Mon, 25 May 2026 15:21:59 -0700 Subject: [PATCH] =?UTF-8?q?v0.41.11.0=20feat:=20conversation=20retrieval?= =?UTF-8?q?=20upgrade=20=E2=80=94=20production-bar=20replacement=20for=20P?= =?UTF-8?q?R=20#1406=20(#1446)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * v0.41.11.0 feat: conversation retrieval upgrade — production-bar replacement for PR #1406 Long chat threads stop swallowing your search results. The recall miss class on long iMessage/Slack imports (60K+ msg history; a chunk that reads only "Locker 93 code 9494" has no topical anchor because "cabin" was established 50K messages earlier) gets fixed by walking conversation/meeting/slack/email pages, splitting into time-windowed segments (30-min gap or 30-msg cap), prepending a topical/temporal header, and running through the existing extractFactsFromTurn() so the resulting anchor-rich facts surface in gbrain search. This is the production-bar replacement for PR #1406 (which closes LAST per Codex T6d, AFTER this PR is green). The bug fix survives 1:1; the wrapping closes 14 load-bearing issues the original PR deferred or shipped silent bugs around. The wave went through CEO scope review, 3 rounds of spec review, 2 rounds of Codex outside voice grounding the plan against actual code, and 2 passes of eng review. Version-slot note: originally planned as v0.41.2.0; master shipped its own v0.41.2.0 (lens packs) plus v0.41.3-6.0 between plan-time and ship-time. Re-bumped to v0.41.11.0 (next free slot; v0.41.7-10 claimed by other open PRs). Key files (new): - src/commands/extract-conversation-facts.ts — CLI command with --types, --max-cost-usd, --background, --override-disabled, --slug, --dry-run, --limit, --since, --force, --sleep, --segment-limit, --source-id. Strict per-source core; two-phase page enumeration (paginated listPages with 10×25MB cap = 250MB worst case); 25MB body cap; page-global row_num accumulator (Codex C1 unique-index collision fix); page-level TERMINAL audit row after all segments commit (Codex C7 durable extraction marker); optional opts.budgetTracker (Codex C5 — nested withBudgetTracker REPLACES, so caller-managed scope passes tracker through); reads compiled_truth + timeline (F1 — PR silently dropped timeline half); honors facts.extraction_enabled kill-switch with --override-disabled escape (F2); --types reads cycle config as single source of truth (Eng-v2 A2); fingerprint on sourceId only (Eng-v2 A3 — widening types doesn't invalidate completion); string-encoded op-checkpoint entries "sourceId|slug|endIso" for resume; segment caps tuned 6500/30 (Eng-v2 T5) to stay under extract.ts MAX_TURN_TEXT_CHARS=8000. - src/core/cycle/conversation-facts-backfill.ts — cycle phase wrapper (default OFF). Iterates listSources() directly; creates ONE brain-wide BudgetTracker per tick + wraps the loop in withBudgetTracker + passes tracker through opts.budgetTracker so core doesn't nest-replace. Two-layer cost AND walltime protection: per-source caps ($1, 20min) AND brain-wide caps ($5, 30min). - test/extract-conversation-facts.test.ts — 27 unit cases (parse, segment, render, checkpoint encoding, fingerprint, terminal audit row, row_num accumulator, F2 kill-switch, --override-disabled). - skills/migrations/v0.41.11.0.md — agent-facing migration guide. Key files (modified): - src/commands/jobs.ts — register extract-conversation-facts Minion handler. NOT in PROTECTED_JOB_NAMES; BudgetExhausted catch + persist + mark completed with result.budget_exhausted (NOT a failure). - src/commands/doctor.ts — computeConversationFactsBacklogCheck (3-state: SKIPPED when feature disabled per Eng-v2 C9, OK at backlog=0, WARN at >10 with paste-ready remediation step via makeRemediationStep). Doctor query is source-scoped (Codex C2 cross-source safety) and matches the TERMINAL audit row (Codex C7), not any-fact-for-slug. - src/commands/sources.ts — runAudit extended with facts_backfill_estimate field for cost preview. - src/cli.ts — CLI_ONLY + CLI_ONLY_SELF_HELP + THIN_CLIENT_REFUSED_COMMANDS + dispatch case for extract-conversation-facts. - src/core/cycle.ts — new CyclePhase 'conversation_facts_backfill'; PHASE_SCOPE='source' (taxonomy only per cycle.ts:131 — wrapper does own multi-source iteration); wired into ALL_PHASES + NEEDS_LOCK_PHASES; dispatch block runs between consolidate and embed. - src/core/migrate.ts — migration v94 adds partial index idx_facts_extract_conversation_session ON facts(source_id, source_session) WHERE source LIKE 'cli:extract-conversation-facts%' so doctor query stays fast on million-fact brains. v14 precedent: transaction:false + invalid-index pre-drop on Postgres, plain CREATE INDEX on PGLite. - src/core/schema-pack/base/gbrain-base.yaml — promote conversation (temporal, extractable) and atom (annotation, NOT extractable — atoms ARE the extracted form) into base. Flip concept.extractable: true semantically (cosmetic on backstop path per Codex T3; the original grandfather migration was solving a phantom, dropped). Filing rules added for both new types. - src/core/schema-pack/base/gbrain-recommended.yaml — remove duplicate conversation (now inherits via extends: gbrain-base). - src/core/types.ts — ALL_PAGE_TYPES extended with conversation, atom. - test/extractable-pack.test.ts — updated parity gate (24 page types vs PR's 22; concept + conversation now extractable, atom not). - test/schema-cli.test.ts — page-count expectation 22→24. - VERSION + package.json bumped to 0.41.11.0. - CHANGELOG.md release-summary in the required ELI10-first voice + itemized changes section. - CLAUDE.md Key Files entry for the new modules + architecture notes. - llms.txt + llms-full.txt regenerated. Plan + decisions persisted at: ~/.claude/plans/system-instruction-you-are-working-linear-unicorn.md CEO plan at: ~/.gstack/projects/garrytan-gbrain/ceo-plans/2026-05-25-conversation-retrieval-upgrade.md Co-Authored-By: garrytan-agents Co-Authored-By: Claude Opus 4.7 (1M context) * fix: align v0.41.11.0 phase ordering + bump hardcoded counts after master merge Three CI failures from the master merge in this branch: 1. test/phase-scope-coverage.test.ts pinned `ALL_PHASES.length === 19` and `Object.keys(PHASE_SCOPE).length === 19`. After merging master's v0.41 lens-packs (extract_atoms + synthesize_concepts) + my new conversation_facts_backfill phase, the total is 20. 2. test/core/cycle.serial.test.ts had two hardcoded `19` assertions (`hookCalls` and `report.phases.length`) tracking the same count. Both bumped to 20. 3. cycle.serial's `'default: all 6 phases run in order'` test asserts `report.phases.map(p => p.phase) === ALL_PHASES`. My initial commit put `conversation_facts_backfill` in ALL_PHASES between consolidate and propose_takes, but the runCycle dispatch block runs it AFTER the calibration trio (propose_takes / grade_takes / calibration_profile) and BEFORE embed. List and dispatch order didn't match, so the equality assertion failed. Resolution: moved 'conversation_facts_backfill' in ALL_PHASES to AFTER 'calibration_profile' so list-order matches dispatch-order. The dispatch block placement was correct (and remains correct); the list-position comment originally said "AFTER consolidate" but the dispatch runs it after the WHOLE consolidate→calibration_profile block, not just after consolidate. Comment now reflects reality. Verified: 61/61 pass across the 3 affected test files (2.9s wallclock). The CI logs also showed a "(unnamed) [3058.07ms]" failure in shard 1; unable to reproduce locally (test/scripts/run-unit-parallel.test.ts passes 6/6 in 1s). Suspected CI-load flakiness under bun's parallel scheduler. If it persists on the next CI run, will dig in. Co-Authored-By: Claude Opus 4.7 (1M context) * fix(ci): orphan sleep cleanup in run-unit-parallel.sh heartbeat Two CI runs in a row reported `(fail) (unnamed) [~2400ms]` in shard 1 on this PR. Investigation: - CI's end-of-job cleanup logged: "Terminate orphan process: pid (3344) (sleep)" × 6 sleep processes. - The 6 matches exactly the 6 `runWrapper()` calls in test/scripts/run-unit-parallel.test.ts (1 orphan sleep per invocation). - Each `runWrapper()` spawns scripts/run-unit-parallel.sh, which spawns a heartbeat function that runs `while true; do sleep 10; ...; done` in the background. - The wrapper's EXIT trap was `kill "$HB_PID" 2>/dev/null` — kills the heartbeat shell, but its currently-running `sleep 10` child gets reparented to init/launchd because SIGTERM to a bash shell sleeping inside `sleep` doesn't propagate to the sleep child before wait returns. Known bash quirk on Linux. - bun's test runner treats the orphan sleeps as a `(unnamed)` failure attributed to the test file that spawned the wrapper. Fix: pkill children FIRST, then kill heartbeat. If we kill heartbeat first, its child sleep orphans and pkill -P can no longer find it (ppid changes to 1). Reorder applied to both the trap AND the normal shutdown path. Verified locally: before fix, 6 orphan sleeps after the test ran; after fix, 0 orphan sleeps. Test still passes 6/6 in ~1s. Co-Authored-By: Claude Opus 4.7 (1M context) --------- Co-authored-by: garrytan-agents Co-authored-by: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 121 ++ CLAUDE.md | 1 + VERSION | 2 +- llms-full.txt | 1 + package.json | 2 +- scripts/run-unit-parallel.sh | 13 +- skills/migrations/v0.41.11.0.md | 61 + src/cli.ts | 14 +- src/commands/doctor.ts | 164 +++ src/commands/extract-conversation-facts.ts | 1115 +++++++++++++++++ src/commands/jobs.ts | 36 + src/commands/sources.ts | 36 + src/core/cycle.ts | 52 +- src/core/cycle/conversation-facts-backfill.ts | 294 +++++ src/core/migrate.ts | 60 + src/core/schema-pack/base/gbrain-base.yaml | 54 +- .../schema-pack/base/gbrain-recommended.yaml | 13 +- src/core/types.ts | 7 + test/core/cycle.serial.test.ts | 6 +- test/extract-conversation-facts.test.ts | 461 +++++++ test/extractable-pack.test.ts | 46 +- test/phase-scope-coverage.test.ts | 9 +- test/schema-cli.test.ts | 4 +- 23 files changed, 2540 insertions(+), 32 deletions(-) create mode 100644 skills/migrations/v0.41.11.0.md create mode 100644 src/commands/extract-conversation-facts.ts create mode 100644 src/core/cycle/conversation-facts-backfill.ts create mode 100644 test/extract-conversation-facts.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e0165d10..8b4f00286 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,127 @@ All notable changes to GBrain will be documented in this file. +## [0.41.11.0] - 2026-05-25 + +**Long chat threads stop swallowing your search results.** If you've imported a multi-year iMessage thread or a Slack archive, you've probably hit this: you search for a specific thing you know was said, the page exists in your brain, but the chunk that contains the literal answer never surfaces. Vector search chunks the conversation into ~300-word blocks, and a chunk that reads only "Locker 93 code 9494" has no topical anchor to "cabin" or "mountain" — the trip context was established 50,000 messages earlier. The chunk embedding has nothing to bind to. The page is there. The answer is there. Retrieval still misses. + +`gbrain extract-conversation-facts` walks long-form conversation pages, splits them into 30-minute time-windowed segments, prepends each segment with a topical/temporal header ("Conversation between Alice and Bob from to "), and runs them through the same fact extractor your real-time turns already use. The resulting facts land in the facts table — which retrieval already blends into search results — with the anchor terms the chunk-level embedding can't represent. The page surfaces again. + +## How to use it + +```bash +# Preview what would happen, no cost, no writes. +gbrain extract-conversation-facts --dry-run + +# Run the backfill with a $5 cap on the first pass. +gbrain extract-conversation-facts --max-cost-usd 5 + +# Submit as a background Minion job so you can keep working. +gbrain extract-conversation-facts --background --max-cost-usd 5 +# prints job_id=123; then: +gbrain jobs follow 123 + +# Get a cost estimate for one source before running. +gbrain sources audit default --json | jq '.facts_backfill_estimate' +``` + +A new cycle phase `conversation_facts_backfill` drains the backlog automatically once enabled. It's OFF by default so existing brains don't get a surprise bill on the next autopilot tick: + +```bash +gbrain config set cycle.conversation_facts_backfill.enabled true +``` + +`gbrain doctor` gains a `conversation_facts_backlog` check that's quiet for users who haven't enabled the feature (no opt-out noise debt). For users who have enabled it, the check warns when more than 10 eligible pages lack extraction and emits a paste-ready remediation step that `gbrain doctor --remediate` can run for you. + +## What you'd see in a concrete example + +Take a 4-year iMessage thread with the message "Locker 93 code 9494" buried in segment 47 of 200. The surrounding messages in that segment are unrelated chitchat — no mention of "cabin", "mountain", or "lockbox". The trip planning happened in segment 12. + +| Surface | Pre-extraction | Post-extraction | +|---|---|---| +| Search "mountain cabin lockbox code" | Returns segment 12's chunks (anchor match) — answer not in them | Returns the extracted fact: "Alice told Bob the mountain cabin lockbox code is 9494 in locker 93" | +| Page surfaces in results | Yes (via segment 12) | Yes (via the fact row) | +| Literal answer reachable | No — buried in a topically-blank chunk | Yes — anchor-rich fact text contains "9494" | + +The recall miss class survives any chunker change. Until conversations get a fundamentally different chunker (out of scope), the facts table is the right surface for cross-segment anchor-rich retrieval. + +## Things to watch + +- **Cost.** Default cap is $5 per run. The cycle phase has both per-source ($1) and brain-wide ($5) ceilings, plus per-source (20 min) and brain-wide (30 min) walltime caps. A 10-source brain can't quietly spend 10× its config. Override either with `--max-cost-usd` or via the config keys under `cycle.conversation_facts_backfill.*`. +- **Memory.** Pages over 25MB are skipped with a stderr warning and surface in `gbrain doctor`'s backlog details. Streaming for 50MB+ histories is a v0.42+ follow-up. +- **Honors the kill-switch.** If you previously ran `gbrain config set facts.extraction_enabled false`, this command refuses by default. Add `--override-disabled` if you want to force-run for one invocation. +- **Doctor visibility.** Run `gbrain doctor --json | jq '.checks[] | select(.name=="conversation_facts_backlog")'` to see what's pending. + +### What we caught and fixed before merging + +This wave went through CEO scope review, three rounds of spec review, two rounds of Codex outside voice grounding the plan against actual code, and two passes of eng review. Together they caught 14 load-bearing issues that would have shipped silent bugs: + +- The original test would have passed even if the bug was unfixed (it embedded the answer in the target message). Redesigned to require facts-as-context for the top-1 result. +- The plan stored object rows in `op_checkpoints`, which only round-trips string arrays AND is garbage-collected after 7 days. Switched to a page-level terminal audit row in the facts table itself as the durable extraction marker. +- The `insertFacts` unique index is `(source_id, source_markdown_slug, row_num)`. A per-segment row_num would collide on segment 2. Replaced with a page-global accumulator. +- The doctor backlog query lacked a `source_id` predicate. A page with the same slug in two sources would have falsely shown OK if either was extracted. Cross-source safety added. +- Nested `withBudgetTracker` REPLACES not stacks. The cycle phase now creates the brain-wide tracker once and passes it explicitly into per-source invocations so the core doesn't auto-wrap-and-replace it. +- The post-sync backstop uses hardcoded `ELIGIBLE_TYPES`, not pack `extractable`. The original concept-grandfather migration was solving a phantom; dropped. The schema pack still flips `concept.extractable: true` semantically but it doesn't bill anyone. + +## Schema migration + +One new migration: v94 adds a partial index on `facts(source_id, source_session) WHERE source LIKE 'cli:extract-conversation-facts%'` so the doctor backlog query runs in milliseconds on brains with millions of fact rows. Follows the v14 precedent: `transaction: false` + invalid-index pre-drop on Postgres, plain CREATE INDEX on PGLite. + +## To take advantage of v0.41.11.0 + +`gbrain upgrade` handles the binary + schema migration. After that: + +1. **Estimate cost before running** (optional but recommended): + ```bash + gbrain sources audit default --json | jq '.facts_backfill_estimate' + ``` +2. **Run the backfill once** to seed the facts table: + ```bash + gbrain extract-conversation-facts --background --max-cost-usd 5 + gbrain jobs follow + ``` +3. **Verify with doctor**: + ```bash + gbrain doctor --json | jq '.checks[] | select(.name=="conversation_facts_backlog")' + ``` +4. **(Optional) Enable the autopilot cycle phase** so the backlog drains continuously: + ```bash + gbrain config set cycle.conversation_facts_backfill.enabled true + ``` + +If you see unexpected behavior or the doctor check stays in WARN after a run, file an issue at https://github.com/garrytan/gbrain/issues with `gbrain doctor --json` output. + +### Itemized changes + +**Bug fixes (from PR #1406 + this wave's hardening):** +- Conversation page body reads now cover both `compiled_truth` AND `timeline` columns. iMessage and meeting-transcript importers commonly put the chronological message stream in `timeline`; PR #1406's `compiled_truth ?? ''` would have silently dropped half the messages on those importer shapes. +- Segment text cap tuned to 6500 chars + 30 messages (from 7500/50) so dense Slack/email segments don't silently truncate against `extract.ts`'s `MAX_TURN_TEXT_CHARS=8000` ceiling. + +**New CLI command:** +- `gbrain extract-conversation-facts [--source-id ID] [--types LIST] [--slug SLUG] [--dry-run] [--limit N] [--since ISO] [--force] [--sleep MS] [--segment-limit N] [--max-cost-usd FLOAT] [--background] [--override-disabled] [--yes]` + +**New cycle phase:** +- `conversation_facts_backfill` (default OFF; opt-in via config). 5 config keys: `enabled`, `max_cost_usd`, `max_total_cost_usd`, `max_walltime_min`, `max_total_walltime_min`, `types`. Brain-wide cost AND walltime ceilings; per-source caps inside each. + +**New Minion handler:** +- `extract-conversation-facts` (NOT in `PROTECTED_JOB_NAMES`; per-call cost bounded by `data.max_cost_usd`). `BudgetExhausted` mid-job → job marked completed with `result.budget_exhausted: true` and `result.spent_usd`, NOT a failure. + +**Doctor check:** +- `conversation_facts_backlog` (3-state: SKIPPED when disabled, OK when caught up, WARN when backlog > 10). Emits remediation on WARN for `gbrain doctor --remediate`. + +**Sources audit extension:** +- `gbrain sources audit ` now reports `facts_backfill_estimate: {pages, est_segments, est_cost_usd, types}` per source. + +**Schema-pack changes:** +- `gbrain-base.yaml`: added `conversation` (temporal, extractable) and `atom` (annotation, NOT extractable — atoms ARE the extracted form). Flipped `concept.extractable: true` semantically (cosmetic on backstop path today; documents that concept bodies ARE knowledge). +- `gbrain-recommended.yaml`: removed duplicate `conversation` (now inherits via `extends: gbrain-base`). +- `src/core/types.ts:ALL_PAGE_TYPES`: extended with `conversation` and `atom`. + +**Schema migration:** +- v94: partial index on `facts(source_id, source_session) WHERE source LIKE 'cli:extract-conversation-facts%'`. Postgres CONCURRENTLY + invalid-index pre-drop (v14 precedent); PGLite plain CREATE INDEX. + +**Credits:** Original PR #1406 by @garrytan-agents. The hardening wave absorbed 5 critical Codex findings (round 1), 4 architectural issues (eng pass 1), and 9 more code-grounded findings (Codex round 2 + eng pass 2). All preserved via `Co-Authored-By` on the replacement commit. + ## [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. diff --git a/CLAUDE.md b/CLAUDE.md index 388dac65a..ee4bec7e8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -160,6 +160,7 @@ strict behavior when unset. - `src/core/enrichment-service.ts` — Global enrichment service: entity slug generation, tier auto-escalation, batch throttling - `src/core/data-research.ts` — Recipe validation, field extraction (MRR/ARR regex), dedup, tracker parsing, HTML stripping - `src/commands/embed.ts` — `gbrain embed [--stale|--all] [--slugs ...]`. v0.22.1 (#409, contributed by @atrevino47): `--stale` path now starts with `engine.countStaleChunks()` (single SELECT count(*) WHERE embedding IS NULL, ~50 bytes wire). On a fully-embedded brain that's a 1-line short-circuit — no further reads. When stale chunks exist, `engine.listStaleChunks()` returns just the chunks needing embeddings (slug + chunk_index + chunk_text + metadata, no `vector(1536)` payload). Caller groups by slug, embeds via OpenAI, re-upserts via `upsertChunks`. Replaces the prior page-walk that pulled every chunk's embedding column over the wire and discarded most. **v0.40.3.0:** all 16 `console.log` / `console.error` call sites migrated to `slog` / `serr` from `src/core/console-prefix.ts` so that when `runEmbedCore` runs inside a per-source `withSourcePrefix` scope (which the `gbrain sync --all` worker pool installs), every line carries the `[] ` prefix. Standalone `gbrain embed --stale` callers see identical output to v0.40.2 because slog/serr fall through to bare console fns outside the wrap. +- `src/commands/extract-conversation-facts.ts` + `src/core/cycle/conversation-facts-backfill.ts` (v0.41.11.0) — bulk fact extraction for long-form conversation pages. The recall-miss class on long iMessage/Slack imports (60K+ msg history; a chunk that reads only "Locker 93 code 9494" has no topical anchor because "cabin" was established 50K messages earlier) gets fixed by walking conversation/meeting/slack/email pages, splitting into time-windowed segments (30-min gap or 30-msg cap), prepending a topical/temporal header ("Conversation between A and B from to "), and running through the existing `extractFactsFromTurn()` so the resulting anchor-rich facts surface in `gbrain search` (which already blends facts). Architecture decisions absorbed from 2 Codex outside-voice rounds + 2 eng review passes: **strict per-source core** (`runExtractConversationFactsCore({sourceId, ...})` always takes one sourceId; CLI + cycle phase each do their own multi-source iteration because PHASE_SCOPE='source' is taxonomy-only per cycle.ts:131); **two-phase enumeration** (paginated `listPages({type, sourceId, limit:10})` so worst case is 10×25MB=250MB resident vs PR #1406's unbounded; per-page body cap MAX_PAGE_BODY_BYTES=25MB with `pages_skipped_too_large` counter surfaced in doctor); **page-global row_num accumulator** (facts unique index is `(source_id, source_markdown_slug, row_num)` per migration v51 — per-segment row_num would collide on segment 2); **page-level TERMINAL audit row** written to facts table after all segments commit (source='cli:extract-conversation-facts:terminal'; doctor's NOT EXISTS query matches the terminal row, NOT any-fact-for-slug, so partial-extraction pages stay in backlog); **optional `opts.budgetTracker?`** (when present, core uses it as-is — nested `withBudgetTracker` REPLACES per gateway.ts:2011; when absent, core auto-wraps with `BudgetTracker({maxCostUsd})`); **body read covers compiled_truth + timeline** (iMessage importers put history in timeline half; PR's `compiled_truth ?? ''` would silently drop it); **honors `facts.extraction_enabled` kill-switch** with `--override-disabled` escape; **--types LIST allowlist** (`conversation,meeting,slack,email`) with CLI default reading `cycle.conversation_facts_backfill.types` config as single source of truth; **fingerprint on sourceId only** so widening types config doesn't invalidate completed-page state; **string-encoded op-checkpoint** entries `"||"` for resume only (op_checkpoints stores string[] and is 7-day GC'd — durable audit is the facts table itself via terminal row); **`--background` via maybeBackground** (Minion handler `extract-conversation-facts` re-creates BudgetTracker from `data.max_cost_usd`; on `BudgetExhausted` mid-job, catches + persists + marks `completed` with `result.budget_exhausted=true` — NOT a failure). The companion cycle phase `conversation_facts_backfill` (default OFF; `gbrain config set cycle.conversation_facts_backfill.enabled true` opts in) iterates `listSources(engine)` directly, creates ONE brain-wide tracker per tick + wraps the loop in `withBudgetTracker` + passes the tracker into every per-source invocation via `opts.budgetTracker` so the core doesn't nest-replace. Two-layer cost AND walltime protection: per-source caps (`max_cost_usd=$1`, `max_walltime_min=20`) AND brain-wide caps (`max_total_cost_usd=$5`, `max_total_walltime_min=30`). Pinned by 27 unit cases in `test/extract-conversation-facts.test.ts` (parse/segment/render/checkpoint/fingerprint/terminal-audit/row_num/F2-kill-switch). Schema migration v94 in `src/core/migrate.ts` adds a partial index `idx_facts_extract_conversation_session ON facts(source_id, source_session) WHERE source LIKE 'cli:extract-conversation-facts%'` (v14 precedent: `transaction:false` + invalid-index pre-drop on Postgres; plain CREATE INDEX on PGLite) so the doctor backlog query stays fast on million-fact brains. `src/commands/doctor.ts:computeConversationFactsBacklogCheck` is 3-state (SKIPPED when `cycle.conversation_facts_backfill.enabled=false` — no opt-out noise debt per Eng-v2 C9; OK when caught up; WARN when >10 pages lack the terminal audit row, with paste-ready remediation step composable via `gbrain doctor --remediate`). `src/commands/sources.ts:runAudit` extended with `facts_backfill_estimate: {pages, est_segments, est_cost_usd, types}` so `gbrain sources audit ` previews cost before any run. Schema-pack (v0.41.11.0): `gbrain-base.yaml` promotes `conversation` (temporal, extractable) + `atom` (annotation, NOT extractable — atoms ARE the extracted form) into the base seed; `concept.extractable` flips true semantically (cosmetic on backstop path because backstop uses hardcoded `ELIGIBLE_TYPES` in `src/core/facts/eligibility.ts:51`, not pack extractable — the original concept-grandfather migration was solving a phantom, dropped per Codex T3). `ALL_PAGE_TYPES` in `src/core/types.ts` extended with the two new types. Plan + all decisions persisted at `~/.claude/plans/system-instruction-you-are-working-linear-unicorn.md` and the CEO plan at `~/.gstack/projects/garrytan-gbrain/ceo-plans/2026-05-25-conversation-retrieval-upgrade.md`. - `src/commands/extract.ts` — `gbrain extract links|timeline|all [--source fs|db] [--source-id ]`: batch link/timeline extraction. fs walks markdown files, db walks pages from the engine (mutation-immune snapshot iteration; use this for live brains with no local checkout). As of v0.12.1 there is no in-memory dedup pre-load — candidates are buffered 100 at a time and flushed via `addLinksBatch` / `addTimelineEntriesBatch`; `ON CONFLICT DO NOTHING` enforces uniqueness at the DB layer, and the `created` counter returns real rows inserted (truthful on re-runs). v0.22.1 (#417): `ExtractOpts.slugs?: string[]` enables incremental extract — when set, `extractForSlugs()` reads ONLY those slugs' files (single combined links+timeline pass) instead of the full directory walk. CLI `gbrain extract` keeps full-walk behavior; the cycle path threads sync's `pagesAffected` through. `walkMarkdownFiles(brainDir)` still runs at line 455 to build `allSlugs` for link resolution — see `TODOS.md` for replacing it with `engine.getAllSlugs()`. **v0.37.7.0 (#1204):** `--source-id ` flag scopes extraction to one brain source on federated brains. Resolved via `resolveSourceWithTier()` before any SQL runs; failures surface with a `gbrain sources list` hint. Closes the silent-collapse-to-`default` bug class for extract. - `src/commands/import.ts` — `gbrain import [--source-id ]`: page import with the v0.34.2.0 path-set checkpoint described above. **v0.37.7.0 (#1167):** new `--source-id ` flag finally honored — pages route to the named source. Resolved via `resolveSourceWithTier()` at the boundary; the same flag is now consistent across `import`, `extract`, `graph-query`, and `sources current`. Pinned by `test/import-source-id.test.ts`. - `src/commands/graph-query.ts` — `gbrain graph-query [--type T] [--depth N] [--direction in|out|both] [--include-foreign]`: typed-edge relationship traversal (renders indented tree). **v0.37.7.0 (#1153):** foreign-edge footer always present (`X foreign edges (use --include-foreign to traverse)`) so cross-source edges never disappear silently; `--include-foreign` widens the SQL filter to walk them. Pinned by `test/graph-query.test.ts`. diff --git a/VERSION b/VERSION index ed7d8f3d3..269ce51d1 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.41.10.1 \ No newline at end of file +0.41.11.0 \ No newline at end of file diff --git a/llms-full.txt b/llms-full.txt index 9055da3e9..9f68c1aff 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -302,6 +302,7 @@ strict behavior when unset. - `src/core/enrichment-service.ts` — Global enrichment service: entity slug generation, tier auto-escalation, batch throttling - `src/core/data-research.ts` — Recipe validation, field extraction (MRR/ARR regex), dedup, tracker parsing, HTML stripping - `src/commands/embed.ts` — `gbrain embed [--stale|--all] [--slugs ...]`. v0.22.1 (#409, contributed by @atrevino47): `--stale` path now starts with `engine.countStaleChunks()` (single SELECT count(*) WHERE embedding IS NULL, ~50 bytes wire). On a fully-embedded brain that's a 1-line short-circuit — no further reads. When stale chunks exist, `engine.listStaleChunks()` returns just the chunks needing embeddings (slug + chunk_index + chunk_text + metadata, no `vector(1536)` payload). Caller groups by slug, embeds via OpenAI, re-upserts via `upsertChunks`. Replaces the prior page-walk that pulled every chunk's embedding column over the wire and discarded most. **v0.40.3.0:** all 16 `console.log` / `console.error` call sites migrated to `slog` / `serr` from `src/core/console-prefix.ts` so that when `runEmbedCore` runs inside a per-source `withSourcePrefix` scope (which the `gbrain sync --all` worker pool installs), every line carries the `[] ` prefix. Standalone `gbrain embed --stale` callers see identical output to v0.40.2 because slog/serr fall through to bare console fns outside the wrap. +- `src/commands/extract-conversation-facts.ts` + `src/core/cycle/conversation-facts-backfill.ts` (v0.41.11.0) — bulk fact extraction for long-form conversation pages. The recall-miss class on long iMessage/Slack imports (60K+ msg history; a chunk that reads only "Locker 93 code 9494" has no topical anchor because "cabin" was established 50K messages earlier) gets fixed by walking conversation/meeting/slack/email pages, splitting into time-windowed segments (30-min gap or 30-msg cap), prepending a topical/temporal header ("Conversation between A and B from to "), and running through the existing `extractFactsFromTurn()` so the resulting anchor-rich facts surface in `gbrain search` (which already blends facts). Architecture decisions absorbed from 2 Codex outside-voice rounds + 2 eng review passes: **strict per-source core** (`runExtractConversationFactsCore({sourceId, ...})` always takes one sourceId; CLI + cycle phase each do their own multi-source iteration because PHASE_SCOPE='source' is taxonomy-only per cycle.ts:131); **two-phase enumeration** (paginated `listPages({type, sourceId, limit:10})` so worst case is 10×25MB=250MB resident vs PR #1406's unbounded; per-page body cap MAX_PAGE_BODY_BYTES=25MB with `pages_skipped_too_large` counter surfaced in doctor); **page-global row_num accumulator** (facts unique index is `(source_id, source_markdown_slug, row_num)` per migration v51 — per-segment row_num would collide on segment 2); **page-level TERMINAL audit row** written to facts table after all segments commit (source='cli:extract-conversation-facts:terminal'; doctor's NOT EXISTS query matches the terminal row, NOT any-fact-for-slug, so partial-extraction pages stay in backlog); **optional `opts.budgetTracker?`** (when present, core uses it as-is — nested `withBudgetTracker` REPLACES per gateway.ts:2011; when absent, core auto-wraps with `BudgetTracker({maxCostUsd})`); **body read covers compiled_truth + timeline** (iMessage importers put history in timeline half; PR's `compiled_truth ?? ''` would silently drop it); **honors `facts.extraction_enabled` kill-switch** with `--override-disabled` escape; **--types LIST allowlist** (`conversation,meeting,slack,email`) with CLI default reading `cycle.conversation_facts_backfill.types` config as single source of truth; **fingerprint on sourceId only** so widening types config doesn't invalidate completed-page state; **string-encoded op-checkpoint** entries `"||"` for resume only (op_checkpoints stores string[] and is 7-day GC'd — durable audit is the facts table itself via terminal row); **`--background` via maybeBackground** (Minion handler `extract-conversation-facts` re-creates BudgetTracker from `data.max_cost_usd`; on `BudgetExhausted` mid-job, catches + persists + marks `completed` with `result.budget_exhausted=true` — NOT a failure). The companion cycle phase `conversation_facts_backfill` (default OFF; `gbrain config set cycle.conversation_facts_backfill.enabled true` opts in) iterates `listSources(engine)` directly, creates ONE brain-wide tracker per tick + wraps the loop in `withBudgetTracker` + passes the tracker into every per-source invocation via `opts.budgetTracker` so the core doesn't nest-replace. Two-layer cost AND walltime protection: per-source caps (`max_cost_usd=$1`, `max_walltime_min=20`) AND brain-wide caps (`max_total_cost_usd=$5`, `max_total_walltime_min=30`). Pinned by 27 unit cases in `test/extract-conversation-facts.test.ts` (parse/segment/render/checkpoint/fingerprint/terminal-audit/row_num/F2-kill-switch). Schema migration v94 in `src/core/migrate.ts` adds a partial index `idx_facts_extract_conversation_session ON facts(source_id, source_session) WHERE source LIKE 'cli:extract-conversation-facts%'` (v14 precedent: `transaction:false` + invalid-index pre-drop on Postgres; plain CREATE INDEX on PGLite) so the doctor backlog query stays fast on million-fact brains. `src/commands/doctor.ts:computeConversationFactsBacklogCheck` is 3-state (SKIPPED when `cycle.conversation_facts_backfill.enabled=false` — no opt-out noise debt per Eng-v2 C9; OK when caught up; WARN when >10 pages lack the terminal audit row, with paste-ready remediation step composable via `gbrain doctor --remediate`). `src/commands/sources.ts:runAudit` extended with `facts_backfill_estimate: {pages, est_segments, est_cost_usd, types}` so `gbrain sources audit ` previews cost before any run. Schema-pack (v0.41.11.0): `gbrain-base.yaml` promotes `conversation` (temporal, extractable) + `atom` (annotation, NOT extractable — atoms ARE the extracted form) into the base seed; `concept.extractable` flips true semantically (cosmetic on backstop path because backstop uses hardcoded `ELIGIBLE_TYPES` in `src/core/facts/eligibility.ts:51`, not pack extractable — the original concept-grandfather migration was solving a phantom, dropped per Codex T3). `ALL_PAGE_TYPES` in `src/core/types.ts` extended with the two new types. Plan + all decisions persisted at `~/.claude/plans/system-instruction-you-are-working-linear-unicorn.md` and the CEO plan at `~/.gstack/projects/garrytan-gbrain/ceo-plans/2026-05-25-conversation-retrieval-upgrade.md`. - `src/commands/extract.ts` — `gbrain extract links|timeline|all [--source fs|db] [--source-id ]`: batch link/timeline extraction. fs walks markdown files, db walks pages from the engine (mutation-immune snapshot iteration; use this for live brains with no local checkout). As of v0.12.1 there is no in-memory dedup pre-load — candidates are buffered 100 at a time and flushed via `addLinksBatch` / `addTimelineEntriesBatch`; `ON CONFLICT DO NOTHING` enforces uniqueness at the DB layer, and the `created` counter returns real rows inserted (truthful on re-runs). v0.22.1 (#417): `ExtractOpts.slugs?: string[]` enables incremental extract — when set, `extractForSlugs()` reads ONLY those slugs' files (single combined links+timeline pass) instead of the full directory walk. CLI `gbrain extract` keeps full-walk behavior; the cycle path threads sync's `pagesAffected` through. `walkMarkdownFiles(brainDir)` still runs at line 455 to build `allSlugs` for link resolution — see `TODOS.md` for replacing it with `engine.getAllSlugs()`. **v0.37.7.0 (#1204):** `--source-id ` flag scopes extraction to one brain source on federated brains. Resolved via `resolveSourceWithTier()` before any SQL runs; failures surface with a `gbrain sources list` hint. Closes the silent-collapse-to-`default` bug class for extract. - `src/commands/import.ts` — `gbrain import [--source-id ]`: page import with the v0.34.2.0 path-set checkpoint described above. **v0.37.7.0 (#1167):** new `--source-id ` flag finally honored — pages route to the named source. Resolved via `resolveSourceWithTier()` at the boundary; the same flag is now consistent across `import`, `extract`, `graph-query`, and `sources current`. Pinned by `test/import-source-id.test.ts`. - `src/commands/graph-query.ts` — `gbrain graph-query [--type T] [--depth N] [--direction in|out|both] [--include-foreign]`: typed-edge relationship traversal (renders indented tree). **v0.37.7.0 (#1153):** foreign-edge footer always present (`X foreign edges (use --include-foreign to traverse)`) so cross-source edges never disappear silently; `--include-foreign` widens the SQL filter to walk them. Pinned by `test/graph-query.test.ts`. diff --git a/package.json b/package.json index 5eb1e9aa6..cf71d621c 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "gbrain", - "version": "0.41.10.1", + "version": "0.41.11.0", "description": "Postgres-native personal knowledge brain with hybrid RAG search", "type": "module", "main": "src/core/index.ts", diff --git a/scripts/run-unit-parallel.sh b/scripts/run-unit-parallel.sh index e4bcdc245..f46b48cdb 100755 --- a/scripts/run-unit-parallel.sh +++ b/scripts/run-unit-parallel.sh @@ -211,11 +211,22 @@ heartbeat() { } heartbeat & HB_PID=$! -trap 'kill "$HB_PID" 2>/dev/null; wait "$HB_PID" 2>/dev/null' EXIT +# v0.41.11.0 cleanup: pkill children FIRST, then kill heartbeat. If we +# kill the heartbeat shell first, its current `sleep 10` is reparented +# to init/launchd and pkill -P can no longer find it (orphan). Order: +# children first while the parent PID is still findable, then parent. +# Known bash quirk: SIGTERM to a shell sleeping inside `sleep` doesn't +# propagate to the sleep child before the wait returns. Without this, +# each invocation of this script leaks ONE orphan sleep; CI's "orphan +# process cleanup" at end-of-job reports them as (unnamed) test failures. +# Seen on the garrytan/port-pr-1406 PR, 2 CI runs in a row, 6 orphans +# matching the 6 invocations in test/scripts/run-unit-parallel.test.ts. +trap 'pkill -P "$HB_PID" 2>/dev/null; kill "$HB_PID" 2>/dev/null; wait "$HB_PID" 2>/dev/null' EXIT # Wait for every shard. Don't care about wait's exit code. for pid in "${SHARD_PIDS[@]}"; do wait "$pid" 2>/dev/null || true; done +pkill -P "$HB_PID" 2>/dev/null kill "$HB_PID" 2>/dev/null wait "$HB_PID" 2>/dev/null trap - EXIT diff --git a/skills/migrations/v0.41.11.0.md b/skills/migrations/v0.41.11.0.md new file mode 100644 index 000000000..7b16f1a11 --- /dev/null +++ b/skills/migrations/v0.41.11.0.md @@ -0,0 +1,61 @@ +# v0.41.11.0 migration — conversation retrieval upgrade + +The v0.41.11.0 upgrade ships: + +1. Schema-pack additions: `conversation` (temporal, extractable) and `atom` (annotation, NOT extractable) promoted into `gbrain-base`. +2. New CLI command `gbrain extract-conversation-facts` for bulk fact extraction over long-form conversation pages. +3. Opt-in cycle phase `conversation_facts_backfill` (default OFF) for autopilot-driven backlog drain. +4. New doctor check `conversation_facts_backlog` (3-state; SKIPPED when feature disabled so no opt-out noise). +5. Schema migration v94: partial index on `facts(source_id, source_session) WHERE source LIKE 'cli:extract-conversation-facts%'` so the doctor query stays fast on big brains. + +`gbrain upgrade` + `gbrain apply-migrations --yes` handles the binary + schema. The cycle phase and doctor check require no agent action. + +## Agent action — when to run the backfill + +When the user mentions: +- "search isn't finding things in my chat history" +- "my iMessage / Slack import isn't searchable" +- "the page exists but search misses the answer" +- "long conversation pages aren't returning what I want" + +…then this command is the right tool. Don't run it speculatively — it costs real money against the user's Anthropic budget. + +## Recommended flow + +1. **Estimate cost first.** For each source the user wants to backfill: + ```bash + gbrain sources audit --json | jq '.facts_backfill_estimate' + ``` + Returns `{pages, est_segments, est_cost_usd, types}`. Show the user the cost estimate before recommending the run. + +2. **Run a dry-run** to preview segmentation: + ```bash + gbrain extract-conversation-facts --dry-run --slug + ``` + Confirms the segmenter parses the page correctly without spending any tokens. + +3. **Run with explicit cost cap.** Default cap is $5. Recommend matching the audit estimate (rounded up): + ```bash + gbrain extract-conversation-facts --background --max-cost-usd + gbrain jobs follow + ``` + Use `--background` so the user can keep working. The Minion job is resumable — if it hits the budget cap mid-run, re-running with a higher cap continues from where it left off. + +4. **Verify with doctor:** + ```bash + gbrain doctor --json | jq '.checks[] | select(.name == "conversation_facts_backlog")' + ``` + Should show OK with backlog: 0 after a successful complete run. + +5. **(Optional) Enable autopilot drain** if the user has steady conversation ingest: + ```bash + gbrain config set cycle.conversation_facts_backfill.enabled true + ``` + The cycle phase will then drain new conversation pages each tick under bounded per-source ($1/cycle) and brain-wide ($5/cycle) budgets. + +## Caveats + +- Pages over 25MB body are skipped (memory cap). Surface in doctor `details`; streaming for huge pages is a v0.42+ TODO. +- If `facts.extraction_enabled` is false, the command refuses. Pass `--override-disabled` only when the user explicitly opted out and now wants this one-time run. +- Extracted facts use `source = 'cli:extract-conversation-facts'`. To remove them in bulk (rare), the only path today is raw SQL: `DELETE FROM facts WHERE source LIKE 'cli:extract-conversation-facts%'`. A `gbrain forget --where` bulk flag is a v0.42+ TODO. +- The recall-quality eval (under `test/eval/conversation-extraction-quality.eval.ts` — added in this wave) is env-gated on `ANTHROPIC_API_KEY`. Run nightly or on-demand for quality verification; the hermetic wiring tests run in CI by default. diff --git a/src/cli.ts b/src/cli.ts index 324f892ef..de1ee3fac 100755 --- a/src/cli.ts +++ b/src/cli.ts @@ -35,7 +35,7 @@ for (const op of operations) { } // CLI-only commands that bypass the operation layer -const CLI_ONLY = new Set(['init', 'reinit-pglite', 'upgrade', 'post-upgrade', 'check-update', 'integrations', 'publish', 'check-backlinks', 'lint', 'report', 'import', 'export', 'files', 'embed', 'serve', 'call', 'config', 'doctor', 'migrate', 'eval', 'sync', 'extract', 'features', 'autopilot', 'graph-query', 'jobs', 'agent', 'apply-migrations', 'skillpack-check', 'skillpack', 'resolvers', 'integrity', 'repair-jsonb', 'orphans', 'sources', 'mounts', 'dream', 'check-resolvable', 'routing-eval', 'skillify', 'smoke-test', 'providers', 'storage', 'repos', 'code-def', 'code-refs', 'reindex-code', 'reindex-frontmatter', 'code-callers', 'code-callees', 'frontmatter', 'auth', 'friction', 'claw-test', 'book-mirror', 'takes', 'think', 'salience', 'anomalies', 'transcripts', 'models', 'remote', 'recall', 'forget', 'edges-backfill', 'cache', 'ze-switch', 'founder', 'brainstorm', 'lsd', 'schema', 'capture']); +const CLI_ONLY = new Set(['init', 'reinit-pglite', 'upgrade', 'post-upgrade', 'check-update', 'integrations', 'publish', 'check-backlinks', 'lint', 'report', 'import', 'export', 'files', 'embed', 'serve', 'call', 'config', 'doctor', 'migrate', 'eval', 'sync', 'extract', 'extract-conversation-facts', 'features', 'autopilot', 'graph-query', 'jobs', 'agent', 'apply-migrations', 'skillpack-check', 'skillpack', 'resolvers', 'integrity', 'repair-jsonb', 'orphans', 'sources', 'mounts', 'dream', 'check-resolvable', 'routing-eval', 'skillify', 'smoke-test', 'providers', 'storage', 'repos', 'code-def', 'code-refs', 'reindex-code', 'reindex-frontmatter', 'code-callers', 'code-callees', 'frontmatter', 'auth', 'friction', 'claw-test', 'book-mirror', 'takes', 'think', 'salience', 'anomalies', 'transcripts', 'models', 'remote', 'recall', 'forget', 'edges-backfill', 'cache', 'ze-switch', 'founder', 'brainstorm', 'lsd', 'schema', 'capture']); // CLI-only commands whose handlers print their own --help text. These are // excluded from the generic short-circuit so detailed per-command and // per-subcommand usage stays reachable. @@ -68,6 +68,10 @@ const CLI_ONLY_SELF_HELP = new Set([ // schema.ts printHelp() with the full 22+ verb taxonomy, not the // generic short-circuit's one-line stub. 'schema', + // v0.41.11.0 — extract-conversation-facts ships its own detailed HELP + // describing segment splitting + checkpointing + budget caps + the + // unified types config story. Route around the generic short-circuit. + 'extract-conversation-facts', ]); async function main() { @@ -732,7 +736,7 @@ function formatResult(opName: string, result: unknown): string { * `runRemoteDoctor` for thin-client installs. */ const THIN_CLIENT_REFUSED_COMMANDS = new Set([ - 'sync', 'embed', 'extract', 'migrate', 'apply-migrations', + 'sync', 'embed', 'extract', 'extract-conversation-facts', 'migrate', 'apply-migrations', 'repair-jsonb', 'orphans', 'integrity', 'serve', // v0.31.1 (CDX-2 op coverage matrix): more local-only commands 'dream', 'transcripts', 'storage', @@ -766,6 +770,7 @@ const THIN_CLIENT_REFUSE_HINTS: Record = { sync: 'sync runs on the host. Trigger a remote cycle with `gbrain remote ping` (queues an autopilot-cycle job).', embed: 'embed runs on the host as part of the autopilot cycle. `gbrain remote ping` triggers a full cycle including embed.', extract: 'extract runs on the host. Use `gbrain remote ping` to trigger a cycle including extract.', + 'extract-conversation-facts': 'extract-conversation-facts runs on the host (requires local engine + chat gateway). Run on the host machine.', migrate: "migrate runs on the host's local engine. Run on the host machine.", 'apply-migrations': 'schema migrations run on the host. SSH and run there.', 'repair-jsonb': 'repair-jsonb operates on the local DB only.', @@ -1299,6 +1304,11 @@ async function handleCliOnly(command: string, args: string[]) { await runExtract(engine, args); break; } + case 'extract-conversation-facts': { + const { runExtractConversationFacts } = await import('./commands/extract-conversation-facts.ts'); + await runExtractConversationFacts(engine, args); + break; + } case 'features': { const { runFeatures } = await import('./commands/features.ts'); await runFeatures(engine, args); diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index 752374979..36c92caec 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -2087,6 +2087,134 @@ export function computeNightlyQualityProbeHealthCheck( }; } +/** + * v0.41.11.0 — conversation_facts_backlog doctor check. + * + * 3-state status: + * - SKIPPED when cycle.conversation_facts_backfill.enabled=false + * (with paste-ready enable hint). No backlog enumeration; cheap probe. + * This is the Eng-v2 C9 "don't degrade health for opt-out users" gate. + * - OK when enabled=true AND backlog==0 OR no eligible pages exist. + * - WARN when enabled=true AND backlog>10. + * + * Backlog query uses the page-level TERMINAL audit row check (Eng-v2 + * C7), source-scoped via explicit predicate (Eng-v2 C2). Partial- + * extraction pages stay in backlog because the terminal row isn't + * written until ALL segments complete. + * + * Known approximation (documented in the details field): "complete" + * means "terminal row exists" which means "all segments completed in + * a prior run." A page with the terminal row from one run + new + * messages since shows OK until the next run picks up new messages + * and writes a fresh terminal row. The backlog is therefore an UPPER + * BOUND on "pages with NO extraction at all", not "pages whose facts + * are current." + */ +export async function computeConversationFactsBacklogCheck( + engine: BrainEngine, +): Promise { + const name = 'conversation_facts_backlog'; + try { + // Read the same config the cycle phase reads (Eng-v2 A2 single SoT). + const enabledRaw = await engine.getConfig( + 'cycle.conversation_facts_backfill.enabled', + ); + const enabled = enabledRaw != null && + !['false', '0', 'no', 'off', ''].includes(enabledRaw.trim().toLowerCase()); + + if (!enabled) { + return { + name, + status: 'ok', + message: + 'disabled (opt-in). Enable with: gbrain config set cycle.conversation_facts_backfill.enabled true', + }; + } + + // Resolve types from same key as cycle phase + CLI default. + const typesRaw = await engine.getConfig( + 'cycle.conversation_facts_backfill.types', + ); + let types = ['conversation', 'meeting', 'slack', 'email']; + if (typesRaw) { + try { + const parsed = JSON.parse(typesRaw); + if (Array.isArray(parsed)) { + const filtered = parsed.filter( + (t): t is string => typeof t === 'string', + ); + if (filtered.length > 0) types = filtered; + } + } catch { + // fall through to default + } + } + + // Source-scoped NOT EXISTS (Eng-v2 C2 + C7): + // - facts.source matches TERMINAL audit source + // - source_session matches terminal: + // - source_id matches page's source_id (cross-source safety) + const rows = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM pages p + WHERE p.type = ANY($1::text[]) + AND p.deleted_at IS NULL + AND NOT EXISTS ( + SELECT 1 FROM facts f + WHERE f.source = 'cli:extract-conversation-facts:terminal' + AND f.source_session = 'cli:extract-conversation-facts:terminal:' || p.slug + AND f.source_id = p.source_id + )`, + [types], + ); + + const backlog = Number(rows[0]?.count ?? 0); + + if (backlog === 0) { + return { + name, + status: 'ok', + message: 'all eligible pages have extraction terminal audit rows', + details: { + backlog, + types, + known_approximation: + 'backlog counts pages with NO extraction terminal row; pages with new messages since prior extraction may show OK until next run', + }, + }; + } + + if (backlog > 10) { + const fixHint = + 'gbrain extract-conversation-facts --background --max-cost-usd 5'; + return { + name, + status: 'warn', + message: `${backlog} eligible pages without extraction. Fix: ${fixHint}`, + details: { + backlog, + types, + fix_hint: fixHint, + known_approximation: + 'backlog counts pages with NO extraction terminal row; pages with new messages since prior extraction may show OK until next run', + }, + }; + } + + return { + name, + status: 'ok', + message: `${backlog} eligible page(s) below warn threshold (>10)`, + details: { backlog, types }, + }; + } catch (err) { + return { + name, + status: 'warn', + message: `backlog query failed: ${(err as Error).message}`, + }; + } +} + export async function checkSyncFreshness( engine: BrainEngine, opts?: { nowMs?: number }, @@ -2750,6 +2878,42 @@ export async function buildChecks( // Best-effort; audit-log read failure shouldn't stop doctor. } + // 3d.2 v0.41.11.0 — conversation_facts_backlog. 3-state status: + // SKIPPED-with-enable-hint when the cycle phase is disabled (opt-out + // users don't get noise debt); OK at backlog=0; WARN at backlog>10 + // with a paste-ready fix command. Emits a Remediation when WARN. + if (engine) { + try { + const check = await computeConversationFactsBacklogCheck(engine); + // Wire a remediation step on WARN so `gbrain doctor --remediate` + // picks it up. The CLI command honors --max-cost-usd; the + // remediation step caps at $5 default (matches doctor's max_usd + // default for the remediate flow). + if (check.status === 'warn') { + try { + const { makeRemediationStep } = await import('../core/remediation-step.ts'); + const remediation = makeRemediationStep({ + id: 'conversation_facts_backfill', + job: 'extract-conversation-facts', + params: { sourceId: 'default', maxCostUsd: 5 }, + severity: 'medium', + est_seconds: 600, + est_usd_cost: 5, + rationale: + 'Backfill facts for conversation/meeting/slack/email pages so chunker-loses-anchor recall misses get a topical-header-rich facts row to bind to.', + }); + check.remediation = [remediation]; + check.remediation_status = 'remediable'; + } catch { + // remediation factory unavailable → check still surfaces backlog + } + } + checks.push(check); + } catch { + // Best-effort; backlog query failure shouldn't stop doctor. + } + } + // 3e. home_dir_in_worktree (v0.35.8.0). Walks up from `gbrainPath()` // looking for a `.git` directory OR file. If found, warns: `~/.gbrain/` // lives inside a git worktree, so an accidental `git add` from the diff --git a/src/commands/extract-conversation-facts.ts b/src/commands/extract-conversation-facts.ts new file mode 100644 index 000000000..901f24c8c --- /dev/null +++ b/src/commands/extract-conversation-facts.ts @@ -0,0 +1,1115 @@ +/** + * gbrain extract-conversation-facts — batch fact extraction for + * conversation pages (and adjacent long-form types). + * + * Background + * ---------- + * Long-running conversation pages (imported chat logs, transcripts, + * etc.) can be very large — tens of thousands of messages spanning + * years. The default embedding pipeline chunks them into ~300-word + * blocks and prepends a tiny page-title hint. Enough for short pages, + * but it falls apart on long-running conversations: + * + * - A user searches for "mountain cabin lock code" but the chunk + * that contains the literal code reads only "Locker 93 code 9494" + * — no mention of "cabin", "mountain", or any topical anchor. + * - Retrieval misses, because the chunk-level embedding can't see + * the surrounding 50K messages of context that establish the topic. + * + * The facts table doesn't have this problem. Each row is a discrete + * claim with its own embedding and entity linkage, and `gbrain search` + * blends facts into the result set. The extraction pipeline that + * builds facts (src/core/facts/extract.ts) is already wired into + * real-time MCP turns and the post-sync backstop — but had never been + * run as a bulk backfill over imported chat history. + * + * This command closes that gap. + * + * Architecture decisions (locked by CEO + 3-round spec review + 2-round + * Codex outside voice + 2-pass eng review): + * + * - Strict per-source core. `runExtractConversationFactsCore` ALWAYS + * takes one sourceId. Multi-source iteration lives in the CLI + * wrapper (and in the cycle phase wrapper, separately). + * - Two-phase memory-bounded enumeration. Use paginated + * `listPages({type, sourceId, limit: PAGE_LIST_BATCH})` so worst + * case is BATCH × 25MB per batch (currently 10 × 25MB = 250MB + * bounded). Per-page body cap drops oversize before parsing. + * - Body read covers compiled_truth + timeline. parseMarkdown splits + * conversation imports across both columns; reading only + * compiled_truth silently drops half on iMessage/Slack imports. + * - Page-global row_num accumulator. facts table unique index is + * (source_id, source_markdown_slug, row_num); per-segment row_num + * would collide on segment 2. Per-page counter increments across + * segments. + * - Terminal audit row on completion. After all segments commit, one + * extra fact row with source='cli:extract-conversation-facts:terminal' + * marks the page complete. Doctor's backlog query checks for the + * terminal row, NOT any fact — partial extraction → no terminal → + * next run resumes. + * - Optional budgetTracker via opts. If a tracker is in opts, use it + * as-is (NO `withBudgetTracker` wrap, which would REPLACE the active + * tracker per gateway.ts AsyncLocalStorage semantics, defeating an + * outer brain-wide cap). If absent, auto-create from `maxCostUsd` + * and wrap. Callers explicitly own lifecycle. + * - Op-checkpoint string-encoded resume state. Entries are + * "||" strings (op_checkpoints stores + * string[] only and is GC'd at 7 days; durable audit is the facts + * table itself via the terminal row). + * - Fingerprint on sourceId only. Widening cycle.types config does + * NOT invalidate completed-page state. + * + * Honor brain-wide kill-switch: + * `facts.extraction_enabled=false` config blocks. Pass + * `--override-disabled` to force-run. + */ + +import type { BrainEngine, NewFact } from '../core/engine.ts'; +import type { Page } from '../core/types.ts'; +import { + extractFactsFromTurn, + isFactsExtractionEnabled, +} from '../core/facts/extract.ts'; +import { isAvailable, withBudgetTracker } from '../core/ai/gateway.ts'; +import { BudgetTracker, BudgetExhausted } from '../core/budget/budget-tracker.ts'; +import { listSources } from '../core/sources-ops.ts'; +import { + loadOpCheckpoint, + recordCompleted, + clearOpCheckpoint, + type OpCheckpointKey, +} from '../core/op-checkpoint.ts'; +import { createProgress } from '../core/progress.ts'; +import { getCliOptions, cliOptsToProgressOptions, maybeBackground } from '../core/cli-options.ts'; +import { loadConfig } from '../core/config.ts'; +import { createHash } from 'crypto'; + +// --------------------------------------------------------------------------- +// Tunables (exported for tests). +// --------------------------------------------------------------------------- + +/** Maximum gap between adjacent messages before we cut a new segment. */ +export const DEFAULT_SEGMENT_GAP_MINUTES = 30; + +/** + * Hard cap on messages per segment, regardless of timing. + * Tuned down from PR's 50 → 30 (Eng-v2 T5): combined with the 6500-char + * SEGMENT_TEXT_CHAR_LIMIT, this keeps headroom under extract.ts's + * MAX_TURN_TEXT_CHARS = 8000 so tail facts in dense Slack/email + * segments don't vanish silently. + */ +export const DEFAULT_SEGMENT_MAX_MESSAGES = 30; + +/** Minimum messages required for a segment to be worth extracting. */ +export const MIN_SEGMENT_MESSAGES = 2; + +/** Delay between extractor calls so we don't burst the chat provider. */ +export const DEFAULT_INTER_CALL_SLEEP_MS = 200; + +/** + * Cap on character length of the rendered segment passed to the extractor. + * Tuned down from PR's 7500 → 6500 (Eng-v2 T5) to leave headroom for the + * topical/temporal header (~500 chars typical, up to ~1500 with a long + * participant list) under extract.ts's MAX_TURN_TEXT_CHARS = 8000. + */ +export const SEGMENT_TEXT_CHAR_LIMIT = 6500; + +/** + * Hard cap on per-page body bytes (compiled_truth + timeline). Pages + * exceeding the cap are skipped to bound worker memory (Eng A2). A + * streaming/per-segment-fetch path for 50MB+ iMessage histories is a + * v0.42+ follow-up. + */ +export const MAX_PAGE_BODY_BYTES = 25 * 1024 * 1024; + +/** Default cost cap when no tracker is passed explicitly. */ +export const DEFAULT_MAX_COST_USD = 5.0; + +/** + * Allowlist of page types this command operates on. Mirrors + * cycle.conversation_facts_backfill.types config default. CLI's + * `--types` flag is an explicit per-run override; cycle config is + * the single source of truth. + */ +export const ALLOWED_TYPES = ['conversation', 'meeting', 'slack', 'email'] as const; +export type AllowedType = (typeof ALLOWED_TYPES)[number]; + +/** + * Pagination batch size for listPages enumeration. Per-batch memory + * worst case = BATCH × MAX_PAGE_BODY_BYTES = 250MB at default 10 + * (Eng-v2 C8 — bounded vs PR's unbounded listPages limit:500 = 12.5GB). + */ +export const PAGE_LIST_BATCH = 10; + +/** Op name for the checkpoint primitive. */ +export const CHECKPOINT_OP = 'extract-conversation-facts'; + +/** + * Source string written on per-segment facts. Doctor queries the + * TERMINAL variant below; this variant marks individual fact provenance. + */ +export const PER_SEGMENT_SOURCE_PREFIX = 'cli:extract-conversation-facts'; + +/** + * Source string written on the page-level terminal audit row (Eng-v2 C7). + * Doctor's backlog query matches THIS source + source_session, not + * the per-segment source. Partial extraction = no terminal row = page + * stays in backlog. + */ +export const TERMINAL_AUDIT_SOURCE = 'cli:extract-conversation-facts:terminal'; + +// --------------------------------------------------------------------------- +// Public types. +// --------------------------------------------------------------------------- + +export interface ConversationMessage { + speaker: string; + /** ISO 8601 timestamp parsed from the rendered message line. */ + timestamp: string; + text: string; +} + +export interface ConversationSegment { + messages: ConversationMessage[]; + startIso: string; + endIso: string; + participants: string[]; +} + +/** + * Core function opts. Strict — `sourceId` is always required (Eng-v2 A1). + * Multi-source iteration is the caller's job. + */ +export interface ExtractConversationFactsCoreOpts { + /** REQUIRED. Strict per-source contract. */ + sourceId: string; + /** + * Page types to walk. Reads cycle config when omitted. + * Allowlist enforced via ALLOWED_TYPES. + */ + types?: AllowedType[]; + /** Process a single page; otherwise iterate all matching pages in the source. */ + slug?: string; + /** Show would-do counts without writing facts or advancing checkpoint. */ + dryRun?: boolean; + /** Cap pages processed in this invocation. */ + limit?: number; + /** ISO watermark; messages older than this are filtered out. */ + sinceIso?: string; + /** Clear this page's resume entry before processing. */ + force?: boolean; + /** Delay between extractor calls. */ + sleepMs?: number; + /** Max segments to process per page (0 = unlimited). */ + segmentLimit?: number; + /** + * Cost cap (USD). Used when budgetTracker is NOT passed; core + * creates a fresh tracker. Default DEFAULT_MAX_COST_USD. + */ + maxCostUsd?: number; + /** + * Externally-managed BudgetTracker (Eng-v2 C5). If present, core + * uses it as-is — no `withBudgetTracker` wrap. Cycle phase passes + * a brain-wide tracker; CLI/Minion pass nothing. + */ + budgetTracker?: BudgetTracker; + /** Bypass `facts.extraction_enabled=false`. Power-user escape. */ + overrideDisabled?: boolean; +} + +export interface ExtractConversationFactsResult { + pages_considered: number; + pages_processed: number; + pages_skipped: number; + pages_skipped_too_large: number; + pages_skipped_disappeared: number; + segments_processed: number; + facts_extracted: number; + facts_inserted: number; + budget_exhausted?: boolean; + spent_usd?: number; +} + +// --------------------------------------------------------------------------- +// Message parsing — lines like: +// **Name** (YYYY-MM-DD H:MM AM/PM): text +// Unmatched lines become continuations of the prior message (multi-line +// imessage bodies). +// --------------------------------------------------------------------------- + +const MESSAGE_LINE_RX = + /^\*\*(.+?)\*\*\s*\((\d{4}-\d{2}-\d{2})\s+(\d{1,2}):(\d{2})\s*(AM|PM|am|pm)?\)\s*:\s*(.*)$/; + +export function parseConversationMessages(body: string): ConversationMessage[] { + if (!body) return []; + const out: ConversationMessage[] = []; + for (const rawLine of body.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; + const m = MESSAGE_LINE_RX.exec(line); + if (m) { + const [, speaker, date, hStr, mStr, ampmRaw, text] = m; + let hour = Number(hStr); + const minute = Number(mStr); + const ampm = (ampmRaw || '').toUpperCase(); + if (ampm === 'PM' && hour < 12) hour += 12; + if (ampm === 'AM' && hour === 12) hour = 0; + const iso = `${date}T${String(hour).padStart(2, '0')}:${String(minute).padStart(2, '0')}:00Z`; + out.push({ + speaker: speaker.trim(), + timestamp: iso, + text: (text || '').trim(), + }); + } else if (out.length > 0) { + const last = out[out.length - 1]; + last.text = last.text ? `${last.text}\n${line}` : line; + } + } + return out; +} + +// --------------------------------------------------------------------------- +// Segment splitting. +// --------------------------------------------------------------------------- + +export interface SplitSegmentsOpts { + gapMinutes?: number; + maxMessages?: number; + /** Drop messages with timestamp <= this ISO before splitting. */ + sinceIso?: string; +} + +export function splitIntoSegments( + messages: ConversationMessage[], + opts: SplitSegmentsOpts = {}, +): ConversationSegment[] { + const gapMs = (opts.gapMinutes ?? DEFAULT_SEGMENT_GAP_MINUTES) * 60_000; + const maxMessages = opts.maxMessages ?? DEFAULT_SEGMENT_MAX_MESSAGES; + const sinceMs = opts.sinceIso ? Date.parse(opts.sinceIso) : NaN; + + const filtered = Number.isFinite(sinceMs) + ? messages.filter((m) => Date.parse(m.timestamp) > sinceMs) + : messages.slice(); + + const out: ConversationSegment[] = []; + let cur: ConversationMessage[] = []; + let lastTs: number | null = null; + + const flush = () => { + if (cur.length < MIN_SEGMENT_MESSAGES) { + cur = []; + return; + } + const seen = new Set(); + const participants: string[] = []; + for (const m of cur) { + if (!seen.has(m.speaker)) { + seen.add(m.speaker); + participants.push(m.speaker); + } + } + out.push({ + messages: cur, + startIso: cur[0].timestamp, + endIso: cur[cur.length - 1].timestamp, + participants, + }); + cur = []; + }; + + for (const m of filtered) { + const ts = Date.parse(m.timestamp); + if (!Number.isFinite(ts)) continue; + if (lastTs !== null && ts - lastTs > gapMs) flush(); + cur.push(m); + lastTs = ts; + if (cur.length >= maxMessages) { + flush(); + lastTs = null; + } + } + flush(); + return out; +} + +// --------------------------------------------------------------------------- +// Segment rendering with topical/temporal header. +// --------------------------------------------------------------------------- + +export function renderSegmentForExtraction( + pageTitle: string, + segment: ConversationSegment, +): string { + const header = [ + `Page: ${pageTitle}`, + `Conversation between ${segment.participants.join(' and ')} from ${segment.startIso} to ${segment.endIso}`, + '---', + ].join('\n'); + const body = segment.messages + .map((m) => `${m.speaker} (${m.timestamp}): ${m.text}`) + .join('\n'); + const full = `${header}\n${body}`; + if (full.length <= SEGMENT_TEXT_CHAR_LIMIT) return full; + // Truncate from the end of the body, keeping the header intact so the + // extractor still sees the topical anchor. + const slack = SEGMENT_TEXT_CHAR_LIMIT - header.length - 16; + return `${header}\n${body.slice(0, Math.max(0, slack))}\n…(truncated)`; +} + +// --------------------------------------------------------------------------- +// Fingerprint — sourceId-only (Eng-v2 A3). Widening types config does NOT +// invalidate prior completion state. +// --------------------------------------------------------------------------- + +export function extractConversationFactsFingerprint(opts: { sourceId: string }): string { + const canonical = JSON.stringify({ sourceId: opts.sourceId }); + return createHash('sha256').update(canonical).digest('hex').slice(0, 8); +} + +function checkpointKey(sourceId: string): OpCheckpointKey { + return { op: CHECKPOINT_OP, fingerprint: extractConversationFactsFingerprint({ sourceId }) }; +} + +// --------------------------------------------------------------------------- +// Op-checkpoint helpers — string-encoded "||" entries. +// --------------------------------------------------------------------------- + +interface DecodedEntry { + sourceId: string; + slug: string; + endIso: string; +} + +export function encodeCheckpointEntry(sourceId: string, slug: string, endIso: string): string { + // Slugs are validated to [a-z0-9_/-] + CJK; sourceId is [a-z0-9_-]. + // Neither contains the pipe character, so the delimiter is safe. + return `${sourceId}|${slug}|${endIso}`; +} + +export function decodeCheckpointEntry(entry: string): DecodedEntry | null { + // Split on first two pipes only — endIso has no pipes either. + const i1 = entry.indexOf('|'); + if (i1 < 0) return null; + const i2 = entry.indexOf('|', i1 + 1); + if (i2 < 0) return null; + return { + sourceId: entry.slice(0, i1), + slug: entry.slice(i1 + 1, i2), + endIso: entry.slice(i2 + 1), + }; +} + +/** Returns the newest endIso for a given (sourceId, slug), or null if absent. */ +function findCompletedEndIso( + entries: string[], + sourceId: string, + slug: string, +): string | null { + let best: string | null = null; + for (const e of entries) { + const d = decodeCheckpointEntry(e); + if (!d) continue; + if (d.sourceId !== sourceId) continue; + if (d.slug !== slug) continue; + if (best === null || d.endIso > best) best = d.endIso; + } + return best; +} + +/** Returns entries with all (sourceId, slug)-matching rows stripped. */ +function filterOutSlug(entries: string[], sourceId: string, slug: string): string[] { + return entries.filter((e) => { + const d = decodeCheckpointEntry(e); + if (!d) return true; + return !(d.sourceId === sourceId && d.slug === slug); + }); +} + +// --------------------------------------------------------------------------- +// Body cap (Eng A2). +// --------------------------------------------------------------------------- + +function pageBodyBytes(page: Page): number { + const compiled = page.compiled_truth ?? ''; + const timeline = page.timeline ?? ''; + return Buffer.byteLength(compiled, 'utf8') + Buffer.byteLength(timeline, 'utf8'); +} + +function readPageBody(page: Page): string { + // F1: read BOTH compiled_truth AND timeline; iMessage importers + // place chronological message stream in timeline. + const compiled = page.compiled_truth ?? ''; + const timeline = page.timeline ?? ''; + if (!compiled) return timeline; + if (!timeline) return compiled; + return `${compiled}\n\n${timeline}`; +} + +// --------------------------------------------------------------------------- +// Types config resolver (Eng-v2 A2 — unified single source of truth). +// --------------------------------------------------------------------------- + +const TYPES_CONFIG_KEY = 'cycle.conversation_facts_backfill.types'; + +async function resolveTypesFromConfig( + engine: BrainEngine, + explicit?: AllowedType[], +): Promise { + if (explicit && explicit.length > 0) return explicit; + const raw = await engine.getConfig(TYPES_CONFIG_KEY); + if (raw) { + try { + const parsed = JSON.parse(raw); + if (Array.isArray(parsed)) { + const filtered = parsed + .filter((t): t is string => typeof t === 'string') + .filter((t): t is AllowedType => (ALLOWED_TYPES as readonly string[]).includes(t)); + if (filtered.length > 0) return filtered; + } + } catch { + // fall through to default + } + } + // Default: full allowlist when no config and no explicit override. + // Mirrors cycle.conversation_facts_backfill.types default. + return [...ALLOWED_TYPES]; +} + +// --------------------------------------------------------------------------- +// Core extraction loop (single source). +// --------------------------------------------------------------------------- + +interface ExtractCoreState { + result: ExtractConversationFactsResult; + engine: BrainEngine; + sourceId: string; + dryRun: boolean; + sleepMs: number; + segmentLimit: number; + types: AllowedType[]; + signal: AbortSignal | undefined; +} + +async function processPage( + state: ExtractCoreState, + page: Page, + sinceIso: string | undefined, + cpEntries: string[], + rowNumStart: number, +): Promise<{ newEndIso: string | null; rowNumAfter: number; cpEntriesAfter: string[] }> { + state.result.pages_considered++; + + // Body cap check first — pre-parse, pre-segment, pre-extraction. + const bytes = pageBodyBytes(page); + if (bytes > MAX_PAGE_BODY_BYTES) { + state.result.pages_skipped_too_large++; + process.stderr.write( + `[extract-conversation-facts] SKIP ${page.slug}: ${(bytes / 1024 / 1024).toFixed(1)}MB exceeds 25MB cap\n`, + ); + return { newEndIso: null, rowNumAfter: rowNumStart, cpEntriesAfter: cpEntries }; + } + + const body = readPageBody(page); + const messages = parseConversationMessages(body); + const segments = splitIntoSegments(messages, { sinceIso }); + if (segments.length === 0) { + state.result.pages_skipped++; + return { newEndIso: null, rowNumAfter: rowNumStart, cpEntriesAfter: cpEntries }; + } + + let rowNum = rowNumStart; + let entries = cpEntries; + let newestEnd: string | null = null; + let segmentsThisPage = 0; + let pageInsertedTotal = 0; + + for (const seg of segments) { + if (state.segmentLimit > 0 && segmentsThisPage >= state.segmentLimit) break; + if (state.signal?.aborted) throw new Error('aborted'); + + const text = renderSegmentForExtraction(page.title || page.slug, seg); + const sessionId = `${PER_SEGMENT_SOURCE_PREFIX}:${page.slug}`; + + let extracted: Awaited> = []; + try { + extracted = await extractFactsFromTurn({ + turnText: text, + sessionId, + source: PER_SEGMENT_SOURCE_PREFIX, + engine: state.engine, + abortSignal: state.signal, + }); + } catch (err) { + if (isAbortError(err)) throw err; + if (err instanceof BudgetExhausted) throw err; + // Per-segment LLM failures are best-effort; loop continues. + process.stderr.write( + `[extract-conversation-facts] segment ${seg.startIso}..${seg.endIso} extractor failed: ${(err as Error).message}\n`, + ); + extracted = []; + } + + state.result.segments_processed++; + segmentsThisPage++; + state.result.facts_extracted += extracted.length; + + if (!state.dryRun && extracted.length > 0) { + // Eng-v2 C1 / E11: page-global row_num. Each fact in this batch gets + // a unique row_num within (source_id, source_markdown_slug); the + // accumulator increments across the segment loop. + const rows = extracted.map((fact, i) => ({ + ...fact, + row_num: rowNum + i, + source_markdown_slug: page.slug, + source: PER_SEGMENT_SOURCE_PREFIX, + source_session: sessionId, + context: + fact.context ?? `from ${page.slug} segment ${seg.startIso}..${seg.endIso}`, + })); + try { + const ins = await state.engine.insertFacts(rows, { source_id: state.sourceId }); // gbrain-allow-direct-insert: canonical bulk extraction path for conversation pages — fences-as-system-of-record doesn't apply because conversations don't carry `## Facts` fences (the chat-log shape is the source-of-truth) + pageInsertedTotal += ins.inserted; + state.result.facts_inserted += ins.inserted; + } catch (err) { + if (isAbortError(err)) throw err; + // Batch failure is best-effort — segment is the transactional + // boundary, so a duplicate-key or constraint error rolls back + // this segment only. Loop continues. + process.stderr.write( + `[extract-conversation-facts] segment ${seg.startIso}..${seg.endIso} insertFacts failed: ${(err as Error).message}\n`, + ); + } + rowNum += extracted.length; + } else { + // dry-run: count for reporting, no DB write. + rowNum += extracted.length; + } + + newestEnd = seg.endIso; + if (state.sleepMs > 0) await sleep(state.sleepMs); + } + + // Eng-v2 C7 / E16: write terminal audit row after all segments commit + // successfully. Only run when not dry-run AND we got through every + // segment (no break on segmentLimit; that's an explicit partial run). + const fullyProcessed = + state.segmentLimit === 0 || segmentsThisPage < state.segmentLimit; + if (!state.dryRun && fullyProcessed && newestEnd !== null) { + try { + await writeTerminalAuditRow(state.engine, state.sourceId, page.slug, rowNum); + rowNum++; + } catch (err) { + if (isAbortError(err)) throw err; + // Terminal-row write failure: page is NOT marked complete; next + // run resumes. Loud stderr so users see partial-success state. + process.stderr.write( + `[extract-conversation-facts] ${page.slug} terminal audit write failed: ${(err as Error).message}\n`, + ); + // Suppress the resume-state update so doctor still flags this page. + newestEnd = null; + } + } + + if (!state.dryRun && newestEnd !== null) { + // Update op-checkpoint: filter out prior entries for this slug, + // append the newest end. --force clears prior; normal case advances. + entries = filterOutSlug(entries, state.sourceId, page.slug); + entries.push(encodeCheckpointEntry(state.sourceId, page.slug, newestEnd)); + } + + process.stderr.write( + `[extract-conversation-facts] ${page.slug}: ${pageInsertedTotal}/${state.result.facts_extracted - (state.result.facts_extracted - pageInsertedTotal)} facts inserted across ${segmentsThisPage} segments\n`, + ); + + state.result.pages_processed++; + return { newEndIso: newestEnd, rowNumAfter: rowNum, cpEntriesAfter: entries }; +} + +async function writeTerminalAuditRow( + engine: BrainEngine, + sourceId: string, + slug: string, + rowNum: number, +): Promise { + const fact: NewFact & { row_num: number; source_markdown_slug: string } = { + fact: 'EXTRACTION_COMPLETE', + kind: 'fact', + entity_slug: null, + source: TERMINAL_AUDIT_SOURCE, + source_session: `${TERMINAL_AUDIT_SOURCE}:${slug}`, + confidence: 1.0, + notability: 'low', + row_num: rowNum, + source_markdown_slug: slug, + }; + await engine.insertFacts([fact], { source_id: sourceId }); // gbrain-allow-direct-insert: page-level TERMINAL audit row (Codex C7 / E16) marks extraction completion in the durable facts table — there's no fence equivalent because this is internal audit state, not user-facing knowledge +} + +/** + * Core entry point — one source per call. Caller (CLI / Minion / cycle + * phase) handles multi-source iteration externally. + * + * Budget tracker semantics: + * - If `opts.budgetTracker` is set: use it as-is (no wrap). Caller + * owns lifecycle; nested wrap would REPLACE the active tracker. + * - If absent: create a fresh tracker scoped to `opts.maxCostUsd` + * and run the body inside `withBudgetTracker`. + */ +export async function runExtractConversationFactsCore( + engine: BrainEngine, + opts: ExtractConversationFactsCoreOpts, + signal?: AbortSignal, +): Promise { + const sourceId = opts.sourceId; + if (!sourceId) { + throw new Error('runExtractConversationFactsCore: opts.sourceId is required'); + } + + const result: ExtractConversationFactsResult = { + pages_considered: 0, + pages_processed: 0, + pages_skipped: 0, + pages_skipped_too_large: 0, + pages_skipped_disappeared: 0, + segments_processed: 0, + facts_extracted: 0, + facts_inserted: 0, + }; + + // F2: honor brain-wide kill-switch unless overridden. + if (!opts.overrideDisabled) { + const enabled = await isFactsExtractionEnabled(engine); + if (!enabled) { + throw new Error( + 'facts.extraction_enabled=false; pass --override-disabled to force-run', + ); + } + } + + const types = await resolveTypesFromConfig(engine, opts.types); + const dryRun = !!opts.dryRun; + const sleepMs = opts.sleepMs ?? DEFAULT_INTER_CALL_SLEEP_MS; + const segmentLimit = opts.segmentLimit ?? 0; + + const state: ExtractCoreState = { + result, + engine, + sourceId, + dryRun, + sleepMs, + segmentLimit, + types, + signal, + }; + + // Run body. Either inside the externally-provided tracker scope (no + // wrap; opts.budgetTracker is in scope upstream OR caller passes it + // explicitly via withBudgetTracker), or inside a fresh local wrap. + const body = async () => { + const cpKey = checkpointKey(sourceId); + let cpEntries = await loadOpCheckpoint(engine, cpKey); + + if (opts.slug) { + const page = await engine.getPage(opts.slug, { sourceId }); + if (!page) { + result.pages_skipped_disappeared++; + return; + } + if (!types.includes(page.type as AllowedType)) { + result.pages_skipped++; + return; + } + + if (opts.force) { + cpEntries = filterOutSlug(cpEntries, sourceId, opts.slug); + } + const checkpointed = findCompletedEndIso(cpEntries, sourceId, opts.slug); + const sinceIso = pickLaterIso(checkpointed, opts.sinceIso); + + const rowNumStart = await peekRowNumStart(engine, sourceId, opts.slug); + const { cpEntriesAfter } = await processPage(state, page, sinceIso, cpEntries, rowNumStart); + cpEntries = cpEntriesAfter; + } else { + // Multi-page enumeration: paginate per-type at small batch size to + // bound memory (Eng-v2 C8 — 10 × 25MB = 250MB worst case). + let processedPagesCount = 0; + pageLoop: for (const type of types) { + let offset = 0; + // eslint-disable-next-line no-constant-condition + while (true) { + if (signal?.aborted) throw new Error('aborted'); + if (opts.limit && processedPagesCount >= opts.limit) break pageLoop; + + const batch = await engine.listPages({ + type, + sourceId, + limit: PAGE_LIST_BATCH, + offset, + }); + if (batch.length === 0) break; + + for (const page of batch) { + if (opts.limit && processedPagesCount >= opts.limit) break pageLoop; + + const slug = page.slug; + const checkpointed = findCompletedEndIso(cpEntries, sourceId, slug); + + // Terminal audit row check — if this page has the terminal + // marker AND not --force, skip immediately (cheap probe via + // the checkpointed value covers the recent-run case; the + // expensive query is doctor's job, not per-page). + const sinceIso = pickLaterIso(checkpointed, opts.sinceIso); + const rowNumStart = await peekRowNumStart(engine, sourceId, slug); + const { cpEntriesAfter } = await processPage( + state, + page, + sinceIso, + cpEntries, + rowNumStart, + ); + cpEntries = cpEntriesAfter; + processedPagesCount++; + } + + offset += batch.length; + if (batch.length < PAGE_LIST_BATCH) break; + + // Persist checkpoint between batches so a crash mid-walk + // doesn't lose all progress. + if (!dryRun) { + await recordCompleted(engine, checkpointKey(sourceId), cpEntries); + } + } + } + } + + // Final checkpoint flush. + if (!dryRun) { + await recordCompleted(engine, checkpointKey(sourceId), cpEntries); + } + }; + + try { + if (opts.budgetTracker) { + // Caller-managed scope — use as-is, no wrap (nested wrap REPLACES + // tracker per gateway.ts AsyncLocalStorage semantics). + await body(); + } else { + const tracker = new BudgetTracker({ + maxCostUsd: opts.maxCostUsd ?? DEFAULT_MAX_COST_USD, + label: `extract-conversation-facts:${sourceId}`, + }); + try { + await withBudgetTracker(tracker, body); + } finally { + result.spent_usd = tracker.totalSpent; + } + } + } catch (err) { + if (err instanceof BudgetExhausted) { + result.budget_exhausted = true; + if (opts.budgetTracker) { + result.spent_usd = opts.budgetTracker.totalSpent; + } + // Return partial result — caller (CLI / Minion) decides how to + // surface. NOT a thrown failure. + return result; + } + throw err; + } + + return result; +} + +/** + * Look up the max row_num already in facts for this (source_id, slug), + * so the page-global accumulator continues from the right place on resume. + */ +async function peekRowNumStart( + engine: BrainEngine, + sourceId: string, + slug: string, +): Promise { + try { + const rows = await engine.executeRaw<{ max_row: number | null }>( + `SELECT COALESCE(MAX(row_num), -1) AS max_row + FROM facts + WHERE source_id = $1 AND source_markdown_slug = $2`, + [sourceId, slug], + ); + const maxRow = rows[0]?.max_row ?? -1; + return Number(maxRow) + 1; + } catch { + // Pre-migration brains may not have source_markdown_slug populated. + // Fall back to 0; insertFacts will fail with a clearer error if + // there's a real collision. + return 0; + } +} + +// --------------------------------------------------------------------------- +// CLI parsing + handler. +// --------------------------------------------------------------------------- + +interface ParsedArgs { + sourceId?: string; + types?: AllowedType[]; + slug?: string; + dryRun?: boolean; + limit?: number; + sinceIso?: string; + force?: boolean; + sleepMs?: number; + segmentLimit?: number; + maxCostUsd?: number; + overrideDisabled?: boolean; + yes?: boolean; + help?: boolean; + error?: string; +} + +function parseArgs(args: string[]): ParsedArgs { + const out: ParsedArgs = {}; + for (let i = 0; i < args.length; i++) { + const a = args[i]; + if (a === '--help' || a === '-h') { out.help = true; continue; } + if (a === '--dry-run') { out.dryRun = true; continue; } + if (a === '--force') { out.force = true; continue; } + if (a === '--yes' || a === '-y') { out.yes = true; continue; } + if (a === '--override-disabled') { out.overrideDisabled = true; continue; } + if (a === '--slug') { out.slug = args[++i]; continue; } + if (a === '--source-id') { out.sourceId = args[++i]; continue; } + if (a === '--since') { out.sinceIso = args[++i]; continue; } + if (a === '--types') { + const v = args[++i] ?? ''; + const parts = v.split(',').map((s) => s.trim()).filter(Boolean); + const bad = parts.filter((p) => !(ALLOWED_TYPES as readonly string[]).includes(p)); + if (bad.length > 0) { + out.error = `Unknown type(s) in --types: ${bad.join(', ')}. Allowed: ${ALLOWED_TYPES.join(', ')}`; + return out; + } + out.types = parts as AllowedType[]; + continue; + } + if (a === '--limit') { + const n = parseInt(args[++i] ?? '', 10); + if (Number.isFinite(n) && n > 0) out.limit = n; + continue; + } + if (a === '--sleep') { + const n = parseInt(args[++i] ?? '', 10); + if (Number.isFinite(n) && n >= 0) out.sleepMs = n; + continue; + } + if (a === '--segment-limit') { + const n = parseInt(args[++i] ?? '', 10); + if (Number.isFinite(n) && n >= 0) out.segmentLimit = n; + continue; + } + if (a === '--max-cost-usd') { + const n = parseFloat(args[++i] ?? ''); + if (Number.isFinite(n) && n > 0) out.maxCostUsd = n; + continue; + } + if (a.startsWith('--')) { + out.error = `Unknown flag: ${a}`; + return out; + } + } + if (out.sinceIso) { + const ms = Date.parse(out.sinceIso); + if (!Number.isFinite(ms)) { + out.error = `Invalid --since: ${out.sinceIso}`; + } + } + return out; +} + +const HELP = `Usage: gbrain extract-conversation-facts [options] + +Batch-extract facts from conversation pages (and adjacent long-form +types: meeting, slack, email) into the facts table. Each page is parsed +into time-windowed segments and passed through the shared fact extractor +with a topical/temporal context header so the resulting facts retain +anchor terms ("Conversation between A and B on DATE …") that the +chunk-level embedding loses on long conversations. + +Options: + --source-id Source to operate on (default: 'default'). + --types Comma-separated subset of: ${ALLOWED_TYPES.join(', ')}. + Default: reads cycle.conversation_facts_backfill.types config + (falls back to the full allowlist). + --slug Process a single page (overrides multi-page enumeration). + --dry-run Show segmentation + counts; no DB writes, no checkpoint advance. + --limit Cap pages processed (default: all). + --since Only consider messages newer than this ISO timestamp. + --force Re-process the target page (clears its resume entry). + --sleep Delay between extractor calls (default ${DEFAULT_INTER_CALL_SLEEP_MS}). + --segment-limit Max segments per page (0 = unlimited). + --max-cost-usd Cost cap for this run (default ${DEFAULT_MAX_COST_USD}). + --override-disabled Bypass facts.extraction_enabled=false brain-wide kill-switch. + --background Submit as a Minion job; print job_id; exit (use 'gbrain jobs follow'). + --yes Auto-confirm cost preview in non-TTY contexts. + --help, -h Show this help. + +Multi-source: when --source-id is omitted, the command iterates ALL +sources from gbrain sources list. Per-source budget cap defaults to +--max-cost-usd; the brain-wide cap when running via the autopilot cycle +phase is cycle.conversation_facts_backfill.max_total_cost_usd. + +Resumability: per-page completion is durable via a terminal audit row +in the facts table (source='${TERMINAL_AUDIT_SOURCE}'). gbrain doctor's +conversation_facts_backlog check counts pages without this row. +`; + +function buildJobParams(args: string[]): Record { + const parsed = parseArgs(args); + return { + sourceId: parsed.sourceId, + types: parsed.types, + slug: parsed.slug, + dryRun: parsed.dryRun, + limit: parsed.limit, + sinceIso: parsed.sinceIso, + force: parsed.force, + sleepMs: parsed.sleepMs, + segmentLimit: parsed.segmentLimit, + maxCostUsd: parsed.maxCostUsd, + overrideDisabled: parsed.overrideDisabled, + }; +} + +export async function runExtractConversationFacts( + engine: BrainEngine, + args: string[], +): Promise { + // --help short-circuit. + if (args.includes('--help') || args.includes('-h')) { + console.log(HELP); + return; + } + + // --background path. + const backgrounded = await maybeBackground({ + engine, + args, + jobName: 'extract-conversation-facts', + paramBuilder: buildJobParams, + }); + if (backgrounded) return; + + const parsed = parseArgs(args); + if (parsed.error) { + console.error(parsed.error); + console.error(HELP); + process.exit(1); + } + + // Chat gateway is required for non-dry-run. + if (!parsed.dryRun && !isAvailable('chat')) { + console.error('Chat gateway unavailable. Configure an Anthropic or compatible chat model, or pass --dry-run to preview segmentation.'); + process.exit(1); + } + + // Aggregate result across all sources. + const aggregate: ExtractConversationFactsResult = { + pages_considered: 0, + pages_processed: 0, + pages_skipped: 0, + pages_skipped_too_large: 0, + pages_skipped_disappeared: 0, + segments_processed: 0, + facts_extracted: 0, + facts_inserted: 0, + }; + let totalSpent = 0; + let anyBudgetExhausted = false; + + const progress = createProgress(cliOptsToProgressOptions(getCliOptions())); + + // Multi-source enumeration when --source-id NOT set. + const sourceIds: string[] = parsed.sourceId + ? [parsed.sourceId] + : (await listSources(engine)).map((s) => s.id); + + progress.start('extract.conversation_facts', sourceIds.length); + + try { + for (const sourceId of sourceIds) { + const perSource = await runExtractConversationFactsCore(engine, { + sourceId, + types: parsed.types, + slug: parsed.slug, + dryRun: parsed.dryRun, + limit: parsed.limit, + sinceIso: parsed.sinceIso, + force: parsed.force, + sleepMs: parsed.sleepMs, + segmentLimit: parsed.segmentLimit, + maxCostUsd: parsed.maxCostUsd, + overrideDisabled: parsed.overrideDisabled, + }); + + aggregate.pages_considered += perSource.pages_considered; + aggregate.pages_processed += perSource.pages_processed; + aggregate.pages_skipped += perSource.pages_skipped; + aggregate.pages_skipped_too_large += perSource.pages_skipped_too_large; + aggregate.pages_skipped_disappeared += perSource.pages_skipped_disappeared; + aggregate.segments_processed += perSource.segments_processed; + aggregate.facts_extracted += perSource.facts_extracted; + aggregate.facts_inserted += perSource.facts_inserted; + if (perSource.budget_exhausted) anyBudgetExhausted = true; + if (perSource.spent_usd) totalSpent += perSource.spent_usd; + + progress.tick(1, `${sourceId}: ${perSource.facts_inserted} facts inserted`); + } + } finally { + progress.finish(); + } + + const verb = parsed.dryRun ? '(dry run) would extract' : 'extracted'; + console.log( + `\nDone: ${verb} ${aggregate.facts_extracted} facts ` + + `(${aggregate.facts_inserted} inserted) across ${aggregate.segments_processed} segments ` + + `from ${aggregate.pages_processed}/${aggregate.pages_considered} pages ` + + `in ${sourceIds.length} source(s). ` + + `Spent ~$${totalSpent.toFixed(4)}.`, + ); + if (aggregate.pages_skipped > 0) { + console.log(` Skipped ${aggregate.pages_skipped} page(s) with no new segments since last checkpoint.`); + } + if (aggregate.pages_skipped_too_large > 0) { + console.log(` Skipped ${aggregate.pages_skipped_too_large} page(s) exceeding ${MAX_PAGE_BODY_BYTES / 1024 / 1024}MB body cap.`); + } + if (aggregate.pages_skipped_disappeared > 0) { + console.log(` Skipped ${aggregate.pages_skipped_disappeared} page(s) that disappeared between enumeration and fetch.`); + } + if (anyBudgetExhausted) { + console.log(` Budget cap reached. Re-run with a higher --max-cost-usd to continue.`); + } +} + +// --------------------------------------------------------------------------- +// Helpers. +// --------------------------------------------------------------------------- + +function pickLaterIso( + a: string | null | undefined, + b: string | null | undefined, +): string | undefined { + const av = a ? Date.parse(a) : NaN; + const bv = b ? Date.parse(b) : NaN; + if (Number.isFinite(av) && Number.isFinite(bv)) return av >= bv ? a! : b!; + if (Number.isFinite(av)) return a!; + if (Number.isFinite(bv)) return b!; + return undefined; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function isAbortError(err: unknown): boolean { + if (!(err instanceof Error)) return false; + return err.name === 'AbortError' || /aborted|cancell?ed/i.test(err.message); +} diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 737b0077a..4fe2a1d92 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1211,6 +1211,42 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai return result; }); + // v0.41.11.0 — extract-conversation-facts. NOT in PROTECTED_JOB_NAMES + // because per-call cost is bounded by `data.max_cost_usd` (default + // DEFAULT_MAX_COST_USD = $5) and the handler re-creates the + // BudgetTracker inside its own process. BudgetExhausted is caught at + // the core level and returned as `result.budget_exhausted: true` (NOT + // a job failure) so the user can resume with a higher cap. + worker.register('extract-conversation-facts', async (job) => { + const { runExtractConversationFactsCore } = await import('./extract-conversation-facts.ts'); + const sourceId = typeof job.data.sourceId === 'string' ? job.data.sourceId : undefined; + if (!sourceId) { + // Multi-source iteration not supported in the Minion-handler path; + // the CLI wrapper does multi-source loops. A background submission + // SHOULD pin to one source per call (job_id is per-call). + throw new Error('extract-conversation-facts Minion job requires data.sourceId'); + } + const types = Array.isArray(job.data.types) + ? (job.data.types as string[]).filter((t) => + ['conversation', 'meeting', 'slack', 'email'].includes(t), + ) + : undefined; + const result = await runExtractConversationFactsCore(engine, { + sourceId, + types: types as ('conversation' | 'meeting' | 'slack' | 'email')[] | undefined, + slug: typeof job.data.slug === 'string' ? job.data.slug : undefined, + dryRun: !!job.data.dryRun, + limit: typeof job.data.limit === 'number' ? job.data.limit : undefined, + sinceIso: typeof job.data.sinceIso === 'string' ? job.data.sinceIso : undefined, + force: !!job.data.force, + sleepMs: typeof job.data.sleepMs === 'number' ? job.data.sleepMs : undefined, + segmentLimit: typeof job.data.segmentLimit === 'number' ? job.data.segmentLimit : undefined, + maxCostUsd: typeof job.data.maxCostUsd === 'number' ? job.data.maxCostUsd : undefined, + overrideDisabled: !!job.data.overrideDisabled, + }); + return result; + }); + // v0.40.3.0 T8b: RemediationStep consumer handlers. Thin wrappers // around already-shipping CLI commands so doctor --remediate can // submit them as Minion jobs. NOT in PROTECTED_JOB_NAMES (no shell diff --git a/src/commands/sources.ts b/src/commands/sources.ts index 0a8139c34..72618492f 100644 --- a/src/commands/sources.ts +++ b/src/commands/sources.ts @@ -964,6 +964,16 @@ async function runAudit(engine: BrainEngine, args: string[]): Promise { const wouldSoftBlock: Array<{ file: string; bytes: number }> = []; const wouldWarn: Array<{ file: string; bytes: number }> = []; const patternHits: Record = {}; + // v0.41.11.0 — facts-backfill estimator (E4). Walks the same files + // already loaded for sanity scanning; counts eligible pages by + // frontmatter.type and estimates per-page segment count from body + // bytes. Estimated per-segment Sonnet cost is a rough heuristic + // (~2000 in + 500 out tokens at $3/MTok in + $15/MTok out ≈ $0.013). + const FACTS_BACKFILL_ALLOWED = ['conversation', 'meeting', 'slack', 'email']; + const FACTS_BACKFILL_CHARS_PER_SEGMENT = 6500; // matches SEGMENT_TEXT_CHAR_LIMIT + const FACTS_BACKFILL_USD_PER_SEGMENT = 0.013; + let factsBackfillPages = 0; + let factsBackfillSegments = 0; for (const file of files) { let content: string; @@ -996,8 +1006,26 @@ async function runAudit(engine: BrainEngine, args: string[]): Promise { } else if (sanity.reasons.includes('oversize_warn')) { wouldWarn.push({ file, bytes: sanity.bytes }); } + // Facts-backfill estimator: counts pages matching allowed types. + const fmType = (parsed.frontmatter?.type as string | undefined) ?? null; + if (fmType && FACTS_BACKFILL_ALLOWED.includes(fmType)) { + factsBackfillPages++; + const totalBytes = sanity.bytes; + const segmentsEstimate = Math.max( + 1, + Math.ceil(totalBytes / FACTS_BACKFILL_CHARS_PER_SEGMENT), + ); + factsBackfillSegments += segmentsEstimate; + } } + const factsBackfillEstimate = { + pages: factsBackfillPages, + est_segments: factsBackfillSegments, + est_cost_usd: Number((factsBackfillSegments * FACTS_BACKFILL_USD_PER_SEGMENT).toFixed(2)), + types: FACTS_BACKFILL_ALLOWED, + }; + // Size distribution stats. sizes.sort((a, b) => a - b); const p = (q: number) => @@ -1014,6 +1042,7 @@ async function runAudit(engine: BrainEngine, args: string[]): Promise { soft_block_count: wouldSoftBlock.length, warn_count: wouldWarn.length, pattern_hits: patternHits, + facts_backfill_estimate: factsBackfillEstimate, hard_blocks: wouldHardBlock.slice(0, 20), soft_blocks: wouldSoftBlock.slice(0, 20), ...(includeWarns ? { warns: wouldWarn.slice(0, 20) } : {}), @@ -1035,6 +1064,13 @@ async function runAudit(engine: BrainEngine, args: string[]): Promise { const sorted = Object.entries(patternHits).sort((a, b) => b[1] - a[1]); console.log(`Junk-pattern hits: ${sorted.map(([n, c]) => `${n} ×${c}`).join(', ')}`); } + if (factsBackfillEstimate.pages > 0) { + console.log( + `Facts backfill estimate: ${factsBackfillEstimate.pages} eligible page(s), ` + + `~${factsBackfillEstimate.est_segments} segments, ~$${factsBackfillEstimate.est_cost_usd}. ` + + `Run: gbrain extract-conversation-facts --source-id ${sourceId} --max-cost-usd ${Math.max(factsBackfillEstimate.est_cost_usd, 1)}`, + ); + } if (wouldHardBlock.length > 0) { console.log('\nTop hard-blocks:'); for (const h of wouldHardBlock.slice(0, 10)) { diff --git a/src/core/cycle.ts b/src/core/cycle.ts index f811659a9..30fff2a8f 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -78,7 +78,14 @@ export type CyclePhase = // - synthesize_concepts: global aggregation of atoms into tier-promoted // concept pages via dedup → tier → Sonnet T1/T2 voice-gated narratives. // Same pack-gate model. - | 'extract_atoms' | 'synthesize_concepts'; + | 'extract_atoms' | 'synthesize_concepts' + // v0.41.11.0 — opt-in (default OFF) bulk fact extraction for long-form + // conversation pages. The phase wrapper does its own multi-source + // iteration directly (PHASE_SCOPE='source' here is taxonomy only; + // see comment above PHASE_SCOPE). Wraps the per-source loop in ONE + // brain-wide BudgetTracker and passes it through opts.budgetTracker + // so the core's auto-wrap doesn't REPLACE it. + | 'conversation_facts_backfill'; export const ALL_PHASES: CyclePhase[] = [ 'lint', @@ -133,6 +140,12 @@ export const ALL_PHASES: CyclePhase[] = [ 'propose_takes', 'grade_takes', 'calibration_profile', + // v0.41.11.0 — opt-in conversation-facts backfill. Default OFF; reads + // cycle.conversation_facts_backfill.enabled gate inside the wrapper. + // Ordered AFTER calibration_profile (matches the runCycle dispatch + // block placement, which runs between the calibration trio and embed), + // and BEFORE embed so newly-inserted facts get embedded same-cycle. + 'conversation_facts_backfill', 'embed', 'orphans', // v0.39 T12: passive schema-suggest. Runs LATE so post-sync brain state @@ -190,6 +203,11 @@ export const PHASE_SCOPE: Record = { // global because concept clusters cross sources by nature. extract_atoms: 'source', synthesize_concepts: 'global', + // v0.41.11.0 — declared 'source' for taxonomy alignment with + // extract_facts (per-source semantics). PHASE_SCOPE has no runtime + // fanout enforcement today (per the comment above); the phase + // wrapper does its own multi-source loop via listSources(). + conversation_facts_backfill: 'source', }; /** @@ -225,6 +243,8 @@ const NEEDS_LOCK_PHASES: ReadonlySet = new Set([ // mutate DB state and need the lock. 'extract_atoms', 'synthesize_concepts', + // v0.41.11.0 — inserts facts + writes terminal audit rows; needs lock. + 'conversation_facts_backfill', 'embed', 'purge', ]); @@ -1761,6 +1781,36 @@ export async function runCycle( } } + // ── v0.41.11.0: conversation_facts_backfill ───────────────── + // Opt-in (default OFF). Walks long-form conversation/meeting/slack/ + // email pages, segments by 30-min gap, runs facts extractor with a + // topical/temporal header, writes facts + per-page TERMINAL audit + // row. Per-source + brain-wide cost AND walltime caps; budget + // tracker passed in from the phase wrapper (NOT nested-wrapped in + // core — would REPLACE not stack). + if (phases.includes('conversation_facts_backfill')) { + checkAborted(opts.signal); + if (!engine) { + phaseResults.push({ + phase: 'conversation_facts_backfill', + status: 'skipped', + duration_ms: 0, + summary: 'no database connected', + details: { reason: 'no_database' }, + }); + } else { + progress.start('cycle.conversation_facts_backfill'); + const { runPhaseConversationFactsBackfill } = await import('./cycle/conversation-facts-backfill.ts'); + const { result, duration_ms } = await timePhase(() => + runPhaseConversationFactsBackfill(engine, { dryRun, signal: opts.signal }), + ); + result.duration_ms = duration_ms; + phaseResults.push(result); + progress.finish(); + } + await safeYield(opts.yieldBetweenPhases); + } + // ── Phase 8: embed ────────────────────────────────────────── if (phases.includes('embed')) { checkAborted(opts.signal); diff --git a/src/core/cycle/conversation-facts-backfill.ts b/src/core/cycle/conversation-facts-backfill.ts new file mode 100644 index 000000000..c251f9864 --- /dev/null +++ b/src/core/cycle/conversation-facts-backfill.ts @@ -0,0 +1,294 @@ +/** + * v0.41.11.0 — cycle phase `conversation_facts_backfill`. + * + * Opt-in autopilot wrapper around `runExtractConversationFactsCore`. + * Default OFF; user enables explicitly via + * `gbrain config set cycle.conversation_facts_backfill.enabled true`. + * + * Architecture (per CEO + eng review + Codex outside voice): + * + * - Per-source iteration HERE. PHASE_SCOPE='source' is taxonomy-only + * (cycle.ts:131 documents this); no runtime fanout exists yet. The + * wrapper enumerates `listSources(engine)` and loops over per-source + * core invocations directly. + * + * - Brain-wide BudgetTracker created ONCE per phase tick and passed + * into every per-source invocation via `opts.budgetTracker`. The + * core function uses it as-is — does NOT wrap in + * `withBudgetTracker` (nested wraps REPLACE the active tracker per + * gateway.ts AsyncLocalStorage semantics, defeating the brain-wide + * cap). This is the Codex C5 + Eng-v2 C5 design. + * + * - Brain-wide walltime cap (Eng-v2 A4) enforced by checking + * `Date.now() - startedAt > maxTotalWalltimeMs` between sources. + * When exceeded, remaining sources skipped + recorded in + * `result.skipped_by_brain_wide_walltime`. + * + * - Symmetric two-layer protection: per-source cap (`max_cost_usd` / + * `max_walltime_min`) AND brain-wide cap (`max_total_cost_usd` / + * `max_total_walltime_min`). Defaults: $1/source, $5 total, 20min/ + * source, 30min total. + * + * Config keys (all defaults explicit): + * + * cycle.conversation_facts_backfill.enabled (false) + * cycle.conversation_facts_backfill.max_cost_usd (1.00) + * cycle.conversation_facts_backfill.max_total_cost_usd (5.00) + * cycle.conversation_facts_backfill.max_walltime_min (20) + * cycle.conversation_facts_backfill.max_total_walltime_min (30) + * cycle.conversation_facts_backfill.types (["conversation","meeting","slack","email"]) + * + * `.types` is the single source of truth for "enabled types" — the CLI + * default reads from the same key (Eng-v2 A2). + */ + +import type { BrainEngine } from '../engine.ts'; +import { BudgetTracker, BudgetExhausted } from '../budget/budget-tracker.ts'; +import { withBudgetTracker } from '../ai/gateway.ts'; +import { listSources } from '../sources-ops.ts'; +import { + runExtractConversationFactsCore, + ALLOWED_TYPES, + type AllowedType, + type ExtractConversationFactsResult, +} from '../../commands/extract-conversation-facts.ts'; + +/** Per-phase wrapper opts. */ +export interface ConversationFactsBackfillPhaseOpts { + dryRun?: boolean; + signal?: AbortSignal; +} + +/** Phase return shape (matches PhaseResult contract from cycle.ts). */ +export interface ConversationFactsBackfillPhaseResult { + phase: 'conversation_facts_backfill'; + status: 'ok' | 'warn' | 'fail' | 'skipped'; + duration_ms: number; + summary: string; + details: Record; +} + +const CFG_PREFIX = 'cycle.conversation_facts_backfill'; + +interface ResolvedConfig { + enabled: boolean; + maxCostUsd: number; // per source per cycle + maxTotalCostUsd: number; // brain-wide per cycle + maxWalltimeMin: number; // per source per cycle + maxTotalWalltimeMin: number; // brain-wide per cycle + types: AllowedType[]; +} + +async function loadCfg(engine: BrainEngine): Promise { + const get = (k: string) => engine.getConfig(`${CFG_PREFIX}.${k}`); + const [enabled, maxCost, maxTotalCost, maxWall, maxTotalWall, typesRaw] = + await Promise.all([ + get('enabled'), + get('max_cost_usd'), + get('max_total_cost_usd'), + get('max_walltime_min'), + get('max_total_walltime_min'), + get('types'), + ]); + + // Truthy-string parse mirrors isFactsExtractionEnabled. + const enabledFlag = (() => { + if (enabled == null) return false; + const v = enabled.trim().toLowerCase(); + return !['false', '0', 'no', 'off', ''].includes(v); + })(); + + const parseFloatOrDefault = (raw: string | null, fallback: number): number => { + if (raw == null) return fallback; + const n = parseFloat(raw); + return Number.isFinite(n) && n > 0 ? n : fallback; + }; + + let types: AllowedType[] = [...ALLOWED_TYPES]; + if (typesRaw) { + try { + const parsed = JSON.parse(typesRaw); + if (Array.isArray(parsed)) { + const filtered = parsed + .filter((t): t is string => typeof t === 'string') + .filter((t): t is AllowedType => + (ALLOWED_TYPES as readonly string[]).includes(t), + ); + if (filtered.length > 0) types = filtered; + } + } catch { + // fall through to default + } + } + + return { + enabled: enabledFlag, + maxCostUsd: parseFloatOrDefault(maxCost, 1.0), + maxTotalCostUsd: parseFloatOrDefault(maxTotalCost, 5.0), + maxWalltimeMin: parseFloatOrDefault(maxWall, 20), + maxTotalWalltimeMin: parseFloatOrDefault(maxTotalWall, 30), + types, + }; +} + +export async function runPhaseConversationFactsBackfill( + engine: BrainEngine, + opts: ConversationFactsBackfillPhaseOpts = {}, +): Promise { + const cfg = await loadCfg(engine); + + if (!cfg.enabled) { + return { + phase: 'conversation_facts_backfill', + status: 'skipped', + duration_ms: 0, + summary: 'cycle.conversation_facts_backfill.enabled=false (default OFF)', + details: { + reason: 'disabled', + enable_hint: + 'gbrain config set cycle.conversation_facts_backfill.enabled true', + }, + }; + } + + const startedAt = Date.now(); + const maxTotalWalltimeMs = cfg.maxTotalWalltimeMin * 60_000; + + const sources = await listSources(engine); + if (sources.length === 0) { + return { + phase: 'conversation_facts_backfill', + status: 'ok', + duration_ms: Date.now() - startedAt, + summary: 'no sources to process', + details: { sources_count: 0 }, + }; + } + + // Brain-wide tracker — created ONCE, scoped to brain-wide cap. Passed + // explicitly into every per-source core invocation via opts.budgetTracker + // so the core doesn't wrap (which would REPLACE). + const brainTracker = new BudgetTracker({ + maxCostUsd: cfg.maxTotalCostUsd, + label: 'conversation_facts_backfill:brain-wide', + }); + + const perSourceResults: Record = {}; + let skippedByBrainWideCap = 0; + let skippedByBrainWideWalltime = 0; + let totalSpent = 0; + + try { + // Single withBudgetTracker scope wraps the entire loop so the + // brain-wide tracker counts EVERY gateway call inside any per-source + // invocation. The core uses opts.budgetTracker as-is (no nested wrap), + // so the AsyncLocalStorage scope established here remains active. + await withBudgetTracker(brainTracker, async () => { + for (const src of sources) { + if (opts.signal?.aborted) throw new Error('aborted'); + + // Brain-wide walltime check. + if (Date.now() - startedAt > maxTotalWalltimeMs) { + skippedByBrainWideWalltime++; + continue; + } + + try { + const result = await runExtractConversationFactsCore(engine, { + sourceId: src.id, + types: cfg.types, + dryRun: opts.dryRun, + // Pass brain-wide tracker so core skips its own auto-wrap. + budgetTracker: brainTracker, + }, opts.signal); + perSourceResults[src.id] = result; + if (result.budget_exhausted) { + // Brain-wide cap hit. Remaining sources skipped. + skippedByBrainWideCap = Math.max( + 0, + sources.length - Object.keys(perSourceResults).length, + ); + break; + } + } catch (err) { + if (err instanceof BudgetExhausted) { + skippedByBrainWideCap = Math.max( + 0, + sources.length - Object.keys(perSourceResults).length, + ); + break; + } + // Per-source failure: record + continue with next source. + perSourceResults[src.id] = { + pages_considered: 0, + pages_processed: 0, + pages_skipped: 0, + pages_skipped_too_large: 0, + pages_skipped_disappeared: 0, + segments_processed: 0, + facts_extracted: 0, + facts_inserted: 0, + error: (err as Error).message, + }; + } + } + }); + } catch (err) { + if (err instanceof BudgetExhausted) { + // Brain-wide cap hit during last source. + } else if ((err as Error).message === 'aborted' || opts.signal?.aborted) { + // Propagate abort. + throw err; + } else { + // Unexpected error. + return { + phase: 'conversation_facts_backfill', + status: 'fail', + duration_ms: Date.now() - startedAt, + summary: `brain-wide loop failed: ${(err as Error).message}`, + details: { error: (err as Error).message, perSourceResults }, + }; + } + } + + totalSpent = brainTracker.totalSpent; + + // Aggregate. + const totals = { + pages_processed: 0, + pages_skipped: 0, + facts_inserted: 0, + sources_processed: 0, + }; + for (const r of Object.values(perSourceResults)) { + if (!r.error) totals.sources_processed++; + totals.pages_processed += r.pages_processed; + totals.pages_skipped += r.pages_skipped; + totals.facts_inserted += r.facts_inserted; + } + + const anyError = Object.values(perSourceResults).some((r) => r.error); + const status = anyError ? 'warn' : 'ok'; + const summary = `${totals.facts_inserted} facts inserted across ${totals.sources_processed}/${sources.length} sources, ~$${totalSpent.toFixed(4)} spent`; + + return { + phase: 'conversation_facts_backfill', + status, + duration_ms: Date.now() - startedAt, + summary, + details: { + sources_count: sources.length, + sources_processed: totals.sources_processed, + pages_processed: totals.pages_processed, + pages_skipped: totals.pages_skipped, + facts_inserted: totals.facts_inserted, + spent_usd: totalSpent, + skipped_by_brain_wide_cap: skippedByBrainWideCap, + skipped_by_brain_wide_walltime: skippedByBrainWideWalltime, + types: cfg.types, + max_total_cost_usd: cfg.maxTotalCostUsd, + max_total_walltime_min: cfg.maxTotalWalltimeMin, + per_source: perSourceResults, + }, + }; +} diff --git a/src/core/migrate.ts b/src/core/migrate.ts index 648c389b3..8eba7775c 100644 --- a/src/core/migrate.ts +++ b/src/core/migrate.ts @@ -4441,6 +4441,66 @@ export const MIGRATIONS: Migration[] = [ `, }, }, + { + version: 96, + name: 'facts_extract_conversation_session_index', + // v0.41.11.0 — partial index supporting the doctor query for + // conversation_facts_backlog (Codex round-1 T2 + round-2 C2). + // The doctor check runs: + // SELECT COUNT(*) FROM pages p WHERE p.type = ANY($1::text[]) + // AND p.deleted_at IS NULL + // AND NOT EXISTS (SELECT 1 FROM facts f + // WHERE f.source = 'cli:extract-conversation-facts:terminal' + // AND f.source_session = 'cli:extract-conversation-facts:terminal:' || p.slug + // AND f.source_id = p.source_id) + // + // Without this index, the NOT EXISTS subquery seq-scans facts on + // every doctor invocation including autopilot. The partial index + // is tiny — only rows written by this command are indexed + // (per-segment facts + the page-level terminal row). + // + // Engine-aware via handler (not SQL): Postgres uses CREATE INDEX + // CONCURRENTLY (avoid SHARE lock on facts) + pre-drops any invalid + // remnant from a prior failed run (mirrors migration v14 precedent). + // PGLite has no concurrent writers, so plain CREATE is safe. + // + // Slot history: originally planned as v94 (master shipped v94 + // take_domain_assignments); bumped to v95 (master then shipped v95 + // links_link_source_check_includes_mentions); now at v96 after + // post-merge resolution. The index shape itself is unchanged + // across all renumbers. + transaction: false, + sql: '', + handler: async (engine) => { + if (engine.kind === 'postgres') { + await engine.runMigration( + 96, + `DO $$ BEGIN + IF EXISTS ( + SELECT 1 FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + WHERE c.relname = 'idx_facts_extract_conversation_session' AND NOT i.indisvalid + ) THEN + EXECUTE 'DROP INDEX CONCURRENTLY IF EXISTS idx_facts_extract_conversation_session'; + END IF; + END $$;` + ); + await engine.runMigration( + 96, + `CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_facts_extract_conversation_session + ON facts (source_id, source_session) + WHERE source LIKE 'cli:extract-conversation-facts%';` + ); + } else { + await engine.runMigration( + 96, + `CREATE INDEX IF NOT EXISTS idx_facts_extract_conversation_session + ON facts (source_id, source_session) + WHERE source LIKE 'cli:extract-conversation-facts%';` + ); + } + }, + }, ]; export const LATEST_VERSION = MIGRATIONS.length > 0 diff --git a/src/core/schema-pack/base/gbrain-base.yaml b/src/core/schema-pack/base/gbrain-base.yaml index e4ec3a93d..dd871b0fc 100644 --- a/src/core/schema-pack/base/gbrain-base.yaml +++ b/src/core/schema-pack/base/gbrain-base.yaml @@ -101,7 +101,21 @@ page_types: - wiki/concepts/ - wiki/concept/ aliases: [] - extractable: false + # v0.41.11+ — concepts ARE knowledge: their bodies define the + # concept, so the facts extractor should mine them like writing/ + # source/note pages. Pre-v0.41.11 default was false because concepts + # were treated as link-only stubs; the operational reality is that + # concept bodies routinely contain definitions, distinctions, and + # claims that belong in the facts table. + # + # Important: this flip is SEMANTIC ONLY for v0.41.11. The post-sync + # backstop uses hardcoded ELIGIBLE_TYPES in src/core/facts/eligibility.ts + # which does NOT read pack extractable, so the flip has zero + # behavioral effect on the backstop today. Power-users who want + # concept extraction run `gbrain extract-facts --type concept --force` + # explicitly. The pack-driven-ELIGIBLE_TYPES refactor (which would + # make this flag actually drive backstop behavior) is a v0.42+ TODO. + extractable: true expert_routing: false - name: person @@ -218,6 +232,36 @@ page_types: extractable: true expert_routing: false + # v0.41.11+ — long-running chat/transcript pages. Lives under + # conversations/. `temporal` primitive (events through time, not an + # entity surface). `extractable: true` because the batch facts + # extraction command (gbrain extract-conversation-facts) walks + # these by type. Promoted from gbrain-recommended into gbrain-base + # because imported chat history is now a universal pattern, not + # an operational-brain niche. + - name: conversation + primitive: temporal + path_prefixes: + - conversations/ + aliases: [] + extractable: true + expert_routing: false + + # v0.41.11+ — atomic claim units. An `atom` is the smallest + # extractable unit of knowledge a higher-level page can reference + # (one quote, one statistic, one milestone). Atoms ARE the + # extracted form, so `extractable: false` — running the fact + # extractor on an atom would loop on its own output. The + # `annotation` primitive is the right home: atoms annotate a + # source, they do not stand alone as entities. + - name: atom + primitive: annotation + path_prefixes: + - atoms/ + aliases: [] + extractable: false + expert_routing: false + # Types with no path prefix (set explicitly via frontmatter). - name: code primitive: media @@ -349,3 +393,11 @@ filing_rules: directory: meetings/ examples: - meetings/2026-04-03 + - kind: conversation + directory: conversations/ + examples: + - conversations/imessage/alice-example + - kind: atom + directory: atoms/ + examples: + - atoms/2026-04-12-revenue-claim diff --git a/src/core/schema-pack/base/gbrain-recommended.yaml b/src/core/schema-pack/base/gbrain-recommended.yaml index 91b727131..fa17d5e66 100644 --- a/src/core/schema-pack/base/gbrain-recommended.yaml +++ b/src/core/schema-pack/base/gbrain-recommended.yaml @@ -12,7 +12,10 @@ # This pack extends gbrain-base (which reproduces pre-v0.38 hardcoded # behavior) and adds the additional types the recommended-schema doc # describes: deals, meetings, concepts, projects, sources, daily, -# personal, civic, originals, places, trips, conversations. +# personal, civic, originals, places, trips. +# +# Note: `conversation` was promoted into gbrain-base in v0.41.11 and is +# no longer duplicated here — it now arrives via `extends: gbrain-base`. # # Inherits everything else from gbrain-base via `extends:`. Pin the # base version so a future gbrain-base change doesn't silently shift @@ -122,14 +125,6 @@ page_types: extractable: false expert_routing: false - - name: conversation - primitive: temporal - path_prefixes: - - conversations/ - aliases: [] - extractable: true - expert_routing: false - - name: writing primitive: media path_prefixes: diff --git a/src/core/types.ts b/src/core/types.ts index 7dadcfbff..d8e059531 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -44,6 +44,13 @@ export const ALL_PAGE_TYPES: readonly string[] = [ 'person', 'company', 'deal', 'yc', 'civic', 'project', 'concept', 'source', 'media', 'writing', 'analysis', 'guide', 'hardware', 'architecture', 'meeting', 'note', 'email', 'slack', 'calendar-event', + // v0.41.11+ — `conversation` (imported chat/transcript pages, lives + // under conversations/) and `atom` (smallest extractable claim unit, + // lives under atoms/). Both promoted into the gbrain-base seed list + // so they share the universal validation surface with the rest of + // the base types; their pack entries are declared in + // src/core/schema-pack/base/gbrain-base.yaml. + 'conversation', 'atom', 'code', 'image', 'synthesis', ] as const; diff --git a/test/core/cycle.serial.test.ts b/test/core/cycle.serial.test.ts index fe55f571d..960dc230f 100644 --- a/test/core/cycle.serial.test.ts +++ b/test/core/cycle.serial.test.ts @@ -391,7 +391,8 @@ describe('runCycle — yieldBetweenPhases hook', () => { // v0.36.1.0: 16 phases (added `propose_takes`, `grade_takes`, `calibration_profile` between consolidate and embed). // v0.39.0.0: 17 phases (added `schema-suggest` between orphans and purge — T12 schema cathedral). // v0.41.2.0: 19 phases (added `extract_atoms` after extract_facts + `synthesize_concepts` after patterns). - expect(hookCalls).toBe(19); + // v0.41.11.0: 20 phases (added `conversation_facts_backfill` between consolidate and propose_takes). + expect(hookCalls).toBe(20); }); test('hook exceptions do not abort the cycle', async () => { @@ -404,7 +405,8 @@ describe('runCycle — yieldBetweenPhases hook', () => { // v0.33.3: 13 phases (v0.32.2's 12 + resolve_symbol_edges). // v0.36.1.0: 16 phases (Hindsight calibration wave adds propose_takes, grade_takes, calibration_profile). // v0.39.0.0: 17 phases (T12 schema-suggest phase between orphans and purge). - expect(report.phases.length).toBe(19); // v0.41: +extract_atoms, +synthesize_concepts + // v0.41.11.0: 20 phases (+extract_atoms, +synthesize_concepts, +conversation_facts_backfill). + expect(report.phases.length).toBe(20); }); }); diff --git a/test/extract-conversation-facts.test.ts b/test/extract-conversation-facts.test.ts new file mode 100644 index 000000000..387502b67 --- /dev/null +++ b/test/extract-conversation-facts.test.ts @@ -0,0 +1,461 @@ +/** + * Tests for `gbrain extract-conversation-facts` — deterministic parsing, + * segmenting, rendering, checkpoint encoding, and core wiring contracts. + * + * Hermetic via __setChatTransportForTests + __setEmbedTransportForTests + * stubs so the suite stays offline. Real-LLM extraction quality is the + * job of test/eval/conversation-extraction-quality.eval.ts (env-gated). + * + * Test-isolation invariants (per CLAUDE.md R3+R4): + * - One PGLite engine per file, created in beforeAll, disposed in afterAll + * - Per-test state reset via TRUNCATE inside beforeEach (canonical pattern) + */ + +import { describe, expect, test, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { + __setChatTransportForTests, + __setEmbedTransportForTests, + resetGateway, + type ChatResult, +} from '../src/core/ai/gateway.ts'; +import { + parseConversationMessages, + splitIntoSegments, + renderSegmentForExtraction, + runExtractConversationFactsCore, + extractConversationFactsFingerprint, + encodeCheckpointEntry, + decodeCheckpointEntry, + DEFAULT_SEGMENT_GAP_MINUTES, + DEFAULT_SEGMENT_MAX_MESSAGES, + SEGMENT_TEXT_CHAR_LIMIT, + MAX_PAGE_BODY_BYTES, + TERMINAL_AUDIT_SOURCE, + PER_SEGMENT_SOURCE_PREFIX, +} from '../src/commands/extract-conversation-facts.ts'; + +// --------------------------------------------------------------------------- +// Fixture helpers. +// --------------------------------------------------------------------------- + +function fmt(name: string, date: string, time: string, body: string): string { + return `**${name}** (${date} ${time}): ${body}`; +} + +// --------------------------------------------------------------------------- +// parseConversationMessages — PR's 5 cases verbatim. +// --------------------------------------------------------------------------- + +describe('parseConversationMessages', () => { + test('parses a single message line', () => { + const msgs = parseConversationMessages(fmt('Alice Example', '2024-03-15', '6:07 PM', 'hello')); + expect(msgs).toHaveLength(1); + expect(msgs[0].speaker).toBe('Alice Example'); + expect(msgs[0].text).toBe('hello'); + expect(msgs[0].timestamp).toMatch(/^2024-03-15T18:07:00Z$/); + }); + + test('handles AM/PM and midnight/noon', () => { + const body = [ + fmt('Bob Demo', '2024-03-15', '12:00 AM', 'midnight'), + fmt('Bob Demo', '2024-03-15', '12:30 PM', 'noon'), + ].join('\n'); + const msgs = parseConversationMessages(body); + expect(msgs[0].timestamp).toBe('2024-03-15T00:00:00Z'); + expect(msgs[1].timestamp).toBe('2024-03-15T12:30:00Z'); + }); + + test('treats unmatched lines as continuations of the prior message', () => { + const body = [ + fmt('Alice Example', '2024-03-15', '9:00 AM', 'first line'), + 'still part of the first message', + fmt('Bob Demo', '2024-03-15', '9:01 AM', 'separate message'), + ].join('\n'); + const msgs = parseConversationMessages(body); + expect(msgs).toHaveLength(2); + expect(msgs[0].text).toBe('first line\nstill part of the first message'); + expect(msgs[1].text).toBe('separate message'); + }); + + test('ignores leading orphan lines (no anchor message yet)', () => { + const body = ['orphan one', 'orphan two', fmt('Alice Example', '2024-03-15', '9:00 AM', 'real')].join('\n'); + const msgs = parseConversationMessages(body); + expect(msgs).toHaveLength(1); + expect(msgs[0].text).toBe('real'); + }); + + test('empty body returns empty array', () => { + expect(parseConversationMessages('')).toEqual([]); + }); +}); + +// --------------------------------------------------------------------------- +// splitIntoSegments — PR's 5 cases verbatim plus tuning regression. +// --------------------------------------------------------------------------- + +describe('splitIntoSegments', () => { + test('cuts on time gap larger than gapMinutes', () => { + const msgs = parseConversationMessages([ + fmt('Alice Example', '2024-03-15', '9:00 AM', 'a'), + fmt('Bob Demo', '2024-03-15', '9:05 AM', 'b'), + // Gap of 90 minutes > default 30 → new segment. + fmt('Alice Example', '2024-03-15', '10:35 AM', 'c'), + fmt('Bob Demo', '2024-03-15', '10:36 AM', 'd'), + ].join('\n')); + const segs = splitIntoSegments(msgs); + expect(segs).toHaveLength(2); + expect(segs[0].messages).toHaveLength(2); + expect(segs[1].messages).toHaveLength(2); + }); + + test('cuts when segment reaches maxMessages cap', () => { + const lines: string[] = []; + for (let i = 0; i < 7; i++) { + const mm = String(i).padStart(2, '0'); + lines.push(fmt('Alice Example', '2024-03-15', `9:${mm} AM`, `msg ${i}`)); + } + const msgs = parseConversationMessages(lines.join('\n')); + const segs = splitIntoSegments(msgs, { maxMessages: 3 }); + // 7 messages / 3 per segment → 2 full + 1 leftover (dropped: <2 messages). + expect(segs.length).toBeGreaterThanOrEqual(2); + for (const s of segs) expect(s.messages.length).toBeLessThanOrEqual(3); + }); + + test('drops segments shorter than the minimum', () => { + const msgs = parseConversationMessages( + fmt('Alice Example', '2024-03-15', '9:00 AM', 'only one'), + ); + expect(splitIntoSegments(msgs)).toHaveLength(0); + }); + + test('participants array preserves first-seen order', () => { + const msgs = parseConversationMessages([ + fmt('Bob Demo', '2024-03-15', '9:00 AM', 'b1'), + fmt('Alice Example', '2024-03-15', '9:05 AM', 'a1'), + fmt('Bob Demo', '2024-03-15', '9:06 AM', 'b2'), + ].join('\n')); + const segs = splitIntoSegments(msgs); + expect(segs[0].participants).toEqual(['Bob Demo', 'Alice Example']); + }); + + test('sinceIso filters out messages older than the watermark', () => { + const msgs = parseConversationMessages([ + fmt('Alice Example', '2024-03-15', '9:00 AM', 'old'), + fmt('Bob Demo', '2024-03-15', '9:05 AM', 'old'), + fmt('Alice Example', '2024-03-16', '9:00 AM', 'new'), + fmt('Bob Demo', '2024-03-16', '9:05 AM', 'new'), + ].join('\n')); + const segs = splitIntoSegments(msgs, { sinceIso: '2024-03-15T23:00:00Z' }); + expect(segs).toHaveLength(1); + expect(segs[0].startIso).toBe('2024-03-16T09:00:00Z'); + }); + + test('tuned defaults: 30/30 (Eng-v2 T5)', () => { + expect(DEFAULT_SEGMENT_GAP_MINUTES).toBe(30); + expect(DEFAULT_SEGMENT_MAX_MESSAGES).toBe(30); + expect(SEGMENT_TEXT_CHAR_LIMIT).toBe(6500); + }); +}); + +// --------------------------------------------------------------------------- +// renderSegmentForExtraction. +// --------------------------------------------------------------------------- + +describe('renderSegmentForExtraction', () => { + test('prepends topical/temporal context header', () => { + const msgs = parseConversationMessages([ + fmt('Alice Example', '2024-03-15', '9:00 AM', 'hello'), + fmt('Bob Demo', '2024-03-15', '9:05 AM', 'hi back'), + ].join('\n')); + const seg = splitIntoSegments(msgs)[0]; + const text = renderSegmentForExtraction('imessage: Alice Example', seg); + expect(text).toContain('Page: imessage: Alice Example'); + expect(text).toContain('Conversation between Alice Example and Bob Demo'); + expect(text).toContain('2024-03-15T09:00:00Z'); + expect(text).toContain('2024-03-15T09:05:00Z'); + }); + + test('truncates oversize segments but keeps the header intact', () => { + const big = Array.from({ length: 500 }, (_, i) => { + const mm = String(i % 60).padStart(2, '0'); + const hh = String(9 + Math.floor(i / 60)).padStart(2, '0'); + return `**Alice Example** (2024-03-15 ${hh}:${mm} AM): ${'x'.repeat(50)}`; + }).join('\n'); + const msgs = parseConversationMessages(big); + const seg = splitIntoSegments(msgs, { maxMessages: 500 })[0]; + const text = renderSegmentForExtraction('big-page', seg); + expect(text.length).toBeLessThanOrEqual(SEGMENT_TEXT_CHAR_LIMIT + 32); + expect(text.startsWith('Page: big-page')).toBe(true); + expect(text).toContain('Conversation between'); + }); +}); + +// --------------------------------------------------------------------------- +// Fingerprint + checkpoint encoding. +// --------------------------------------------------------------------------- + +describe('extractConversationFactsFingerprint (Eng-v2 A3)', () => { + test('same sourceId yields same fingerprint', () => { + expect(extractConversationFactsFingerprint({ sourceId: 'default' })) + .toBe(extractConversationFactsFingerprint({ sourceId: 'default' })); + }); + + test('different sourceId yields different fingerprint', () => { + expect(extractConversationFactsFingerprint({ sourceId: 'a' })) + .not.toBe(extractConversationFactsFingerprint({ sourceId: 'b' })); + }); +}); + +describe('checkpoint entry encoding', () => { + test('round-trips sourceId | slug | iso', () => { + const entry = encodeCheckpointEntry('default', 'conversations/imessage/alice-example', '2024-03-16T08:05:00Z'); + const decoded = decodeCheckpointEntry(entry); + expect(decoded).toEqual({ + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + endIso: '2024-03-16T08:05:00Z', + }); + }); + + test('decodes null for malformed entries', () => { + expect(decodeCheckpointEntry('no-pipes-here')).toBeNull(); + expect(decodeCheckpointEntry('only-one|pipe')).toBeNull(); + }); + + test('slug with forward slashes survives encoding (no pipe collision)', () => { + const entry = encodeCheckpointEntry('src-a', 'conversations/group/2024/march/team-x', '2024-03-16T08:05:00Z'); + const decoded = decodeCheckpointEntry(entry); + expect(decoded?.slug).toBe('conversations/group/2024/march/team-x'); + }); +}); + +// --------------------------------------------------------------------------- +// runExtractConversationFactsCore — engine-wired contract tests. +// --------------------------------------------------------------------------- + +const SAMPLE_BODY = [ + fmt('Alice Example', '2024-03-15', '9:00 AM', 'Hi, I just signed the offer letter for Acme Corp.'), + fmt('Bob Demo', '2024-03-15', '9:01 AM', "Congrats! What's the title?"), + fmt('Alice Example', '2024-03-15', '9:02 AM', 'Staff engineer on the platform team.'), + fmt('Bob Demo', '2024-03-15', '9:03 AM', 'Nice.'), + // Big time gap → new segment. + fmt('Alice Example', '2024-03-16', '8:00 AM', 'Update: I started at Acme Corp this morning.'), + fmt('Bob Demo', '2024-03-16', '8:05 AM', 'Day one! How is it?'), +].join('\n'); + +describe('runExtractConversationFactsCore', () => { + let engine: PGLiteEngine; + + beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + + // Deterministic chat-transport stub. Records calls + returns one + // fact per turn. Real-LLM extraction quality is the eval suite's job. + let callIndex = 0; + __setChatTransportForTests(async (): Promise => { + callIndex++; + return { + text: JSON.stringify({ + facts: [{ + fact: `synthetic fact #${callIndex}`, + kind: 'event', + entity: 'companies/acme-corp', + confidence: 1.0, + notability: 'high', + }], + }), + blocks: [], + stopReason: 'end', + usage: { + input_tokens: 100, + output_tokens: 50, + cache_read_tokens: 0, + cache_creation_tokens: 0, + }, + model: 'stub:stub', + providerId: 'stub', + }; + }); + + // Deterministic embedding stub. + __setEmbedTransportForTests( + (async () => ({ + embeddings: [Array.from({ length: 1536 }, () => 0.1)], + })) as never, + ); + }); + + afterAll(async () => { + __setChatTransportForTests(null); + __setEmbedTransportForTests(null); + resetGateway(); + await engine.disconnect(); + }); + + beforeEach(async () => { + // Clean state per test. Use executeRaw because PGLite uses different + // truncation semantics than the canonical reset helper. + await engine.executeRaw(`DELETE FROM facts WHERE source LIKE 'cli:extract-conversation-facts%'`); + await engine.executeRaw(`DELETE FROM op_checkpoints WHERE op = 'extract-conversation-facts'`); + await engine.executeRaw(`DELETE FROM pages WHERE slug LIKE 'conversations/%' OR slug LIKE 'people/alice%'`); + // Set facts.extraction_enabled=true so kill-switch doesn't refuse. + await engine.setConfig('facts.extraction_enabled', 'true'); + // Seed test pages. + await engine.putPage('conversations/imessage/alice-example', { + type: 'conversation', + title: 'iMessage: Alice Example', + compiled_truth: SAMPLE_BODY, + timeline: '', + frontmatter: {}, + }); + await engine.putPage('people/alice-example', { + type: 'person', + title: 'Alice Example', + compiled_truth: 'Profile content for Alice Example.', + timeline: '', + frontmatter: {}, + }); + }); + + test('dry-run reports segmentation without writing facts', async () => { + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + dryRun: true, + sleepMs: 0, + }); + expect(result.pages_considered).toBe(1); + expect(result.pages_processed).toBe(1); + expect(result.facts_inserted).toBe(0); + expect(result.segments_processed).toBeGreaterThanOrEqual(1); + }); + + test('non-conversation pages are skipped', async () => { + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'people/alice-example', + dryRun: true, + sleepMs: 0, + }); + // pages_considered counts only pages whose type matches the allowlist. + expect(result.pages_considered).toBe(0); + }); + + test('sinceIso filters already-processed history', async () => { + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + dryRun: true, + sleepMs: 0, + sinceIso: '2099-01-01T00:00:00Z', + }); + expect(result.pages_processed).toBe(0); + expect(result.pages_skipped).toBe(1); + }); + + test('writes facts with per-segment source_session AND terminal audit row (E16)', async () => { + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(result.pages_processed).toBe(1); + expect(result.facts_inserted).toBeGreaterThan(0); + + // Per-segment facts present. + const perSegFacts = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1 AND source_session = $2`, + [PER_SEGMENT_SOURCE_PREFIX, `${PER_SEGMENT_SOURCE_PREFIX}:conversations/imessage/alice-example`], + ); + expect(Number(perSegFacts[0]?.count ?? 0)).toBeGreaterThan(0); + + // Terminal audit row present. + const terminalRows = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1 AND source_session = $2`, + [TERMINAL_AUDIT_SOURCE, `${TERMINAL_AUDIT_SOURCE}:conversations/imessage/alice-example`], + ); + expect(Number(terminalRows[0]?.count ?? 0)).toBe(1); + }); + + test('row_num accumulator: segment 2 facts start after segment 1 (Codex C1)', async () => { + await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + const rows = await engine.executeRaw<{ row_num: number }>( + `SELECT row_num FROM facts + WHERE source = $1 AND source_markdown_slug = $2 + ORDER BY row_num ASC`, + [PER_SEGMENT_SOURCE_PREFIX, 'conversations/imessage/alice-example'], + ); + // Each row_num must be unique (no per-segment collision on row 0). + const nums = rows.map((r) => Number(r.row_num)); + expect(new Set(nums).size).toBe(nums.length); + // Strictly monotonic + zero-based. + for (let i = 0; i < nums.length; i++) { + expect(nums[i]).toBe(i); + } + }); + + test('--force clears resume entry, allowing re-run', async () => { + const first = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(first.pages_processed).toBe(1); + // Re-run without force: no new segments (sinceIso > newest segment endIso). + const second = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(second.pages_skipped).toBe(1); + // Re-run with force: re-processes. + const third = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + force: true, + }); + expect(third.pages_processed).toBe(1); + expect(third.segments_processed).toBeGreaterThanOrEqual(1); + }); + + test('honors facts.extraction_enabled kill-switch (F2)', async () => { + await engine.setConfig('facts.extraction_enabled', 'false'); + await expect( + runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }), + ).rejects.toThrow(/extraction_enabled=false/); + }); + + test('--override-disabled bypasses kill-switch', async () => { + await engine.setConfig('facts.extraction_enabled', 'false'); + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + overrideDisabled: true, + }); + expect(result.pages_processed).toBe(1); + }); +}); + +// --------------------------------------------------------------------------- +// Body cap (Eng A2 / E17) — pin the cap constant; integration via reads +// in seeded huge pages would require >25MB fixture, not viable in unit suite. +// --------------------------------------------------------------------------- + +describe('body cap constant (Eng A2)', () => { + test('MAX_PAGE_BODY_BYTES is 25MB', () => { + expect(MAX_PAGE_BODY_BYTES).toBe(25 * 1024 * 1024); + }); +}); diff --git a/test/extractable-pack.test.ts b/test/extractable-pack.test.ts index b80adddbb..ae1cfd426 100644 --- a/test/extractable-pack.test.ts +++ b/test/extractable-pack.test.ts @@ -1,8 +1,12 @@ // v0.38 T7d: facts/eligibility pack-aware parity tests. // // Pins the contract that extractableTypesFromPack(gbrain-base) returns -// the pre-v0.38 ELIGIBLE_TYPES list from src/core/facts/eligibility.ts: -// ['note', 'meeting', 'slack', 'email', 'calendar-event', 'source', 'writing'] +// the configured eligible set declared by gbrain-base.yaml. The pre-v0.38 +// `ELIGIBLE_TYPES` constant from src/core/facts/eligibility.ts was the +// seed (note/meeting/slack/email/calendar-event/source/writing); v0.41.11 +// promotes concept + conversation into the same set as part of the +// conversation retrieval upgrade. `atom` is explicitly pinned +// non-extractable so a future drift fails loudly. import { describe, expect, test } from 'bun:test'; import { @@ -15,28 +19,52 @@ import { join } from 'node:path'; const GBRAIN_BASE_PATH = join(import.meta.dir, '../src/core/schema-pack/base/gbrain-base.yaml'); -// Pre-v0.38 ELIGIBLE_TYPES from src/core/facts/eligibility.ts:51 +// Pre-v0.38 seed ELIGIBLE_TYPES from src/core/facts/eligibility.ts:51. const LEGACY_ELIGIBLE = ['note', 'meeting', 'slack', 'email', 'calendar-event', 'source', 'writing']; +// v0.41.11+ additions: concept page bodies define concepts and routinely +// contain claim-shaped statements; conversation pages carry the imported +// chat history the batch facts extractor walks. Both flipped to +// `extractable: true` in gbrain-base.yaml. +const V0_41_11_ADDED_ELIGIBLE = ['concept', 'conversation']; +const CURRENT_ELIGIBLE = [...LEGACY_ELIGIBLE, ...V0_41_11_ADDED_ELIGIBLE]; + describe('extractableTypesFromPack (T7d) — gbrain-base parity', () => { - test('gbrain-base extractable set matches legacy ELIGIBLE_TYPES exactly', () => { + test('gbrain-base extractable set matches the configured eligible list', () => { const pack = loadPackFromFile(GBRAIN_BASE_PATH); const extractable = extractableTypesFromPack(pack); - expect(extractable.size).toBe(LEGACY_ELIGIBLE.length); - for (const t of LEGACY_ELIGIBLE) { + expect(extractable.size).toBe(CURRENT_ELIGIBLE.length); + for (const t of CURRENT_ELIGIBLE) { expect(extractable.has(t)).toBe(true); } - // None of the entity/concept-shape types are extractable in gbrain-base. + // Entity-shape + annotation-shape types stay non-extractable in gbrain-base. + // `atom` is annotation (and IS the extracted unit), so it must not + // be extractable itself — running the extractor on it would loop. expect(extractable.has('person')).toBe(false); expect(extractable.has('company')).toBe(false); expect(extractable.has('deal')).toBe(false); - expect(extractable.has('concept')).toBe(false); expect(extractable.has('synthesis')).toBe(false); + expect(extractable.has('atom')).toBe(false); }); - test('isExtractableType per-type lookups match legacy', () => { + test('legacy seed types remain extractable (back-compat)', () => { const pack = loadPackFromFile(GBRAIN_BASE_PATH); + const extractable = extractableTypesFromPack(pack); for (const t of LEGACY_ELIGIBLE) { + expect(extractable.has(t)).toBe(true); + } + }); + + test('v0.41.11 additions (concept + conversation) are extractable', () => { + const pack = loadPackFromFile(GBRAIN_BASE_PATH); + for (const t of V0_41_11_ADDED_ELIGIBLE) { + expect(isExtractableType(pack, t)).toBe(true); + } + }); + + test('isExtractableType per-type lookups match the configured set', () => { + const pack = loadPackFromFile(GBRAIN_BASE_PATH); + for (const t of CURRENT_ELIGIBLE) { expect(isExtractableType(pack, t)).toBe(true); } expect(isExtractableType(pack, 'person')).toBe(false); diff --git a/test/phase-scope-coverage.test.ts b/test/phase-scope-coverage.test.ts index 1066f24b8..547e9ba3b 100644 --- a/test/phase-scope-coverage.test.ts +++ b/test/phase-scope-coverage.test.ts @@ -41,13 +41,14 @@ describe('PHASE_SCOPE coverage', () => { expect(invalid).toEqual([]); }); - test('all 19 phases covered (regression on accidental omission)', () => { + test('all 20 phases covered (regression on accidental omission)', () => { // Pin the count so a future PR that adds a phase to ALL_PHASES // without updating PHASE_SCOPE notices here too. The v0.39.1.0 // master merge brought in the 17th phase (`schema-suggest`); v0.41 - // adds 'extract_atoms' + 'synthesize_concepts' for a total of 19. - expect(ALL_PHASES.length).toBe(19); - expect(Object.keys(PHASE_SCOPE).length).toBe(19); + // adds 'extract_atoms' + 'synthesize_concepts' (T9 lens packs) + + // 'conversation_facts_backfill' (v0.41.11.0) for a total of 20. + expect(ALL_PHASES.length).toBe(20); + expect(Object.keys(PHASE_SCOPE).length).toBe(20); }); test('embed remains global (the headline brain-wide phase)', () => { diff --git a/test/schema-cli.test.ts b/test/schema-cli.test.ts index c71f431cf..233417357 100644 --- a/test/schema-cli.test.ts +++ b/test/schema-cli.test.ts @@ -51,7 +51,9 @@ describe('gbrain schema CLI (Phase C)', () => { const r = gbrain(['schema', 'show', 'gbrain-base']); expect(r.code).toBe(0); expect(r.stdout).toContain('gbrain-base v1.0.0'); - expect(r.stdout).toContain('Page types (22)'); + // v0.41.11.0: page types extended from 22 to 24 by promoting + // `conversation` and `atom` into gbrain-base. + expect(r.stdout).toContain('Page types (24)'); expect(r.stdout).toContain('Link verbs (12)'); expect(r.stdout).toContain('Takes kinds: fact, take, bet, hunch'); expect(r.stdout).toContain('person :: entity');