feat(8.0): brain.graph.export() + noun-walk cursor + noun visibility hydration
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:<shard>:<id> 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.
This commit is contained in:
parent
8c2b57a488
commit
c2a84c9d1f
4 changed files with 369 additions and 75 deletions
|
|
@ -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) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue