diff --git a/src/core/engine.ts b/src/core/engine.ts index 317931fcd..d12b839c6 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -16,6 +16,7 @@ import type { EmotionalWeightInputRow, EmotionalWeightWriteRow, DomainBankSampleOpts, CorpusSampleOpts, DomainBankRow, AdjacencyRow, + EnrichCandidatesOpts, EnrichCandidate, } from './types.ts'; /** @@ -1976,6 +1977,20 @@ export interface BrainEngine { */ getRecentSalience(opts: SalienceOpts): Promise; + /** + * v0.41.39 (issue #1700) — enrich candidate selection for + * `gbrain enrich --thin`. ONE source-aware SQL query: thin-filter + + * per-page inbound-link count (source-correct via `to_page_id = p.id`, + * `link_source='mentions'` excluded) + optional `enriched_at` recency + * guard + whitelisted ORDER BY (ENRICH_ORDER_SQL) + LIMIT. Returns a + * lightweight projection (NO page bodies) so ranking 100K stubs doesn't + * pull every body into memory. + * + * Empty `opts.types` → empty result, no SQL. Source scope follows the + * canonical scalar/array precedence (sourceIds wins over sourceId). + */ + listEnrichCandidates(opts: EnrichCandidatesOpts): Promise; + /** * Anomaly detection: cohorts (tag, type) with unusually-high page activity * on a target day vs baseline mean+stddev over the previous N days. Year diff --git a/src/core/pglite-engine.ts b/src/core/pglite-engine.ts index c82c3457d..09e6348a1 100644 --- a/src/core/pglite-engine.ts +++ b/src/core/pglite-engine.ts @@ -40,11 +40,12 @@ import type { SalienceOpts, SalienceResult, AnomaliesOpts, AnomalyResult, EmotionalWeightInputRow, EmotionalWeightWriteRow, DomainBankSampleOpts, CorpusSampleOpts, DomainBankRow, + EnrichCandidatesOpts, EnrichCandidate, } from './types.ts'; import { validateSlug, contentHash, rowToPage, 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 } from './types.ts'; +import { GBrainError, PAGE_SORT_SQL, ENRICH_ORDER_SQL } from './types.ts'; import { computeAnomaliesFromBuckets } from './cycle/anomaly.ts'; import { resolveBoostMap, resolveHardExcludes } from './search/source-boost.ts'; import { buildSourceFactorCase, buildHardExcludeClause, buildVisibilityClause, buildRecencyComponentSql, buildBestPerPagePoolCte } from './search/sql-ranking.ts'; @@ -5012,6 +5013,78 @@ export class PGLiteEngine implements BrainEngine { })); } + async listEnrichCandidates(opts: EnrichCandidatesOpts): Promise { + // v0.41.39 (issue #1700). Parity with postgres-engine.listEnrichCandidates. + if (!opts.types || opts.types.length === 0) return []; + const limit = Math.max(1, Math.min(opts.limit ?? 50, 5000)); + const threshold = Math.max(0, opts.thinThreshold); + + const params: unknown[] = []; + params.push(opts.types); + const typesParam = `$${params.length}`; + params.push(threshold); + const thresholdParam = `$${params.length}`; + + const where: string[] = [ + 'p.deleted_at IS NULL', + `p.type = ANY(${typesParam}::text[])`, + `(char_length(p.compiled_truth) + char_length(COALESCE(p.timeline, ''))) < ${thresholdParam}`, + ]; + + // Source scope: array wins over scalar. + if (opts.sourceIds && opts.sourceIds.length > 0) { + params.push(opts.sourceIds); + where.push(`p.source_id = ANY($${params.length}::text[])`); + } else if (opts.sourceId) { + params.push(opts.sourceId); + where.push(`p.source_id = $${params.length}`); + } + + // Re-enrich recency guard. Lexical text compare on the ISO `enriched_at` + // (never cast → can't throw on a malformed value). NULL → eligible. + const reenrichMs = opts.reenrichAfterMs ?? 0; + if (reenrichMs > 0) { + params.push(new Date(Date.now() - reenrichMs).toISOString()); + where.push( + `NOT (p.frontmatter ->> 'enriched_at' IS NOT NULL AND p.frontmatter ->> 'enriched_at' > $${params.length})`, + ); + } + + const orderKey = ENRICH_ORDER_SQL[opts.order] ? opts.order : 'inbound-links'; + const orderBy = ENRICH_ORDER_SQL[orderKey]; + + params.push(limit); + const limitParam = `$${params.length}`; + + const { rows } = await this.db.query( + `SELECT + p.slug, + p.source_id, + p.title, + p.type, + (char_length(p.compiled_truth) + char_length(COALESCE(p.timeline, ''))) AS body_len, + COALESCE(( + SELECT COUNT(*) + FROM links l + WHERE l.to_page_id = p.id + AND l.link_source IS DISTINCT FROM 'mentions' + ), 0)::int AS inbound_count + FROM pages p + WHERE ${where.join(' AND ')} + ORDER BY ${orderBy} + LIMIT ${limitParam}`, + params, + ); + return (rows as Record[]).map((r) => ({ + slug: String(r.slug), + source_id: String(r.source_id), + title: String(r.title ?? ''), + type: r.type as EnrichCandidate['type'], + body_len: Number(r.body_len ?? 0), + inbound_count: Number(r.inbound_count ?? 0), + })); + } + async findAnomalies(opts: AnomaliesOpts): Promise { const sigma = opts.sigma ?? 3.0; const lookbackDays = Math.max(1, opts.lookback_days ?? 30); diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 61d081a19..10dff72f9 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -47,8 +47,9 @@ import type { EvalCaptureFailure, EvalCaptureFailureReason, SalienceOpts, SalienceResult, AnomaliesOpts, AnomalyResult, EmotionalWeightInputRow, EmotionalWeightWriteRow, + EnrichCandidatesOpts, EnrichCandidate, } from './types.ts'; -import { GBrainError, PAGE_SORT_SQL } from './types.ts'; +import { GBrainError, PAGE_SORT_SQL, ENRICH_ORDER_SQL } from './types.ts'; import { computeAnomaliesFromBuckets } from './cycle/anomaly.ts'; import * as db from './db.ts'; import { ConnectionManager } from './connection-manager.ts'; @@ -5074,6 +5075,67 @@ export class PostgresEngine implements BrainEngine { })); } + async listEnrichCandidates(opts: EnrichCandidatesOpts): Promise { + // v0.41.39 (issue #1700). Empty types → no rows (no SQL). + if (!opts.types || opts.types.length === 0) return []; + const sql = this.sql; + const limit = Math.max(1, Math.min(opts.limit ?? 50, 5000)); + const threshold = Math.max(0, opts.thinThreshold); + + // Source scope: array wins over scalar (canonical precedence). + const sourceCondition = opts.sourceIds && opts.sourceIds.length > 0 + ? sql`AND p.source_id = ANY(${opts.sourceIds}::text[])` + : opts.sourceId + ? sql`AND p.source_id = ${opts.sourceId}` + : sql``; + + // Re-enrich recency guard. enriched_at is written as toISOString() so a + // lexical text comparison is correct AND can't throw on a malformed value + // (a ::timestamptz cast would). Pages never enriched (NULL) are eligible. + const reenrichMs = opts.reenrichAfterMs ?? 0; + const recencyCondition = reenrichMs > 0 + ? sql`AND NOT ( + p.frontmatter ->> 'enriched_at' IS NOT NULL + AND p.frontmatter ->> 'enriched_at' > ${new Date(Date.now() - reenrichMs).toISOString()} + )` + : sql``; + + // Whitelisted ORDER BY (no injection — enum maps to a literal fragment). + const orderKey = ENRICH_ORDER_SQL[opts.order] ? opts.order : 'inbound-links'; + const orderBy = sql.unsafe(ENRICH_ORDER_SQL[orderKey]); + + const rows = await sql` + SELECT + p.slug, + p.source_id, + p.title, + p.type, + (char_length(p.compiled_truth) + char_length(COALESCE(p.timeline, ''))) AS body_len, + COALESCE(( + SELECT COUNT(*) + FROM links l + WHERE l.to_page_id = p.id + AND l.link_source IS DISTINCT FROM 'mentions' + ), 0)::int AS inbound_count + FROM pages p + WHERE p.deleted_at IS NULL + AND p.type = ANY(${opts.types}::text[]) + AND (char_length(p.compiled_truth) + char_length(COALESCE(p.timeline, ''))) < ${threshold} + ${sourceCondition} + ${recencyCondition} + ORDER BY ${orderBy} + LIMIT ${limit} + `; + return rows.map((r: Record) => ({ + slug: String(r.slug), + source_id: String(r.source_id), + title: String(r.title ?? ''), + type: r.type as EnrichCandidate['type'], + body_len: Number(r.body_len ?? 0), + inbound_count: Number(r.inbound_count ?? 0), + })); + } + async findAnomalies(opts: AnomaliesOpts): Promise { const sql = this.sql; const sigma = opts.sigma ?? 3.0; diff --git a/src/core/types.ts b/src/core/types.ts index 59cf88930..d9b0cb3ae 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -431,6 +431,70 @@ export interface SalienceResult { score: number; } +/** + * v0.41.39 (issue #1700) — `gbrain enrich --thin` candidate selection. + * + * One source-aware SQL query enumerates thin (stub) pages of the given + * types, computes a source-correct inbound-link count per page, applies an + * optional re-enrich recency guard, orders by the chosen signal, and slices + * to `limit`. Returns a LIGHTWEIGHT projection (no page bodies) so a 100K- + * page brain doesn't pull every stub body into memory just to rank them. + * + * Why a dedicated engine method instead of composing `listPages` + + * `getBacklinkCounts` in memory: + * - `getBacklinkCounts` groups by bare `slug`, so the same slug in two + * sources collapses/contaminates the count. This query counts inbound + * links per page row (`to_page_id = p.id`), which is source-correct by + * construction. + * - `listPages` returns full `Page` rows (bodies). 500 stub bodies per + * type per source is not a memory guarantee. This projection carries + * only `body_len`, never the body. + */ +export interface EnrichCandidatesOpts { + /** Page types to consider (e.g. ['person', 'company']). Empty → no rows. */ + types: PageType[]; + /** Body-length (chars) below which a page is "thin". */ + thinThreshold: number; + /** Ordering signal. Whitelisted via ENRICH_ORDER_SQL. */ + order: 'inbound-links' | 'updated' | 'salience'; + /** Max rows to return. */ + limit: number; + /** + * Skip pages whose frontmatter `enriched_at` is newer than + * `now - reenrichAfterMs`. Omitted/0 → no recency guard (every thin page + * is eligible). Pages never enriched (no `enriched_at`) are always eligible. + */ + reenrichAfterMs?: number; + /** Single-source scope (canonical scalar form). */ + sourceId?: string; + /** Federated read scope (array form, wins over scalar). */ + sourceIds?: string[]; +} + +/** v0.41.39 — one row per enrich candidate. Lightweight: NO page body. */ +export interface EnrichCandidate { + slug: string; + source_id: string; + title: string; + type: PageType; + /** char_length(compiled_truth) + char_length(timeline). */ + body_len: number; + /** Source-correct inbound-link count (excludes `link_source='mentions'`). */ + inbound_count: number; +} + +/** + * v0.41.39 — whitelisted ORDER BY fragments for EnrichCandidatesOpts.order. + * No SQL-injection risk: callers pass the enum, engines map to these literal + * fragments. Every fragment ends with the (source_id, slug) tiebreaker so + * tied scores produce a deterministic order across engines and runs. + */ +export const ENRICH_ORDER_SQL: Record = { + 'inbound-links': 'inbound_count DESC, p.source_id ASC, p.slug ASC', + 'updated': 'p.updated_at DESC, p.source_id ASC, p.slug ASC', + 'salience': 'p.emotional_weight DESC, inbound_count DESC, p.source_id ASC, p.slug ASC', +}; + /** * v0.29 — Anomaly detection: cohorts (tag, type) with unusually-high activity in a window. * Cohort baseline is computed over `lookback_days` excluding `since`; current count is diff --git a/test/e2e/engine-parity.test.ts b/test/e2e/engine-parity.test.ts index 70faefe85..5ee39d7d8 100644 --- a/test/e2e/engine-parity.test.ts +++ b/test/e2e/engine-parity.test.ts @@ -412,4 +412,50 @@ describeBoth('Engine parity — Postgres vs PGLite', () => { expect(pgFed).toEqual(['people/op-linker-b', 'people/op-orphan-a']); expect(pgliteFed).toEqual(pgFed); }); + + test('v0.41.39 listEnrichCandidates parity (thin filter + source-aware inbound + order)', async () => { + const stub = 'Stub page.'; + const pageSql = ` + INSERT INTO pages (source_id, slug, type, title, compiled_truth, timeline, frontmatter) + VALUES ('default', $1, $2, $3, $4, '', '{}'::jsonb) + ON CONFLICT (source_id, slug) DO NOTHING + `; + for (const eng of [pgEngine, pgliteEngine]) { + // Two thin people (ec-alice ← 2 inbound, ec-bob ← 1), one thin company + // (ec-widget ← 0), one long page (must be excluded by the thin filter). + await eng.executeRaw(pageSql, ['ep/ec-alice', 'person', 'EC Alice', stub]); + await eng.executeRaw(pageSql, ['ep/ec-bob', 'person', 'EC Bob', stub]); + await eng.executeRaw(pageSql, ['companies/ec-widget', 'company', 'EC Widget', stub]); + await eng.executeRaw(pageSql, ['ep/ec-long', 'person', 'EC Long', 'x'.repeat(900)]); + // Linker pages + inbound links (link_source NULL → counted). + await eng.executeRaw(pageSql, ['ep/ec-l1', 'note', 'L1', 'links']); + await eng.executeRaw(pageSql, ['ep/ec-l2', 'note', 'L2', 'links']); + await eng.executeRaw(pageSql, ['ep/ec-l3', 'note', 'L3', 'links']); + await eng.addLink('ep/ec-l1', 'ep/ec-alice', 'ctx a1'); + await eng.addLink('ep/ec-l2', 'ep/ec-alice', 'ctx a2'); + await eng.addLink('ep/ec-l3', 'ep/ec-bob', 'ctx b1'); + } + + const run = async (eng: BrainEngine) => + (await eng.listEnrichCandidates({ + types: ['person', 'company'], + thinThreshold: 400, + order: 'inbound-links', + limit: 10, + sourceId: 'default', + })).filter((c) => c.slug.startsWith('ep/') || c.slug === 'companies/ec-widget'); + + const pg = await run(pgEngine); + const pglite = await run(pgliteEngine); + + const shape = (rows: typeof pg) => rows.map((r) => `${r.slug}:${r.inbound_count}:${r.body_len}`); + expect(shape(pg)).toEqual(shape(pglite)); + + // Concrete contract: long page excluded; ordering alice(2) > bob(1) > widget(0). + const slugs = pg.map((r) => r.slug); + expect(slugs).not.toContain('ep/ec-long'); + expect(slugs.indexOf('ep/ec-alice')).toBeLessThan(slugs.indexOf('ep/ec-bob')); + expect(slugs.indexOf('ep/ec-bob')).toBeLessThan(slugs.indexOf('companies/ec-widget')); + expect(pg.find((r) => r.slug === 'ep/ec-alice')!.inbound_count).toBe(2); + }); });