From c2a84c9d1f9a320cafb6741af9f1fc16c86af4f5 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Sun, 21 Jun 2026 10:22:12 -0700 Subject: [PATCH] feat(8.0): brain.graph.export() + noun-walk cursor + noun visibility hydration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit brain.graph.export() streams the WHOLE graph in one O(N+E) pass — node chunks then edge chunks, async-iterable — the right primitive for visualizing all data (vs. paging per node). Native snapshot-consistent graphCursor when present, else a TS cursor walk over nouns + verbs. Building it surfaced two latent noun bugs, both fixed here (each a correctness win beyond export): - getNounsWithPagination IGNORED its cursor ('offset-based, cursor planned') — the noun mirror of the verb bug fixed in the cursor-pagination commit. Latent because the only multi-page consumer (aggregate backfill) uses one big page; a small chunkSize needed page 2 and, trusting the returned-but-ignored nextCursor, re-fetched page 0 forever (infinite loop). Ported the proven verb cursor: opaque cn1:: token, stable within-shard id order, resume-after, O(N) at any chunk size. (Generalized verbIdFromVectorPath → idFromVectorPath.) - hydrateNounWithMetadata DROPPED 'visibility' (noun mirror of the Fix #1 verb hydration bug) — so getNouns-fed visibility filters saw nothing and leaked system (VFS root) / internal nodes. Now hydrated. - New: GraphApi.export + GraphExportOptions (chunkSize, includeInternal/System, includeNodes/Edges). hydrateNativeSubgraph extracted + shared by subgraph+export. Test: graph-export.test.ts — full-graph completeness incl. isolated nodes, default-hides-internal, node/edge include toggles, chunkSize chunking (the case that exposed the cursor hang). Full gate green. --- src/brainy.ts | 134 ++++++++++++++++-- src/storage/baseStorage.ts | 180 ++++++++++++++++--------- src/types/brainy.types.ts | 33 ++++- tests/unit/brainy/graph-export.test.ts | 97 +++++++++++++ 4 files changed, 369 insertions(+), 75 deletions(-) create mode 100644 tests/unit/brainy/graph-export.test.ts diff --git a/src/brainy.ts b/src/brainy.ts index 590d21f1..74377e29 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -37,7 +37,7 @@ import { setGlobalCache } from './utils/unifiedCache.js' import type { UnifiedCache } from './utils/unifiedCache.js' import { rankIndicesByScore, reorderByIndices } from './utils/resultRanking.js' import { PluginRegistry, isVersionedIndexProvider, isGraphAccelerationProvider } from './plugin.js' -import type { GraphAccelerationProvider, TraverseOptions } from './plugin.js' +import type { GraphAccelerationProvider, TraverseOptions, Subgraph } from './plugin.js' import type { BrainyPlugin, BrainyPluginContext, @@ -93,6 +93,7 @@ import { GraphView, GraphNode, SubgraphOptions, + GraphExportOptions, GetOptions, AddManyParams, RemoveManyParams, @@ -3178,7 +3179,8 @@ export class Brainy implements BrainyInterface { get graph(): GraphApi { return { subgraph: (seeds: string | string[], options?: SubgraphOptions) => - this.graphSubgraph(seeds, options) + this.graphSubgraph(seeds, options), + export: (options?: GraphExportOptions) => this.graphExport(options) } } @@ -3255,7 +3257,12 @@ export class Brainy implements BrainyInterface { frontier = next } - return this.buildGraphView(nodeDepth, [...edgesById.values()], options, truncated) + return this.buildGraphView( + nodeDepth, + [...edgesById.values()], + options?.hydrateNodes !== false, + truncated + ) } /** One frontier hop's incident edges in the requested direction (filters applied by `related()`). */ @@ -3318,16 +3325,26 @@ export class Brainy implements BrainyInterface { } const sub = await accel.traverse(seedInts, traverseOptions) + return this.hydrateNativeSubgraph(sub, options?.hydrateNodes !== false) + } - // Edge hydration: verb ints → ids → Relation objects. + /** + * @description Hydrate a columnar native `Subgraph` into a {@link GraphView}: + * edge verb-ints → `Relation`s (via `verbIntsToIds` + `getVerbsBatchCached`), + * and node ints paired with their depth (position-preserving — `entityIntsToUuids` + * drops deleted ints, which would desync the depth column). Shared by the native + * `subgraph` and `export` paths. + * @param sub - The provider's columnar subgraph / cursor chunk. + * @param hydrateNodes - Whether to batch-load each node's `type` / `subtype`. + * @returns The hydrated {@link GraphView}. + */ + private async hydrateNativeSubgraph(sub: Subgraph, hydrateNodes: boolean): Promise> { const verbIds = (await this.graphIndex.verbIntsToIds(Array.from(sub.edgeVerbInts))).filter( (id): id is string => id !== null ) const verbsMap = await this.graphIndex.getVerbsBatchCached(verbIds) const edges = this.verbsToRelations([...verbsMap.values()]) - // Node hydration: pair each node int with its depth (position-preserving — - // entityIntsToUuids drops deleted ints, which would desync the depth column). const idMapper = this.metadataIndex.getIdMapper() const nodeDepth = new Map() for (let i = 0; i < sub.nodes.length; i++) { @@ -3335,7 +3352,7 @@ export class Brainy implements BrainyInterface { if (uuid !== undefined) nodeDepth.set(uuid, sub.nodeDepth ? sub.nodeDepth[i] : 0) } - return this.buildGraphView(nodeDepth, edges, options, sub.truncated ?? false) + return this.buildGraphView(nodeDepth, edges, hydrateNodes, sub.truncated ?? false) } /** @@ -3346,12 +3363,12 @@ export class Brainy implements BrainyInterface { private async buildGraphView( nodeDepth: Map, edges: Relation[], - options: SubgraphOptions | undefined, + hydrateNodes: boolean, truncated: boolean ): Promise> { const ids = [...nodeDepth.keys()] let entities: Map> | undefined - if (options?.hydrateNodes !== false && ids.length > 0) { + if (hydrateNodes && ids.length > 0) { entities = await this.batchGet(ids) } const nodes: GraphNode[] = ids.map((id) => { @@ -3368,6 +3385,105 @@ export class Brainy implements BrainyInterface { return { nodes, edges, truncated } } + /** + * @description {@link GraphApi.export} dispatch: native snapshot-consistent + * graph cursor when present, else the TS cursor walk over nouns + verbs. Both + * are O(N+E) single passes (cursor pagination — no re-scan). + */ + private async *graphExport(options?: GraphExportOptions): AsyncGenerator> { + await this.ensureInitialized() + const accel = this.graphAccelerationProvider() + if (accel) { + yield* this.graphExportNative(accel, options) + } else { + yield* this.graphExportFallback(options) + } + } + + /** + * @description TS export: stream all nouns (node chunks) then all verbs (edge + * chunks) via cursor pagination — O(N+E), no re-scan. Node refs carry + * `type`/`subtype` straight from the noun records (no extra read). Hidden tiers + * are excluded by default. + */ + private async *graphExportFallback(options?: GraphExportOptions): AsyncGenerator> { + const chunkSize = options?.chunkSize ?? 1000 + const excludedTiers = this.excludedVisibilityTiers(options ?? {}) + const excludedSet = excludedTiers ? new Set(excludedTiers) : null + + if (options?.includeNodes !== false) { + let cursor: string | undefined + let offset = 0 + for (;;) { + const page = await this.storage.getNouns({ + pagination: cursor ? { limit: chunkSize, cursor } : { limit: chunkSize, offset } + }) + const nodes: GraphNode[] = [] + for (const noun of page.items) { + if (excludedSet && noun.visibility && excludedSet.has(noun.visibility)) continue + nodes.push({ + id: noun.id, + type: noun.type, + ...(noun.subtype !== undefined && { subtype: noun.subtype }) + }) + } + if (nodes.length > 0) yield { nodes, edges: [], truncated: false } + if (!page.hasMore || page.items.length === 0) break + if (page.nextCursor) cursor = page.nextCursor + else offset += page.items.length + } + } + + if (options?.includeEdges !== false) { + const filter = excludedTiers + ? { excludeVisibility: excludedTiers as unknown as string[] } + : undefined + let cursor: string | undefined + let offset = 0 + for (;;) { + const page = await this.storage.getVerbs({ + pagination: cursor ? { limit: chunkSize, cursor } : { limit: chunkSize, offset }, + filter + }) + if (page.items.length > 0) { + yield { nodes: [], edges: this.verbsToRelations(page.items), truncated: false } + } + if (!page.hasMore || page.items.length === 0) break + if (page.nextCursor) cursor = page.nextCursor + else offset += page.items.length + } + } + } + + /** + * @description Native export: open a snapshot-consistent graph cursor (pins a + * generation — no dup/skip under concurrent writes), pull light chunks, and + * hydrate each into a {@link GraphView}. The cursor is always closed (even on + * early break / error). Cross-layer tested against the native provider; the TS + * fallback exercises the same chunk shape in CI. + */ + private async *graphExportNative( + accel: GraphAccelerationProvider, + options?: GraphExportOptions + ): AsyncGenerator> { + const excludedTiers = this.excludedVisibilityTiers(options ?? {}) + const handle = await accel.graphCursorOpen({ + direction: 'both', + ...(excludedTiers && { excludeVisibility: excludedTiers as unknown as string[] }), + projection: 'light' + }) + try { + for (;;) { + const chunk = await accel.graphCursorNext(handle, options?.chunkSize ?? 1000) + const view = await this.hydrateNativeSubgraph(chunk.subgraph, true) + if (view.nodes.length > 0 || view.edges.length > 0) yield view + if (chunk.done) break + } + } finally { + await accel.graphCursorClose(handle) + } + } + // ============= SEARCH & DISCOVERY ============= /** diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 2f708a1e..768d1920 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -185,14 +185,14 @@ function getVerbVectorPath(id: string): string { } /** - * @description Extract the verb id embedded in a verb vector path - * (`entities/verbs/{shard}/{id}/vectors.json`). Used by the cursored verb walk - * to order and skip candidates by id WITHOUT reading each file, which is what - * keeps a full cursored pagination O(N) instead of O(N²). - * @param path - A verb vector path (full or prefix-relative; must end with `/vectors.json`). - * @returns The verb id (the path segment immediately before `/vectors.json`). + * @description Extract the entity id embedded in a vector path + * (`entities/{nouns|verbs}/{shard}/{id}/vectors.json`). Used by the cursored + * noun/verb walks to order and skip candidates by id WITHOUT reading each file, + * which is what keeps a full cursored pagination O(N) instead of O(N²). + * @param path - A vector path (full or prefix-relative; must end with `/vectors.json`). + * @returns The entity id (the path segment immediately before `/vectors.json`). */ -function verbIdFromVectorPath(path: string): string { +function idFromVectorPath(path: string): string { const withoutSuffix = path.replace(/\/vectors\.json$/, '') const lastSlash = withoutSuffix.lastIndexOf('/') return lastSlash >= 0 ? withoutSuffix.slice(lastSlash + 1) : withoutSuffix @@ -1100,6 +1100,11 @@ export abstract class BaseStorage extends BaseStorageAdapter { // Standard fields at top-level type: (reserved.noun as NounType) || NounType.Thing, subtype: reserved.subtype as string | undefined, + // visibility is a reserved top-level field (absent === 'public'). Surfacing it + // here lets visibility-aware reads (export node streaming, count/find candidate + // filters) see a noun's tier from getNouns without re-reading metadata — the + // noun mirror of the verb-hydration fix. + visibility: reserved.visibility as HNSWNounWithMetadata['visibility'], createdAt: normalizeStoredTimestamp(reserved.createdAt), updatedAt: normalizeStoredTimestamp(reserved.updatedAt), confidence: reserved.confidence as number | undefined, @@ -1600,7 +1605,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { public async getNounsWithPagination(options: { limit: number offset: number - cursor?: string // Currently ignored (offset-based pagination). Cursor support planned + cursor?: string // Opaque resume token from a prior page's nextCursor; when set it supersedes offset (O(N) walk, no re-scan) filter?: { nounType?: string | string[] service?: string | string[] @@ -1615,56 +1620,67 @@ export abstract class BaseStorage extends BaseStorageAdapter { await this.ensureInitialized() const { limit, offset = 0, filter } = options - const collectedNouns: HNSWNounWithMetadata[] = [] - // Collect ONE item past the requested window so `hasMore` is decidable: - // stopping exactly at offset+limit cannot distinguish "page full, nothing - // after it" from "page full, more behind it" — which made hasMore - // permanently false and silently truncated every multi-page walk - // (audit/migrateField/fillSubtypes, graph verb-id recovery, count rebuilds). - const targetCount = offset + limit - const peekCount = targetCount + 1 + // Cursor (8.0): resume token carrying the (shard, nounId) of the last returned + // noun — the noun mirror of getVerbsWithPagination. When present it supersedes + // `offset` and resumes the shard walk immediately AFTER that position, so a full + // walk is O(N) instead of the O(N²) of offset paging. Malformed/foreign tokens + // decode to null → offset fallback. (Previously the cursor was ignored, which + // was latent — the only multi-page consumer used a single big page — until small + // chunk sizes needed page 2 and an offset-0-on-every-call walk never terminated.) + const cursor = this.decodeNounWalkCursor(options.cursor) + const collected: Array<{ noun: HNSWNounWithMetadata; shard: number }> = [] - // Iterate by shards (0x00-0xFF) instead of types - for (let shard = 0; shard < 256 && collectedNouns.length < peekCount; shard++) { + // Peek one past the window so `hasMore` is decidable. Cursor mode collects one + // page (+1); offset mode keeps the full [0, offset+limit] window (+1). + const peekCount = cursor ? limit + 1 : offset + limit + 1 + const startShard = cursor ? cursor.shard : 0 + + // Iterate by shards (0x00-0xFF), early-terminating at peekCount. + for (let shard = startShard; shard < 256 && collected.length < peekCount; shard++) { const shardHex = shard.toString(16).padStart(2, '0') const shardDir = `entities/nouns/${shardHex}` try { const nounFiles = await this.listCanonicalObjects(shardDir) - for (const nounPath of nounFiles) { - if (collectedNouns.length >= peekCount) break - if (!nounPath.includes('/vectors.json')) continue + // Stable within-shard order (by noun id) so offset windows and cursor resume + // are deterministic; ids come from the path so skipped nouns are never read. + const entries = nounFiles + .filter((p) => p.includes('/vectors.json')) + .map((p) => ({ path: p, id: idFromVectorPath(p) })) + .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)) + + for (const { path: nounPath, id: nounId } of entries) { + if (collected.length >= peekCount) break + // Resume: in the cursor's own shard, skip up to AND INCLUDING the cursor id. + if (cursor && shard === cursor.shard && nounId <= cursor.id) continue try { const noun = await this.readCanonicalObject(nounPath) - if (noun) { - const deserialized = this.deserializeNoun(noun) - const metadata = await this.getNounMetadata(deserialized.id) + if (!noun) continue + const deserialized = this.deserializeNoun(noun) + const metadata = await this.getNounMetadata(deserialized.id) + if (!metadata) continue - if (metadata) { - // Apply type filter - if (filter?.nounType && metadata.noun) { - const types = Array.isArray(filter.nounType) ? filter.nounType : [filter.nounType] - if (!types.includes(metadata.noun)) { - continue - } - } - - // Apply service filter - if (filter?.service) { - const services = Array.isArray(filter.service) ? filter.service : [filter.service] - if (metadata.service && !services.includes(metadata.service)) { - continue - } - } - - // Combine noun + metadata via the canonical hydration helper — - // reserved fields top-level, ONLY custom fields in `metadata` - // (this site previously echoed the full flat record). - collectedNouns.push(this.hydrateNounWithMetadata(deserialized, metadata)) + // Apply type filter + if (filter?.nounType && metadata.noun) { + const types = Array.isArray(filter.nounType) ? filter.nounType : [filter.nounType] + if (!types.includes(metadata.noun)) { + continue } } + + // Apply service filter + if (filter?.service) { + const services = Array.isArray(filter.service) ? filter.service : [filter.service] + if (metadata.service && !services.includes(metadata.service)) { + continue + } + } + + // Combine noun + metadata via the canonical hydration helper — + // reserved fields top-level, ONLY custom fields in `metadata`. + collected.push({ noun: this.hydrateNounWithMetadata(deserialized, metadata), shard }) } catch (error) { // Skip nouns that fail to load } @@ -1674,35 +1690,69 @@ export abstract class BaseStorage extends BaseStorageAdapter { } } - // Apply pagination. The peeked extra item (if any) is dropped by the slice; - // its existence is exactly what makes hasMore true. - const paginatedNouns = collectedNouns.slice(offset, offset + limit) - const hasMore = collectedNouns.length > targetCount + // Window selection. Cursor mode already starts at the resume point (window + // [0, limit)); offset mode slices [offset, offset+limit). The peeked extra + // entry (if any) is dropped — its existence is exactly what makes hasMore true. + const windowStart = cursor ? 0 : offset + const pagePairs = collected.slice(windowStart, windowStart + limit) + const paginatedNouns = pagePairs.map((p) => p.noun) + const hasMore = collected.length > windowStart + limit - // totalCount must be the TRUE dataset total, not the size of this (peeked) - // page. The shard scan early-terminates at `peekCount = offset + limit + 1` - // for memory efficiency, so `collectedNouns.length` only ever reaches the - // page size + 1 — returning it as `totalCount` made every non-empty brain - // look like it held ~`limit` items (e.g. `getNouns({ pagination: { limit: 1 } - // }).totalCount` was 2, tripping the index-rebuild gate's "Small dataset" - // path). For the unfiltered case the authoritative total is the O(1) counter - // maintained on every add/delete and rehydrated on init; `Math.max` guards a - // stale counter from under-reporting below what we collected. A filtered scan - // has no cheap exact total, so it keeps the collected length (a lower bound). + // totalCount must be the TRUE dataset total, not this peeked page. For the + // unfiltered case the authoritative total is the O(1) counter maintained on + // every add/delete (rehydrated on init); `Math.max` guards a stale counter. A + // filtered scan has no cheap exact total, so it keeps the collected length. const totalCount = filter - ? collectedNouns.length - : Math.max(this.totalNounCount, collectedNouns.length) + ? collected.length + : Math.max(this.totalNounCount, collected.length) + + // nextCursor = the (shard, id) of the last RETURNED noun, so the next call + // resumes immediately after it (works for both cursor and offset callers). + let nextCursor: string | undefined = undefined + if (hasMore && pagePairs.length > 0) { + const lastPair = pagePairs[pagePairs.length - 1] + nextCursor = this.encodeNounWalkCursor(lastPair.shard, lastPair.noun.id) + } return { items: paginatedNouns, totalCount, hasMore, - nextCursor: hasMore && paginatedNouns.length > 0 - ? paginatedNouns[paginatedNouns.length - 1].id - : undefined + nextCursor } } + /** + * @description Encode a noun-walk resume cursor — the `(shard, nounId)` of the + * last returned noun — as an opaque, version-tagged token (`cn1:` prefix lets + * {@link decodeNounWalkCursor} reject foreign tokens, e.g. a bare-id cursor). + * `nounId` is placed last and the decoder re-joins on `:` so any id format survives. + * @param shard - The shard (0–255) the noun lives in. + * @param id - The noun id. + * @returns The opaque cursor token. + */ + private encodeNounWalkCursor(shard: number, id: string): string { + return `cn1:${shard}:${id}` + } + + /** + * @description Decode a noun-walk cursor from {@link encodeNounWalkCursor}; + * returns `null` for an absent / malformed / foreign token (caller falls back + * to offset paging rather than mis-resuming). + * @param cursor - The opaque cursor token, or undefined. + * @returns `{ shard, id }` resume position, or `null`. + */ + private decodeNounWalkCursor(cursor?: string): { shard: number; id: string } | null { + if (!cursor) return null + const parts = cursor.split(':') + if (parts.length < 3 || parts[0] !== 'cn1') return null + const shard = Number(parts[1]) + if (!Number.isInteger(shard) || shard < 0 || shard > 255) return null + const id = parts.slice(2).join(':') + if (id.length === 0) return null + return { shard, id } + } + /** * Get verbs with pagination (Type-first implementation with billion-scale optimizations) * @@ -1799,7 +1849,7 @@ export abstract class BaseStorage extends BaseStorageAdapter { // verbs skipped by the cursor are never read. const entries = verbFiles .filter((p) => p.includes('/vectors.json')) - .map((p) => ({ path: p, id: verbIdFromVectorPath(p) })) + .map((p) => ({ path: p, id: idFromVectorPath(p) })) .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)) for (const { path: verbPath, id: verbId } of entries) { diff --git a/src/types/brainy.types.ts b/src/types/brainy.types.ts index 7d8debdc..d8eaa94a 100644 --- a/src/types/brainy.types.ts +++ b/src/types/brainy.types.ts @@ -821,13 +821,29 @@ export interface SubgraphOptions { hydrateNodes?: boolean } +/** + * Options for {@link GraphApi.export}. + */ +export interface GraphExportOptions { + /** Nodes/edges per streamed chunk (default `1000`). */ + chunkSize?: number + /** Also include `internal`-visibility nodes/edges (hidden by default). */ + includeInternal?: boolean + /** Also include `system`-visibility nodes/edges (hidden by default). */ + includeSystem?: boolean + /** Stream the nodes (default `true`; set `false` to stream only edges). */ + includeNodes?: boolean + /** Stream the edges (default `true`; set `false` to stream only nodes). */ + includeEdges?: boolean +} + /** * The `brain.graph` namespace — graph-shaped reads over the knowledge graph. * Routes to a native {@link import('../plugin.js').GraphAccelerationProvider} * when one is registered, otherwise serves the same results from Brainy's * pure-TS adjacency (correct at small/medium scale). * - * (Grows with the engine: `subgraph` ships first; `export` / `rank` / `path` / + * (Grows with the engine: `subgraph` + `export` ship first; `rank` / `path` / * `communities` follow.) */ export interface GraphApi { @@ -843,6 +859,21 @@ export interface GraphApi { * // view.nodes = the 2-hop neighborhood; view.edges = the relations among them */ subgraph(seeds: string | string[], options?: SubgraphOptions): Promise> + + /** + * @description Stream the WHOLE graph in one O(N+E) pass as a sequence of + * {@link GraphView} chunks — the right primitive for visualizing all data + * (vs. paging per node). Node chunks come first, then edge chunks; assemble + * them on the consumer side. Backed by cursor pagination, so a full walk is + * O(N+E) at any chunk size (no re-scan). + * @param options - Chunk size, visibility, and node/edge inclusion toggles. + * @returns An async-iterable of graph chunks. + * @example + * for await (const chunk of brain.graph.export()) { + * addNodes(chunk.nodes); addEdges(chunk.edges) + * } + */ + export(options?: GraphExportOptions): AsyncIterable> } // ============= Batch Operations ============= diff --git a/tests/unit/brainy/graph-export.test.ts b/tests/unit/brainy/graph-export.test.ts new file mode 100644 index 00000000..73d66c6a --- /dev/null +++ b/tests/unit/brainy/graph-export.test.ts @@ -0,0 +1,97 @@ +/** + * @module tests/unit/brainy/graph-export + * @description Graph engine #22: `brain.graph.export()` — stream the whole graph + * in one O(N+E) pass (the right primitive for visualizing all data, vs. paging + * per node). Exercises the pure-TS fallback (no native provider in CI), which + * streams all nouns then all verbs via cursor pagination. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { Brainy } from '../../../src/index.js' +import { NounType, VerbType } from '../../../src/types/graphTypes.js' +import { createTestConfig } from '../../helpers/test-factory.js' +import type { GraphView } from '../../../src/index.js' + +async function collect(stream: AsyncIterable): Promise<{ + nodes: Map + edges: Map + chunks: number +}> { + const nodes = new Map() + const edges = new Map() + let chunks = 0 + for await (const chunk of stream) { + chunks++ + for (const n of chunk.nodes) nodes.set(n.id, n) + for (const e of chunk.edges) edges.set(e.id, e) + } + return { nodes, edges, chunks } +} + +describe('brain.graph.export() (graph engine #22)', () => { + let brain: Brainy + let a: string, b: string, c: string, iso: string + + beforeEach(async () => { + brain = new Brainy(createTestConfig()) + await brain.init() + a = await brain.add({ type: NounType.Person, subtype: 'employee', data: 'A' }) + b = await brain.add({ type: NounType.Person, subtype: 'employee', data: 'B' }) + c = await brain.add({ type: NounType.Project, subtype: 'milestone', data: 'C' }) + iso = await brain.add({ type: NounType.Concept, subtype: 'general', data: 'isolated' }) + await brain.relate({ from: a, to: b, type: VerbType.RelatedTo, subtype: 'colleague' }) + await brain.relate({ from: b, to: c, type: VerbType.ParticipatesIn, subtype: 'assignment' }) + }) + + afterEach(async () => { + await brain.close() + }) + + it('streams the WHOLE graph — every node (incl. isolated) and every edge', async () => { + const { nodes, edges } = await collect(brain.graph.export()) + // All four nouns, including the isolated one (the node stream catches it). + expect(new Set(nodes.keys())).toEqual(new Set([a, b, c, iso])) + // Node refs carry type/subtype straight from the noun records. + expect(nodes.get(c)?.type).toBe(NounType.Project) + expect(nodes.get(c)?.subtype).toBe('milestone') + // Both edges. + const pairs = [...edges.values()].map((e) => `${e.from}->${e.to}`).sort() + expect(pairs).toEqual([`${a}->${b}`, `${b}->${c}`].sort()) + }) + + it('excludes internal-visibility nodes/edges by default; includeInternal surfaces them', async () => { + const hidden = await brain.add({ + type: NounType.Person, + subtype: 'employee', + data: 'hidden', + visibility: 'internal' + }) + await brain.relate({ from: a, to: hidden, type: VerbType.RelatedTo, subtype: 'colleague', visibility: 'internal' }) + + const visible = await collect(brain.graph.export()) + expect(visible.nodes.has(hidden)).toBe(false) + expect([...visible.edges.values()].some((e) => e.to === hidden)).toBe(false) + + const all = await collect(brain.graph.export({ includeInternal: true })) + expect(all.nodes.has(hidden)).toBe(true) + expect([...all.edges.values()].some((e) => e.to === hidden)).toBe(true) + }) + + it('includeNodes:false streams only edges; includeEdges:false streams only nodes', async () => { + const edgesOnly = await collect(brain.graph.export({ includeNodes: false })) + expect(edgesOnly.nodes.size).toBe(0) + expect(edgesOnly.edges.size).toBe(2) + + const nodesOnly = await collect(brain.graph.export({ includeEdges: false })) + expect(nodesOnly.edges.size).toBe(0) + expect(nodesOnly.nodes.size).toBe(4) + }) + + it('chunks the stream by chunkSize (multiple passes, same total)', async () => { + const { nodes, edges, chunks } = await collect(brain.graph.export({ chunkSize: 1 })) + // 4 nodes + 2 edges at 1 per chunk → several chunks. + expect(chunks).toBeGreaterThan(1) + expect(nodes.size).toBe(4) + expect(edges.size).toBe(2) + }) +})