mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 21:19:18 +00:00
v0.42.7.0 feat(extract): link/timeline extraction freshness watermark — gbrain extract --stale + doctor lag check (#1696) (#1755)
* feat(extract): link/timeline extraction freshness watermark (#1696) Closes the "imported != curated" gap: plain `gbrain sync` only extracts CHANGED pages, so a brain with autopilot off accumulated a links table that was ~99.7% untyped `mentions` with nothing surfacing it. Adds a per-page freshness watermark (pages.links_extracted_at, migration v112) and three things built on it: - `gbrain extract --stale [--source-id] [--catch-up] [--dry-run] [--json]`: incremental DB-source link+timeline sweep over pages whose extraction is stale (never extracted, edited since, or extractor version bumped). Small byte-bounded batches, non-swallowing flush, stamp-after-flush so a crash re-extracts idempotently. Stamps with the row's READ updated_at (not now()) so a concurrent edit during the sweep stays stale instead of being lost. - `links_extraction_lag` doctor check (local + remote): warn-only by default (>20%), hard-fail only via GBRAIN_EXTRACTION_LAG_FAIL_PCT. Vacuous-skip <100 pages; pre-v112 brains graceful-skip. - `gbrain sync --no-extract` flag + end-of-sync nudge (fires on synced|first_sync|up_to_date so the initial import surfaces its backlog). Three new BrainEngine methods (countStalePagesForExtraction / listStalePagesForExtraction / markPagesExtractedBatch) with Postgres<->PGLite parity + bootstrap probes. Schema parity: schema.sql + regenerated pglite-schema.ts + schema-embedded.ts + bootstrap-coverage test. Migration v112 (composite (source_id, links_extracted_at) index, no backfill so the real backlog surfaces on first doctor run). * test(audit): hermetic GBRAIN_AUDIT_DIR override for prune ENOENT case The "no-op when audit dir does not exist (ENOENT)" case called pruneOldBatchRetryAuditFiles without a GBRAIN_AUDIT_DIR override, so it read the developer's real ~/.gbrain/audit and flaked (kept>0) on any machine with prior gbrain audit history. Point it at a guaranteed-nonexistent temp path so it tests the real missing-dir branch hermetically — matching the file header's "never touches ~/.gbrain/audit" contract. Pre-existing flake (introduced by v0.41.19.0 #1537), unrelated to #1696. * chore: bump version and changelog (v0.42.2.0) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * docs: CLAUDE.md key-files entry for the #1696 extract-stale wave + regen llms-full --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
662a6e27d4
commit
ca68a551db
+101
@@ -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 <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 <id>` 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`
|
||||
|
||||
@@ -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
|
||||
|
||||
+1
-1
File diff suppressed because one or more lines are too long
+1
-1
@@ -142,5 +142,5 @@
|
||||
"bun": ">=1.3.10"
|
||||
},
|
||||
"license": "MIT",
|
||||
"version": "0.42.6.0"
|
||||
"version": "0.42.7.0"
|
||||
}
|
||||
|
||||
+125
-6
@@ -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<DoctorRep
|
||||
// v0.41.19.0 (Issue 5): sync --all consolidation nudge for multi-source brains.
|
||||
checks.push(await checkSyncConsolidation(engine));
|
||||
|
||||
// v0.42.7 (#1696): link-extraction lag. Strictly SQL (single indexed COUNT),
|
||||
// safe on the thin-client/remote path — remote operators on checkout-less
|
||||
// Postgres brains are exactly who can't otherwise see the extraction backlog.
|
||||
// Brain-wide here (remote --source scoping is a separate TODO, like orphan_ratio).
|
||||
checks.push(await checkLinksExtractionLag(engine));
|
||||
|
||||
// v0.39 T7 + T9 — schema-pack health checks (3 checks per v0.38 plan):
|
||||
// schema_pack_active — active pack resolves cleanly
|
||||
// schema_pack_consistency — % of pages typed against active pack
|
||||
@@ -2298,18 +2306,38 @@ async function checkSubagentCapability(engine: BrainEngine): Promise<Check> {
|
||||
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<string>();
|
||||
|
||||
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<Check> {
|
||||
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.
|
||||
|
||||
+275
-33
@@ -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<void> {
|
||||
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<string>,
|
||||
slugToSources: Map<string, string[]>,
|
||||
): { 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 <links|timeline|all> --ner --source db
|
||||
gbrain extract <timeline|all> --from-meetings
|
||||
|
||||
Incremental sweep (v0.42.7):
|
||||
gbrain extract --stale [--source-id <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 <kind> [--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<string>();
|
||||
const slugToSources = new Map<string, string[]>();
|
||||
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.
|
||||
|
||||
+77
-2
@@ -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<Sy
|
||||
const extractOpts = opts.sourceId ? { sourceId: opts.sourceId } : undefined;
|
||||
if (!opts.noExtract && pagesAffected.length > 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<void> {
|
||||
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.
|
||||
*
|
||||
|
||||
@@ -85,6 +85,7 @@ export const BRAIN_CHECK_NAMES: ReadonlySet<string> = new Set([
|
||||
'image_assets',
|
||||
'integrity',
|
||||
'jsonb_integrity',
|
||||
'links_extraction_lag',
|
||||
'markdown_body_completeness',
|
||||
'nightly_quality_probe_health',
|
||||
'ocr_health',
|
||||
|
||||
+43
-1
@@ -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<void>;
|
||||
|
||||
// ============================================================
|
||||
// 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<number>;
|
||||
/**
|
||||
* 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<StalePageRow[]>;
|
||||
/**
|
||||
* 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<void>;
|
||||
|
||||
// Links
|
||||
/**
|
||||
* Single-row link insert. linkSource defaults to 'markdown' for back-compat
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<T>(fn: (conn: ReservedConnection) => Promise<T>): Promise<T> {
|
||||
@@ -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<number> {
|
||||
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<StalePageRow[]> {
|
||||
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<string, unknown>[]).map(rowToStalePage);
|
||||
}
|
||||
|
||||
async markPagesExtractedBatch(refs: Array<{ slug: string; source_id: string; extractedAt?: string }>, defaultExtractedAt: string): Promise<void> {
|
||||
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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<T>(fn: (engine: BrainEngine) => Promise<T>): Promise<T> {
|
||||
@@ -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<number> {
|
||||
const { where, params } = this.buildStalePagesWhere(opts);
|
||||
const rows = await this.sql.unsafe(
|
||||
`SELECT count(*)::int AS count FROM pages WHERE ${where}`,
|
||||
params as Parameters<typeof this.sql.unsafe>[1],
|
||||
);
|
||||
return Number((rows[0] as { count?: number } | undefined)?.count ?? 0);
|
||||
}
|
||||
|
||||
async listStalePagesForExtraction(opts: {
|
||||
batchSize: number;
|
||||
afterPageId?: number;
|
||||
sourceId?: string;
|
||||
versionTs?: string;
|
||||
}): Promise<StalePageRow[]> {
|
||||
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<typeof this.sql.unsafe>[1],
|
||||
);
|
||||
return (rows as Record<string, unknown>[]).map(rowToStalePage);
|
||||
}
|
||||
|
||||
async markPagesExtractedBatch(refs: Array<{ slug: string; source_id: string; extractedAt?: string }>, defaultExtractedAt: string): Promise<void> {
|
||||
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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
updated_at: Date;
|
||||
}
|
||||
|
||||
export interface ChunkInput {
|
||||
chunk_index: number;
|
||||
chunk_text: string;
|
||||
|
||||
+22
-1
@@ -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<string, unknown>): 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<string, unknown>): 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<string, unknown>,
|
||||
updated_at: new Date(row.updated_at as string),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize an embedding value into a Float32Array.
|
||||
*
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -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<void> {
|
||||
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`);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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 = `
|
||||
|
||||
@@ -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<string | null> {
|
||||
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');
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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<string | null> {
|
||||
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);
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user