/** * @module db/portableGraph * @description Portable graph export/import engine for Brainy 8.0 — the * `PortableGraph v1` document format plus the `export`/`import` engines that power * `db.export()` / `brain.export()` (read, at a pinned generation) and * `brain.import(graph)` (write, applied as one atomic transaction). * * A `PortableGraph` is a self-describing, versioned, **portable** representation * of a graph (entities + relations, optionally vectors and file blobs) — the unit * Brainy exports for interchange between instances, versions, and products, and the * payload other artifacts embed (e.g. a workbench file's `graph` field). It is NOT * a backup: that role belongs to `db.persist()`/`Brainy.load()`, the native * whole-brain snapshot which preserves generation history. `PortableGraph` is the * human-readable, partial-or-whole, cross-version round-trip; distinct also from * `brain.import(file)` (foreign-file ingestion). * * Every document carries `format: 'brainy-portable-graph'` and `formatVersion: 1`; * import gates on `formatVersion` for forward migration. The format is shared * verbatim across the 7.x and 8.0 lines. * * **Reserved-field split:** each `PortableGraphEntity` carries Brainy's standard fields * (`subtype`, `visibility`, `data`, `confidence`, `weight`, `service`, `createdBy`, * `createdAt`) at the TOP LEVEL and `metadata` holds ONLY custom user fields — * mirroring the in-memory `Entity`, so import maps each to its dedicated `add`/ * `relate` parameter rather than the metadata bag. */ import { Entity, Relation, Result } from '../types/brainy.types.js' import { NounType, VerbType } from '../types/graphTypes.js' import { StorageAdapter, HNSWVerbWithMetadata } from '../coreTypes.js' import { getBrainyVersion } from '../utils/version.js' import { TxOperation } from './types.js' /** Fixed entity id of the VFS root collection (a `system` entity; excluded unless `includeSystem`). */ const VFS_ROOT_ID = '00000000-0000-0000-0000-000000000000' /** Magic string identifying a Brainy `PortableGraph` document (the `format` tag). */ export const PORTABLE_GRAPH_FORMAT = 'brainy-portable-graph' /** Current portable-format version (import gates on this for cross-version migration). */ export const PORTABLE_GRAPH_FORMAT_VERSION = 1 /** Default embedding model label (informational; `dimensions` is the real compat gate). */ const DEFAULT_EMBED_MODEL = 'all-MiniLM-L6-v2' /** Page size for find-based enumeration (whole-brain / predicate selectors). */ const ENUM_PAGE = 1000 /** Per-node relation fetch ceiling (source/target-filtered, so no unfiltered-scan warning). */ const RELATION_FETCH_LIMIT = 100_000 // ============================================================================ // Public types — identical wire shape to the 7.x line // ============================================================================ /** * @description Selects WHICH part of the graph to export. Reuses `find()`'s grammar * plus export-specific selectors. Omit (or pass `{}`) to export the whole brain. * Structural selectors (`ids`/`collection`/`connected`/`vfsPath`) and predicate * selectors (`type`/`subtype`/`where`/`service`/`visibility`) **compose**. */ export interface ExportSelector { /** Exactly these entity ids. */ ids?: string[] /** A collection id → the collection + its transitive `Contains` members. */ collection?: string /** Alias for `collection`. */ memberOf?: string /** An entity + its N-hop neighbourhood. */ connected?: { from: string depth?: number verbs?: VerbType[] direction?: 'out' | 'in' | 'both' } /** A VFS path → the directory/file + (for a directory) its `Contains` subtree. */ vfsPath?: string /** For `vfsPath` directories: include the whole subtree (default: true). */ recursive?: boolean /** For `collection`/`vfsPath`: cap traversal depth (default: unbounded). */ depth?: number /** Predicate: entity type(s). */ type?: NounType | NounType[] /** Predicate: entity subtype(s). */ subtype?: string | string[] /** Predicate: exact-match metadata fields. */ where?: Record /** Predicate: multi-tenancy service id. */ service?: string /** Predicate: visibility (`'public'` matches entities with no explicit visibility). */ visibility?: string } /** Controls HOW the selected graph is serialized. */ export interface ExportOptions { /** Include embedding vectors verbatim (default: false → `import()` re-embeds from `data`). */ includeVectors?: boolean /** Include VFS file bytes in `blobs` so files round-trip byte-identically (default: false). */ includeContent?: boolean /** Include `visibility:'system'` entities (e.g. the VFS root) (default: false). */ includeSystem?: boolean /** Which edges to include (default: `'induced'`). */ edges?: 'induced' | 'incident' | 'none' /** * How the whole-brain / predicate selector (no `ids`/`collection`/`connected`/ * `vfsPath`) resolves its candidate id set: * * - `'index'` (DEFAULT — unchanged behavior) — the generation-correct * paginated `find()` walk. Fast (O(matches), not O(N)) but rides the * metadata index as an acceleration structure: a lost/stale index posting * can silently OMIT a canonical record, and a stale posting pointing at a * record that no longer matches can silently produce a phantom (dropped * later by the same predicate re-check `'canonical'` mode also runs, so * phantoms never reach `entities` — but they ARE lost silently unless * {@link reportIndexDrift} is set). * - `'canonical'` — walks every live noun/verb directly off the storage * adapter's canonical shard layout (`storage.getNouns()`/`getVerbs()` — * the same primitive `repairIndex()`'s recount and every index-heal walk * use), then applies the selector as a plain predicate over the walked * records. This GUARANTEES canon-completeness — the metadata/graph * indexes are never consulted, so their corruption cannot hide a live * record — at the cost of an O(N) walk regardless of selector * selectivity (unlike the index path's O(matches)). Structural selectors * (`ids`/`collection`/`connected`/`vfsPath`) resolve their node set * exactly as in `'index'` mode either way (they never rode the metadata * index); `'canonical'` additionally walks relations canonically for * EVERY selector, since a lost adjacency-index posting can hide a * relation regardless of how the node set was produced. Requires a * storage adapter and the CURRENT generation — throws on a historical * `asOf()` view or a speculative `with()` overlay (see * {@link CanonicalEnumerationUnavailableError}), because the canonical * walk has no notion of "as of a past generation." */ enumeration?: 'index' | 'canonical' /** * Only meaningful with `enumeration:'canonical'` (ignored otherwise): ALSO * run the `'index'` enumeration in parallel and diff it against the * canonical ground truth, attaching the result as {@link PortableGraph.drift}. * Never auto-heals anything — this is migration-audit evidence, reported * loudly (`console.warn` with the counts) whenever either list is non-empty, * never silently. Default: false. */ reportIndexDrift?: boolean } /** * @description Index-vs-canonical drift for one `export({ enumeration: 'canonical', * reportIndexDrift: true })` call. Only populated for the whole-brain / predicate * selector (structural selectors never consulted the metadata index for their node * set, so there is nothing to diff — both lists are empty for those). */ export interface ExportIndexDrift { /** * Ids the canonical storage walk confirmed (live, selector-matching) that the * index-based `find()` enumeration did NOT return — canonical records the * metadata index has lost track of. */ canonicalOnly: string[] /** * Ids the index-based `find()` enumeration returned for this selector that * canonical ground truth (the storage walk + the same predicate check) does * NOT support — phantom index rows (stale or cross-bucket postings). */ indexOnly: string[] } /** Controls how a `PortableGraph` is applied on `import()`. */ export interface ImportOptions { /** Conflict policy for an existing id (default: `'merge'` — dedup-by-id). */ onConflict?: 'merge' | 'replace' | 'skip' /** Vector policy (default: `'auto'` — re-embed from `data` when no vector is carried). */ reembed?: 'auto' | 'never' /** Rewrite every id on the way in (e.g. to clone a template subgraph under fresh ids). */ remapIds?: (id: string) => string /** Transaction metadata, recorded in the tx-log alongside the new generation. */ meta?: Record } /** One entity in a `PortableGraph`. Standard fields top-level; `metadata` is custom-only. */ export interface PortableGraphEntity { id: string type: string subtype?: string visibility?: string data?: any confidence?: number weight?: number service?: string createdBy?: any createdAt?: number vector?: number[] metadata?: any } /** One relation (edge) in a `PortableGraph`. */ export interface PortableGraphRelation { id: string from: string to: string type: string subtype?: string visibility?: string weight?: number confidence?: number metadata?: any } /** A self-describing, versioned, portable graph document (identical on 7.x and 8.0). */ export interface PortableGraph { format: typeof PORTABLE_GRAPH_FORMAT formatVersion: number brainyVersion: string createdAt: string embedding: { model: string; dimensions: number } selector?: ExportSelector entities: PortableGraphEntity[] relations: PortableGraphRelation[] blobs?: Record danglingIds?: string[] /** Present only when `export()` was called with `reportIndexDrift: true`. */ drift?: ExportIndexDrift stats: { entityCount: number; relationCount: number; blobCount: number; vectorDimensions?: number } } /** Outcome of an `import()`. */ export interface ImportResult { imported: number merged: number skipped: number reembedded: number blobsWritten: number errors: Array<{ id: string; error: string }> } /** The minimal read surface the export engine needs — satisfied by `Db` and `Brainy`. */ export interface PortableGraphReader { get(id: string, options?: { includeVectors?: boolean }): Promise | null> find(query: any): Promise[]> related(paramsOrId?: any): Promise[]> } /** The minimal write surface the import engine needs — satisfied by `Brainy`. */ export interface PortableGraphWriter { get(id: string, options?: { includeVectors?: boolean }): Promise | null> transact(ops: TxOperation[], options?: { meta?: Record }): Promise } /** Type guard: is this value a Brainy `PortableGraph` document (vs a file/foreign object)? */ export function isPortableGraph(value: unknown): value is PortableGraph { return ( typeof value === 'object' && value !== null && (value as any).format === PORTABLE_GRAPH_FORMAT && Array.isArray((value as any).entities) ) } /** Result of {@link validatePortableGraph}. `valid` is true iff `errors` is empty. */ export interface PortableGraphValidation { valid: boolean errors: string[] warnings: string[] } /** * @description Dry-run check a value before `import()`, without mutating the brain: * structural validity, a supported `formatVersion`, entity-id uniqueness + required * fields, and relation-endpoint coverage. `import()` throws on a non-`PortableGraph` or a * too-new `formatVersion`; `validatePortableGraph()` lets a consumer (e.g. a serializer checking * a `.wbench` graph payload) surface issues first. * @param data - The value to validate (need not be a `PortableGraph`). * @returns `{ valid, errors, warnings }`. * @example * const { valid, errors } = validatePortableGraph(payload) * if (!valid) throw new Error(`invalid PortableGraph: ${errors.join('; ')}`) * await brain.import(payload) */ export function validatePortableGraph(data: unknown): PortableGraphValidation { const errors: string[] = [] const warnings: string[] = [] if (!isPortableGraph(data)) { errors.push(`not a PortableGraph document (expected format: '${PORTABLE_GRAPH_FORMAT}')`) return { valid: false, errors, warnings } } if (typeof data.formatVersion !== 'number') { errors.push('formatVersion is missing or not a number') } else if (data.formatVersion > PORTABLE_GRAPH_FORMAT_VERSION) { errors.push( `formatVersion ${data.formatVersion} is newer than supported (max ${PORTABLE_GRAPH_FORMAT_VERSION})` ) } const ids = new Set() for (const e of data.entities) { if (!e || typeof e.id !== 'string' || e.id.length === 0) { errors.push('an entity is missing an id') continue } if (ids.has(e.id)) errors.push(`duplicate entity id: ${e.id}`) ids.add(e.id) if (e.type === undefined || e.type === null || (e.type as string) === '') { errors.push(`entity ${e.id} is missing a type`) } } const dangling = new Set(data.danglingIds ?? []) for (const r of data.relations) { if (!r || typeof r.id !== 'string') { errors.push('a relation is missing an id') continue } if (r.type === undefined || r.type === null || (r.type as string) === '') { errors.push(`relation ${r.id} is missing a type`) } if (!r.from || !r.to) { errors.push(`relation ${r.id} is missing from/to`) continue } if (!ids.has(r.from) && !dangling.has(r.from)) { warnings.push(`relation ${r.id}: from-endpoint ${r.from} is not in entities`) } if (!ids.has(r.to) && !dangling.has(r.to)) { warnings.push(`relation ${r.id}: to-endpoint ${r.to} is not in entities`) } } if (!data.embedding || typeof data.embedding.dimensions !== 'number') { warnings.push('embedding manifest is missing or has no dimensions') } return { valid: errors.length === 0, errors, warnings } } // ============================================================================ // EXPORT // ============================================================================ /** * @description Serialize part or all of a graph (read through `reader` at its pinned * generation) into a portable `PortableGraph` document. * * `enumeration:'canonical'` (see {@link ExportOptions.enumeration}) requires a * storage adapter and the current generation: it throws * {@link CanonicalEnumerationUnavailableError} if `storage` is absent, and the * caller (`Db.export()`) throws the same error before this runs if the view is * historical or a speculative overlay — the canonical storage walk has no * generation parameter, so it can only ever answer "as of right now." * * @param reader - Generation-correct read surface (`Db` or `Brainy`). * @param storage - Storage adapter (VFS blob bytes when `includeContent`; the * canonical noun/verb walk when `enumeration:'canonical'`). * @param selector - WHAT to export (omit for the whole brain). * @param options - HOW to export (vectors / file bytes / edge policy / enumeration mode). */ export async function exportGraph( reader: PortableGraphReader, storage: StorageAdapter | undefined, selector: ExportSelector, options: ExportOptions ): Promise { const { includeVectors = false, includeContent = false, includeSystem = false, edges = 'induced', enumeration = 'index', reportIndexDrift = false } = options if (enumeration === 'canonical' && !storage) { throw new Error( `export(): enumeration:'canonical' requires a storage adapter, but none was supplied ` + `to this reader. Use enumeration:'index' (the default), or export through a Db/Brainy ` + `that carries its storage adapter.` ) } const wantDrift = enumeration === 'canonical' && reportIndexDrift // 1. Resolve the node-id set (+ the index's raw candidate set, only when diffing it). const { idSet, indexCandidateIds } = await resolveSelector( reader, storage, selector, includeSystem, enumeration, wantDrift ) // 2. Read canonical entities (reserved fields top-level), applying any predicate filter. // Identical for both enumeration modes: 'canonical' only changes WHICH ids reach this // loop, never how a candidate is verified — so the two modes can disagree on candidacy, // never on what counts as a match. const usePredicate = hasPredicate(selector) const entityMap = new Map>() const entities: PortableGraphEntity[] = [] for (const id of idSet) { const e = await reader.get(id, { includeVectors }) if (!e) continue if (!includeSystem && (e as any).visibility === 'system') continue if (usePredicate && !matchesPredicate(e, selector)) continue entityMap.set(id, e) entities.push(toPortableGraphEntity(e, includeVectors)) } const keptIds = new Set(entityMap.keys()) // 2b. Finalize the drift report now that ground truth (keptIds) is known. let drift: ExportIndexDrift | undefined if (wantDrift) { const indexIds = indexCandidateIds ?? new Set() const canonicalOnly = [...keptIds].filter((id) => !indexIds.has(id)) const indexOnly = [...indexIds].filter((id) => !keptIds.has(id)) drift = { canonicalOnly, indexOnly } if (canonicalOnly.length > 0 || indexOnly.length > 0) { console.warn( `[Brainy] export() index drift: ${canonicalOnly.length} canonical-only id(s) ` + `(canon-present, the index-based enumeration missed them) and ${indexOnly.length} ` + `index-only id(s) (index-visible, canon-absent — phantom rows). ` + `See the returned PortableGraph's 'drift' field for the exact ids. Nothing was ` + `auto-healed — run brain.repairIndex() to reconcile the metadata index.` ) } } // 3. Edges per policy. Canonical mode ALSO walks verbs canonically for every // selector (not just whole-brain) — a lost adjacency-index posting can hide a // relation regardless of how the node set was produced. const { relations, danglingIds } = enumeration === 'canonical' ? await collectEdgesCanonical(storage!, keptIds, edges) : await collectEdges(reader, keptIds, edges) // 4. VFS blob bytes (only when requested). let blobs: Record | undefined if (includeContent) blobs = await collectBlobs(storage, entityMap) const blobCount = blobs ? Object.keys(blobs).length : 0 // Embedding dimensionality for the manifest: detect from a carried vector, // else the default model's 384 (informational — `dimensions` is the compat gate). const dimensions = entities.find((e) => e.vector && e.vector.length)?.vector?.length ?? 384 return { format: PORTABLE_GRAPH_FORMAT, formatVersion: PORTABLE_GRAPH_FORMAT_VERSION, brainyVersion: getBrainyVersion(), createdAt: new Date().toISOString(), embedding: { model: DEFAULT_EMBED_MODEL, dimensions }, selector, entities, relations, ...(blobs && blobCount > 0 ? { blobs } : {}), ...(danglingIds && danglingIds.length > 0 ? { danglingIds } : {}), ...(drift ? { drift } : {}), stats: { entityCount: entities.length, relationCount: relations.length, blobCount, vectorDimensions: dimensions } } } // ============================================================================ // IMPORT // ============================================================================ /** * @description Apply a `PortableGraph` to the brain as ONE atomic transaction (a single * new generation). Dedup-by-id merge by default; re-embeds from `data` when no vector * is carried. VFS blob bytes are written first (content-addressed, idempotent). * @param writer - The brain (`get` + `transact`). * @param storage - Storage adapter (for VFS blob bytes). * @param data - A `PortableGraph` document. * @param options - Conflict / vector / id-remap policy. * @throws If `data` is not a `PortableGraph`, or its `formatVersion` is newer than supported. */ export async function importGraph( writer: PortableGraphWriter, storage: StorageAdapter | undefined, data: PortableGraph, options: ImportOptions = {} ): Promise { if (!isPortableGraph(data)) { throw new Error( `import() expects a PortableGraph document (format:'${PORTABLE_GRAPH_FORMAT}'). ` + `For file ingestion (CSV/PDF/Excel/JSON), pass a file/buffer instead.` ) } if (typeof data.formatVersion === 'number' && data.formatVersion > PORTABLE_GRAPH_FORMAT_VERSION) { throw new Error( `PortableGraph formatVersion ${data.formatVersion} is newer than this Brainy supports ` + `(max ${PORTABLE_GRAPH_FORMAT_VERSION}). Upgrade Brainy to import this document.` ) } const { onConflict = 'merge', reembed = 'auto', remapIds, meta } = options const result: ImportResult = { imported: 0, merged: 0, skipped: 0, reembedded: 0, blobsWritten: 0, errors: [] } const mapId = (id: string) => (remapIds ? remapIds(id) : id) // 1. Blob bytes first (content-addressed → idempotent), so file entities resolve content. if (data.blobs && Object.keys(data.blobs).length > 0) { const blobStorage = (storage as any)?.blobStorage if (blobStorage?.write) { for (const [hash, b64] of Object.entries(data.blobs)) { try { await blobStorage.write(Buffer.from(b64, 'base64')) result.blobsWritten++ } catch (e) { result.errors.push({ id: hash, error: `blob: ${(e as Error).message}` }) } } } else { for (const hash of Object.keys(data.blobs)) { result.errors.push({ id: hash, error: 'blob: storage does not support binary blobs' }) } } } // 2. Build the operation batch (one atomic transaction → one generation). const ops: TxOperation[] = [] for (const be of data.entities || []) { const id = mapId(be.id) try { const exists = (await writer.get(id)) !== null if (exists) { if (onConflict === 'skip') { result.skipped++ continue } if (onConflict === 'merge') { ops.push({ op: 'update', id, ...entityUpdateFields(be), merge: true } as TxOperation) result.merged++ continue } ops.push({ op: 'remove', id } as TxOperation) // 'replace' } const useVector = Array.isArray(be.vector) && be.vector.length > 0 if (!useVector && reembed === 'never') { result.errors.push({ id, error: 'no vector carried and reembed:never' }) continue } ops.push({ op: 'add', id, ...entityAddFields(be), ...(useVector ? { vector: be.vector } : {}) } as TxOperation) result.imported++ if (!useVector) result.reembedded++ } catch (e) { result.errors.push({ id, error: (e as Error).message }) } } for (const br of data.relations || []) { ops.push({ op: 'relate', from: mapId(br.from), to: mapId(br.to), type: br.type as VerbType, ...(br.subtype !== undefined ? { subtype: br.subtype } : {}), ...(br.weight !== undefined ? { weight: br.weight } : {}), ...(br.confidence !== undefined ? { confidence: br.confidence } : {}), ...(br.metadata !== undefined ? { metadata: br.metadata } : {}) } as TxOperation) } // 3. Apply atomically — one generation for the whole graph, or none. if (ops.length > 0) { await writer.transact(ops, meta ? { meta } : undefined) } return result } // ============================================================================ // Selector resolution (private) // ============================================================================ function hasStructural(s: ExportSelector): boolean { return !!(s.ids || s.collection || s.memberOf || s.connected || s.vfsPath) } function hasPredicate(s: ExportSelector): boolean { return ( s.type !== undefined || s.subtype !== undefined || s.where !== undefined || s.service !== undefined || s.visibility !== undefined ) } /** * @param reader - Generation-correct read surface. * @param storage - Storage adapter (only touched when `enumeration:'canonical'` * resolves the whole-brain/predicate branch). * @param s - The export selector. * @param includeSystem - Whether `visibility:'system'` entities are wanted. * @param enumeration - `'index'` (default) or `'canonical'` — see {@link ExportOptions.enumeration}. * Only affects the whole-brain/predicate branch (the `else` below): structural * selectors (`ids`/`collection`/`connected`/`vfsPath`) never rode the metadata * index for their node set, so they resolve identically either way. * @param wantIndexCandidates - When true (only meaningful with `enumeration:'canonical'` * on the whole-brain/predicate branch), ALSO run the index-based walk and return * its raw candidate set as `indexCandidateIds`, for {@link ExportIndexDrift}. */ async function resolveSelector( reader: PortableGraphReader, storage: StorageAdapter | undefined, s: ExportSelector, includeSystem: boolean, enumeration: 'index' | 'canonical', wantIndexCandidates: boolean ): Promise<{ idSet: Set; indexCandidateIds?: Set }> { let idSet: Set let indexCandidateIds: Set | undefined if (s.ids && s.ids.length) { idSet = new Set(s.ids) } else if (s.collection ?? s.memberOf) { idSet = await resolveCollectionSubtree(reader, (s.collection ?? s.memberOf)!, s.depth) } else if (s.connected) { idSet = await resolveConnected(reader, s.connected) } else if (s.vfsPath) { idSet = await resolveVfsPath(reader, s.vfsPath, s.recursive ?? true, s.depth) } else if (enumeration === 'canonical') { idSet = await enumerateAllCanonical(storage!) if (wantIndexCandidates) indexCandidateIds = await enumerateAllIndexed(reader, s) } else { idSet = await enumerateAllIndexed(reader, s) } if (!includeSystem) idSet.delete(VFS_ROOT_ID) return { idSet, indexCandidateIds } } /** * @description Whole-brain / predicate enumeration via generation-correct * paginated `find()`. The metadata index is an acceleration structure over * this candidate set — see {@link enumerateAllCanonical} for the storage-level * counterpart that never consults it. */ async function enumerateAllIndexed(reader: PortableGraphReader, s: ExportSelector): Promise> { const params: any = {} if (s.type !== undefined) params.type = s.type if (s.subtype !== undefined) params.subtype = s.subtype if (s.where !== undefined) params.where = s.where if (s.service !== undefined) params.service = s.service const ids = new Set() let offset = 0 // eslint-disable-next-line no-constant-condition while (true) { const batch = await reader.find({ ...params, limit: ENUM_PAGE, offset }) for (const r of batch) ids.add(r.id) if (batch.length < ENUM_PAGE) break offset += ENUM_PAGE } return ids } /** * @description Canonical (storage-level) counterpart of {@link enumerateAllIndexed}: * walks every live noun directly off the storage adapter's canonical shard layout * (`storage.getNouns()` — the same primitive `repairIndex()`'s recount and every * index-heal walk use) instead of going through the metadata index. Guarantees * canon-completeness — a lost or stale metadata-index posting cannot cause a * canonical record to be silently missing from the returned set — at the cost of * an O(N) walk regardless of selector selectivity (unlike the index path's * O(matches)). Returns the RAW candidate id set; `exportGraph`'s caller applies * `matchesPredicate` per-entity via `reader.get()` afterward, exactly as the index * path does, so both paths share one predicate-evaluation code path and can only * disagree on candidacy, never on what a match means. * * Mirrors `find()`'s default hidden-tier policy (always hides `'internal'` and * `'system'` here — `enumerateAllIndexed` never opts either back in via `find()` * either, since `ExportOptions.includeSystem` is applied later, per-entity, and * only reachable for ids a selector already named directly) so the two * enumeration modes produce identical id sets when the index is healthy. */ async function enumerateAllCanonical(storage: StorageAdapter): Promise> { const ids = new Set() let offset = 0 let cursor: string | undefined // eslint-disable-next-line no-constant-condition while (true) { const page = await storage.getNouns({ pagination: { limit: ENUM_PAGE, offset, cursor } }) for (const item of page.items) { if (item.visibility === 'internal' || item.visibility === 'system') continue ids.add(item.id) } if (!page.hasMore || page.items.length === 0) break if (page.nextCursor !== undefined) { cursor = page.nextCursor } else { offset += ENUM_PAGE } } return ids } async function resolveCollectionSubtree( reader: PortableGraphReader, rootId: string, depth?: number ): Promise> { const set = new Set([rootId]) const maxDepth = depth ?? Infinity let frontier = [rootId] let d = 0 while (frontier.length && d < maxDepth) { const next: string[] = [] for (const id of frontier) { const rels = await reader.related({ from: id, type: VerbType.Contains, limit: RELATION_FETCH_LIMIT }) for (const r of rels) { if (!set.has(r.to)) { set.add(r.to) next.push(r.to) } } } frontier = next d++ } return set } async function resolveConnected( reader: PortableGraphReader, c: NonNullable ): Promise> { const { from, depth = 1, verbs, direction = 'out' } = c const set = new Set([from]) let frontier = [from] for (let d = 0; d < depth; d++) { const next: string[] = [] for (const id of frontier) { const neighbours = await neighboursOf(reader, id, direction, verbs) for (const n of neighbours) { if (!set.has(n)) { set.add(n) next.push(n) } } } frontier = next if (!next.length) break } return set } async function neighboursOf( reader: PortableGraphReader, id: string, direction: 'out' | 'in' | 'both', verbs?: VerbType[] ): Promise { const out: string[] = [] if (direction === 'out' || direction === 'both') { const rels = await reader.related({ from: id, limit: RELATION_FETCH_LIMIT }) for (const r of rels) if (!verbs || verbs.includes(r.type)) out.push(r.to) } if (direction === 'in' || direction === 'both') { const rels = await reader.related({ to: id, limit: RELATION_FETCH_LIMIT }) for (const r of rels) if (!verbs || verbs.includes(r.type)) out.push(r.from) } return out } async function resolveVfsPath( reader: PortableGraphReader, path: string, recursive: boolean, depth?: number ): Promise> { const dirId = await resolveVfsPathToId(reader, path) if (!dirId) return new Set() if (!recursive) { const set = new Set([dirId]) const rels = await reader.related({ from: dirId, type: VerbType.Contains, limit: RELATION_FETCH_LIMIT }) for (const r of rels) set.add(r.to) return set } return resolveCollectionSubtree(reader, dirId, depth) } async function resolveVfsPathToId(reader: PortableGraphReader, path: string): Promise { const segments = path.split('/').filter(Boolean) let currentId = VFS_ROOT_ID for (const seg of segments) { const rels = await reader.related({ from: currentId, type: VerbType.Contains, limit: RELATION_FETCH_LIMIT }) let found: string | null = null for (const r of rels) { const child = await reader.get(r.to) if ((child?.metadata as any)?.name === seg) { found = r.to break } } if (!found) return null currentId = found } return currentId } function matchesPredicate(e: Entity, s: ExportSelector): boolean { if (s.type !== undefined) { const types = Array.isArray(s.type) ? s.type : [s.type] if (!types.includes(e.type)) return false } if (s.subtype !== undefined) { const subs = Array.isArray(s.subtype) ? s.subtype : [s.subtype] if (e.subtype === undefined || !subs.includes(e.subtype)) return false } if (s.service !== undefined && e.service !== s.service) return false if (s.visibility !== undefined) { const vis = (e as any).visibility ?? 'public' if (vis !== s.visibility) return false } if (s.where && !matchesWhere(e.metadata, s.where)) return false return true } function matchesWhere(metadata: any, where: Record): boolean { if (!metadata) return false for (const [key, value] of Object.entries(where)) { if (metadata[key] !== value) return false } return true } // ============================================================================ // Serialization helpers (private) // ============================================================================ function toPortableGraphEntity(e: Entity, includeVectors: boolean): PortableGraphEntity { const be: PortableGraphEntity = { id: e.id, type: e.type as string } if (e.subtype !== undefined) be.subtype = e.subtype const vis = (e as any).visibility if (vis !== undefined && vis !== 'public') be.visibility = vis if (e.data !== undefined) be.data = e.data if (e.confidence !== undefined) be.confidence = e.confidence if (e.weight !== undefined) be.weight = e.weight if (e.service !== undefined) be.service = e.service if ((e as any).createdBy !== undefined) be.createdBy = (e as any).createdBy if (e.createdAt !== undefined) be.createdAt = e.createdAt if (includeVectors && e.vector && e.vector.length) be.vector = e.vector if (e.metadata && Object.keys(e.metadata as any).length) be.metadata = e.metadata return be } function toPortableGraphRelation(r: Relation): PortableGraphRelation { const br: PortableGraphRelation = { id: r.id, from: r.from, to: r.to, type: r.type as string } if (r.subtype !== undefined) br.subtype = r.subtype const vis = (r as any).visibility if (vis !== undefined && vis !== 'public') br.visibility = vis if (r.weight !== undefined) br.weight = r.weight if (r.confidence !== undefined) br.confidence = r.confidence if (r.metadata && Object.keys(r.metadata as any).length) br.metadata = r.metadata return br } async function collectEdges( reader: PortableGraphReader, idSet: Set, edges: 'induced' | 'incident' | 'none' ): Promise<{ relations: PortableGraphRelation[]; danglingIds?: string[] }> { if (edges === 'none') return { relations: [] } const relations: PortableGraphRelation[] = [] const dangling = new Set() const seen = new Set() for (const id of idSet) { const rels = await reader.related({ from: id, limit: RELATION_FETCH_LIMIT }) for (const r of rels) { if (seen.has(r.id)) continue const toIn = idSet.has(r.to) if (edges === 'induced' && !toIn) continue if (!toIn) dangling.add(r.to) seen.add(r.id) relations.push(toPortableGraphRelation(r)) } } if (edges === 'incident') { for (const id of idSet) { const rels = await reader.related({ to: id, limit: RELATION_FETCH_LIMIT }) for (const r of rels) { if (seen.has(r.id)) continue if (!idSet.has(r.from)) { dangling.add(r.from) seen.add(r.id) relations.push(toPortableGraphRelation(r)) } } } } return dangling.size > 0 ? { relations, danglingIds: Array.from(dangling) } : { relations } } /** Converts a canonical verb record (as returned by `storage.getVerbs()`) into the wire shape. */ function hnswVerbToPortableGraphRelation(v: HNSWVerbWithMetadata): PortableGraphRelation { const br: PortableGraphRelation = { id: v.id, from: v.sourceId, to: v.targetId, type: v.verb as string } if (v.subtype !== undefined) br.subtype = v.subtype if (v.visibility !== undefined && v.visibility !== 'public') br.visibility = v.visibility if (v.weight !== undefined) br.weight = v.weight if (v.confidence !== undefined) br.confidence = v.confidence if (v.metadata && Object.keys(v.metadata as any).length) br.metadata = v.metadata return br } /** * @description Canonical (storage-level) counterpart of {@link collectEdges}: * walks every live verb directly off `storage.getVerbs()` — the same primitive * `repairIndex()`'s recount and every index-heal walk use — instead of the graph * adjacency index (`reader.related()`), so a lost/stale adjacency posting cannot * cause a canonical relationship to be silently dropped from the export. Used for * EVERY selector in `enumeration:'canonical'` mode, not just the whole-brain * branch: relations can be blinded by adjacency-index corruption regardless of * how `idSet` (the kept node ids) was produced. * * Mirrors `related()`'s default hidden-tier policy (always hides `'internal'` * and `'system'` — `collectEdges` never opts either back in via `related()` * either) so the two enumeration modes produce identical relation sets when the * index is healthy. */ async function collectEdgesCanonical( storage: StorageAdapter, idSet: Set, edges: 'induced' | 'incident' | 'none' ): Promise<{ relations: PortableGraphRelation[]; danglingIds?: string[] }> { if (edges === 'none') return { relations: [] } const relations: PortableGraphRelation[] = [] const dangling = new Set() const seen = new Set() let offset = 0 let cursor: string | undefined // eslint-disable-next-line no-constant-condition while (true) { const page = await storage.getVerbs({ pagination: { limit: ENUM_PAGE, offset, cursor } }) for (const v of page.items) { if (seen.has(v.id)) continue if (v.visibility === 'internal' || v.visibility === 'system') continue const fromIn = idSet.has(v.sourceId) const toIn = idSet.has(v.targetId) if (edges === 'induced') { if (!fromIn || !toIn) continue } else if (!fromIn && !toIn) { continue // 'incident': neither endpoint kept — irrelevant to this export } if (fromIn && !toIn) dangling.add(v.targetId) if (toIn && !fromIn) dangling.add(v.sourceId) seen.add(v.id) relations.push(hnswVerbToPortableGraphRelation(v)) } if (!page.hasMore || page.items.length === 0) break if (page.nextCursor !== undefined) { cursor = page.nextCursor } else { offset += ENUM_PAGE } } return dangling.size > 0 ? { relations, danglingIds: Array.from(dangling) } : { relations } } async function collectBlobs( storage: StorageAdapter | undefined, entityMap: Map> ): Promise> { const blobs: Record = {} const blobStorage = (storage as any)?.blobStorage if (!blobStorage?.read) return blobs for (const e of entityMap.values()) { const storageMeta = (e.metadata as any)?.storage const hash = storageMeta?.hash if (storageMeta?.type === 'blob' && hash && !blobs[hash]) { try { const buf = await blobStorage.read(hash) blobs[hash] = Buffer.from(buf).toString('base64') } catch { // Referenced blob bytes unreadable (storage drift): skip — file structure still // travels, and stats.blobCount reflects what was actually captured. } } } return blobs } // ============================================================================ // add()/update() param mapping (private) // ============================================================================ function entityAddFields(be: PortableGraphEntity): Record { const fields: Record = { data: be.data, type: be.type as NounType } if (be.subtype !== undefined) fields.subtype = be.subtype if (be.visibility !== undefined) fields.visibility = be.visibility if (be.service !== undefined) fields.service = be.service if (be.confidence !== undefined) fields.confidence = be.confidence if (be.weight !== undefined) fields.weight = be.weight if (be.metadata !== undefined) fields.metadata = be.metadata return fields } function entityUpdateFields(be: PortableGraphEntity): Record { const fields: Record = {} if (be.data !== undefined) fields.data = be.data if (be.type !== undefined) fields.type = be.type as NounType if (be.subtype !== undefined) fields.subtype = be.subtype if (be.visibility !== undefined) fields.visibility = be.visibility if (be.confidence !== undefined) fields.confidence = be.confidence if (be.weight !== undefined) fields.weight = be.weight if (be.metadata !== undefined) fields.metadata = be.metadata if (Array.isArray(be.vector) && be.vector.length) fields.vector = be.vector return fields }