soulcraft/brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions 2

feat: queryAggregate() + HAVING, plus aggregate backfill, traversal depth/via, extraction typing (BR-ADV-FEATURES-BUN)

Browse source 
New report APIs:
- brain.queryAggregate(name, params) returns the clean AggregateResult[] shape
  ({ groupKey, metrics, count }) directly, instead of the search-Result wrapper that
  find({ aggregate }) returns.
- HAVING: find({ aggregate, having }) and queryAggregate filter groups by computed
  metric values (e.g. revenue > 1000), complementing where (which filters group keys).
  Evaluated per group: O(groups), independent of entity count.

Fixes (all reproducible on Node; surfaced under Cortex+Bun by Memory):
- Aggregate backfill-on-define: defining an aggregate over a store that already holds
  matching entities returned []. It now backfills from existing entities on first query
  (storage-agnostic via getNouns), so it works under durable backends that reopen
  pre-populated. groupBy:['noun'] resolves to the entity type; find({ aggregate }) rows
  expose groupKey/metrics/count at the top level.
- Multi-hop find({ connected: { depth, via } }) honors depth and via at every hop
  (was 1-hop only, with verb filtering applied to hop 1 only); BFS bounded by limit.
- extractEntities no longer bleeds a neighbour's type indicator across candidates; each
  candidate is typed by its own span. Also fixes the "Dr." title pattern.

Adds real-embedding regression tests; full unit suite green.
This commit is contained in:
David Snelling 2026-05-26 13:55:43 -07:00
parent 513186d951
commit 1a98e4276a
9 changed files with 481 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

68
src/aggregation/AggregationIndex.ts
View file

@ -208,6 +208,15 @@ export class AggregationIndex {
/** Track which aggregates have dirty state needing persistence */
private dirty = new Set<string>()
/**
* Aggregates whose state must be backfilled from entities that already existed
* when the aggregate was defined (or whose persisted state was missing/stale at
* init). Write-time hooks only capture entities added *after* a definition, so
* without backfill an aggregate defined over a populated store stays empty.
* Drained by the owner (Brainy) which has the entity iterator; see `getPendingBackfills`.
*/
private needsBackfill = new Set<string>()
/** Track aggregates with stale MIN/MAX (need lazy recompute) */
private staleMinMax = new Map<string, Set<string>>()
@ -243,8 +252,10 @@ export class AggregationIndex {
}
this.states.set(def.name, groupMap)
} else {
// Definition changed or no saved state — start fresh (will be rebuilt)
// Definition changed or no saved state — start fresh and backfill from
// existing entities (the owner drains needsBackfill on first query).
this.states.set(def.name, new Map())
this.needsBackfill.add(def.name)
}
this.definitionHashes.set(def.name, currentHash)
@ -332,9 +343,12 @@ export class AggregationIndex {
this.definitions.set(def.name, def)
this.definitionHashes.set(def.name, newHash)
// Reset state if definition changed or doesn't exist yet
// Reset state if definition changed or doesn't exist yet, and flag it for
// backfill so already-stored entities are counted (write-time hooks only see
// future writes). The owner drains this on the next query via getPendingBackfills().
if (!this.states.has(def.name) || (oldHash && oldHash !== newHash)) {
this.states.set(def.name, new Map())
this.needsBackfill.add(def.name)
}
// Notify native provider of definition (caches compiled form for hot path)
@ -376,6 +390,50 @@ export class AggregationIndex {
return this.definitions.has(name)
}
// ============= Backfill =============
//
// Write-time hooks only capture entities added after a definition exists, so an
// aggregate defined over a populated store would stay empty. The owner (Brainy) has
// the entity iterator, so backfill is driven from there: it reads the pending set,
// clears the aggregate, streams every existing entity through `backfillEntity`, then
// calls `finishBackfill`. Clearing first means a concurrent write that landed via the
// incremental hook is wiped and re-counted exactly once by the rescan.
/** Names of aggregates whose state must be (re)built from existing entities. */
getPendingBackfills(): string[] {
return Array.from(this.needsBackfill)
}
/** Clear an aggregate's state so a full rescan cannot double-count. */
beginBackfill(name: string): void {
this.states.set(name, new Map())
// Reset native provider state for this aggregate too, if present.
const def = this.definitions.get(name)
if (def && this.nativeProvider?.removeAggregate && this.nativeProvider?.defineAggregate) {
this.nativeProvider.removeAggregate(name)
this.nativeProvider.defineAggregate(def)
}
}
/** Feed one already-stored entity into a single aggregate during backfill. */
backfillEntity(name: string, entity: Record<string, unknown>): void {
if (isAggregateEntity(entity)) return
const def = this.definitions.get(name)
if (!def || !matchesSource(entity, def.source)) return
if (this.nativeProvider) {
this.applyNativeResults(name, this.nativeProvider.incrementalUpdate(name, def, entity, 'add'))
} else {
this.addContribution(name, def, entity)
}
}
/** Mark an aggregate's backfill complete; rebuilt state persists on next flush(). */
finishBackfill(name: string): void {
this.needsBackfill.delete(name)
this.dirty.add(name)
}
// ============= Write-Time Hooks =============
/**
@ -549,6 +607,12 @@ export class AggregationIndex {
}
}
// HAVING: filter groups by computed metric values (post-compute, O(groups), before
// sort/pagination). Reuses the where-operator engine over metrics + `count`.
if (params.having && Object.keys(params.having).length > 0) {
if (!matchesMetadataFilter({ ...metrics, count: totalCount } as any, params.having)) continue
}
results.push({
groupKey: { ...group.groupKey },
metrics,

185
src/brainy.ts
View file

@ -2238,6 +2238,35 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
/**
* Query a named aggregate, returning the documented `AggregateResult[]` shape
* (`{ groupKey, metrics, count }`) directly — the report-friendly view.
*
* `find({ aggregate })` returns the same data wrapped as search `Result<T>` rows (with
* `score`/`type`/`entity`) for uniformity with the rest of `find()`; this method is the
* first-class analytics path for dashboards and reports.
*
* @param name - Aggregate name (must be defined via `defineAggregate`)
* @param params - Optional `where` (group-key filter), `having` (metric filter), `orderBy`, `order`, `limit`, `offset`
* @returns Computed group rows
*
* @example
* const rows = await brain.queryAggregate('sales_by_category', { orderBy: 'revenue', order: 'desc', limit: 10 })
* // [{ groupKey: { category: 'food' }, metrics: { revenue: 17.5, count: 2 }, count: 2 }, ...]
*/
async queryAggregate(
name: string,
params?: Omit<AggregateQueryParams, 'name'>
): Promise<AggregateResult[]> {
await this.ensureInitialized()
this.ensureAggregationIndex()
if (!this._aggregationIndex!.hasAggregate(name)) {
throw new Error(`Aggregate '${name}' is not defined. Call defineAggregate() first.`)
}
await this.backfillAggregateIfNeeded(name)
return this._aggregationIndex!.queryAggregate({ name, ...params })
}
/**
* Lazily create the AggregationIndex on first use.
* Checks for a native 'aggregation' provider from plugins.
@ -4479,12 +4508,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* @param options - Extraction options
* @returns Array of extracted entities with types and confidence
*
* Fast heuristic ensemble (pattern + type-embedding + context), not a trained NER —
* each candidate is typed by its own span (no cross-candidate bleed). Confidences are
* approximate; pass `types` to constrain results when precision matters.
*
* @param text - Text to extract entities from
* @param options - Extraction options
* @returns Array of extracted entities with types and confidence
*
* @example
* const entities = await brain.extract('John Smith founded Acme Corp in New York')
* const entities = await brain.extract('Sarah Chen founded Acme Corp')
* // [
* // { text: 'John Smith', type: NounType.Person, confidence: 0.95 },
* // { text: 'Acme Corp', type: NounType.Organization, confidence: 0.92 },
* // { text: 'New York', type: NounType.Location, confidence: 0.88 }
* // { text: 'Sarah Chen', type: NounType.Person, confidence: 0.68 },
* // { text: 'Acme Corp', type: NounType.Organization, confidence: 0.85 }
* // ]
*/
async extract(
@ -6275,7 +6311,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
options?: {
direction?: 'outgoing' | 'incoming' | 'both'
depth?: number
verbType?: VerbType
verbType?: VerbType | VerbType[]
limit?: number
}
): Promise<string[]> {
@ -6283,66 +6319,43 @@ export class Brainy<T = any> implements BrainyInterface<T> {
const direction = options?.direction || 'both'
const limit = options?.limit
const verbTypes = options?.verbType === undefined
? undefined
: new Set(Array.isArray(options.verbType) ? options.verbType : [options.verbType])
// Map our API direction to graphIndex direction
const graphDirection = direction === 'outgoing' ? 'out' :
direction === 'incoming' ? 'in' : 'both'
// Get neighbors from graph index
let neighbors = await this.graphIndex.getNeighbors(entityId, {
direction: graphDirection,
limit
})
let neighbors = await this.getTypedNeighbors(entityId, graphDirection, verbTypes, limit)
// Filter by verb type if specified
if (options?.verbType) {
const filteredNeighbors: string[] = []
const verbIds = direction !== 'incoming'
? await this.graphIndex.getVerbIdsBySource(entityId)
: await this.graphIndex.getVerbIdsByTarget(entityId)
// Load verbs to check their types
const verbs = await this.graphIndex.getVerbsBatchCached(verbIds)
for (const [, verb] of verbs) {
if (verb.type === options.verbType || verb.verb === options.verbType) {
const neighborId = verb.sourceId === entityId ? verb.targetId : verb.sourceId
if (neighbors.includes(neighborId)) {
filteredNeighbors.push(neighborId)
}
}
}
neighbors = filteredNeighbors
}
// Handle depth > 1 (multi-hop traversal)
// Handle depth > 1 (multi-hop traversal). The verb-type filter is applied at EVERY hop
// (previously only the first), and the BFS is bounded by `limit` so a dense graph can't
// expand without limit before the final slice.
if (options?.depth && options.depth > 1) {
const visited = new Set<string>([entityId, ...neighbors])
let currentLevel = neighbors
for (let d = 1; d < options.depth; d++) {
if (limit && neighbors.length >= limit) break
const nextLevel: string[] = []
for (const nodeId of currentLevel) {
const nodeNeighbors = await this.graphIndex.getNeighbors(nodeId, {
direction: graphDirection
})
outer: for (const nodeId of currentLevel) {
const nodeNeighbors = await this.getTypedNeighbors(nodeId, graphDirection, verbTypes)
for (const neighbor of nodeNeighbors) {
if (!visited.has(neighbor)) {
visited.add(neighbor)
nextLevel.push(neighbor)
neighbors.push(neighbor)
if (limit && neighbors.length >= limit) break outer
}
}
}
if (nextLevel.length === 0) break
currentLevel = nextLevel
neighbors.push(...nextLevel)
}
// Apply limit after multi-hop traversal
if (limit && neighbors.length > limit) {
neighbors = neighbors.slice(0, limit)
}
@ -6351,6 +6364,39 @@ export class Brainy<T = any> implements BrainyInterface<T> {
return neighbors
}
/**
* Neighbours of a single node, optionally filtered to a set of verb types.
*
* Extracted so depth traversal can apply the verb-type filter at every hop (not just the
* first). For `direction: 'both'` the verb scan unions out-edges and in-edges so the filter
* isn't silently limited to outgoing edges.
*/
private async getTypedNeighbors(
nodeId: string,
graphDirection: 'in' | 'out' | 'both',
verbTypes?: Set<VerbType>,
limit?: number
): Promise<string[]> {
const neighbors = await this.graphIndex.getNeighbors(nodeId, { direction: graphDirection, limit })
if (!verbTypes || verbTypes.size === 0) return neighbors
// Gather candidate edges in the relevant direction(s).
const verbIds: string[] = []
if (graphDirection !== 'in') verbIds.push(...await this.graphIndex.getVerbIdsBySource(nodeId))
if (graphDirection !== 'out') verbIds.push(...await this.graphIndex.getVerbIdsByTarget(nodeId))
const verbs = await this.graphIndex.getVerbsBatchCached(verbIds)
const neighborSet = new Set(neighbors)
const filtered: string[] = []
for (const [, verb] of verbs) {
if (verbTypes.has(verb.type as VerbType) || verbTypes.has(verb.verb as VerbType)) {
const neighborId = verb.sourceId === nodeId ? verb.targetId : verb.sourceId
if (neighborSet.has(neighborId)) filtered.push(neighborId)
}
}
return filtered
}
/**
* Find semantic duplicates in the database
*
@ -6805,18 +6851,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
const toNeighborDir = (d: 'in' | 'out' | 'both'): 'incoming' | 'outgoing' | 'both' =>
d === 'in' ? 'incoming' : d === 'out' ? 'outgoing' : 'both'
// Collect reachable ids honoring depth + verb-type filter. `via` may be a single VerbType
// or an array; reuse the depth-aware BFS in neighbors() (one pass per requested verb type).
const collect = async (id: string, dir: 'incoming' | 'outgoing' | 'both'): Promise<string[]> => {
const verbTypes: Array<VerbType | undefined> =
via === undefined ? [undefined] : Array.isArray(via) ? via : [via]
const ids = new Set<string>()
for (const verbType of verbTypes) {
const hops = await this.neighbors(id, { direction: dir, depth: depth ?? 1, verbType })
for (const n of hops) ids.add(n)
}
return [...ids]
}
// Collect reachable ids honoring depth + verb-type filter via the depth-aware BFS in
// neighbors(), which takes the whole verb-type set and filters every hop against it — so a
// mixed-verb path (a-[likes]->b-[knows]->c with via:[likes,knows]) is followed correctly.
const collect = (id: string, dir: 'incoming' | 'outgoing' | 'both'): Promise<string[]> =>
this.neighbors(id, { direction: dir, depth: depth ?? 1, verbType: via })
const connectedIds = new Set<string>()
@ -7909,6 +7948,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
aggParams.offset = params.offset
}
// Backfill from already-stored entities if this aggregate was defined over a
// populated store (write-time hooks only capture entities added after define).
await this.backfillAggregateIfNeeded(aggParams.name)
const aggregateResults = this._aggregationIndex.queryAggregate(aggParams)
// Convert AggregateResult[] to Result<T>[] for API consistency
@ -7946,6 +7989,44 @@ export class Brainy<T = any> implements BrainyInterface<T> {
})
}
/**
* Backfill a named aggregate from entities already in storage.
*
* Write-time hooks (`onEntityAdded` etc.) only capture entities added *after* an
* aggregate is defined, so an aggregate defined over a populated store — the common
* case under durable storage, where a brain reopens pre-populated — would otherwise
* return `[]`. On first query we clear the aggregate's state and stream every stored
* noun back through it. Storage-agnostic: `getNouns()` works for in-memory, the
* filesystem adapter, and native (Cortex) storage alike. One-time per definition —
* the rebuilt state is persisted on flush() and reloaded on the next session.
*/
private async backfillAggregateIfNeeded(name: string): Promise<void> {
const index = this._aggregationIndex
if (!index || !index.getPendingBackfills().includes(name)) return
index.beginBackfill(name)
const PAGE = 500
let offset = 0
let cursor: string | undefined
for (;;) {
const page = await this.storage.getNouns({
pagination: cursor ? { limit: PAGE, cursor } : { limit: PAGE, offset }
})
for (const noun of page.items) {
index.backfillEntity(name, noun as unknown as Record<string, unknown>)
}
if (!page.hasMore || page.items.length === 0) break
if (page.nextCursor) {
cursor = page.nextCursor
} else {
offset += page.items.length
}
}
index.finishBackfill(name)
}
/**
* Close and cleanup
*

8
src/coreTypes.ts
View file

@ -279,6 +279,14 @@ export function resolveEntityField(
entity: HNSWNounWithMetadata,
field: string
): unknown {
// `noun` is an alias for the entity's type — some shapes carry `noun`, the
// canonical HNSWNounWithMetadata carries `type`. Resolve it consistently so
// groupBy/source/sort on 'noun' isn't silently null (it otherwise falls through
// to metadata.noun, which doesn't exist). Mirrors matchesSource's `type ?? noun`.
if (field === 'noun') {
const e = entity as unknown as Record<string, unknown>
return e.type ?? e.noun
}
if (STANDARD_ENTITY_FIELDS.has(field)) {
return (entity as unknown as Record<string, unknown>)[field]
}

86
src/neural/signals/PatternSignal.ts
View file

@ -135,7 +135,7 @@ export class PatternSignal {
// Person patterns - SPECIFIC INDICATORS (high confidence)
this.addPatterns(NounType.Person, 0.82, [
/\b(?:Dr|Prof|Mr|Mrs|Ms|Sir|Lady|Lord)\s+[A-Z][a-z]+/, // Titles
/\b(?:Dr|Prof|Mr|Mrs|Ms|Sir|Lady|Lord)\.?\s+[A-Z][a-z]+/, // Titles (optional period: "Dr." / "Dr")
/\b(?:CEO|CTO|CFO|COO|VP|Director|Manager|Engineer|Developer|Designer)\b/i, // Roles
/\b(?:author|creator|founder|inventor|contributor|maintainer)\b/i
])
@ -294,7 +294,8 @@ export class PatternSignal {
return cached
}
// Try regex patterns (primary method)
// Try regex patterns (primary method). The candidate span is authoritative for its
// type; context can only reinforce the same type, never override it (see matchRegexPatterns).
const regexMatch = this.matchRegexPatterns(candidate, context?.definition)
if (regexMatch && regexMatch.confidence >= this.options.minConfidence) {
this.stats.regexMatches++
@ -328,49 +329,76 @@ export class PatternSignal {
}
/**
* Match against pre-compiled regex patterns
* Match against pre-compiled regex patterns.
*
* Checks candidate and optional definition text
* Matches the CANDIDATE SPAN ONLY — never the surrounding context. Folding the
* candidate's own span decides the type; the surrounding `definition`/context may only
* REINFORCE that same type (small boost), never override it with a different one.
*
* This is the fix for cross-candidate type bleed: folding the 30-char context window into
* one match let an indicator in a neighbour's span (e.g. "Corp" in "Sarah Chen founded
* Acme Corp") flip "Sarah Chen" to Organization. Now a different-type context match is
* ignored. When the candidate has no pattern of its own, the context's type is used as a
* fallback — a descriptive cue (a role, "conference", …) still classifies an otherwise
* shapeless candidate, which is the legitimate use of context the ContextSignal complements.
*/
private matchRegexPatterns(
candidate: string,
definition?: string
): TypeSignal | null {
const textToMatch = definition ? `${candidate} ${definition}` : candidate
const matches: Array<{ pattern: CompiledPattern, count: number }> = []
const candidateMatch = this.bestPatternMatch(candidate)
const contextMatch = definition ? this.bestPatternMatch(definition) : null
// Check all patterns
for (const pattern of this.patterns) {
const matchCount = (textToMatch.match(pattern.regex) || []).length
if (matchCount > 0) {
matches.push({ pattern, count: matchCount })
if (candidateMatch) {
const chosen = candidateMatch.pattern
let confidence = Math.min(chosen.confidence, 0.85) // Cap at 0.85
let evidence = `Pattern match: ${chosen.name} (${candidateMatch.count} occurrence${candidateMatch.count > 1 ? 's' : ''})`
// Context of the SAME type reinforces (use the stronger evidence + a small agreement
// boost); a different-type context match is deliberately ignored — no override.
if (contextMatch && contextMatch.pattern.type === chosen.type) {
confidence = Math.min(Math.max(confidence, contextMatch.pattern.confidence) + 0.05, 0.95)
evidence += ' + context'
}
return {
source: 'pattern-regex',
type: chosen.type,
confidence,
evidence,
metadata: { patternName: chosen.name, matchCount: candidateMatch.count }
}
}
if (matches.length === 0) return null
// Candidate has no shape of its own — fall back to the context's descriptive cue.
if (contextMatch) {
const chosen = contextMatch.pattern
return {
source: 'pattern-regex',
type: chosen.type,
confidence: Math.min(chosen.confidence, 0.85),
evidence: `Context pattern match: ${chosen.name} (${contextMatch.count} occurrence${contextMatch.count > 1 ? 's' : ''})`,
metadata: { patternName: chosen.name, matchCount: contextMatch.count }
}
}
// Find best match (highest confidence * match count)
let best = matches[0]
let bestScore = best.pattern.confidence * Math.log(best.count + 1)
return null
}
for (const match of matches.slice(1)) {
const score = match.pattern.confidence * Math.log(match.count + 1)
/** Best pattern match (ranked by confidence × log(count+1)) for a single text span. */
private bestPatternMatch(text: string): { pattern: CompiledPattern, count: number } | null {
let best: { pattern: CompiledPattern, count: number } | null = null
let bestScore = -1
for (const pattern of this.patterns) {
const count = (text.match(pattern.regex) || []).length
if (count === 0) continue
const score = pattern.confidence * Math.log(count + 1)
if (score > bestScore) {
best = match
best = { pattern, count }
bestScore = score
}
}
return {
source: 'pattern-regex',
type: best.pattern.type,
confidence: Math.min(best.pattern.confidence, 0.85), // Cap at 0.85
evidence: `Pattern match: ${best.pattern.name} (${best.count} occurrence${best.count > 1 ? 's' : ''})`,
metadata: {
patternName: best.pattern.name,
matchCount: best.count
}
}
return best
}
/**

7
src/types/brainy.types.ts
View file

@ -783,6 +783,13 @@ export interface AggregateQueryParams {
name: string
/** Filter aggregate groups by their key values */
where?: Record<string, unknown>
/**
* Filter groups by their computed METRIC values (SQL HAVING). Same BFO operators as
* `where`, but applied to the derived metric results plus `count`, e.g.
* `{ revenue: { greaterThan: 1000 } }`. Evaluated per group (O(groups), independent of
* entity count), before sort/pagination.
*/
having?: Record<string, unknown>
/** Sort by metric name or group key field */
orderBy?: string
/** Sort direction */