diff --git a/CHANGELOG.md b/CHANGELOG.md index 4818a512f..7cdeb3abb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,107 @@ All notable changes to GBrain will be documented in this file. +## [0.42.7.0] - 2026-06-01 + +**Your brain now tells you when it has imported pages but never connected them — and gives you a one-command fix.** + +Importing a page and curating it are two different things. Import drops the +text in. Extraction is the step that reads the text and builds the graph: +the typed edges (`founded`, `works_at`, `invested_in`, `advises`) and the +dated timeline entries that make `gbrain think`, graph traversal, and link +search actually work. On a lot of brains that second step had quietly never +run at scale. One real 280K-page brain had a links table that was 99.7% +untyped `mentions` — a bag of name-drops with almost no real relationships — +and nothing anywhere told the owner. A single manual extraction added 12,500 +typed edges that should have been there all along. + +The reason: plain `gbrain sync` already extracts the pages it changes, but +there was no way to sweep the historical backlog, and no health signal when +extraction fell behind. If you weren't running the autopilot cycle, the gap +grew invisibly. + +Three things fix that: + +- **`gbrain extract --stale`** — a new incremental sweep. It re-extracts only + the pages whose links are stale (never extracted, edited since they were + last extracted, or extracted by an older version), in small batches, safe to + cron. Reads page content straight from the database, so it works on + checkout-less Postgres/Supabase brains too. Pass `--catch-up` to run past + the default 30-minute budget until the backlog is empty; `--dry-run` to just + see the count. + + ``` + gbrain extract --stale # sweep the backlog incrementally + gbrain extract --stale --catch-up # run until 0 remain + gbrain extract --stale --dry-run # how many pages are behind? + ``` + +- **A `links_extraction_lag` doctor check** — `gbrain doctor` now warns when a + meaningful fraction of your pages have un-extracted edges, and tells you to + run `gbrain extract --stale`. Warn-only by default: it will never break a + CI/cron pipeline that gates on the doctor exit code. Set + `GBRAIN_EXTRACTION_LAG_FAIL_PCT` if you want a hard failure above some + threshold. + +- **An end-of-sync nudge** — after a sync that leaves a backlog, `gbrain sync` + prints one line on stderr pointing you at `gbrain extract --stale`. Silence + it with `GBRAIN_SYNC_NO_EXTRACT_NUDGE=1`. + +Plus `gbrain sync --no-extract` to skip inline extraction on purpose (the +pages then show as stale until you sweep them). + +The mechanism behind all of this is a per-page freshness watermark +(`pages.links_extracted_at`). It treats a page as needing extraction when it +was never extracted, when it was edited after its last extraction (the exact +"I wrote a page via the MCP API and it never got connected" case), or when the +extractor logic itself changed. Existing brains correctly show their real +backlog on the first `gbrain doctor` after upgrade — that visibility is the +whole point. + +### Itemized changes + +- New `gbrain extract --stale [--source-id ] [--catch-up] [--dry-run] [--json]`: + DB-source incremental link + timeline sweep. Small batches with a wall-clock + budget; stamps every processed page (including zero-link pages) only after the + link/timeline writes succeed, so a crash mid-sweep leaves pages stale and they + re-extract idempotently on the next run. +- New `links_extraction_lag` doctor check, wired into both local `gbrain doctor` + and the thin-client/remote report. Warn at >20% stale + (`GBRAIN_EXTRACTION_LAG_WARN_PCT`), hard-fail only when + `GBRAIN_EXTRACTION_LAG_FAIL_PCT` is set. Vacuous-skips brains under 100 pages; + `--source ` scopes it. Strictly a SQL count — no filesystem/git access. +- `gbrain sync` now stamps the extraction watermark for the pages it extracts + inline, and prints a one-line stderr nudge after a sync that leaves a backlog. +- New `gbrain sync --no-extract` flag to skip inline extraction. +- A manual `gbrain extract links --source db` / `extract all --source db` now + also clears the watermark for the pages it processes. +- Migration v112 adds `pages.links_extracted_at` (nullable timestamp) plus a + composite `(source_id, links_extracted_at)` index. No backfill — existing + pages start unstamped so the real backlog surfaces. Metadata-only column add; + instant on both Postgres and PGLite. + +## To take advantage of v0.42.7.0 + +`gbrain upgrade` applies migration v112 automatically. If `gbrain doctor` warns +about a partial migration: + +1. **Apply migrations manually:** + ```bash + gbrain apply-migrations --yes + ``` +2. **Check your extraction backlog and sweep it:** + ```bash + gbrain doctor --json # look for links_extraction_lag + gbrain extract --stale --dry-run # how many pages are behind + gbrain extract --stale --catch-up # sweep them all + ``` +3. **Verify the graph filled in:** + ```bash + gbrain doctor # links_extraction_lag should now be ok + ``` + Typed edges (`SELECT link_type, count(*) FROM links GROUP BY 1`) should jump. +4. **If anything looks wrong,** file an issue at + https://github.com/garrytan/gbrain/issues with `gbrain doctor` output. ## [0.42.6.0] - 2026-06-01 **Most of your people and company pages are one-line stubs. `gbrain enrich --thin` diff --git a/CLAUDE.md b/CLAUDE.md index 3b3b07a35..aa44abb60 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -193,7 +193,7 @@ strict behavior when unset. - `src/core/conversation-parser/` (v0.41.16.0, v0.41.24.0, v0.41.29.0) — 14-pattern built-in chat-format registry + opt-in LLM polish/fallback. Replaces the single inline regex that PR #1461 was trying to extend with one more shape. **v0.41.29.0** added the 14th pattern `bold-name-no-time` (regex `/^\*\*(?!\[)(.+?):\*\*\s*(.*)$/`, declared at index 3 after `bold-paren-time`): parses `**Speaker:** text` with NO per-line timestamp — the shape Circleback / Granola / Zoom emit. Anchors every message at `T00:00:00Z` of the frontmatter date (no fabricated wall-clock; line order preserves sequence), same no-time convention as `irc-classic`. The non-shadow guarantee is the colon-INSIDE-bold regex (NOT declaration order — `parse.ts` scores every candidate independently, order is only the tie-break); the `(?!\[)` lookahead rejects telegram-bracket `**[18:37] Name:**` so disabling telegram-bracket yields an honest `no_match` instead of `speaker="[18:37] Name"`. Because `**Label:** text` is a common prose idiom, the pattern sets the new optional `PatternEntry.score_full_body: true`: `parse.ts` recomputes the winner's acceptance score over the FULL body (not just the 10-line head window) before the `SCORING_MIN_ACCEPTANCE` floor, so a notes page with bold labels clustered in its first 10 lines (head score 0.3, which is NOT `< SCORING_HEAD_TRIGGER_THRESHOLD` so the head fallback never fires) drops to its true low density and stays `no_match` instead of mis-parsing as a conversation. The same wave scrubbed the pre-existing real names in `bold-paren-time`'s `test_positive`. Pinned by the `bold-name-no-time` + clustered-head `no_match` cases in `test/conversation-parser/parse.test.ts` + `bold-name-no-time.jsonl` / `adversarial.jsonl` fixtures. **v0.41.24.0 (closes #1533)** added the 13th pattern `bold-paren-time` (`**Speaker** (HH:MM): text` and `**Speaker** (HH:MM:SS): text` — the time-only shape an OpenClaw meeting-ingestion pipeline produces from Circleback transcripts; date_source: frontmatter) AND tightened the parser's fallback gates in `parse.ts`: `SCORING_HEAD_TRIGGER_THRESHOLD = 0.3` triggers a full-body re-score when the head pass scores below that (closes the bug class where a stray head match at 0.1 — e.g. blockquote that accidentally matches an unrelated pattern — suppressed the fallback entirely); `SCORING_MIN_ACCEPTANCE = 0.05` blocks essay false-positives (a 300-line essay with one stray chat-shape line scores ~0.003, below the floor, stays `no_match` instead of returning a 1-message regex_match). Both gates pinned by 11 new unit cases in `test/conversation-parser/parse.test.ts`. New exported `scorePatternFull(body, entry)` for direct full-body scoring; private `getNonBlankLines(body, headCap?)` + `scoreFromLines(lines, entry)` helpers DRY the quick_reject+regex loop. Empirical against all Circleback meetings in `~/git/brain/meetings/`: 113 of 367 parse correctly post-fix (was 0); 20,167 messages flow through to the fact extractor (was 0). The other 254 are notes-only meetings whose transcripts live in separate files. Plan + Codex absorption at `~/.claude/plans/system-instruction-you-are-working-starry-frost.md`. Modules: `types.ts` (PatternEntry + ParseResult + DateContext + CaptureMap + TimezonePolicy), `builtins.ts` (13 hand-vetted patterns sourced from public format docs — iMessage/Slack, Telegram bracket + text-export, bold-paren-time, Discord classic + export, WhatsApp ISO + US, Signal, Matrix/Element, IRC classic + weechat, Teams export; module-load validation runs every `test_positive[]` + `test_negative[]` sample at startup, so a typo in any built-in regex makes gbrain refuse to start; `DEFAULT_SPEAKER_CLEAN` exported as the PR #1461 helper promoted to a module-level default), `parse.ts` (orchestrator with D18 pattern-priority scoring across the first 10 lines + D8 date derivation chain `explicit > frontmatter.date > effective_date > '1970-01-01'` + D5 multi-line continuation + D19 timezone warning), `llm-base.ts` (shared `runLlmCall` with content-hash cache in-process + DB-persistent via migration v97 + 4-strategy JSON repair + Anthropic-key probe; both polish and fallback are thin wrappers per D6 DRY), `llm-polish.ts` (per D15 opt-IN; headroom guard skips when tracker within $0.10 of cap; pure `applyPolish` for merge/drop/edit ops), `llm-fallback.ts` (per D15 opt-IN; per D17 NO regex inference + NO persistence — silent corruption avoided), `eval.ts` (`scoreFixture` + `aggregateScores` + `parseFixtureJsonl` for the fixture-corpus CI gate), `nightly-probe.ts` (DI-stubbed; mode-gated default per D10 — tokenmax=ON, conservative/balanced opt-in; adversarial false-positive detection). CLI surfaces at `src/commands/eval-conversation-parser.ts` (`gbrain eval conversation-parser ` exit 0/1/2 — wired into `bun run verify` via `check:conversation-parser`) and `src/commands/conversation-parser.ts` (`scan ` for debug, `list-builtins` for operator discoverability, `validate ` emits "deferred to v0.42+" notice). Doctor checks: `conversation_format_coverage` (per-pattern hit count + unmatched %), `progressive_batch_audit_health` (abort-verdict counts from last 7d), `conversation_parser_probe_health` (opt-in nightly probe state). Closes PR #1461 — contributor's `BRACKET_TIME_RX` + `cleanSpeaker` survive verbatim as the `telegram-bracket` built-in + `DEFAULT_SPEAKER_CLEAN` export, Co-Authored-By preserved. Plan + 23 decisions + codex outside-voice absorption at `~/.claude/plans/system-instruction-you-are-working-cuddly-hollerith.md`. Pinned by 74 unit cases across `test/conversation-parser/{parse,llm-base,llm-fallback,llm-polish,nightly-probe}.test.ts` + the 27-case PR #1461 baseline at `test/extract-conversation-facts.test.ts` (back-compat invariant). Migration v97 (`conversation_parser_llm_cache_table`). Fixtures at `test/fixtures/conversation-formats/{imessage,telegram-bracket,whatsapp-iso,whatsapp-us,signal-export,irc-classic,irc-weechat,matrix-element,teams-export,all,adversarial}.jsonl` with `scripts/check-fixture-privacy.sh` banning real-name leaks. - `src/core/progressive-batch/` (v0.41.16.0) — shared ramp-up + cost-cap + verification primitive extracted after rule-of-three was comfortably past (12+ ad-hoc cost-prompt patterns scattered across reindex, reindex-multimodal, book-mirror, brainstorm, eval-suspected-contradictions, post-upgrade-reembed, eval-contradictions/cost-prompt). Wintermute-inspired ramp shape (trial 10 → ramp 100 → ramp 500 → full with verification at each stage), productionized as a runtime primitive with verifier+policy injection (callers describe HOW TO MEASURE SUCCESS, not WHEN TO WAIT FOR CTRL-C). Modules: `types.ts` (Stage, StageVerdict, AbortReason, discriminated `Verifier` union per D20: `OutputCountVerifier | IdempotentMutationVerifier | NoopVerifier`, Policy, StageReport), `orchestrator.ts` (`runProgressiveBatch(items, verifier, policy, runner)` — reads `getCurrentBudgetTracker()` ahead of `Policy.maxCostUsd` per D3 fail-closed; null both ways triggers `abort_cost_cap reason='no_budget_safety_net'`), `audit.ts` (ISO-week JSONL at `~/.gbrain/audit/progressive-batch-YYYY-Www.jsonl` via the shared `audit-writer` primitive), `stage-report.ts` (ASCII stage-report formatter for the default `Policy.onStageReport` handler). Env knobs: `GBRAIN_PROGRESSIVE_BATCH_DISABLED=1` (skip ramp + stderr warn), `GBRAIN_PROGRESSIVE_BATCH_AUTO=1` (skip Ctrl-C grace), `GBRAIN_PROGRESSIVE_BATCH_STAGES=10,100,500` (override). v0.41.16.0 ships with ONE proven consumer (the parser cathedral via extract-conversation-facts retrofit); the 9-site retrofit of `reindex.ts`, `reindex-multimodal.ts`, `reindex-code.ts`, `post-upgrade-reembed.ts`, `book-mirror.ts`, `brainstorm/orchestrator.ts`, `eval-suspected-contradictions.ts`, `eval-contradictions/cost-prompt.ts`, `contextual-reindex-per-chunk.ts` lands as the v0.41.14.0+ follow-up wave per D2's bisectability requirement (each retrofit gets its own commit; behavior parity tested before/after migration). Per D21: sites that previously "jumped straight to full" keep doing so by default; ramp is opt-in per-site via `Policy.interactiveAbortMs > 0`. Pinned by 35 unit cases in `test/progressive-batch/orchestrator.test.ts` covering every verdict path. - `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/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. **v0.42.7.0 (#1696) — link-extraction freshness watermark + `extract --stale`:** the "imported ≠ curated" wave. New `gbrain extract --stale [--source-id ] [--catch-up] [--dry-run] [--json]` branch (`extractStaleFromDB`) — incremental DB-source link+timeline sweep over pages whose `pages.links_extracted_at` watermark is stale. Stale predicate (shared by both engines + the doctor check): `links_extracted_at IS NULL OR links_extracted_at < LINK_EXTRACTOR_VERSION_TS::timestamptz OR updated_at > links_extracted_at` — the `updated_at` arm catches the MCP `put_page` / `sync --no-extract` edited-since-extract path. Three new `BrainEngine` methods (parity in postgres-engine.ts + pglite-engine.ts + bootstrap probes): `countStalePagesForExtraction(opts?)`, `listStalePagesForExtraction({batchSize, afterPageId?, sourceId?, versionTs?})` (returns page CONTENT so the sweep avoids an N+1 `getPage`; `rowToStalePage` in utils.ts maps the row, `StalePageRow` in types.ts), and `markPagesExtractedBatch(refs, defaultExtractedAt)` (3-array unnest `slug[],source_id[],ts[]`; each ref may carry its own `extractedAt`). `STALE_BATCH_SIZE` default 25 (`GBRAIN_EXTRACT_STALE_BATCH`) — small because page bodies are unbounded (the LIMIT is the only fetch-time memory bound); `STALE_TIME_BUDGET_MS` 30min wall-clock (`--catch-up` removes the cap). CDX-4 non-swallowing flush: link/timeline flush throws propagate and abort the batch; stamp LAST so a crash leaves pages unstamped and they re-extract idempotently (`addLinksBatch` ON CONFLICT DO NOTHING + timeline dedup). D4 race fix: `extractStaleFromDB` stamps with each row's READ `updated_at` (not `now()`), so a concurrent edit during the sweep advances `updated_at` past the stamp → page stays stale → re-extracted next run rather than marked fresh-with-old-content. Source-correct stamping at the DB-extract sites via `stampExtracted` (best-effort wrapper, never throws); `extractLinksFromDB` only stamps the combined watermark when `subcommand === 'all'` (both links AND timeline ran — a links-only run must not hide timeline staleness, CDX C3). `LINK_EXTRACTOR_VERSION_TS` lives in `src/core/link-extraction.ts` (bump like `CHUNKER_VERSION` when the extractor shape changes → invalidates all prior stamps). Migration v112 (`pages_links_extracted_at`) adds the nullable `TIMESTAMPTZ` column + composite `(source_id, links_extracted_at)` index (CONCURRENTLY + invalid-remnant pre-drop on Postgres, plain on PGLite), NO backfill so the real backlog surfaces on first `gbrain doctor`. Schema parity in schema.sql + regenerated pglite-schema.ts + schema-embedded.ts + `REQUIRED_BOOTSTRAP_COVERAGE`. `src/commands/doctor.ts:checkLinksExtractionLag` (the `links_extraction_lag` check, also in `doctorReportRemote`) is warn-only by default (>`GBRAIN_EXTRACTION_LAG_WARN_PCT`, default 20%; shared `EXTRACTION_LAG_WARN_PCT_DEFAULT` + `EXTRACTION_LAG_MIN_PAGES=100` + exported `_resolveEnvNumber`), hard-fails only when `GBRAIN_EXTRACTION_LAG_FAIL_PCT` is set; vacuous-skips <100 pages (no `--source`); pre-v112 brains graceful-skip via `isUndefinedColumnError`; strictly a SQL COUNT (safe on the remote/thin-client path). `src/commands/sync.ts` gains `--no-extract` (threaded through single-source + `--all` + `syncOneSource`), stamps `links_extracted_at` for `pagesAffected` at the inline-extract call site (after `extractLinksForSlugs`/`extractTimelineForSlugs`, CDX-6), and `maybeExtractionNudge` prints a one-line stderr nudge after a `synced | first_sync | up_to_date` sync that leaves a backlog (the `first_sync` case is the initial-import scenario, D5; `shouldNudgeAfterSync` pure predicate; `GBRAIN_SYNC_NO_EXTRACT_NUDGE` suppresses; shares the doctor vacuous-skip predicate). `src/core/retry.ts` adds `'extract.stale'` to `BATCH_AUDIT_SITES`; `src/core/doctor-categories.ts` adds `links_extraction_lag` to `BRAIN_CHECK_NAMES`. Pinned by `test/extract-stale.test.ts` (incl. CDX-1 edited-after-stamp regression + crash-contract), `test/sync-inline-extract-stamps.serial.test.ts` (IRON-RULE), `test/sync-nudge-status-gate.test.ts`, `test/doctor-links-extraction-lag.test.ts`, engine-parity (Postgres↔PGLite) for the 3 methods + v112 round-trip. Plan: `~/.claude/plans/system-instruction-you-are-working-squishy-crayon.md`. - `src/core/extract/receipt-writer.ts` + `src/core/extract/rollup-writer.ts` + `src/commands/extract-status.ts` + `src/commands/extract-explain.ts` + `src/commands/extract-benchmark.ts` + `src/core/schema-pack/scaffold-extractable.ts` (v0.41.23.0) — unified extract operator-surface wave. Every shipped extractor (deterministic `facts.conversation` in `src/commands/extract-conversation-facts.ts` + three LLM-backed cycle phases at `src/core/cycle/{extract-atoms,synthesize-concepts,propose-takes,extract-facts}.ts`) now writes ONE receipt page per run (`writeReceipt`) + UPSERTs a row to `extract_rollup_7d` (`upsertExtractRollup`). Receipt slug per D-EXTRACT-17: `extracts/{date}/{kind}/{source_id}/{run_id_short}/round-{N}.md`; frontmatter stamps BOTH `type: extract_receipt` AND `dream_generated: true` per D-EXTRACT-19 (belt+suspenders against extraction-loop guard drift). `extract_receipt` joins `ALL_PAGE_TYPES` in `src/core/types.ts`; `extracts/` prefix gets a 0.3x source-boost demote in `src/core/search/source-boost.ts` so receipts surface in search but never dominate user content. Migration v104 adds `extract_rollup_7d (kind, source_id, day, cost_usd, halt_count, eval_pass_count, eval_fail_count, round_completed_count, rollup_write_failures, updated_at)` with PK `(kind, source_id, day)` + `idx_extract_rollup_7d_day`. Rollup writes are best-effort with process-scoped error-dedup so transient DB failures bump `rollup_write_failures` instead of crashing the cycle. New `extract_health` doctor check reads the rollup for last 7 days, warns at halt-rate > 10% AND when rollup_write_failures > 0; pre-v104 brains report `ok` silently. Operator CLI surface: `gbrain extract status [--source-id ID] [--kind X] [--verbose] [--json]` (7-day rollup, sorted by halt_rate desc + cost desc, kubectl-style top-5 + "more rows" hint, stable `schema_version: 1` JSON envelope); `gbrain extract --explain ` (resolution chain: pack-declared vs built-in cycle phase, prompt_template + fixture_corpus paths with `✓`/`(missing)` markers, last 7d rollup); `gbrain extract benchmark --pack X --kind Y` (loads pack fixture corpus through strict D-EXTRACT-21 path validation — rejects absolute paths, `..` traversal, null bytes, AND symlinks resolving outside pack root; v0.41.23 ships as a stub-reporter — LLM dispatch deferred to a follow-up release). Pack-author authoring loop: `src/core/schema-pack/manifest-v1.ts` widens `extractable` from `z.boolean()` to `z.union([z.boolean(), ExtractableSpecSchema])` (carries `prompt_template`, `fixture_corpus`, `eval_dimensions`, `benchmark_min_recall`, plus reserved `verifier_path` per D-EXTRACT-37 — parses but refuses at runtime in v0.41.23); new `extractableSpecsFromPack` + `getExtractableSpec` + `refuseVerifierPathInV042` helpers in `src/core/schema-pack/extractable.ts`; `gbrain schema scaffold-extractable --pack ` declares the type extractable in one verb, generates 5 placeholder fixtures + a prompt template stub under `packs//{fixtures,prompts}/extract/`, refuses to overwrite without `--force`. Tests: 22 cases in `test/extractable-spec-widening.test.ts` (back-compat boolean shape + struct shape + D-EXTRACT-37 verifier_path refuse), 12 cases in `test/extract/receipt-writer.test.ts` (uses canonical PGLite block per CLAUDE.md R3+R4 — one engine per file with `resetPgliteState` in `beforeEach`), 17 cases in `test/extract/benchmark.test.ts` (path validation rejections + JSONL contract), 15 cases in `test/extract/status.test.ts` (pure aggregation + formatting), 15 cases in `test/schema-pack/scaffold-extractable.test.ts` (scaffold mechanics + explicit privacy-rule guards against real-name leakage), 8 cases in `test/doctor-extract-health.test.ts`. Plan persisted at `~/.claude/plans/system-instruction-you-are-working-stateless-dragonfly.md`. Deferred to a follow-up release: replay versioning + `gbrain extract replay --since v`; unified LLM dispatcher (deliberately respects v0.41.13 head comment that extraction is regex + cost-cap value-add lives at embed step). - `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/TODOS.md b/TODOS.md index 3fe7ac7ee..b15d338d4 100644 --- a/TODOS.md +++ b/TODOS.md @@ -1,5 +1,40 @@ # TODOS +## v0.42.7.0 extract-in-default-loop follow-ups (v0.42+) + +Filed from the v0.42.2.0 wave (#1696 link/timeline extraction freshness +watermark). Both surfaced by the Codex review (P1-D, P1-C) and deliberately +scoped OUT — neither is a #1696 regression. See plan + GSTACK REVIEW REPORT at +`~/.claude/plans/system-instruction-you-are-working-squishy-crayon.md`. + +- [ ] **P2 — Repo-wide: `DROP INDEX CONCURRENTLY` inside a `DO $$` block is + Postgres-invalid.** `CONCURRENTLY` cannot run inside a transaction, and a `DO` + block IS a transaction — so the invalid-index pre-drop guard throws + `cannot run inside a transaction block` IF the branch ever fires (only on a + retry after a prior failed concurrent build). Migration v112 + (`pages_links_extracted_at`) copies this pattern verbatim from shipped + precedent: `idx_pages_updated_at_desc` (migrate.ts:~502), + `pages_deleted_at_purge_idx` (~1619), `pages_coalesce_date_idx` (~1967). It is + latent (the IF-EXISTS check returns false on a clean build → EXECUTE never + runs) and has never been hit in production. Fix repo-wide in ONE sweep: replace + each `DO $$ ... EXECUTE 'DROP INDEX CONCURRENTLY ...'` with a plain top-level + `SELECT indisvalid` probe + a bare top-level `DROP INDEX CONCURRENTLY IF EXISTS` + statement (the migration runner already runs these `transaction: false`). Do + NOT single out v112 — fixing one diverges from the precedent; sweep all of them + together with a shared helper. Needs its own review (touches every CONCURRENTLY + migration). +- [ ] **P3 — Add-only extraction never deletes obsolete edges; the watermark now + asserts a currency it can't fully deliver.** All gbrain extraction is add-only + (`addLinksBatch` ON CONFLICT DO NOTHING, inline sync + `extractLinksFromDB` + + `extract --stale`). A page edit that REMOVES a link adds nothing and never + deletes the now-absent edge, yet `links_extracted_at` marks the page current, + so `gbrain doctor` reports OK while the graph carries a stale edge. Pre-existing + architectural property (not new in #1696), but the watermark makes it more + visible. Real fix needs a link-provenance column (`link_source` / extracted-by + marker) so a re-extract can safely DELETE extracted-but-now-absent edges for a + page+source without clobbering manually-added or auto-link edges — mirrors the + v0.41.37.0 tag-provenance deferral (#1621-followup). Defer until that column + lands; until then `extract --stale` is reconcile-add-only by design. ## v0.42.5.0 watchdog / pooler-reap / lens-backlog follow-ups (v0.42+) Deferred from the v0.42.5.0 wave (issue #1678). The shipped fixes are complete diff --git a/VERSION b/VERSION index 6890820d0..218cef5f8 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.42.6.0 \ No newline at end of file +0.42.7.0 diff --git a/llms-full.txt b/llms-full.txt index 764d17afa..ab8c9a530 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -335,7 +335,7 @@ strict behavior when unset. - `src/core/conversation-parser/` (v0.41.16.0, v0.41.24.0, v0.41.29.0) — 14-pattern built-in chat-format registry + opt-in LLM polish/fallback. Replaces the single inline regex that PR #1461 was trying to extend with one more shape. **v0.41.29.0** added the 14th pattern `bold-name-no-time` (regex `/^\*\*(?!\[)(.+?):\*\*\s*(.*)$/`, declared at index 3 after `bold-paren-time`): parses `**Speaker:** text` with NO per-line timestamp — the shape Circleback / Granola / Zoom emit. Anchors every message at `T00:00:00Z` of the frontmatter date (no fabricated wall-clock; line order preserves sequence), same no-time convention as `irc-classic`. The non-shadow guarantee is the colon-INSIDE-bold regex (NOT declaration order — `parse.ts` scores every candidate independently, order is only the tie-break); the `(?!\[)` lookahead rejects telegram-bracket `**[18:37] Name:**` so disabling telegram-bracket yields an honest `no_match` instead of `speaker="[18:37] Name"`. Because `**Label:** text` is a common prose idiom, the pattern sets the new optional `PatternEntry.score_full_body: true`: `parse.ts` recomputes the winner's acceptance score over the FULL body (not just the 10-line head window) before the `SCORING_MIN_ACCEPTANCE` floor, so a notes page with bold labels clustered in its first 10 lines (head score 0.3, which is NOT `< SCORING_HEAD_TRIGGER_THRESHOLD` so the head fallback never fires) drops to its true low density and stays `no_match` instead of mis-parsing as a conversation. The same wave scrubbed the pre-existing real names in `bold-paren-time`'s `test_positive`. Pinned by the `bold-name-no-time` + clustered-head `no_match` cases in `test/conversation-parser/parse.test.ts` + `bold-name-no-time.jsonl` / `adversarial.jsonl` fixtures. **v0.41.24.0 (closes #1533)** added the 13th pattern `bold-paren-time` (`**Speaker** (HH:MM): text` and `**Speaker** (HH:MM:SS): text` — the time-only shape an OpenClaw meeting-ingestion pipeline produces from Circleback transcripts; date_source: frontmatter) AND tightened the parser's fallback gates in `parse.ts`: `SCORING_HEAD_TRIGGER_THRESHOLD = 0.3` triggers a full-body re-score when the head pass scores below that (closes the bug class where a stray head match at 0.1 — e.g. blockquote that accidentally matches an unrelated pattern — suppressed the fallback entirely); `SCORING_MIN_ACCEPTANCE = 0.05` blocks essay false-positives (a 300-line essay with one stray chat-shape line scores ~0.003, below the floor, stays `no_match` instead of returning a 1-message regex_match). Both gates pinned by 11 new unit cases in `test/conversation-parser/parse.test.ts`. New exported `scorePatternFull(body, entry)` for direct full-body scoring; private `getNonBlankLines(body, headCap?)` + `scoreFromLines(lines, entry)` helpers DRY the quick_reject+regex loop. Empirical against all Circleback meetings in `~/git/brain/meetings/`: 113 of 367 parse correctly post-fix (was 0); 20,167 messages flow through to the fact extractor (was 0). The other 254 are notes-only meetings whose transcripts live in separate files. Plan + Codex absorption at `~/.claude/plans/system-instruction-you-are-working-starry-frost.md`. Modules: `types.ts` (PatternEntry + ParseResult + DateContext + CaptureMap + TimezonePolicy), `builtins.ts` (13 hand-vetted patterns sourced from public format docs — iMessage/Slack, Telegram bracket + text-export, bold-paren-time, Discord classic + export, WhatsApp ISO + US, Signal, Matrix/Element, IRC classic + weechat, Teams export; module-load validation runs every `test_positive[]` + `test_negative[]` sample at startup, so a typo in any built-in regex makes gbrain refuse to start; `DEFAULT_SPEAKER_CLEAN` exported as the PR #1461 helper promoted to a module-level default), `parse.ts` (orchestrator with D18 pattern-priority scoring across the first 10 lines + D8 date derivation chain `explicit > frontmatter.date > effective_date > '1970-01-01'` + D5 multi-line continuation + D19 timezone warning), `llm-base.ts` (shared `runLlmCall` with content-hash cache in-process + DB-persistent via migration v97 + 4-strategy JSON repair + Anthropic-key probe; both polish and fallback are thin wrappers per D6 DRY), `llm-polish.ts` (per D15 opt-IN; headroom guard skips when tracker within $0.10 of cap; pure `applyPolish` for merge/drop/edit ops), `llm-fallback.ts` (per D15 opt-IN; per D17 NO regex inference + NO persistence — silent corruption avoided), `eval.ts` (`scoreFixture` + `aggregateScores` + `parseFixtureJsonl` for the fixture-corpus CI gate), `nightly-probe.ts` (DI-stubbed; mode-gated default per D10 — tokenmax=ON, conservative/balanced opt-in; adversarial false-positive detection). CLI surfaces at `src/commands/eval-conversation-parser.ts` (`gbrain eval conversation-parser ` exit 0/1/2 — wired into `bun run verify` via `check:conversation-parser`) and `src/commands/conversation-parser.ts` (`scan ` for debug, `list-builtins` for operator discoverability, `validate ` emits "deferred to v0.42+" notice). Doctor checks: `conversation_format_coverage` (per-pattern hit count + unmatched %), `progressive_batch_audit_health` (abort-verdict counts from last 7d), `conversation_parser_probe_health` (opt-in nightly probe state). Closes PR #1461 — contributor's `BRACKET_TIME_RX` + `cleanSpeaker` survive verbatim as the `telegram-bracket` built-in + `DEFAULT_SPEAKER_CLEAN` export, Co-Authored-By preserved. Plan + 23 decisions + codex outside-voice absorption at `~/.claude/plans/system-instruction-you-are-working-cuddly-hollerith.md`. Pinned by 74 unit cases across `test/conversation-parser/{parse,llm-base,llm-fallback,llm-polish,nightly-probe}.test.ts` + the 27-case PR #1461 baseline at `test/extract-conversation-facts.test.ts` (back-compat invariant). Migration v97 (`conversation_parser_llm_cache_table`). Fixtures at `test/fixtures/conversation-formats/{imessage,telegram-bracket,whatsapp-iso,whatsapp-us,signal-export,irc-classic,irc-weechat,matrix-element,teams-export,all,adversarial}.jsonl` with `scripts/check-fixture-privacy.sh` banning real-name leaks. - `src/core/progressive-batch/` (v0.41.16.0) — shared ramp-up + cost-cap + verification primitive extracted after rule-of-three was comfortably past (12+ ad-hoc cost-prompt patterns scattered across reindex, reindex-multimodal, book-mirror, brainstorm, eval-suspected-contradictions, post-upgrade-reembed, eval-contradictions/cost-prompt). Wintermute-inspired ramp shape (trial 10 → ramp 100 → ramp 500 → full with verification at each stage), productionized as a runtime primitive with verifier+policy injection (callers describe HOW TO MEASURE SUCCESS, not WHEN TO WAIT FOR CTRL-C). Modules: `types.ts` (Stage, StageVerdict, AbortReason, discriminated `Verifier` union per D20: `OutputCountVerifier | IdempotentMutationVerifier | NoopVerifier`, Policy, StageReport), `orchestrator.ts` (`runProgressiveBatch(items, verifier, policy, runner)` — reads `getCurrentBudgetTracker()` ahead of `Policy.maxCostUsd` per D3 fail-closed; null both ways triggers `abort_cost_cap reason='no_budget_safety_net'`), `audit.ts` (ISO-week JSONL at `~/.gbrain/audit/progressive-batch-YYYY-Www.jsonl` via the shared `audit-writer` primitive), `stage-report.ts` (ASCII stage-report formatter for the default `Policy.onStageReport` handler). Env knobs: `GBRAIN_PROGRESSIVE_BATCH_DISABLED=1` (skip ramp + stderr warn), `GBRAIN_PROGRESSIVE_BATCH_AUTO=1` (skip Ctrl-C grace), `GBRAIN_PROGRESSIVE_BATCH_STAGES=10,100,500` (override). v0.41.16.0 ships with ONE proven consumer (the parser cathedral via extract-conversation-facts retrofit); the 9-site retrofit of `reindex.ts`, `reindex-multimodal.ts`, `reindex-code.ts`, `post-upgrade-reembed.ts`, `book-mirror.ts`, `brainstorm/orchestrator.ts`, `eval-suspected-contradictions.ts`, `eval-contradictions/cost-prompt.ts`, `contextual-reindex-per-chunk.ts` lands as the v0.41.14.0+ follow-up wave per D2's bisectability requirement (each retrofit gets its own commit; behavior parity tested before/after migration). Per D21: sites that previously "jumped straight to full" keep doing so by default; ramp is opt-in per-site via `Policy.interactiveAbortMs > 0`. Pinned by 35 unit cases in `test/progressive-batch/orchestrator.test.ts` covering every verdict path. - `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/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. **v0.42.7.0 (#1696) — link-extraction freshness watermark + `extract --stale`:** the "imported ≠ curated" wave. New `gbrain extract --stale [--source-id ] [--catch-up] [--dry-run] [--json]` branch (`extractStaleFromDB`) — incremental DB-source link+timeline sweep over pages whose `pages.links_extracted_at` watermark is stale. Stale predicate (shared by both engines + the doctor check): `links_extracted_at IS NULL OR links_extracted_at < LINK_EXTRACTOR_VERSION_TS::timestamptz OR updated_at > links_extracted_at` — the `updated_at` arm catches the MCP `put_page` / `sync --no-extract` edited-since-extract path. Three new `BrainEngine` methods (parity in postgres-engine.ts + pglite-engine.ts + bootstrap probes): `countStalePagesForExtraction(opts?)`, `listStalePagesForExtraction({batchSize, afterPageId?, sourceId?, versionTs?})` (returns page CONTENT so the sweep avoids an N+1 `getPage`; `rowToStalePage` in utils.ts maps the row, `StalePageRow` in types.ts), and `markPagesExtractedBatch(refs, defaultExtractedAt)` (3-array unnest `slug[],source_id[],ts[]`; each ref may carry its own `extractedAt`). `STALE_BATCH_SIZE` default 25 (`GBRAIN_EXTRACT_STALE_BATCH`) — small because page bodies are unbounded (the LIMIT is the only fetch-time memory bound); `STALE_TIME_BUDGET_MS` 30min wall-clock (`--catch-up` removes the cap). CDX-4 non-swallowing flush: link/timeline flush throws propagate and abort the batch; stamp LAST so a crash leaves pages unstamped and they re-extract idempotently (`addLinksBatch` ON CONFLICT DO NOTHING + timeline dedup). D4 race fix: `extractStaleFromDB` stamps with each row's READ `updated_at` (not `now()`), so a concurrent edit during the sweep advances `updated_at` past the stamp → page stays stale → re-extracted next run rather than marked fresh-with-old-content. Source-correct stamping at the DB-extract sites via `stampExtracted` (best-effort wrapper, never throws); `extractLinksFromDB` only stamps the combined watermark when `subcommand === 'all'` (both links AND timeline ran — a links-only run must not hide timeline staleness, CDX C3). `LINK_EXTRACTOR_VERSION_TS` lives in `src/core/link-extraction.ts` (bump like `CHUNKER_VERSION` when the extractor shape changes → invalidates all prior stamps). Migration v112 (`pages_links_extracted_at`) adds the nullable `TIMESTAMPTZ` column + composite `(source_id, links_extracted_at)` index (CONCURRENTLY + invalid-remnant pre-drop on Postgres, plain on PGLite), NO backfill so the real backlog surfaces on first `gbrain doctor`. Schema parity in schema.sql + regenerated pglite-schema.ts + schema-embedded.ts + `REQUIRED_BOOTSTRAP_COVERAGE`. `src/commands/doctor.ts:checkLinksExtractionLag` (the `links_extraction_lag` check, also in `doctorReportRemote`) is warn-only by default (>`GBRAIN_EXTRACTION_LAG_WARN_PCT`, default 20%; shared `EXTRACTION_LAG_WARN_PCT_DEFAULT` + `EXTRACTION_LAG_MIN_PAGES=100` + exported `_resolveEnvNumber`), hard-fails only when `GBRAIN_EXTRACTION_LAG_FAIL_PCT` is set; vacuous-skips <100 pages (no `--source`); pre-v112 brains graceful-skip via `isUndefinedColumnError`; strictly a SQL COUNT (safe on the remote/thin-client path). `src/commands/sync.ts` gains `--no-extract` (threaded through single-source + `--all` + `syncOneSource`), stamps `links_extracted_at` for `pagesAffected` at the inline-extract call site (after `extractLinksForSlugs`/`extractTimelineForSlugs`, CDX-6), and `maybeExtractionNudge` prints a one-line stderr nudge after a `synced | first_sync | up_to_date` sync that leaves a backlog (the `first_sync` case is the initial-import scenario, D5; `shouldNudgeAfterSync` pure predicate; `GBRAIN_SYNC_NO_EXTRACT_NUDGE` suppresses; shares the doctor vacuous-skip predicate). `src/core/retry.ts` adds `'extract.stale'` to `BATCH_AUDIT_SITES`; `src/core/doctor-categories.ts` adds `links_extraction_lag` to `BRAIN_CHECK_NAMES`. Pinned by `test/extract-stale.test.ts` (incl. CDX-1 edited-after-stamp regression + crash-contract), `test/sync-inline-extract-stamps.serial.test.ts` (IRON-RULE), `test/sync-nudge-status-gate.test.ts`, `test/doctor-links-extraction-lag.test.ts`, engine-parity (Postgres↔PGLite) for the 3 methods + v112 round-trip. Plan: `~/.claude/plans/system-instruction-you-are-working-squishy-crayon.md`. - `src/core/extract/receipt-writer.ts` + `src/core/extract/rollup-writer.ts` + `src/commands/extract-status.ts` + `src/commands/extract-explain.ts` + `src/commands/extract-benchmark.ts` + `src/core/schema-pack/scaffold-extractable.ts` (v0.41.23.0) — unified extract operator-surface wave. Every shipped extractor (deterministic `facts.conversation` in `src/commands/extract-conversation-facts.ts` + three LLM-backed cycle phases at `src/core/cycle/{extract-atoms,synthesize-concepts,propose-takes,extract-facts}.ts`) now writes ONE receipt page per run (`writeReceipt`) + UPSERTs a row to `extract_rollup_7d` (`upsertExtractRollup`). Receipt slug per D-EXTRACT-17: `extracts/{date}/{kind}/{source_id}/{run_id_short}/round-{N}.md`; frontmatter stamps BOTH `type: extract_receipt` AND `dream_generated: true` per D-EXTRACT-19 (belt+suspenders against extraction-loop guard drift). `extract_receipt` joins `ALL_PAGE_TYPES` in `src/core/types.ts`; `extracts/` prefix gets a 0.3x source-boost demote in `src/core/search/source-boost.ts` so receipts surface in search but never dominate user content. Migration v104 adds `extract_rollup_7d (kind, source_id, day, cost_usd, halt_count, eval_pass_count, eval_fail_count, round_completed_count, rollup_write_failures, updated_at)` with PK `(kind, source_id, day)` + `idx_extract_rollup_7d_day`. Rollup writes are best-effort with process-scoped error-dedup so transient DB failures bump `rollup_write_failures` instead of crashing the cycle. New `extract_health` doctor check reads the rollup for last 7 days, warns at halt-rate > 10% AND when rollup_write_failures > 0; pre-v104 brains report `ok` silently. Operator CLI surface: `gbrain extract status [--source-id ID] [--kind X] [--verbose] [--json]` (7-day rollup, sorted by halt_rate desc + cost desc, kubectl-style top-5 + "more rows" hint, stable `schema_version: 1` JSON envelope); `gbrain extract --explain ` (resolution chain: pack-declared vs built-in cycle phase, prompt_template + fixture_corpus paths with `✓`/`(missing)` markers, last 7d rollup); `gbrain extract benchmark --pack X --kind Y` (loads pack fixture corpus through strict D-EXTRACT-21 path validation — rejects absolute paths, `..` traversal, null bytes, AND symlinks resolving outside pack root; v0.41.23 ships as a stub-reporter — LLM dispatch deferred to a follow-up release). Pack-author authoring loop: `src/core/schema-pack/manifest-v1.ts` widens `extractable` from `z.boolean()` to `z.union([z.boolean(), ExtractableSpecSchema])` (carries `prompt_template`, `fixture_corpus`, `eval_dimensions`, `benchmark_min_recall`, plus reserved `verifier_path` per D-EXTRACT-37 — parses but refuses at runtime in v0.41.23); new `extractableSpecsFromPack` + `getExtractableSpec` + `refuseVerifierPathInV042` helpers in `src/core/schema-pack/extractable.ts`; `gbrain schema scaffold-extractable --pack ` declares the type extractable in one verb, generates 5 placeholder fixtures + a prompt template stub under `packs//{fixtures,prompts}/extract/`, refuses to overwrite without `--force`. Tests: 22 cases in `test/extractable-spec-widening.test.ts` (back-compat boolean shape + struct shape + D-EXTRACT-37 verifier_path refuse), 12 cases in `test/extract/receipt-writer.test.ts` (uses canonical PGLite block per CLAUDE.md R3+R4 — one engine per file with `resetPgliteState` in `beforeEach`), 17 cases in `test/extract/benchmark.test.ts` (path validation rejections + JSONL contract), 15 cases in `test/extract/status.test.ts` (pure aggregation + formatting), 15 cases in `test/schema-pack/scaffold-extractable.test.ts` (scaffold mechanics + explicit privacy-rule guards against real-name leakage), 8 cases in `test/doctor-extract-health.test.ts`. Plan persisted at `~/.claude/plans/system-instruction-you-are-working-stateless-dragonfly.md`. Deferred to a follow-up release: replay versioning + `gbrain extract replay --since v`; unified LLM dispatcher (deliberately respects v0.41.13 head comment that extraction is regex + cost-cap value-add lives at embed step). - `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 c878a4c0a..170cf3c28 100644 --- a/package.json +++ b/package.json @@ -142,5 +142,5 @@ "bun": ">=1.3.10" }, "license": "MIT", - "version": "0.42.6.0" + "version": "0.42.7.0" } diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index ed665a7dd..cd0a575a9 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -32,6 +32,8 @@ import { isSourceUnchangedSinceSync } from '../core/git-head.ts'; // this pure comparator (no git subprocess on the HTTP MCP doctor path). import { lagFromContentMs } from '../core/source-health.ts'; import { CHUNKER_VERSION } from '../core/chunkers/code.ts'; +import { LINK_EXTRACTOR_VERSION_TS } from '../core/link-extraction.ts'; +import { isUndefinedColumnError } from '../core/utils.ts'; export interface Check { name: string; @@ -669,6 +671,12 @@ export async function doctorReportRemote(engine: BrainEngine): Promise { const checkSubagentProvider = checkSubagentCapability; void checkSubagentProvider; -// Module-scoped flag so the NaN-fallback warning fires once per process. -let _syncFreshnessEnvWarned = false; +// Module-scoped set so each invalid-env-var warning fires once per process, +// per variable name (v0.42.7 #1696: was a single bool shared across all vars). +const _envNumberWarned = new Set(); -function _resolveSyncFreshnessHours(varName: string, fallback: number): number { +/** + * v0.42.7 (#1696): single source of truth for the extraction-lag warn + * threshold (percent). Both the `links_extraction_lag` doctor check AND the + * end-of-sync nudge (`sync.ts:maybeExtractionNudge`) resolve through this + + * `_resolveEnvNumber` so "the nudge fires iff doctor would warn" can't drift. + */ +export const EXTRACTION_LAG_WARN_PCT_DEFAULT = 20; +/** Min non-deleted page count below which extraction-lag is vacuous-skipped + * (unless an explicit --source scope is set). Shared by doctor + the sync + * nudge (D6/C4) so their skip predicates match exactly. */ +export const EXTRACTION_LAG_MIN_PAGES = 100; + +/** + * v0.42.7 (#1696, C1): generic "read a positive number from an env var, warn + * once + fall back on garbage." Extracted from _resolveSyncFreshnessHours so + * the percent-threshold doctor checks don't reuse a `...Hours`-named helper. + * `opts.unit` is purely cosmetic for the warning string ('h', '%', ''). + * Exported (D3) so the sync nudge resolves the threshold the same way. + */ +export function _resolveEnvNumber(varName: string, fallback: number, opts?: { unit?: string }): number { const raw = process.env[varName]; if (raw === undefined || raw === '') return fallback; const n = Number(raw); if (!Number.isFinite(n) || n <= 0) { - if (!_syncFreshnessEnvWarned) { - _syncFreshnessEnvWarned = true; + if (!_envNumberWarned.has(varName)) { + _envNumberWarned.add(varName); console.warn( - `[gbrain doctor] Ignoring invalid ${varName}=${raw}; using default ${fallback}h.`, + `[gbrain doctor] Ignoring invalid ${varName}=${raw}; using default ${fallback}${opts?.unit ?? ''}.`, ); } return fallback; @@ -2317,6 +2345,10 @@ function _resolveSyncFreshnessHours(varName: string, fallback: number): number { return n; } +function _resolveSyncFreshnessHours(varName: string, fallback: number): number { + return _resolveEnvNumber(varName, fallback, { unit: 'h' }); +} + /** * Sync freshness check (v0.32.4) — verify that sources with local_path have * been synced recently. Detects the silent failure mode where `gbrain sync` @@ -2526,6 +2558,89 @@ export async function computeConversationFactsBacklogCheck( } } +/** + * v0.42.7 (#1696) — links_extraction_lag doctor check. + * + * The signal that surfaces the "imported ≠ curated" root cause: pages whose + * link/timeline extraction is stale (never run, edited-since, or extractor + * bumped). Without it, a brain can run for months at 0% typed-edge coverage + * with nothing warning the operator. + * + * Warn-only by DEFAULT (>20% stale). Hard-fail ONLY when the operator opts in + * via GBRAIN_EXTRACTION_LAG_FAIL_PCT — so a just-upgraded 280K-page brain + * (every page NULL → 100% stale) gets a loud WARN, never a non-zero exit that + * would break a CI/cron pipeline gating on `gbrain doctor`. + * + * Vacuous-skip on tiny brains (<100 pages, no --source) like orphan_ratio. + * Pre-v112 brains (column missing) degrade to OK via isUndefinedColumnError. + * Strictly SQL — no filesystem/git access — so it's safe to wire into the + * thin-client doctorReportRemote path (CDX-5 trust boundary). + * + * `opts.sourceId` scopes both the denominator and the stale count to one + * source (the explicit-only `--source` parse, like orphan_ratio). + */ +export async function checkLinksExtractionLag( + engine: BrainEngine, + opts?: { sourceId?: string }, +): Promise { + const name = 'links_extraction_lag'; + const sourceId = opts?.sourceId; + const fix = "Run: gbrain extract --stale"; + try { + const totalRows = await engine.executeRaw<{ count: number }>( + sourceId + ? `SELECT count(*)::int AS count FROM pages WHERE deleted_at IS NULL AND source_id = $1` + : `SELECT count(*)::int AS count FROM pages WHERE deleted_at IS NULL`, + sourceId ? [sourceId] : [], + ); + const total = Number(totalRows[0]?.count ?? 0); + if (total === 0) { + return { name, status: 'ok', message: 'Extraction lag not applicable (no pages)' }; + } + // Vacuous-skip tiny brains unless explicitly source-scoped. Shared floor + // const so the sync nudge (D6/C4) skips on the exact same predicate. + if (total < EXTRACTION_LAG_MIN_PAGES && !sourceId) { + return { name, status: 'ok', message: `Extraction lag not applicable (${total} pages — too few to assess)` }; + } + + const stale = await engine.countStalePagesForExtraction({ sourceId, versionTs: LINK_EXTRACTOR_VERSION_TS }); + const pct = (stale / total) * 100; + const pctStr = pct.toFixed(0); + const scope = sourceId ? ` in source '${sourceId}'` : ''; + + const warnPct = _resolveEnvNumber('GBRAIN_EXTRACTION_LAG_WARN_PCT', EXTRACTION_LAG_WARN_PCT_DEFAULT, { unit: '%' }); + // Fail threshold is DISABLED unless explicitly set (warn-only default). A + // bare unset env var → no hard-fail; invalid value → warn-once + disabled. + let failPct: number | undefined; + const failRaw = process.env.GBRAIN_EXTRACTION_LAG_FAIL_PCT; + if (failRaw !== undefined && failRaw !== '') { + const n = Number(failRaw); + if (Number.isFinite(n) && n > 0) { + failPct = n; + } else if (!_envNumberWarned.has('GBRAIN_EXTRACTION_LAG_FAIL_PCT')) { + _envNumberWarned.add('GBRAIN_EXTRACTION_LAG_FAIL_PCT'); + console.warn(`[gbrain doctor] Ignoring invalid GBRAIN_EXTRACTION_LAG_FAIL_PCT=${failRaw}; hard-fail stays disabled.`); + } + } + + const details = { total, stale, pct: Number(pctStr), warn_pct: warnPct, fail_pct: failPct ?? null, source_id: sourceId ?? null }; + if (failPct !== undefined && pct > failPct) { + return { name, status: 'fail', message: `${stale}/${total} pages (${pctStr}%)${scope} need link/timeline extraction (> ${failPct}% fail threshold). ${fix}`, details }; + } + if (pct > warnPct) { + return { name, status: 'warn', message: `${stale}/${total} pages (${pctStr}%)${scope} have un-extracted edges. ${fix}`, details }; + } + return { name, status: 'ok', message: `Extraction current: ${stale}/${total} pages (${pctStr}%) stale${scope}`, details }; + } catch (e) { + // Pre-v112 brain: links_extracted_at column doesn't exist yet. Graceful OK + // (migration/bootstrap adds it; nothing to assess until then). + if (isUndefinedColumnError(e, 'links_extracted_at')) { + return { name, status: 'ok', message: 'links_extracted_at not present (pre-v112 brain)' }; + } + return { name, status: 'warn', message: `Could not check links_extraction_lag: ${(e as Error).message}` }; + } +} + /** * issue #1678 — extract_atoms_backlog doctor check. * @@ -5975,6 +6090,10 @@ export async function buildChecks( // v0.41.19.0 (Issue 5): sync --all consolidation nudge. progress.heartbeat('sync_consolidation'); checks.push(await checkSyncConsolidation(engine)); + // v0.42.7 (#1696): link-extraction lag. --source scopes it (explicit-only + // parse, like orphan_ratio); bare doctor stays brain-wide. Fix: extract --stale. + progress.heartbeat('links_extraction_lag'); + checks.push(await checkLinksExtractionLag(engine, { sourceId: orphanRatioSourceId })); // v0.38 — full-cycle freshness, sibling to sync_freshness. Reads // last_full_cycle_at from sources.config; mirrors what autopilot's // per-source dispatch gate sees. diff --git a/src/commands/extract.ts b/src/commands/extract.ts index 64db1e6f9..0f157c44d 100644 --- a/src/commands/extract.ts +++ b/src/commands/extract.ts @@ -35,8 +35,8 @@ import type { PageType } from '../core/types.ts'; import { parseMarkdown } from '../core/markdown.ts'; import { extractPageLinks, parseTimelineEntries, inferLinkType, makeResolver, - extractFrontmatterLinks, - type UnresolvedFrontmatterRef, + extractFrontmatterLinks, LINK_EXTRACTOR_VERSION_TS, + type UnresolvedFrontmatterRef, type LinkCandidate, } from '../core/link-extraction.ts'; import { createProgress } from '../core/progress.ts'; import { getCliOptions, cliOptsToProgressOptions } from '../core/cli-options.ts'; @@ -67,6 +67,71 @@ import { parseWorkers, resolveWorkersWithClamp } from '../core/sync-concurrency. // small (a malformed row aborts at most 100, not thousands). const BATCH_SIZE = 100; +// v0.42.7 (#1696): keyset batch size for `extract --stale`. SMALL by design — +// listStalePagesForExtraction returns page CONTENT (compiled_truth + timeline), +// which is unbounded (25MB transcript pages exist). The LIMIT is the only memory +// bound: the per-batch byte cap CDX-5 described can't run post-fetch (the fetch +// itself is the OOM point), so a small default count is the real safety net — +// 25 caps the worst case at ~625MB even if every page is a 25MB transcript. +// Normal pages are KBs; raise via GBRAIN_EXTRACT_STALE_BATCH for throughput. +const STALE_BATCH_SIZE = Math.max(1, Number(process.env.GBRAIN_EXTRACT_STALE_BATCH) || 25); +// v0.42.7: wall-clock budget for one `extract --stale` invocation (default +// 30 min). `--catch-up` removes the cap (loops until 0 stale). Mirrors +// embedAllStale's time-budget shape. +const STALE_TIME_BUDGET_MS = Math.max(1000, Number(process.env.GBRAIN_EXTRACT_TIME_BUDGET_MS) || 30 * 60 * 1000); + +/** + * v0.42.7 (#1696): best-effort extraction stamp for the source-correct write + * sites (inline sync, `extract --source db`). Wraps `markPagesExtractedBatch` + * and NEVER throws — a stamp failure here just means the page stays "stale" and + * gets swept by `extract --stale` later. Do NOT use this in the `--stale` sweep + * itself: there the stamp is the resume mechanism and a failure must surface + * (CDX-4 — see extractStaleFromDB). + */ +export async function stampExtracted( + engine: BrainEngine, + refs: Array<{ slug: string; source_id: string }>, + at: string = new Date().toISOString(), +): Promise { + if (refs.length === 0) return; + try { + await engine.markPagesExtractedBatch(refs, at); + } catch { /* best-effort: page stays stale, extract --stale re-sweeps it */ } +} + +/** + * v0.42.7 (#1696): pure cross-source resolution for one extracted link + * candidate. Validates both endpoints exist (else the batch JOIN drops the row), + * then picks from_source_id / to_source_id: prefer the origin page's source, + * fall back to 'default', else skip (never push a wrong-source edge). Returns + * null when the candidate should be skipped. Shared by extractLinksFromDB and + * extractStaleFromDB so the F10 multi-source resolution can't drift. + */ +export function resolveCandidateSources( + c: LinkCandidate, + pageSlug: string, + pageSourceId: string, + allSlugs: Set, + slugToSources: Map, +): { fromSlug: string; fromSourceId: string; toSourceId: string } | null { + const fromSlug = c.fromSlug ?? pageSlug; + if (!allSlugs.has(c.targetSlug)) return null; + if (!allSlugs.has(fromSlug)) return null; + const fromSources = slugToSources.get(fromSlug) ?? []; + const fromSourceId = fromSources.includes(pageSourceId) ? pageSourceId + : (fromSources.includes('default') ? 'default' : fromSources[0]); + const targetSources = slugToSources.get(c.targetSlug) ?? []; + let toSourceId: string; + if (targetSources.includes(fromSourceId)) { + toSourceId = fromSourceId; + } else if (targetSources.includes('default')) { + toSourceId = 'default'; + } else { + return null; + } + return { fromSlug, fromSourceId, toSourceId }; +} + // isRetryableConnError reference retained for any inline classification at // call sites. Engine-level retry uses the same predicate via core/retry.ts. void isRetryableConnError; @@ -458,6 +523,33 @@ export async function runExtract(engine: BrainEngine, args: string[]) { return runExtractExplain(engine, args); } + // v0.42.7 (#1696): `gbrain extract --stale` — incremental link+timeline sweep + // over pages whose links_extracted_at watermark is stale. Intercepts BEFORE + // the links|timeline|all subcommand validation so `gbrain extract --stale` + // works with no subcommand (and `gbrain extract all --stale` too). DB-source + // only — reads page content from the DB so it runs on checkout-less brains. + if (args.includes('--stale')) { + const sIdx = args.indexOf('--source'); + const src = (sIdx >= 0 && sIdx + 1 < args.length) ? args[sIdx + 1] : 'db'; + if (src === 'fs') { + console.error( + `extract --stale is DB-source only (reads page content from the database\n` + + `so it works on checkout-less brains). Drop '--source fs' or pass '--source db'.`, + ); + process.exit(1); + } + const sidIdx = args.indexOf('--source-id'); + const staleSourceId = (sidIdx >= 0 && sidIdx + 1 < args.length) ? args[sidIdx + 1] : undefined; + await extractStaleFromDB(engine, { + dryRun: args.includes('--dry-run'), + jsonMode: args.includes('--json'), + includeFrontmatter: args.includes('--include-frontmatter'), + sourceIdFilter: staleSourceId, + catchUp: args.includes('--catch-up'), + }); + return; + } + const dirIdx = args.indexOf('--dir'); const explicitDir = dirIdx >= 0 && dirIdx + 1 < args.length; // When --dir is not passed, resolve from the configured brain source @@ -540,6 +632,12 @@ Extraction (existing): gbrain extract --ner --source db gbrain extract --from-meetings +Incremental sweep (v0.42.7): + gbrain extract --stale [--source-id ] [--catch-up] [--dry-run] [--json] + Re-extract links + timeline ONLY for pages whose extraction is stale + (never extracted, edited since, or extractor bumped). DB-source; safe to + cron. --catch-up loops past the 30-min wall-clock budget until 0 remain. + Inspection (v0.42): gbrain extract --explain [--json] Print resolution chain for one pack-declared extractable kind. @@ -691,7 +789,9 @@ Status (v0.42): } } else { if (subcommand === 'links' || subcommand === 'all') { - const r = await extractLinksFromDB(engine, dryRun, jsonMode, typeFilter, since, { includeFrontmatter, sourceIdFilter }); + // C3 (D6): only stamp the combined links+timeline watermark when BOTH + // ran ('all'); a links-only run must not mark timeline fresh. + const r = await extractLinksFromDB(engine, dryRun, jsonMode, typeFilter, since, { includeFrontmatter, sourceIdFilter, stampWatermark: subcommand === 'all' }); result.links_created = r.created; result.pages_processed = r.pages; } @@ -1058,10 +1158,15 @@ async function extractLinksFromDB( jsonMode: boolean, typeFilter: PageType | undefined, since: string | undefined, - opts?: { includeFrontmatter?: boolean; sourceIdFilter?: string }, + opts?: { includeFrontmatter?: boolean; sourceIdFilter?: string; stampWatermark?: boolean }, ): Promise<{ created: number; pages: number; unresolved: UnresolvedFrontmatterRef[] }> { const includeFrontmatter = opts?.includeFrontmatter ?? false; const sourceIdFilter = opts?.sourceIdFilter; + // C3 (D6): the links_extracted_at watermark covers links AND timeline, so a + // links-ONLY run must NOT stamp it (that would hide timeline staleness for + // `gbrain extract links --source db`). Only stamp when the caller ran BOTH + // (subcommand 'all'). Caller passes stampWatermark accordingly. + const stampWatermark = opts?.stampWatermark ?? false; // Batch resolver: pg_trgm + exact only, NO search fallback. Dodges the // N-thousand API call trap on 46K-page brains. Resolver has a per-run // cache so duplicate names (same person appearing on many pages) resolve @@ -1105,6 +1210,10 @@ async function extractLinksFromDB( slugToSources.set(ref.slug, list); } let processed = 0, created = 0; + // v0.42.7 (#1696): pages whose links we extracted this run — stamped after + // the loop so a manual `gbrain extract links|all --source db` clears the + // links_extraction_lag doctor signal. Non-dry-run only. + const processedRefs: Array<{ slug: string; source_id: string }> = []; const progress = createProgress(cliOptsToProgressOptions(getCliOptions())); progress.start('extract.links_db', allRefs.length); @@ -1150,35 +1259,13 @@ async function extractLinksFromDB( unresolved.push(...extracted.unresolved); for (const c of extracted.candidates) { - // Validate BOTH endpoints exist. Incoming frontmatter edges have - // fromSlug !== the page being processed; we need that page to exist - // too or the JOIN drops the row anyway. - const fromSlug = c.fromSlug ?? slug; - if (!allSlugs.has(c.targetSlug)) continue; - if (!allSlugs.has(fromSlug)) continue; - - // v0.32.8 F10: cross-source link resolution. - // from_source_id = origin page's source_id (this loop's source_id, or - // the candidate's fromSlug source if it lives in a different source). - // to_source_id = priority: origin's source > 'default' > skip (don't - // silently push a wrong-source edge). - const fromSources = slugToSources.get(fromSlug) ?? []; - const fromSourceId = fromSources.includes(source_id) ? source_id - : (fromSources.includes('default') ? 'default' : fromSources[0]); - const targetSources = slugToSources.get(c.targetSlug) ?? []; - let toSourceId: string; - if (targetSources.includes(fromSourceId)) { - toSourceId = fromSourceId; - } else if (targetSources.includes('default')) { - toSourceId = 'default'; - } else { - // Target exists ONLY in non-origin/non-default sources. Skip — don't - // silently push a wrong-source edge. Tracking this as an unresolved - // ref would require expanding UnresolvedFrontmatterRef; for v0.32.8 - // a quiet skip is the conservative choice (matches existing - // "target missing" semantics where allSlugs.has() returns false). - continue; - } + // v0.32.8 F10 cross-source link resolution, extracted to the shared pure + // helper in v0.42.7 (#1696) so extract --stale reuses the exact same + // endpoint-validation + from/to source-id picking (null = skip: missing + // endpoint OR target only in a non-origin/non-default source). + const resolved = resolveCandidateSources(c, slug, source_id, allSlugs, slugToSources); + if (!resolved) continue; + const { fromSlug, fromSourceId, toSourceId } = resolved; if (dryRunSeen) { const key = `${fromSourceId}::${fromSlug}::${toSourceId}::${c.targetSlug}::${c.linkType}::${c.linkSource ?? 'markdown'}`; @@ -1214,9 +1301,21 @@ async function extractLinksFromDB( } } processed++; + if (!dryRun) processedRefs.push({ slug, source_id }); progress.tick(1); } await flush(); + // v0.42.7 (#1696): stamp the extraction watermark for every page we + // processed (incl. zero-link pages — they WERE extracted). Chunked so the + // unnest UPDATE stays bounded on big brains. Best-effort (stampExtracted + // swallows): a stamp miss just leaves the page for extract --stale. + // C3 (D6): ONLY when both links + timeline ran (stampWatermark) — a + // links-only run leaves the combined watermark untouched. + if (!dryRun && stampWatermark) { + for (let i = 0; i < processedRefs.length; i += BATCH_SIZE) { + await stampExtracted(engine, processedRefs.slice(i, i + BATCH_SIZE)); + } + } progress.finish(); if (!jsonMode) { @@ -1330,6 +1429,149 @@ async function extractTimelineFromDB( return { created, pages: processed }; } +/** + * v0.42.7 (#1696) — `gbrain extract --stale`: incremental link + timeline + * extraction over pages whose `links_extracted_at` watermark is stale (NULL, + * older than LINK_EXTRACTOR_VERSION_TS, or older than the page's updated_at). + * DB-source (works on checkout-less Postgres/Supabase brains). Mirrors + * embedAllStale's count → keyset-list → flush → stamp shape. + * + * Crash-safety + CDX-4: per keyset batch we extract ALL links+timeline, flush + * them (NON-swallowing — a flush throw propagates and aborts the sweep), THEN + * stamp the batch's pages. A page is never stamped fresh with lost edges; a + * crash mid-sweep leaves the unflushed/unstamped pages stale and they + * re-extract next run (addLinksBatch ON CONFLICT DO NOTHING + timeline dedup + * make re-extraction idempotent). EVERY processed page is stamped, including + * zero-link pages — they WERE processed. + */ +async function extractStaleFromDB( + engine: BrainEngine, + opts: { + dryRun: boolean; + jsonMode: boolean; + includeFrontmatter: boolean; + sourceIdFilter?: string; + catchUp: boolean; + }, +): Promise<{ linksCreated: number; timelineCreated: number; pagesProcessed: number; staleRemaining: number }> { + const { dryRun, jsonMode, includeFrontmatter, sourceIdFilter, catchUp } = opts; + const versionTs = LINK_EXTRACTOR_VERSION_TS; + + // Pre-flight count — cheap indexed COUNT. dry-run reports and returns. + const totalStale = await engine.countStalePagesForExtraction({ sourceId: sourceIdFilter, versionTs }); + if (dryRun) { + if (jsonMode) { + process.stdout.write(JSON.stringify({ action: 'extract_stale_dry_run', stale_pages: totalStale }) + '\n'); + } else { + console.log(`(dry run) ${totalStale} page(s) need link/timeline extraction. Run without --dry-run to extract.`); + } + return { linksCreated: 0, timelineCreated: 0, pagesProcessed: 0, staleRemaining: totalStale }; + } + if (totalStale === 0) { + if (!jsonMode) console.log('No stale pages — extraction is up to date.'); + return { linksCreated: 0, timelineCreated: 0, pagesProcessed: 0, staleRemaining: 0 }; + } + + // Resolver + cross-source resolution map built ONCE before the loop (the + // extractLinksFromDB:1069 precedent — avoids O(pages) rebuild per batch). + // Batch mode = pg_trgm + exact only, NO per-name search fallback. The + // resolution map sees ALL sources so qualified cross-source wikilinks resolve + // even when --source-id scopes the stale SCAN. + const resolver = makeResolver(engine, { mode: 'batch' }); + const nullResolver = { resolve: async () => null as string | null }; + const activeResolver = includeFrontmatter ? resolver : nullResolver; + const allRefs = await engine.listAllPageRefs(); + const allSlugs = new Set(); + const slugToSources = new Map(); + for (const ref of allRefs) { + allSlugs.add(ref.slug); + const list = slugToSources.get(ref.slug) ?? []; + list.push(ref.source_id); + slugToSources.set(ref.slug, list); + } + + const progress = createProgress(cliOptsToProgressOptions(getCliOptions())); + progress.start('extract.stale', totalStale); + + const startMs = Date.now(); + let afterPageId = 0; + let linksCreated = 0, timelineCreated = 0, pagesProcessed = 0; + let budgetHit = false; + + for (;;) { + const rows = await engine.listStalePagesForExtraction({ + batchSize: STALE_BATCH_SIZE, afterPageId, sourceId: sourceIdFilter, versionTs, + }); + if (rows.length === 0) break; + + const linkRows: LinkBatchInput[] = []; + const timelineRows: TimelineBatchInput[] = []; + const processedRefs: Array<{ slug: string; source_id: string; extractedAt: string }> = []; + + for (const page of rows) { + const fullContent = page.compiled_truth + '\n' + page.timeline; + const extracted = await extractPageLinks( + page.slug, fullContent, page.frontmatter, page.type, activeResolver, + ); + for (const c of extracted.candidates) { + const r = resolveCandidateSources(c, page.slug, page.source_id, allSlugs, slugToSources); + if (!r) continue; + linkRows.push({ + from_slug: r.fromSlug, to_slug: c.targetSlug, link_type: c.linkType, + context: c.context, link_source: c.linkSource, origin_slug: c.originSlug, + origin_field: c.originField, from_source_id: r.fromSourceId, + to_source_id: r.toSourceId, origin_source_id: page.source_id, + }); + } + for (const entry of parseTimelineEntries(fullContent)) { + timelineRows.push({ slug: page.slug, date: entry.date, summary: entry.summary, detail: entry.detail || '', source_id: page.source_id }); + } + // EVERY processed page is stamped (incl. zero-link pages). D4 race fix: + // stamp with the row's READ updated_at, NOT now() — a concurrent edit + // landing between this SELECT and the stamp advances updated_at past the + // stamped value, so the page stays stale and re-extracts next run instead + // of being marked fresh-with-stale-content. + processedRefs.push({ slug: page.slug, source_id: page.source_id, extractedAt: page.updated_at.toISOString() }); + } + + // Flush NON-swallowing (CDX-4): a throw here propagates out of the sweep so + // the batch's pages stay unstamped and re-extract next run. addLinksBatch is + // ON CONFLICT DO NOTHING + timeline dedups, so partial-chunk writes are + // idempotent on re-extraction. + for (let i = 0; i < linkRows.length; i += BATCH_SIZE) { + linksCreated += await engine.addLinksBatch(linkRows.slice(i, i + BATCH_SIZE), { auditSite: 'extract.stale' }); // gbrain-allow-direct-insert: gbrain extract --stale — canonical link reconciliation from markdown body + } + for (let i = 0; i < timelineRows.length; i += BATCH_SIZE) { + timelineCreated += await engine.addTimelineEntriesBatch(timelineRows.slice(i, i + BATCH_SIZE), { auditSite: 'extract.stale' }); + } + // Stamp LAST, directly (not the swallowing stampExtracted) so a stamp + // failure surfaces instead of looping forever. + await engine.markPagesExtractedBatch(processedRefs, new Date().toISOString()); + + pagesProcessed += rows.length; + progress.tick(rows.length); + afterPageId = rows[rows.length - 1]!.id; + + if (!catchUp && Date.now() - startMs > STALE_TIME_BUDGET_MS) { budgetHit = true; break; } + } + + progress.finish(); + const staleRemaining = await engine.countStalePagesForExtraction({ sourceId: sourceIdFilter, versionTs }); + + if (!jsonMode) { + console.log(`Extract --stale: ${linksCreated} link(s) + ${timelineCreated} timeline entr(ies) from ${pagesProcessed} page(s).`); + if (budgetHit && staleRemaining > 0) { + console.log(`Time budget reached — ${staleRemaining} page(s) still stale. Re-run 'gbrain extract --stale' (or pass --catch-up) to continue.`); + } + } else { + process.stdout.write(JSON.stringify({ + action: 'extract_stale_done', links_created: linksCreated, timeline_created: timelineCreated, + pages_processed: pagesProcessed, stale_remaining: staleRemaining, budget_hit: budgetHit, + }) + '\n'); + } + return { linksCreated, timelineCreated, pagesProcessed, staleRemaining }; +} + /** * v0.41.18.0 Part B (migration #1 of #1409) — auto-link body-text entity * mentions to known entity pages. diff --git a/src/commands/sync.ts b/src/commands/sync.ts index b5d887a90..1c820c01f 100644 --- a/src/commands/sync.ts +++ b/src/commands/sync.ts @@ -61,6 +61,19 @@ import { getDefaultSourcePath } from '../core/source-resolver.ts'; import { newestCommitMs, lagFromContentMs } from '../core/source-health.ts'; import { sortNewestFirst } from '../core/sort-newest-first.ts'; +/** + * v0.42.7 (#1696, D5): which terminal sync statuses warrant the end-of-sync + * extraction-lag nudge. Fires on every non-error completion — crucially + * `first_sync` (a fresh / --full import is the BIGGEST un-extracted backlog) and + * `up_to_date` (a no-op sync over a brain with a pre-existing backlog still + * warrants the nudge). Excludes `dry_run` (preview) / `blocked_by_failures` / + * `partial` (inconsistent state). Pure so D5's contract is unit-testable + * without driving the CLI. + */ +export function shouldNudgeAfterSync(status: SyncResult['status']): boolean { + return status === 'synced' || status === 'first_sync' || status === 'up_to_date'; +} + export interface SyncResult { status: 'up_to_date' | 'synced' | 'first_sync' | 'dry_run' | 'blocked_by_failures' | 'partial'; fromCommit: string | null; @@ -1827,12 +1840,22 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 0) { try { - const { extractLinksForSlugs, extractTimelineForSlugs } = await import('./extract.ts'); + const { extractLinksForSlugs, extractTimelineForSlugs, stampExtracted } = await import('./extract.ts'); const linksCreated = await extractLinksForSlugs(engine, repoPath, pagesAffected, extractOpts); const timelineCreated = await extractTimelineForSlugs(engine, repoPath, pagesAffected, extractOpts); if (linksCreated > 0 || timelineCreated > 0) { slog(` Extracted: ${linksCreated} links, ${timelineCreated} timeline entries`); } + // v0.42.7 (#1696, CDX-6): stamp the links_extracted_at watermark for the + // pages we just extracted, AFTER the import set their updated_at, so + // links_extracted_at >= updated_at (page is now fresh, not flagged stale). + // Source-correct via opts.sourceId. Stamp at the CALL SITE (not inside + // extractLinksForSlugs) so we use the per-source sourceId the sync owns. + // Best-effort: a stamp miss just means extract --stale re-sweeps later. + await stampExtracted( + engine, + pagesAffected.map((slug) => ({ slug, source_id: opts.sourceId ?? 'default' })), + ); } catch { /* extraction is best-effort */ } } @@ -2086,6 +2109,9 @@ Options: --no-embed Skip the embed step. Use this when the embed provider is misconfigured or you want to defer embedding (run 'gbrain embed --stale' later). + --no-extract Skip the link/timeline extraction step. Pages will + show as stale in 'gbrain doctor'; run + 'gbrain extract --stale' later to catch up. --workers N Run the import phase with N parallel workers (alias: --concurrency). Default: 4 when the diff is >100 files, else serial. @@ -2140,6 +2166,7 @@ See also: const full = args.includes('--full'); const noPull = args.includes('--no-pull'); const noEmbed = args.includes('--no-embed'); + const noExtract = args.includes('--no-extract'); // v0.42.7 #1696 const skipFailed = args.includes('--skip-failed'); const retryFailed = args.includes('--retry-failed'); const noSchemaPack = args.includes('--no-schema-pack'); // v0.41.37.0 #1569 @@ -2570,6 +2597,7 @@ See also: repoPath: src.local_path!, dryRun, full, noPull, noEmbed: effectiveNoEmbed, + noExtract, skipFailed, retryFailed, noSchemaPack, sourceId: src.id, strategy: cfg.strategy, @@ -2758,6 +2786,10 @@ See also: })); } + // v0.42.7 (#1696): brain-wide extraction-lag nudge after the --all wave. + // Best-effort, stderr-only; skipped on dry-run. + if (!dryRun) await maybeExtractionNudge(engine); + if (errCount > 0) process.exit(1); return; } @@ -2771,7 +2803,7 @@ See also: : undefined; singleSourceTimer?.unref?.(); const opts: SyncOpts = { - repoPath, dryRun, full, noPull, noEmbed, skipFailed, retryFailed, noSchemaPack, sourceId, + repoPath, dryRun, full, noPull, noEmbed, noExtract, skipFailed, retryFailed, noSchemaPack, sourceId, strategy: strategyArg, concurrency, signal: singleSourceController?.signal, }; @@ -2800,6 +2832,11 @@ See also: if (singleSourceTimer !== undefined) clearTimeout(singleSourceTimer); } printSyncResult(result); + // v0.42.7 (#1696, D5): extraction-lag nudge after a completed single-source + // sync. Fire on every non-error completion (synced | first_sync | up_to_date) + // — NOT just 'synced'; a fresh/--full import (`first_sync`) is the biggest + // un-extracted backlog. Scoped to this source; best-effort, stderr-only. + if (shouldNudgeAfterSync(result.status)) await maybeExtractionNudge(engine, sourceId); // Issue #2 + eng-review pass-2 finding #1 + Codex P1: manage .gitignore ONLY // on successful sync. Skip on dry-run (don't mutate disk in preview mode) // and blocked_by_failures (sync state is inconsistent — defer .gitignore @@ -2933,6 +2970,8 @@ export async function syncOneSource( concurrency: number | undefined; /** v0.41.37.0 #1569: propagate --no-schema-pack into every per-source sync. */ noSchemaPack?: boolean; + /** v0.42.7 #1696: propagate --no-extract into every per-source sync. */ + noExtract?: boolean; }, ): Promise<{ result: SyncResult; log: string }> { const cfg = (src.config || {}) as { strategy?: 'markdown' | 'code' | 'auto' }; @@ -2943,6 +2982,7 @@ export async function syncOneSource( full: shared.full, noPull: shared.noPull, noEmbed: shared.noEmbed, + noExtract: shared.noExtract, skipFailed: shared.skipFailed, retryFailed: shared.retryFailed, noSchemaPack: shared.noSchemaPack, @@ -3410,6 +3450,41 @@ export function manageGitignore( } } +/** + * v0.42.7 (#1696): one-line end-of-sync nudge when the brain (or a source) + * carries a meaningful link/timeline extraction backlog. Reuses the same warn + * threshold (GBRAIN_EXTRACTION_LAG_WARN_PCT, default 20%) the doctor check uses + * so the nudge fires iff doctor would warn — one source of truth. Always + * stderr (never stdout — keeps `--json` clean), suppressible via + * GBRAIN_SYNC_NO_EXTRACT_NUDGE, best-effort (never throws). Source-prefix-aware + * via serr when called inside a withSourcePrefix scope. + */ +async function maybeExtractionNudge(engine: BrainEngine, sourceId?: string): Promise { + if (process.env.GBRAIN_SYNC_NO_EXTRACT_NUDGE) return; + try { + const { LINK_EXTRACTOR_VERSION_TS } = await import('../core/link-extraction.ts'); + // D3/C4: resolve the warn threshold + vacuous-skip floor through the SAME + // helpers the doctor check uses (dynamic import keeps doctor.ts off sync's + // eager-load path) so "the nudge fires iff doctor would warn" can't drift. + const { _resolveEnvNumber, EXTRACTION_LAG_WARN_PCT_DEFAULT, EXTRACTION_LAG_MIN_PAGES } = await import('./doctor.ts'); + const totalRows = await engine.executeRaw<{ count: number }>( + sourceId + ? `SELECT count(*)::int AS count FROM pages WHERE deleted_at IS NULL AND source_id = $1` + : `SELECT count(*)::int AS count FROM pages WHERE deleted_at IS NULL`, + sourceId ? [sourceId] : [], + ); + const total = Number(totalRows[0]?.count ?? 0); + // Match doctor's predicate EXACTLY (C4): skip tiny brains only when NOT + // source-scoped (a small explicit source IS assessed, like orphan_ratio). + if (total < EXTRACTION_LAG_MIN_PAGES && !sourceId) return; + const stale = await engine.countStalePagesForExtraction({ sourceId, versionTs: LINK_EXTRACTOR_VERSION_TS }); + const warnPct = _resolveEnvNumber('GBRAIN_EXTRACTION_LAG_WARN_PCT', EXTRACTION_LAG_WARN_PCT_DEFAULT, { unit: '%' }); + if ((stale / total) * 100 > warnPct) { + serr(`[sync] ${stale} page(s) have un-extracted edges — run 'gbrain extract --stale'`); + } + } catch { /* nudge is best-effort — never block sync on it */ } +} + /** * Render a SyncResult to a Writable sink. * diff --git a/src/core/doctor-categories.ts b/src/core/doctor-categories.ts index 680794e21..b2f7ee236 100644 --- a/src/core/doctor-categories.ts +++ b/src/core/doctor-categories.ts @@ -85,6 +85,7 @@ export const BRAIN_CHECK_NAMES: ReadonlySet = new Set([ 'image_assets', 'integrity', 'jsonb_integrity', + 'links_extraction_lag', 'markdown_body_completeness', 'nightly_quality_probe_health', 'ocr_health', diff --git a/src/core/engine.ts b/src/core/engine.ts index d12b839c6..9db813488 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -1,6 +1,6 @@ import type { Page, PageInput, PageFilters, GetPageOpts, - Chunk, ChunkInput, StaleChunkRow, + Chunk, ChunkInput, StaleChunkRow, StalePageRow, SearchResult, SearchOpts, Link, GraphNode, GraphPath, TimelineEntry, TimelineInput, TimelineOpts, @@ -1028,6 +1028,48 @@ export interface BrainEngine { */ deleteChunks(slug: string, opts?: { sourceId?: string }): Promise; + // ============================================================ + // v0.42.7 (#1696): link/timeline extraction freshness watermark. + // A page is stale for extraction when its links_extracted_at is NULL, + // older than the extractor version stamp, or older than its updated_at + // (edited-since-extract — the MCP put_page / sync --no-extract path). + // Powers `gbrain extract --stale` + the `links_extraction_lag` doctor check. + // ============================================================ + /** + * Count pages needing (re)extraction. `versionTs` is the ISO-8601 + * `LINK_EXTRACTOR_VERSION_TS` string (bound `::timestamptz`); when omitted, + * only the NULL + edited-since arms apply. Soft-deleted pages excluded. + */ + countStalePagesForExtraction(opts?: { sourceId?: string; versionTs?: string }): Promise; + /** + * List a keyset page (ordered by `id`, `id > afterPageId`) of stale pages + * WITH their content so the caller extracts without an N+1 `getPage`. Same + * stale predicate as countStalePagesForExtraction. Caller passes a SMALL + * batchSize (page bodies are unbounded — 25MB transcript pages exist) and + * applies its own per-batch byte budget. + */ + listStalePagesForExtraction(opts: { + batchSize: number; + afterPageId?: number; + sourceId?: string; + versionTs?: string; + }): Promise; + /** + * Stamp `links_extracted_at` for a batch of pages keyed on the unique + * `(slug, source_id)` pair (unnest idiom, mirrors addLinksBatch). + * Short-circuits on empty input. Called AFTER the link/timeline flush so a + * crash mid-batch leaves pages unstamped and they re-extract next run. + * + * Each ref may carry its own `extractedAt`; refs that omit it use the + * `defaultExtractedAt` arg. `gbrain extract --stale` passes the row's READ + * `updated_at` per ref (v0.42.7 D4 race fix) so a concurrent edit landing + * between the SELECT and this stamp advances `updated_at` past the stamped + * value → the page stays stale → re-extracted next run, never marked + * fresh-with-the-old-content. Sync / DB-extract sites omit per-ref values and + * pass `now()` (the page was just imported, so `now() >= updated_at`). + */ + markPagesExtractedBatch(refs: Array<{ slug: string; source_id: string; extractedAt?: string }>, defaultExtractedAt: string): Promise; + // Links /** * Single-row link insert. linkSource defaults to 'markdown' for back-compat diff --git a/src/core/link-extraction.ts b/src/core/link-extraction.ts index 12e71746a..684919191 100644 --- a/src/core/link-extraction.ts +++ b/src/core/link-extraction.ts @@ -14,6 +14,21 @@ import type { BrainEngine } from './engine.ts'; import type { PageType } from './types.ts'; +/** + * v0.42.7 — link-extraction version stamp. Bump this ISO timestamp whenever the + * shape of `extractPageLinks` / `inferLinkType` / `parseTimelineEntries` changes + * meaningfully, so the extraction freshness watermark (`pages.links_extracted_at`) + * treats every previously-stamped page as stale and re-extracts it on the next + * `gbrain extract --stale` sweep. Same role CHUNKER_VERSION plays for chunking. + * + * Consumed by `countStalePagesForExtraction` / `listStalePagesForExtraction` + * (both engines) and the `links_extraction_lag` doctor check: a page is stale + * when `links_extracted_at IS NULL OR links_extracted_at < LINK_EXTRACTOR_VERSION_TS + * OR updated_at > links_extracted_at`. It is an ISO-8601 string (NOT a number) — + * the column is TIMESTAMPTZ and the predicate binds it as `::timestamptz`. + */ +export const LINK_EXTRACTOR_VERSION_TS = '2026-05-31T00:00:00Z'; + // ─── Entity references ────────────────────────────────────────── export interface EntityRef { diff --git a/src/core/migrate.ts b/src/core/migrate.ts index 511397053..18ac0d979 100644 --- a/src/core/migrate.ts +++ b/src/core/migrate.ts @@ -5023,6 +5023,71 @@ export const MIGRATIONS: Migration[] = [ ALTER TABLE search_telemetry ADD COLUMN IF NOT EXISTS rank1_high INTEGER NOT NULL DEFAULT 0; `, }, + { + version: 112, + name: 'pages_links_extracted_at', + // v0.42.7 (#1696) — link-extraction freshness watermark. + // + // Closes the "imported ≠ curated" root cause: extraction is the silent third + // leg of `sync → extract → embed`, and a brain with autopilot off (the common + // CLI / external-cron case) accumulated 0% typed-edge coverage with nothing + // surfacing it. This column lets `gbrain extract --stale` sweep the historical + // backlog incrementally and the `links_extraction_lag` doctor check warn when + // extraction has fallen behind. + // + // A page is stale for extraction when: + // links_extracted_at IS NULL (never extracted) + // OR links_extracted_at < LINK_EXTRACTOR_VERSION_TS (extractor logic bumped) + // OR updated_at > links_extracted_at (edited since last extract — + // MCP put_page / sync --no-extract) + // + // GRANDFATHER: no backfill. After this migration every existing page has NULL + // links_extracted_at, so the first `gbrain doctor` correctly surfaces the real + // backlog (the whole point). The doctor check is warn-only by default; it only + // hard-fails if GBRAIN_EXTRACTION_LAG_FAIL_PCT is set — so the upgrade never + // breaks a CI/cron pipeline that gates on `gbrain doctor` exit code. + // + // Composite index (source_id, links_extracted_at) backs the source-scoped + // staleness scans. Postgres path uses CREATE INDEX CONCURRENTLY (+ invalid- + // remnant pre-drop, mirroring v97); PGLite uses plain CREATE INDEX. ADD COLUMN + // with no DEFAULT (NULL) is metadata-only on Postgres 11+ / PGLite 17.5. + // + // Mirror lives in src/schema.sql + pglite-schema.ts (fresh-install column + + // index) and the applyForwardReferenceBootstrap probe set in both engines. + sql: '', + transaction: false, + handler: async (engine) => { + await engine.runMigration( + 112, + `ALTER TABLE pages ADD COLUMN IF NOT EXISTS links_extracted_at TIMESTAMPTZ;` + ); + if (engine.kind === 'postgres') { + await engine.runMigration( + 112, + `DO $$ BEGIN + IF EXISTS ( + SELECT 1 FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + WHERE c.relname = 'pages_links_extracted_at_idx' AND NOT i.indisvalid + ) THEN + EXECUTE 'DROP INDEX CONCURRENTLY IF EXISTS pages_links_extracted_at_idx'; + END IF; + END $$;` + ); + await engine.runMigration( + 112, + `CREATE INDEX CONCURRENTLY IF NOT EXISTS pages_links_extracted_at_idx + ON pages (source_id, links_extracted_at);` + ); + } else { + await engine.runMigration( + 112, + `CREATE INDEX IF NOT EXISTS pages_links_extracted_at_idx + ON pages (source_id, links_extracted_at);` + ); + } + }, + }, ]; export const LATEST_VERSION = MIGRATIONS.length > 0 diff --git a/src/core/pglite-engine.ts b/src/core/pglite-engine.ts index 09e6348a1..b46d38867 100644 --- a/src/core/pglite-engine.ts +++ b/src/core/pglite-engine.ts @@ -26,7 +26,7 @@ import { DELETE_BATCH_SIZE } from './engine-constants.ts'; import { acquireLock, releaseLock, type LockHandle } from './pglite-lock.ts'; import type { Page, PageInput, PageFilters, PageType, - Chunk, ChunkInput, StaleChunkRow, + Chunk, ChunkInput, StaleChunkRow, StalePageRow, SearchResult, SearchOpts, Link, GraphNode, GraphPath, TimelineEntry, TimelineInput, TimelineOpts, @@ -42,7 +42,7 @@ import type { DomainBankSampleOpts, CorpusSampleOpts, DomainBankRow, EnrichCandidatesOpts, EnrichCandidate, } from './types.ts'; -import { validateSlug, contentHash, rowToPage, rowToChunk, rowToSearchResult, takeRowToTake, isUndefinedTableError, warnOncePerProcess } from './utils.ts'; +import { validateSlug, contentHash, rowToPage, rowToStalePage, rowToChunk, rowToSearchResult, takeRowToTake, isUndefinedTableError, warnOncePerProcess } from './utils.ts'; import { deriveResolutionTuple, finalizeScorecard } from './takes-resolution.ts'; import { normalizeWeightForStorage } from './takes-fence.ts'; import { GBrainError, PAGE_SORT_SQL, ENRICH_ORDER_SQL } from './types.ts'; @@ -425,7 +425,9 @@ export class PGLiteEngine implements BrainEngine { EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema='public' AND table_name='pages' AND column_name='generation') AS pages_generation_exists, EXISTS (SELECT 1 FROM information_schema.columns - WHERE table_schema='public' AND table_name='pages' AND column_name='embedding_signature') AS pages_embedding_signature_exists + WHERE table_schema='public' AND table_name='pages' AND column_name='embedding_signature') AS pages_embedding_signature_exists, + EXISTS (SELECT 1 FROM information_schema.columns + WHERE table_schema='public' AND table_name='pages' AND column_name='links_extracted_at') AS pages_links_extracted_at_exists `); const probe = rows[0] as { pages_exists: boolean; @@ -467,6 +469,7 @@ export class PGLiteEngine implements BrainEngine { sources_trust_fm_exists: boolean; pages_generation_exists: boolean; pages_embedding_signature_exists: boolean; + pages_links_extracted_at_exists: boolean; }; const needsPagesBootstrap = probe.pages_exists && !probe.source_id_exists; @@ -538,6 +541,11 @@ export class PGLiteEngine implements BrainEngine { // No SCHEMA_SQL index references it today; bootstrap is defense-in-depth // so future schema work doesn't wedge pre-v108 brains. const needsPagesEmbeddingSignature = probe.pages_exists && !probe.pages_embedding_signature_exists; + // v0.42.7 (v112): pages.links_extracted_at link-extraction freshness + // watermark. pages_links_extracted_at_idx in PGLITE_SCHEMA_SQL references + // it; pre-v112 brains crash without the column, so bootstrap adds it before + // the CREATE INDEX runs. v112 runs later via runMigrations and is idempotent. + const needsPagesLinksExtractedAt = probe.pages_exists && !probe.pages_links_extracted_at_exists; // Fresh installs (no tables yet) and modern brains both no-op. if (!needsPagesBootstrap && !needsLinksBootstrap && !needsChunksBootstrap @@ -548,7 +556,8 @@ export class PGLiteEngine implements BrainEngine { && !needsSourcesArchive && !needsPagesLastRetrievedAt && !needsPagesProvenance && !needsContextualRetrievalColumns && !needsPagesGeneration - && !needsPagesEmbeddingSignature) return; + && !needsPagesEmbeddingSignature + && !needsPagesLinksExtractedAt) return; process.stderr.write(' Pre-v0.21 brain detected, applying forward-reference bootstrap\n'); @@ -785,6 +794,16 @@ export class PGLiteEngine implements BrainEngine { ALTER TABLE pages ADD COLUMN IF NOT EXISTS embedding_signature TEXT; `); } + + if (needsPagesLinksExtractedAt) { + // v112 (pages_links_extracted_at): link-extraction freshness watermark. + // PGLITE_SCHEMA_SQL CREATE INDEX pages_links_extracted_at_idx references + // it, so bootstrap adds the column before the blob's CREATE INDEX runs. + // v112 runs later via runMigrations and is idempotent. + await this.db.exec(` + ALTER TABLE pages ADD COLUMN IF NOT EXISTS links_extracted_at TIMESTAMPTZ; + `); + } } async withReservedConnection(fn: (conn: ReservedConnection) => Promise): Promise { @@ -2296,6 +2315,74 @@ export class PGLiteEngine implements BrainEngine { ); } + // ── v0.42.7 (#1696): link/timeline extraction freshness watermark ── + + /** Shared stale-for-extraction predicate (mirrors PostgresEngine). */ + private buildStalePagesWhere(opts?: { sourceId?: string; versionTs?: string }): { where: string; params: unknown[] } { + const conds: string[] = ['deleted_at IS NULL']; + const params: unknown[] = []; + if (opts?.versionTs) { + params.push(opts.versionTs); + conds.push(`(links_extracted_at IS NULL OR links_extracted_at < $${params.length}::timestamptz OR updated_at > links_extracted_at)`); + } else { + conds.push('(links_extracted_at IS NULL OR updated_at > links_extracted_at)'); + } + if (opts?.sourceId) { + params.push(opts.sourceId); + conds.push(`source_id = $${params.length}`); + } + return { where: conds.join(' AND '), params }; + } + + async countStalePagesForExtraction(opts?: { sourceId?: string; versionTs?: string }): Promise { + const { where, params } = this.buildStalePagesWhere(opts); + const { rows } = await this.db.query<{ count: number }>( + `SELECT count(*)::int AS count FROM pages WHERE ${where}`, + params, + ); + return rows[0]?.count ?? 0; + } + + async listStalePagesForExtraction(opts: { + batchSize: number; + afterPageId?: number; + sourceId?: string; + versionTs?: string; + }): Promise { + const { where, params } = this.buildStalePagesWhere(opts); + let afterClause = ''; + if (opts.afterPageId != null) { + params.push(opts.afterPageId); + afterClause = ` AND id > $${params.length}`; + } + params.push(opts.batchSize); + const limitIdx = params.length; + const { rows } = await this.db.query( + `SELECT id, slug, source_id, type, title, compiled_truth, timeline, frontmatter, updated_at + FROM pages + WHERE ${where}${afterClause} + ORDER BY id + LIMIT $${limitIdx}`, + params, + ); + return (rows as Record[]).map(rowToStalePage); + } + + async markPagesExtractedBatch(refs: Array<{ slug: string; source_id: string; extractedAt?: string }>, defaultExtractedAt: string): Promise { + if (refs.length === 0) return; + const slugs = refs.map(r => r.slug); + const srcs = refs.map(r => r.source_id); + // Per-ref timestamp (D4 race fix): extract --stale passes each row's read + // updated_at; sites that omit it fall back to defaultExtractedAt. + const tss = refs.map(r => r.extractedAt ?? defaultExtractedAt); + await this.db.query( + `UPDATE pages p SET links_extracted_at = v.ts::timestamptz + FROM unnest($1::text[], $2::text[], $3::text[]) AS v(slug, source_id, ts) + WHERE p.slug = v.slug AND p.source_id = v.source_id`, + [slugs, srcs, tss], + ); + } + // Links async addLink( from: string, diff --git a/src/core/pglite-schema.ts b/src/core/pglite-schema.ts index 149f8edd0..ae8e4ef98 100644 --- a/src/core/pglite-schema.ts +++ b/src/core/pglite-schema.ts @@ -99,6 +99,10 @@ CREATE TABLE IF NOT EXISTS pages ( -- v0.37.0 (migration v79): real stale-page signal for gbrain lsd -- (mirrors src/schema.sql). NULL = never retrieved. last_retrieved_at TIMESTAMPTZ, + -- v0.42.7 (migration v112): link-extraction freshness watermark + -- (mirrors src/schema.sql). NULL = never extracted. Powers + -- gbrain extract --stale + the links_extraction_lag doctor check. + links_extracted_at TIMESTAMPTZ, -- v0.40.3.0 contextual retrieval (renumbered from v81 to v90 on master -- merge; mirrors src/schema.sql). -- contextual_retrieval_mode is the tier the page was last embedded under; @@ -187,6 +191,12 @@ CREATE INDEX IF NOT EXISTS pages_coalesce_date_idx -- query (mirrors src/schema.sql). Postgres handles NULL in B-tree indexes. CREATE INDEX IF NOT EXISTS pages_last_retrieved_at_idx ON pages (last_retrieved_at); +-- v0.42.7 (migration v112): composite B-tree backing extract --stale and the +-- links_extraction_lag doctor check (mirrors src/schema.sql). source_id leads so +-- source-scoped staleness scans are indexed; NOT partial-NULL (predicate has a +-- NULL arm AND a version-timestamp arm). +CREATE INDEX IF NOT EXISTS pages_links_extracted_at_idx + ON pages (source_id, links_extracted_at); -- ============================================================ -- content_chunks: chunked content with embeddings diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 5bdc97193..4c8dd7e35 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -34,7 +34,7 @@ import { } from './search/embedding-column.ts'; import type { Page, PageInput, PageFilters, PageType, - Chunk, ChunkInput, StaleChunkRow, + Chunk, ChunkInput, StaleChunkRow, StalePageRow, SearchResult, SearchOpts, Link, GraphNode, GraphPath, TimelineEntry, TimelineInput, TimelineOpts, @@ -54,7 +54,7 @@ import { computeAnomaliesFromBuckets } from './cycle/anomaly.ts'; import * as db from './db.ts'; import { ConnectionManager } from './connection-manager.ts'; import { logConnectionEvent } from './connection-audit.ts'; -import { validateSlug, contentHash, rowToPage, rowToChunk, rowToSearchResult, parseEmbedding, tryParseEmbedding, takeRowToTake, isUndefinedTableError, warnOncePerProcess } from './utils.ts'; +import { validateSlug, contentHash, rowToPage, rowToStalePage, rowToChunk, rowToSearchResult, parseEmbedding, tryParseEmbedding, takeRowToTake, isUndefinedTableError, warnOncePerProcess } from './utils.ts'; import { resolveBoostMap, resolveHardExcludes } from './search/source-boost.ts'; import { buildSourceFactorCase, buildHardExcludeClause, buildVisibilityClause, buildRecencyComponentSql, buildBestPerPagePoolCte } from './search/sql-ranking.ts'; import { DEFAULT_EMBEDDING_MODEL, DEFAULT_EMBEDDING_DIMENSIONS } from './ai/defaults.ts'; @@ -474,7 +474,9 @@ export class PostgresEngine implements BrainEngine { EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'pages' AND column_name = 'generation') AS pages_generation_exists, EXISTS (SELECT 1 FROM information_schema.columns - WHERE table_schema = current_schema() AND table_name = 'pages' AND column_name = 'embedding_signature') AS pages_embedding_signature_exists + WHERE table_schema = current_schema() AND table_name = 'pages' AND column_name = 'embedding_signature') AS pages_embedding_signature_exists, + EXISTS (SELECT 1 FROM information_schema.columns + WHERE table_schema = current_schema() AND table_name = 'pages' AND column_name = 'links_extracted_at') AS pages_links_extracted_at_exists `; const probe = probeRows[0]!; @@ -550,6 +552,7 @@ export class PostgresEngine implements BrainEngine { sources_trust_fm_exists?: boolean; pages_generation_exists?: boolean; pages_embedding_signature_exists?: boolean; + pages_links_extracted_at_exists?: boolean; }; const needsContextualRetrievalColumns = (probe.pages_exists && (!probeCr.pages_cr_mode_exists || !probeCr.pages_corpus_generation_exists)) @@ -563,6 +566,12 @@ export class PostgresEngine implements BrainEngine { // v0.41.31 (v108): pages.embedding_signature for real stale semantics. // No SCHEMA_SQL index references it; bootstrap is defense-in-depth. const needsPagesEmbeddingSignature = probe.pages_exists && !probeCr.pages_embedding_signature_exists; + // v0.42.7 (v112): pages.links_extracted_at link-extraction freshness + // watermark. pages_links_extracted_at_idx in SCHEMA_SQL references it; + // pre-v112 brains crash without the column, so bootstrap adds it before + // SCHEMA_SQL replay creates the index. v112 runs later via runMigrations + // and is idempotent. + const needsPagesLinksExtractedAt = probe.pages_exists && !probeCr.pages_links_extracted_at_exists; if (!needsPagesBootstrap && !needsLinksBootstrap && !needsChunksBootstrap && !needsPagesDeletedAt && !needsMcpLogBootstrap && !needsSubagentProviderId @@ -572,7 +581,8 @@ export class PostgresEngine implements BrainEngine { && !needsPagesLastRetrievedAt && !needsPagesProvenance && !needsContextualRetrievalColumns && !needsPagesGeneration - && !needsPagesEmbeddingSignature) return; + && !needsPagesEmbeddingSignature + && !needsPagesLinksExtractedAt) return; process.stderr.write(' Pre-v0.21 brain detected, applying forward-reference bootstrap\n'); @@ -807,6 +817,18 @@ export class PostgresEngine implements BrainEngine { ALTER TABLE pages ADD COLUMN IF NOT EXISTS embedding_signature TEXT; `); } + + if (needsPagesLinksExtractedAt) { + // v112 (pages_links_extracted_at): link-extraction freshness watermark. + // pages_links_extracted_at_idx in SCHEMA_SQL references it, so bootstrap + // adds the column before the blob's CREATE INDEX runs. The index itself + // lands via the blob (CREATE INDEX IF NOT EXISTS) and v112 (CONCURRENTLY); + // bootstrap only adds the column. v112 runs later via runMigrations and is + // idempotent. + await conn.unsafe(` + ALTER TABLE pages ADD COLUMN IF NOT EXISTS links_extracted_at TIMESTAMPTZ; + `); + } } async transaction(fn: (engine: BrainEngine) => Promise): Promise { @@ -2353,6 +2375,74 @@ export class PostgresEngine implements BrainEngine { `; } + // ── v0.42.7 (#1696): link/timeline extraction freshness watermark ── + + /** Shared stale-for-extraction predicate. Returns `{ where, params }`. */ + private buildStalePagesWhere(opts?: { sourceId?: string; versionTs?: string }): { where: string; params: unknown[] } { + const conds: string[] = ['deleted_at IS NULL']; + const params: unknown[] = []; + if (opts?.versionTs) { + params.push(opts.versionTs); + conds.push(`(links_extracted_at IS NULL OR links_extracted_at < $${params.length}::timestamptz OR updated_at > links_extracted_at)`); + } else { + conds.push('(links_extracted_at IS NULL OR updated_at > links_extracted_at)'); + } + if (opts?.sourceId) { + params.push(opts.sourceId); + conds.push(`source_id = $${params.length}`); + } + return { where: conds.join(' AND '), params }; + } + + async countStalePagesForExtraction(opts?: { sourceId?: string; versionTs?: string }): Promise { + const { where, params } = this.buildStalePagesWhere(opts); + const rows = await this.sql.unsafe( + `SELECT count(*)::int AS count FROM pages WHERE ${where}`, + params as Parameters[1], + ); + return Number((rows[0] as { count?: number } | undefined)?.count ?? 0); + } + + async listStalePagesForExtraction(opts: { + batchSize: number; + afterPageId?: number; + sourceId?: string; + versionTs?: string; + }): Promise { + const { where, params } = this.buildStalePagesWhere(opts); + let afterClause = ''; + if (opts.afterPageId != null) { + params.push(opts.afterPageId); + afterClause = ` AND id > $${params.length}`; + } + params.push(opts.batchSize); + const limitIdx = params.length; + const rows = await this.sql.unsafe( + `SELECT id, slug, source_id, type, title, compiled_truth, timeline, frontmatter, updated_at + FROM pages + WHERE ${where}${afterClause} + ORDER BY id + LIMIT $${limitIdx}`, + params as Parameters[1], + ); + return (rows as Record[]).map(rowToStalePage); + } + + async markPagesExtractedBatch(refs: Array<{ slug: string; source_id: string; extractedAt?: string }>, defaultExtractedAt: string): Promise { + if (refs.length === 0) return; + const slugs = refs.map(r => r.slug); + const srcs = refs.map(r => r.source_id); + // Per-ref timestamp (D4 race fix): extract --stale passes each row's read + // updated_at; sites that omit it fall back to defaultExtractedAt. + const tss = refs.map(r => r.extractedAt ?? defaultExtractedAt); + const sql = this.sql; + await sql` + UPDATE pages p SET links_extracted_at = v.ts::timestamptz + FROM unnest(${slugs}::text[], ${srcs}::text[], ${tss}::text[]) AS v(slug, source_id, ts) + WHERE p.slug = v.slug AND p.source_id = v.source_id + `; + } + // Links async addLink( from: string, diff --git a/src/core/retry.ts b/src/core/retry.ts index 5bc93cde5..4b7160bad 100644 --- a/src/core/retry.ts +++ b/src/core/retry.ts @@ -85,6 +85,8 @@ export const BATCH_AUDIT_SITES = [ 'extract.links_db', 'extract.timeline_db', 'extract.by_mention', + // v0.42.7 (#1696): extract --stale incremental sweep. + 'extract.stale', // operations.ts MCP put_page auto-link path. 'mcp.put_page.autolink', // sync.ts/reindex.ts orchestrator labels. diff --git a/src/core/schema-embedded.ts b/src/core/schema-embedded.ts index e0d7cdedd..963257a44 100644 --- a/src/core/schema-embedded.ts +++ b/src/core/schema-embedded.ts @@ -128,6 +128,14 @@ CREATE TABLE IF NOT EXISTS pages ( -- (NOT inside engine methods — internal callers must not pollute the -- signal). NULL = never retrieved (LSD prioritizes these first). last_retrieved_at TIMESTAMPTZ, + -- v0.42.7 (migration v112): link-extraction freshness watermark. Set when + -- link/timeline extraction last ran for this page (inline sync, \`extract + -- --source db\`, or \`extract --stale\`). A page is stale for extraction when + -- this is NULL, older than LINK_EXTRACTOR_VERSION_TS, or older than + -- updated_at (edited-since-extract — the MCP put_page / sync --no-extract + -- path). Powers \`gbrain extract --stale\` + the \`links_extraction_lag\` doctor + -- check. NULL = never extracted. + links_extracted_at TIMESTAMPTZ, -- v0.40.3.0 contextual retrieval (renumbered from v81 to v90 on master -- merge). contextual_retrieval_mode is what tier the page was last embedded -- under (NULL = pre-v90 = treated as 'none' for drift detection). @@ -253,6 +261,15 @@ CREATE INDEX IF NOT EXISTS pages_deleted_at_purge_idx -- would miss the NULL branch that LSD prioritizes (codex round 2 #6). CREATE INDEX IF NOT EXISTS pages_last_retrieved_at_idx ON pages (last_retrieved_at); +-- v0.42.7 (migration v112): composite B-tree backing \`extract --stale\` and the +-- \`links_extraction_lag\` doctor check. source_id leads so source-scoped staleness +-- scans (\`extract --stale --source X\`, \`gbrain doctor --source X\`) are indexed; +-- the brain-wide COUNT still uses it via the leading column. NOT partial-NULL — +-- the staleness predicate has a NULL arm AND a \`< \$versionTs\` arm (B-tree sorts +-- NULLs to one end, covering both). The \`updated_at > links_extracted_at\` arm is +-- a cross-column filter no index covers; acceptable for a watermark COUNT. +CREATE INDEX IF NOT EXISTS pages_links_extracted_at_idx + ON pages (source_id, links_extracted_at); -- v0.29.1: expression index used by since/until date-range filters that read -- COALESCE(effective_date, updated_at). A partial index on effective_date -- alone would NOT help — the planner can't use it for the negative side of diff --git a/src/core/types.ts b/src/core/types.ts index 9eadf7f8a..c401666a4 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -590,6 +590,26 @@ export interface StaleChunkRow { page_id: number; } +/** + * v0.42.7 (#1696) — a page that needs link/timeline extraction, returned by + * `listStalePagesForExtraction`. Carries the page CONTENT (compiled_truth + + * timeline + frontmatter) so `gbrain extract --stale` extracts in ~1 query per + * batch instead of an N+1 `getPage` per page (mirrors how StaleChunkRow carries + * chunk_text). `id` is the keyset cursor; `updated_at` lets callers reason about + * the edited-since-extract staleness arm. + */ +export interface StalePageRow { + id: number; + slug: string; + source_id: string; + type: string; + title: string; + compiled_truth: string; + timeline: string; + frontmatter: Record; + updated_at: Date; +} + export interface ChunkInput { chunk_index: number; chunk_text: string; diff --git a/src/core/utils.ts b/src/core/utils.ts index 768a11c9f..98daff844 100644 --- a/src/core/utils.ts +++ b/src/core/utils.ts @@ -1,5 +1,5 @@ import { createHash, randomBytes } from 'crypto'; -import type { Page, PageInput, PageType, Chunk, SearchResult } from './types.ts'; +import type { Page, PageInput, PageType, Chunk, SearchResult, StalePageRow } from './types.ts'; import type { Take, TakeKind } from './engine.ts'; /** @@ -121,6 +121,27 @@ export function rowToPage(row: Record): Page { }; } +/** + * v0.42.7 (#1696) — map a DB row to a StalePageRow for the extraction + * freshness sweep. Shared by both engines so frontmatter JSONB parsing can't + * drift. Mirrors rowToPage's `typeof === 'string' ? JSON.parse` idiom; tolerates + * NULL compiled_truth/timeline/frontmatter (empty-string / {} fallback). + */ +export function rowToStalePage(row: Record): StalePageRow { + const fm = row.frontmatter; + return { + id: row.id as number, + slug: row.slug as string, + source_id: (row.source_id as string | undefined) ?? 'default', + type: row.type as string, + title: (row.title as string | null) ?? '', + compiled_truth: (row.compiled_truth as string | null) ?? '', + timeline: (row.timeline as string | null) ?? '', + frontmatter: (fm == null ? {} : (typeof fm === 'string' ? JSON.parse(fm) : fm)) as Record, + updated_at: new Date(row.updated_at as string), + }; +} + /** * Normalize an embedding value into a Float32Array. * diff --git a/src/schema.sql b/src/schema.sql index 1eb53512b..1a10af1fa 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -124,6 +124,14 @@ CREATE TABLE IF NOT EXISTS pages ( -- (NOT inside engine methods — internal callers must not pollute the -- signal). NULL = never retrieved (LSD prioritizes these first). last_retrieved_at TIMESTAMPTZ, + -- v0.42.7 (migration v112): link-extraction freshness watermark. Set when + -- link/timeline extraction last ran for this page (inline sync, `extract + -- --source db`, or `extract --stale`). A page is stale for extraction when + -- this is NULL, older than LINK_EXTRACTOR_VERSION_TS, or older than + -- updated_at (edited-since-extract — the MCP put_page / sync --no-extract + -- path). Powers `gbrain extract --stale` + the `links_extraction_lag` doctor + -- check. NULL = never extracted. + links_extracted_at TIMESTAMPTZ, -- v0.40.3.0 contextual retrieval (renumbered from v81 to v90 on master -- merge). contextual_retrieval_mode is what tier the page was last embedded -- under (NULL = pre-v90 = treated as 'none' for drift detection). @@ -249,6 +257,15 @@ CREATE INDEX IF NOT EXISTS pages_deleted_at_purge_idx -- would miss the NULL branch that LSD prioritizes (codex round 2 #6). CREATE INDEX IF NOT EXISTS pages_last_retrieved_at_idx ON pages (last_retrieved_at); +-- v0.42.7 (migration v112): composite B-tree backing `extract --stale` and the +-- `links_extraction_lag` doctor check. source_id leads so source-scoped staleness +-- scans (`extract --stale --source X`, `gbrain doctor --source X`) are indexed; +-- the brain-wide COUNT still uses it via the leading column. NOT partial-NULL — +-- the staleness predicate has a NULL arm AND a `< $versionTs` arm (B-tree sorts +-- NULLs to one end, covering both). The `updated_at > links_extracted_at` arm is +-- a cross-column filter no index covers; acceptable for a watermark COUNT. +CREATE INDEX IF NOT EXISTS pages_links_extracted_at_idx + ON pages (source_id, links_extracted_at); -- v0.29.1: expression index used by since/until date-range filters that read -- COALESCE(effective_date, updated_at). A partial index on effective_date -- alone would NOT help — the planner can't use it for the negative side of diff --git a/test/core/retry.test.ts b/test/core/retry.test.ts index 9714775fe..6bd214709 100644 --- a/test/core/retry.test.ts +++ b/test/core/retry.test.ts @@ -344,6 +344,7 @@ describe('BATCH_AUDIT_SITES typed enum + isBatchAuditSite guard (D10c codex)', ( 'extract.links_fs', 'extract.timeline_fs', 'extract.links_db', 'extract.timeline_db', 'extract.by_mention', + 'extract.stale', 'mcp.put_page.autolink', 'sync.import_file', 'reindex.markdown', 'reindex.multimodal', diff --git a/test/doctor-links-extraction-lag.test.ts b/test/doctor-links-extraction-lag.test.ts new file mode 100644 index 000000000..8be0b8640 --- /dev/null +++ b/test/doctor-links-extraction-lag.test.ts @@ -0,0 +1,121 @@ +/** + * Tests for the `links_extraction_lag` doctor check (v0.42.7, #1696). + * Hermetic PGLite. Bulk-seeds pages via raw SQL (the check only does COUNT + + * countStalePagesForExtraction — no real ingestion needed). + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { withEnv } from './helpers/with-env.ts'; +import { checkLinksExtractionLag } from '../src/commands/doctor.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}, 60_000); + +afterAll(async () => { + if (engine) await engine.disconnect(); +}, 60_000); + +beforeEach(async () => { + await resetPgliteState(engine); +}); + +/** Bulk-insert N pages under a source; all start with links_extracted_at NULL. */ +async function seedPages(n: number, sourceId = 'default', prefix = 'p'): Promise { + await engine.executeRaw( + `INSERT INTO pages (slug, source_id, type, title, compiled_truth, timeline, frontmatter, content_hash, created_at, updated_at) + SELECT $1 || '/' || g, $2, 'concept', 'T' || g, '', '', '{}'::jsonb, $1 || 'h' || g, now(), now() + FROM generate_series(1, $3) g`, + [prefix, sourceId, n], + ); +} + +describe('links_extraction_lag doctor check', () => { + test('no pages → ok (not applicable)', async () => { + const c = await checkLinksExtractionLag(engine); + expect(c.status).toBe('ok'); + expect(c.message).toContain('no pages'); + }); + + test('<100 pages, no --source → ok (vacuous-skip)', async () => { + await seedPages(50); + const c = await checkLinksExtractionLag(engine); + expect(c.status).toBe('ok'); + expect(c.message).toContain('too few'); + }); + + test('>100 pages, all un-extracted → warn (>20%)', async () => { + await seedPages(120); + const c = await checkLinksExtractionLag(engine); + expect(c.status).toBe('warn'); + expect(c.message).toContain('un-extracted edges'); + expect(c.message).toContain('gbrain extract --stale'); + expect((c.details as any).pct).toBe(100); + }); + + test('>100 pages, all stamped fresh → ok', async () => { + await seedPages(120); + await engine.executeRaw(`UPDATE pages SET links_extracted_at = now()`); + const c = await checkLinksExtractionLag(engine); + expect(c.status).toBe('ok'); + expect(c.message).toContain('Extraction current'); + }); + + test('warn-only by default: 100% stale does NOT fail without fail-pct', async () => { + await seedPages(120); + const c = await checkLinksExtractionLag(engine); + expect(c.status).toBe('warn'); // never 'fail' by default + }); + + test('GBRAIN_EXTRACTION_LAG_FAIL_PCT opts into hard fail', async () => { + await seedPages(120); + await withEnv({ GBRAIN_EXTRACTION_LAG_FAIL_PCT: '50' }, async () => { + const c = await checkLinksExtractionLag(engine); + expect(c.status).toBe('fail'); + expect(c.message).toContain('fail threshold'); + }); + }); + + test('GBRAIN_EXTRACTION_LAG_WARN_PCT raises the warn bar', async () => { + // 10 of 120 stale = ~8%. Default warn 20% → ok. Lower to 5% → warn. + await seedPages(120); + await engine.executeRaw(`UPDATE pages SET links_extracted_at = now() WHERE slug NOT IN (SELECT slug FROM pages ORDER BY id LIMIT 10)`); + const ok = await checkLinksExtractionLag(engine); + expect(ok.status).toBe('ok'); + await withEnv({ GBRAIN_EXTRACTION_LAG_WARN_PCT: '5' }, async () => { + const warn = await checkLinksExtractionLag(engine); + expect(warn.status).toBe('warn'); + }); + }); + + test('--source scope: small source IS assessed (no vacuous-skip)', async () => { + // 10 pages under source 'dept-x' — below the 100 floor, but explicit + // --source means we assess it anyway (mirrors orphan_ratio). + await engine.executeRaw(`INSERT INTO sources (id, name, config) VALUES ('dept-x', 'Dept X', '{}'::jsonb) ON CONFLICT DO NOTHING`); + await seedPages(10, 'dept-x', 'dx'); + const c = await checkLinksExtractionLag(engine, { sourceId: 'dept-x' }); + expect(c.status).toBe('warn'); // all 10 stale, assessed despite < 100 + expect(c.message).toContain("source 'dept-x'"); + }); + + test('pre-v112 brain (column missing) → ok (graceful)', async () => { + await seedPages(120); + // Simulate a pre-v112 brain by dropping the column. + await engine.executeRaw(`ALTER TABLE pages DROP COLUMN links_extracted_at`); + try { + const c = await checkLinksExtractionLag(engine); + expect(c.status).toBe('ok'); + expect(c.message).toContain('pre-v112'); + } finally { + // Restore so resetPgliteState's TRUNCATE-only reset leaves a valid schema + // for the next test (the column is re-added; data is wiped by beforeEach). + await engine.executeRaw(`ALTER TABLE pages ADD COLUMN links_extracted_at TIMESTAMPTZ`); + } + }); +}); diff --git a/test/e2e/engine-parity.test.ts b/test/e2e/engine-parity.test.ts index 5ee39d7d8..76a3e891e 100644 --- a/test/e2e/engine-parity.test.ts +++ b/test/e2e/engine-parity.test.ts @@ -413,6 +413,66 @@ describeBoth('Engine parity — Postgres vs PGLite', () => { expect(pgliteFed).toEqual(pgFed); }); + // v0.42.7 (#1696): stale-page extraction watermark parity. Isolated under a + // dedicated source so other tests' mutations don't perturb the counts. + test('stale-page extraction methods: Postgres ↔ PGLite parity', async () => { + const SRC = 'stale-parity'; + const VER = '2026-05-31T00:00:00Z'; + for (const eng of [pgEngine, pgliteEngine]) { + await eng.executeRaw(`INSERT INTO sources (id, name, config) VALUES ($1, 'Stale Parity', '{}'::jsonb) ON CONFLICT DO NOTHING`, [SRC]); + await eng.executeRaw( + `INSERT INTO pages (slug, source_id, type, title, compiled_truth, timeline, frontmatter, content_hash, created_at, updated_at) + SELECT 'sp/' || g, $1, 'concept', 'SP' || g, 'body ' || g, '', '{}'::jsonb, 'sph' || g, now(), now() + FROM generate_series(1, 3) g`, + [SRC], + ); + } + + // NULL arm: all 3 stale on both engines. + expect(await pgEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(3); + expect(await pgliteEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(3); + + // listStalePagesForExtraction: same slugs + content columns populated. + const pgList = (await pgEngine.listStalePagesForExtraction({ batchSize: 10, sourceId: SRC })).map(r => r.slug).sort(); + const plList = (await pgliteEngine.listStalePagesForExtraction({ batchSize: 10, sourceId: SRC })).map(r => r.slug).sort(); + expect(pgList).toEqual(['sp/1', 'sp/2', 'sp/3']); + expect(plList).toEqual(pgList); + const pgRow = (await pgEngine.listStalePagesForExtraction({ batchSize: 1, sourceId: SRC }))[0]; + expect(pgRow.compiled_truth).toBeTruthy(); + expect(pgRow.updated_at).toBeInstanceOf(Date); + + // markPagesExtractedBatch: stamp one → count drops to 2 on both. + const stampAt = new Date().toISOString(); + await pgEngine.markPagesExtractedBatch([{ slug: 'sp/1', source_id: SRC }], stampAt); + await pgliteEngine.markPagesExtractedBatch([{ slug: 'sp/1', source_id: SRC }], stampAt); + expect(await pgEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(2); + expect(await pgliteEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(2); + + // version arm: stamp sp/2 old + set updated_at old (isolate version arm) → + // flagged only when versionTs is passed. Parity on both engines. + for (const eng of [pgEngine, pgliteEngine]) { + await eng.markPagesExtractedBatch([{ slug: 'sp/2', source_id: SRC }], '2000-01-01T00:00:00Z'); + await eng.executeRaw(`UPDATE pages SET updated_at = '2000-01-01T00:00:00Z' WHERE slug = 'sp/2' AND source_id = $1`, [SRC]); + } + // Without versionTs: sp/2 not stale (stamp == updated, not NULL). sp/3 still NULL-stale. + expect(await pgEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(1); + expect(await pgliteEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(1); + // With versionTs: sp/2's old stamp (< VER) re-flags it → 2 stale. + expect(await pgEngine.countStalePagesForExtraction({ sourceId: SRC, versionTs: VER })).toBe(2); + expect(await pgliteEngine.countStalePagesForExtraction({ sourceId: SRC, versionTs: VER })).toBe(2); + + // edited-since arm: stamp sp/1 in the recent past, updated_at slightly after → + // re-flagged on both engines (updated_at > links_extracted_at). + for (const eng of [pgEngine, pgliteEngine]) { + await eng.executeRaw( + `UPDATE pages SET links_extracted_at = now() - interval '2 hours', updated_at = now() - interval '1 hour' WHERE slug = 'sp/1' AND source_id = $1`, + [SRC], + ); + } + expect(await pgEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(2); // sp/1 (edited) + sp/3 (NULL) + expect(await pgliteEngine.countStalePagesForExtraction({ sourceId: SRC })).toBe(2); + }); + test('v0.41.39 listEnrichCandidates parity (thin filter + source-aware inbound + order)', async () => { const stub = 'Stub page.'; const pageSql = ` diff --git a/test/extract-stale.test.ts b/test/extract-stale.test.ts new file mode 100644 index 000000000..a1209bdbc --- /dev/null +++ b/test/extract-stale.test.ts @@ -0,0 +1,261 @@ +/** + * Tests for `gbrain extract --stale` + the link-extraction freshness watermark + * (v0.42.7, #1696). Hermetic PGLite — no DATABASE_URL, no API keys. + * + * Covers: + * - engine methods: countStalePagesForExtraction (NULL / version / edited-since + * arms + source scope), listStalePagesForExtraction (content + keyset), + * markPagesExtractedBatch (composite-key stamp). + * - `extract --stale`: sweep creates typed edges + stamps every processed page + * (incl. zero-link), second run finds 0 stale (idempotent), --dry-run writes + * nothing, --source-id scope. + * - CRITICAL regression (CDX-1): a page edited after a prior stamp + * (updated_at > links_extracted_at) is re-flagged stale and re-extracted. + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { runExtract } from '../src/commands/extract.ts'; +import { LINK_EXTRACTOR_VERSION_TS } from '../src/core/link-extraction.ts'; +import type { PageInput } from '../src/core/types.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}, 60_000); + +afterAll(async () => { + if (engine) await engine.disconnect(); +}, 60_000); + +async function truncateAll() { + for (const t of ['content_chunks', 'links', 'tags', 'raw_data', 'timeline_entries', 'page_versions', 'ingest_log', 'pages']) { + await (engine as any).db.exec(`DELETE FROM ${t}`); + } +} +beforeEach(truncateAll); + +const personPage = (title: string, body = ''): PageInput => ({ type: 'person', title, compiled_truth: body, timeline: '' }); +const companyPage = (title: string, body = ''): PageInput => ({ type: 'company', title, compiled_truth: body, timeline: '' }); + +async function stampOf(slug: string, sourceId = 'default'): Promise { + const rows = await engine.executeRaw<{ links_extracted_at: string | null }>( + `SELECT links_extracted_at FROM pages WHERE slug = $1 AND source_id = $2`, [slug, sourceId], + ); + return rows[0]?.links_extracted_at ?? null; +} + +describe('engine: stale-page extraction methods', () => { + test('countStalePagesForExtraction: NULL arm counts never-extracted pages', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.putPage('people/bob', personPage('Bob')); + expect(await engine.countStalePagesForExtraction()).toBe(2); + }); + + test('countStalePagesForExtraction: stamped pages drop out', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.markPagesExtractedBatch([{ slug: 'people/alice', source_id: 'default' }], new Date().toISOString()); + expect(await engine.countStalePagesForExtraction()).toBe(0); + }); + + test('countStalePagesForExtraction: version arm flags pre-version stamps', async () => { + await engine.putPage('people/alice', personPage('Alice')); + // Stamp with an OLD timestamp (before LINK_EXTRACTOR_VERSION_TS). + await engine.markPagesExtractedBatch([{ slug: 'people/alice', source_id: 'default' }], '2000-01-01T00:00:00Z'); + // Without versionTs: only NULL/edited arms → not stale (stamp >= updated? no: + // stamp is 2000, updated is now → updated_at > stamp → STALE via edited arm). + // So set updated_at back too, isolating the version arm: + await engine.executeRaw(`UPDATE pages SET updated_at = '2000-01-01T00:00:00Z' WHERE slug = 'people/alice'`); + expect(await engine.countStalePagesForExtraction()).toBe(0); // no version, stamp==updated, not NULL + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(1); // version arm + }); + + test('countStalePagesForExtraction: edited-since arm (CDX-1)', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.markPagesExtractedBatch([{ slug: 'people/alice', source_id: 'default' }], new Date().toISOString()); + expect(await engine.countStalePagesForExtraction()).toBe(0); + // Simulate an edit AFTER the stamp (put_page / sync --no-extract). + await engine.executeRaw(`UPDATE pages SET updated_at = '2099-01-01T00:00:00Z' WHERE slug = 'people/alice'`); + expect(await engine.countStalePagesForExtraction()).toBe(1); + }); + + test('listStalePagesForExtraction: returns content columns + keyset paginates', async () => { + await engine.putPage('people/alice', personPage('Alice', 'Body A')); + await engine.putPage('people/bob', personPage('Bob', 'Body B')); + const batch1 = await engine.listStalePagesForExtraction({ batchSize: 1 }); + expect(batch1.length).toBe(1); + expect(batch1[0].compiled_truth).toBeTruthy(); + expect(batch1[0].title).toBeTruthy(); + expect(batch1[0].frontmatter).toBeDefined(); + const batch2 = await engine.listStalePagesForExtraction({ batchSize: 10, afterPageId: batch1[0].id }); + expect(batch2.length).toBe(1); + expect(batch2[0].id).toBeGreaterThan(batch1[0].id); + }); + + test('markPagesExtractedBatch: empty input is a no-op', async () => { + await engine.markPagesExtractedBatch([], new Date().toISOString()); + expect(true).toBe(true); // no throw + }); +}); + +describe('gbrain extract --stale', () => { + test('extracts typed edges + stamps every processed page (incl. zero-link)', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.putPage('companies/acme', companyPage('Acme', '[Alice](people/alice) is the CEO of [Acme](companies/acme).')); + await engine.putPage('people/lonely', personPage('Lonely', 'No links here.')); + + await runExtract(engine, ['--stale']); + + const links = await engine.getLinks('companies/acme'); + expect(links.some(l => l.to_slug === 'people/alice')).toBe(true); + // EVERY processed page stamped — including the zero-link one. + expect(await stampOf('people/alice')).not.toBeNull(); + expect(await stampOf('companies/acme')).not.toBeNull(); + expect(await stampOf('people/lonely')).not.toBeNull(); + // Nothing left stale. + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(0); + }); + + test('idempotent: second run finds 0 stale and creates no new links', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.putPage('companies/acme', companyPage('Acme', '[Alice](people/alice) advises [Acme](companies/acme).')); + + await runExtract(engine, ['--stale']); + const after1 = (await engine.getLinks('companies/acme')).length; + const stamp1 = await stampOf('companies/acme'); + + await runExtract(engine, ['--stale']); + const after2 = (await engine.getLinks('companies/acme')).length; + expect(after2).toBe(after1); + // Second run had 0 stale → did not re-stamp (stamp unchanged is acceptable; + // the key invariant is no duplicate links). + expect(stamp1).not.toBeNull(); + }); + + test('--dry-run reports count and writes nothing', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.putPage('companies/acme', companyPage('Acme', '[Alice](people/alice) joined [Acme](companies/acme).')); + + await runExtract(engine, ['--stale', '--dry-run']); + + expect(await engine.getLinks('companies/acme')).toHaveLength(0); + expect(await stampOf('people/alice')).toBeNull(); + expect(await stampOf('companies/acme')).toBeNull(); + // Still stale after dry-run. + expect(await engine.countStalePagesForExtraction()).toBe(2); + }); + + test('CRITICAL (CDX-1): page edited after stamp is re-extracted', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.putPage('companies/acme', companyPage('Acme', 'No links yet.')); + await runExtract(engine, ['--stale']); + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(0); + + // Simulate an edit that adds a link WITHOUT extracting (MCP put_page / + // sync --no-extract). Use relative intervals so it's clock-agnostic: the + // stamp + edit both land in the recent past (after LINK_EXTRACTOR_VERSION_TS), + // with updated_at AFTER the stamp — and crucially both BEFORE real-now, so + // the re-extract's now()-stamp deterministically supersedes the edit. + await engine.executeRaw( + `UPDATE pages + SET compiled_truth = $1, + links_extracted_at = now() - interval '2 hours', + updated_at = now() - interval '1 hour' + WHERE slug = 'companies/acme'`, + ['[Alice](people/alice) now works at [Acme](companies/acme).'], + ); + // Re-flagged stale by the updated_at arm (updated > stamp). + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(1); + + // extract --stale picks it up, creates the now-present edge, and re-stamps + // at now() (> the edit's updated_at) so the page is fresh again. + await runExtract(engine, ['--stale']); + const links = await engine.getLinks('companies/acme'); + expect(links.some(l => l.to_slug === 'people/alice')).toBe(true); + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(0); + }); + + test('CDX-4 (D2): a link-flush throw aborts the sweep and leaves pages UNSTAMPED', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.putPage('companies/acme', companyPage('Acme', '[Alice](people/alice) founded [Acme](companies/acme).')); + + // Make the link flush throw mid-sweep. The --stale path flushes + // NON-swallowing (no try/catch), so the throw must propagate AND no page in + // the batch may be stamped (stamp runs only AFTER a successful flush). + const origBatch = engine.addLinksBatch.bind(engine); + let threw = false; + (engine as unknown as { addLinksBatch: unknown }).addLinksBatch = async () => { throw new Error('__flush_boom__'); }; + try { + await runExtract(engine, ['--stale']); + } catch (e) { + if ((e as Error).message === '__flush_boom__') threw = true; else throw e; + } finally { + (engine as unknown as { addLinksBatch: unknown }).addLinksBatch = origBatch; + } + expect(threw).toBe(true); + // Pages whose edges were lost are NOT stamped fresh — they stay stale. + expect(await stampOf('people/alice')).toBeNull(); + expect(await stampOf('companies/acme')).toBeNull(); + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(2); + + // A clean re-run re-extracts idempotently (ON CONFLICT DO NOTHING). + await runExtract(engine, ['--stale']); + expect((await engine.getLinks('companies/acme')).some(l => l.to_slug === 'people/alice')).toBe(true); + expect(await stampOf('companies/acme')).not.toBeNull(); + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(0); + }); + + test('D4 race: a concurrent edit landing during the sweep is NOT masked', async () => { + await engine.putPage('people/alice', personPage('Alice')); + await engine.putPage('companies/acme', companyPage('Acme', '[Alice](people/alice) backs [Acme](companies/acme).')); + // Anchor acme's updated_at in the past so the read value is well-defined. + await engine.executeRaw(`UPDATE pages SET updated_at = now() - interval '3 hours' WHERE slug = 'companies/acme'`); + + // Simulate an edit landing BETWEEN the list read (updated_at = now-3h) and + // the stamp: bump acme's updated_at to now-1h just before the real stamp. + // D4 stamps with the READ updated_at (now-3h), so now-1h > now-3h → acme + // stays stale (edit preserved). The OLD now()-stamp would set + // links_extracted_at = now > now-1h → acme marked fresh, edit silently lost. + const origStamp = engine.markPagesExtractedBatch.bind(engine); + let hooked = false; + (engine as unknown as { markPagesExtractedBatch: unknown }).markPagesExtractedBatch = async ( + refs: Array<{ slug: string; source_id: string; extractedAt?: string }>, def: string, + ) => { + if (!hooked) { + hooked = true; + await engine.executeRaw(`UPDATE pages SET updated_at = now() - interval '1 hour' WHERE slug = 'companies/acme'`); + } + return origStamp(refs, def); + }; + try { + await runExtract(engine, ['--stale']); + } finally { + (engine as unknown as { markPagesExtractedBatch: unknown }).markPagesExtractedBatch = origStamp; + } + expect(hooked).toBe(true); + // acme stays stale (only the concurrently-edited page); alice was stamped + // with its own read updated_at and is fresh. + expect(await engine.countStalePagesForExtraction({ versionTs: LINK_EXTRACTOR_VERSION_TS })).toBe(1); + }); + + test('--source fs is rejected (DB-source only)', async () => { + const origErr = console.error; + const origExit = process.exit; + let exited = false; let msg = ''; + console.error = (m?: unknown) => { msg += String(m); }; + process.exit = ((_code?: number) => { exited = true; throw new Error('__exit__'); }) as unknown as typeof process.exit; + try { + await runExtract(engine, ['--stale', '--source', 'fs']); + } catch (e) { + if ((e as Error).message !== '__exit__') throw e; + } finally { + console.error = origErr; + process.exit = origExit; + } + expect(exited).toBe(true); + expect(msg).toContain('DB-source only'); + }); +}); diff --git a/test/migrate.test.ts b/test/migrate.test.ts index faf5fb56c..67fd83248 100644 --- a/test/migrate.test.ts +++ b/test/migrate.test.ts @@ -2139,3 +2139,43 @@ describe('migrate v89 — round-trip on PGLite', () => { }); }); +// v0.42.7 (#1696): pages_links_extracted_at watermark migration. +describe('v112 — pages_links_extracted_at', () => { + let engine: PGLiteEngine; + beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + }, 60_000); + afterAll(async () => { if (engine) await engine.disconnect(); }, 60_000); + + test('v112 entry exists with the documented name + transaction:false + handler', () => { + const m = MIGRATIONS.find(x => x.version === 112); + expect(m).toBeDefined(); + expect(m!.name).toBe('pages_links_extracted_at'); + expect(m!.transaction).toBe(false); + expect(typeof m!.handler).toBe('function'); + }); + + test('LATEST_VERSION is at or above 112', () => { + expect(LATEST_VERSION).toBeGreaterThanOrEqual(112); + }); + + test('links_extracted_at column exists after initSchema, nullable, TIMESTAMPTZ', async () => { + const rows = await engine.executeRaw<{ is_nullable: string; data_type: string }>( + `SELECT is_nullable, data_type FROM information_schema.columns + WHERE table_name = 'pages' AND column_name = 'links_extracted_at'`, [], + ); + expect(rows.length).toBe(1); + expect(rows[0].is_nullable).toBe('YES'); + expect(rows[0].data_type.toLowerCase()).toContain('timestamp'); + }); + + test('composite index pages_links_extracted_at_idx exists after initSchema', async () => { + const rows = await engine.executeRaw<{ indexname: string }>( + `SELECT indexname FROM pg_indexes WHERE tablename = 'pages' AND indexname = 'pages_links_extracted_at_idx'`, [], + ); + expect(rows.length).toBe(1); + }); +}); + diff --git a/test/schema-bootstrap-coverage.test.ts b/test/schema-bootstrap-coverage.test.ts index d6c41826e..63bde3a18 100644 --- a/test/schema-bootstrap-coverage.test.ts +++ b/test/schema-bootstrap-coverage.test.ts @@ -162,6 +162,12 @@ const REQUIRED_BOOTSTRAP_COVERAGE: ForwardReference[] = [ // semantics. No SCHEMA_SQL index references it; bootstrap probe is // defense-in-depth (and satisfies the MIGRATIONS ADD COLUMN coverage gate). { kind: 'column', table: 'pages', column: 'embedding_signature' }, + // v0.42.7 (v112) — forward-referenced by `CREATE INDEX + // pages_links_extracted_at_idx ON pages (source_id, links_extracted_at)`. + // Pre-v112 brains have pages without this column; bootstrap adds it before + // SCHEMA_SQL replay creates the index. Powers `gbrain extract --stale` + the + // `links_extraction_lag` doctor check. + { kind: 'column', table: 'pages', column: 'links_extracted_at' }, ]; test('applyForwardReferenceBootstrap covers every forward reference declared in REQUIRED_BOOTSTRAP_COVERAGE', async () => { diff --git a/test/sync-inline-extract-stamps.serial.test.ts b/test/sync-inline-extract-stamps.serial.test.ts new file mode 100644 index 000000000..c10f86b47 --- /dev/null +++ b/test/sync-inline-extract-stamps.serial.test.ts @@ -0,0 +1,114 @@ +/** + * CRITICAL regression (v0.42.7, #1696, CDX-6) — inline sync extract stamps the + * link-extraction watermark. + * + * `performSync`'s INCREMENTAL path runs link/timeline extraction inline for the + * changed pages (the `gbrain sync` default — see performSyncInner's auto-extract + * block). v0.42.7 adds a `stampExtracted` call at that call site (after + * extractLinksForSlugs/extractTimelineForSlugs) so a normal incremental sync + * marks the pages it just extracted as fresh — otherwise every synced page + * would show as stale forever in the links_extraction_lag doctor check. + * + * NOTE: a FULL / first sync routes to performFullSync which does NOT extract + * inline (pre-existing behavior — exactly the "imported ≠ curated" gap that + * `extract --stale` closes). So this regression test drives the INCREMENTAL + * path: full sync to seed, then edit + incremental sync. + * + * IRON RULE: pins (a) an incremental sync NOW stamps links_extracted_at for the + * pages it processed, and (b) the existing link extraction is unchanged. Plus + * --no-extract: the changed page stays unstamped AND no links are created. + * + * Marked .serial.test.ts — spawns git subprocesses + shares one PGLite engine. + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test'; +import { mkdtempSync, writeFileSync, rmSync, mkdirSync } from 'fs'; +import { execSync } from 'child_process'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; + +let engine: PGLiteEngine; +let repoPath: string; + +function git(cmd: string): void { execSync(cmd, { cwd: repoPath, stdio: 'pipe' }); } + +function writeAcme(body: string): void { + writeFileSync(join(repoPath, 'companies/acme.md'), [ + '---', 'type: company', 'title: Acme', '---', '', body, + ].join('\n')); +} + +async function stampOf(slug: string): Promise { + const rows = await engine.executeRaw<{ links_extracted_at: string | null }>( + `SELECT links_extracted_at FROM pages WHERE slug = $1 AND source_id = 'default'`, [slug], + ); + return rows[0]?.links_extracted_at ?? null; +} + +describe('#1696 — inline sync extract stamps links_extracted_at', () => { + beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + }, 60_000); + + afterAll(async () => { + if (engine) await engine.disconnect(); + }, 60_000); + + beforeEach(async () => { + await resetPgliteState(engine); + repoPath = mkdtempSync(join(tmpdir(), 'gbrain-stamp-')); + execSync('git init', { cwd: repoPath, stdio: 'pipe' }); + execSync('git config user.email "t@t.com"', { cwd: repoPath, stdio: 'pipe' }); + execSync('git config user.name "T"', { cwd: repoPath, stdio: 'pipe' }); + mkdirSync(join(repoPath, 'people'), { recursive: true }); + mkdirSync(join(repoPath, 'companies'), { recursive: true }); + writeFileSync(join(repoPath, 'people/alice.md'), [ + '---', 'type: person', 'title: Alice', '---', '', 'Alice is a founder.', + ].join('\n')); + writeAcme('Acme is a company.'); + git('git add -A && git commit -m "initial"'); + + // Seed: full first sync imports both pages (no inline extract on this path). + const { performSync } = await import('../src/commands/sync.ts'); + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + }); + + afterEach(() => { + if (repoPath) rmSync(repoPath, { recursive: true, force: true }); + }); + + test('(a) incremental sync stamps the watermark AND (b) still extracts the link', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + // Edit acme to add the link, commit → incremental sync processes it. + // Disk-relative link form (how real brain files reference each other on + // disk; the FS extractor resolves relative to the file's directory). + writeAcme('[Alice](../people/alice.md) is the CEO of Acme.'); + git('git add -A && git commit -m "add link"'); + const result = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(['synced', 'first_sync']).toContain(result.status); + + // (b) extraction unchanged — the CEO-of link is created. + const links = await engine.getLinks('companies/acme'); + expect(links.some(l => l.to_slug === 'people/alice')).toBe(true); + + // (a) the changed page sync extracted is now stamped (not stale). + expect(await stampOf('companies/acme')).not.toBeNull(); + }, 60_000); + + test('--no-extract: changed page is NOT stamped and no links are created', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + // Disk-relative link form (how real brain files reference each other on + // disk; the FS extractor resolves relative to the file's directory). + writeAcme('[Alice](../people/alice.md) is the CEO of Acme.'); + git('git add -A && git commit -m "add link"'); + const result = await performSync(engine, { repoPath, noPull: true, noEmbed: true, noExtract: true }); + expect(['synced', 'first_sync']).toContain(result.status); + + expect(await engine.getLinks('companies/acme')).toHaveLength(0); + expect(await stampOf('companies/acme')).toBeNull(); + }, 60_000); +}); diff --git a/test/sync-nudge-status-gate.test.ts b/test/sync-nudge-status-gate.test.ts new file mode 100644 index 000000000..e38aa2917 --- /dev/null +++ b/test/sync-nudge-status-gate.test.ts @@ -0,0 +1,40 @@ +/** + * v0.42.7 (#1696, D5) — the end-of-sync extraction-lag nudge status gate. + * + * Codex caught that the single-source nudge was gated `=== 'synced'`, which + * skips `first_sync` (a fresh / --full import — the BIGGEST un-extracted + * backlog, the exact 280K-page scenario #1696 exists for) and `up_to_date`. + * `shouldNudgeAfterSync` is the pure predicate the call site now uses; this + * pins its contract so the gate can't silently narrow back to `synced` only. + * + * Pure predicate test — no PGLite, no env mutation, no module stubbing (R1-R4 compliant). + */ + +import { describe, test, expect } from 'bun:test'; +import { shouldNudgeAfterSync } from '../src/commands/sync.ts'; + +describe('shouldNudgeAfterSync (D5 status gate)', () => { + test('fires on first_sync — the biggest-backlog initial-import case', () => { + expect(shouldNudgeAfterSync('first_sync')).toBe(true); + }); + + test('fires on synced (incremental)', () => { + expect(shouldNudgeAfterSync('synced')).toBe(true); + }); + + test('fires on up_to_date (no-op sync over a pre-existing backlog)', () => { + expect(shouldNudgeAfterSync('up_to_date')).toBe(true); + }); + + test('does NOT fire on dry_run (preview, no real sync)', () => { + expect(shouldNudgeAfterSync('dry_run')).toBe(false); + }); + + test('does NOT fire on blocked_by_failures (inconsistent state)', () => { + expect(shouldNudgeAfterSync('blocked_by_failures')).toBe(false); + }); + + test('does NOT fire on partial (inconsistent state)', () => { + expect(shouldNudgeAfterSync('partial')).toBe(false); + }); +});