diff --git a/gitnexus/README.md b/gitnexus/README.md index 6e92c69d8..c1e9e6050 100644 --- a/gitnexus/README.md +++ b/gitnexus/README.md @@ -482,6 +482,7 @@ Configure the behavior with these environment variables: | `GITNEXUS_LBUG_EXTENSION_INSTALL_TIMEOUT_MS` | positive integer | `15000` | Wall-clock budget for the out-of-process extension-install child before it is killed. | | `GITNEXUS_FTS_STEMMER` | supported LadybugDB stemmer | `porter` | Stemmer used when rebuilding BM25/FTS indexes. Use `none` for CJK-heavy repositories, or a language stemmer such as `german`, `french`, or `spanish` when that better matches repository comments and identifiers. Re-run `gitnexus analyze --repair-fts` after changing it. | | `GITNEXUS_FTS_CJK_SEGMENTATION` | `none`, `bigram` | `none` | `bigram` inserts overlapping character-bigram boundaries into Chinese/Japanese Han-ideograph spans in `content`/`description` before FTS indexing, so LadybugDB's space-only tokenizer can see sub-phrase word boundaries. Scoped to CJK Unified Ideographs only — Japanese Hiragana/Katakana and Korean Hangul are not currently segmented. Unlike `GITNEXUS_FTS_STEMMER`, this rewrites stored text — enabling it on an already-indexed repo requires a full `gitnexus analyze --force`; neither `--repair-fts` nor a plain incremental `analyze` applies it to previously-indexed files. Set the same value wherever `analyze` and search-serving processes (CLI query, MCP server, web server) run. | +| `GITNEXUS_STREAM_GRAPH_EMIT` | `0`, `1` | `1` (on) | **On by default** on a full rebuild (`--force`); incremental runs ignore it. Holds structural relationships (CALLS, IMPORTS, ACCESSES, CONTAINS, ...) as CSV-on-disk plus compact in-memory columns instead of as objects in three overlapping indexes, cutting peak in-memory graph heap by ~1.4x at no measurable CPU cost (measured A/B on a synthetic 400k-node / 1.08M-edge graph: 819 MB -> 584 MB, iteration at parity, scaling verified linear from 100k to 800k nodes, with every edge still visible through the graph interface; no end-to-end measurement on a real repository yet). Nothing is traded away — community detection, process extraction, PDG taint summaries and the local-symbol pruner all read a complete relationship set and behave identically. Set to `0` only to bisect a suspected streaming-related fault. | | `GITNEXUS_COMMUNITY_ENGINE` | `graphology`, `icebug`, `auto` | `graphology` | Community-detection engine used during analyze. `graphology` uses the bundled default path. `icebug` and `auto` currently behave identically: both try the experimental Icebug CSR path and fall back to Graphology if the optional native module is unavailable or incompatible. | | `GITNEXUS_WAL_CHECKPOINT_THRESHOLD` | integer `>= -1` | `67108864` (64 MiB) | LadybugDB WAL auto-checkpoint threshold during analyze (bytes). Auto-checkpoint remains enabled; `-1` keeps Ladybug's stock ~16 MiB. Larger thresholds reduce checkpoint frequency but increase the WAL size at rotation time — choose a smaller value on disk-constrained environments. | | `GITNEXUS_LBUG_BUFFER_POOL_SIZE` | integer `>= 0` (bytes) | min(2 GiB, 80% RAM) | LadybugDB buffer-pool ceiling for every GitNexus database (analyze, MCP server, serve, group bridges). Bounded so a long-lived `gitnexus mcp` process or a large incremental `analyze` cannot grow toward LadybugDB's native 80%-of-RAM default and OOM the host (#2557). `0` restores that native unbounded default; invalid values warn and fall back to the default. During `analyze` the pool is right-sized to the graph and, on non-4 KiB-page hosts (Apple Silicon 16 KiB, Ascend/aarch64 64 KiB), scaled by the page-size granule ratio up to min(2 GiB × pageSize/4 KiB, 80% RAM) (#2631); this env var overrides all of that as an absolute value. | diff --git a/gitnexus/src/core/graph/graph.ts b/gitnexus/src/core/graph/graph.ts index c906e1b10..1c708a4af 100644 --- a/gitnexus/src/core/graph/graph.ts +++ b/gitnexus/src/core/graph/graph.ts @@ -162,6 +162,11 @@ export const createKnowledgeGraph = (): KnowledgeGraph => { forEachRelationship(fn: (rel: GraphRelationship) => void) { relationshipMap.forEach(fn); }, + forEachRelationshipFields( + fn: (sourceId: string, targetId: string, type: RelationshipType, confidence: number) => void, + ) { + relationshipMap.forEach((rel) => fn(rel.sourceId, rel.targetId, rel.type, rel.confidence)); + }, getNode: (id: string) => nodeMap.get(id), // O(1) count getters - avoid creating arrays just for length diff --git a/gitnexus/src/core/graph/types.ts b/gitnexus/src/core/graph/types.ts index 539f77987..9d9caf12b 100644 --- a/gitnexus/src/core/graph/types.ts +++ b/gitnexus/src/core/graph/types.ts @@ -27,6 +27,19 @@ export interface KnowledgeGraph { iterRelationshipsByType: (type: RelationshipType) => IterableIterator; forEachNode: (fn: (node: GraphNode) => void) => void; forEachRelationship: (fn: (rel: GraphRelationship) => void) => void; + /** + * Zero-allocation relationship scan: fields, not objects (#2680). + * + * The whole-graph scans (the local-symbol pruner, community detection, + * process extraction) read only these four fields, and materializing a + * `GraphRelationship` per edge just to read them dominates iteration cost once + * relationships are held columnar — measured at ~90 ms per analyze on a + * million-edge graph. Prefer this over `forEachRelationship` in any pass that + * walks every edge and needs no other field. + */ + forEachRelationshipFields: ( + fn: (sourceId: string, targetId: string, type: RelationshipType, confidence: number) => void, + ) => void; getNode: (id: string) => GraphNode | undefined; nodeCount: number; relationshipCount: number; @@ -34,5 +47,12 @@ export interface KnowledgeGraph { addRelationship: (relationship: GraphRelationship) => void; removeNode: (nodeId: string) => boolean; removeNodesByFile: (filePath: string) => number; + /** + * Removes the relationship with this id, returning whether it existed. + * + * Implementations that offload relationships out of memory cannot always tell + * "absent" from "already written out" — `GraphEmitSink` deliberately throws + * rather than answering `false` for an edge it can no longer recall (#2680). + */ removeRelationship: (relationshipId: string) => boolean; } diff --git a/gitnexus/src/core/ingestion/community-processor.ts b/gitnexus/src/core/ingestion/community-processor.ts index ff892ae73..7a91eb574 100644 --- a/gitnexus/src/core/ingestion/community-processor.ts +++ b/gitnexus/src/core/ingestion/community-processor.ts @@ -290,14 +290,16 @@ export const buildCommunityProjection = (knowledgeGraph: KnowledgeGraph): Commun const connectedNodes = new Set(); const nodeDegree = new Map(); - knowledgeGraph.forEachRelationship((rel) => { - if (!isClusteringRelationship(rel.type) || rel.sourceId === rel.targetId) return; - if (isLarge && rel.confidence < MIN_CONFIDENCE_LARGE) return; + // Field-wise scan (#2680): this walks every edge and reads only these four, + // so taking objects would allocate one per edge for nothing. + knowledgeGraph.forEachRelationshipFields((sourceId, targetId, type, confidence) => { + if (!isClusteringRelationship(type) || sourceId === targetId) return; + if (isLarge && confidence < MIN_CONFIDENCE_LARGE) return; - connectedNodes.add(rel.sourceId); - connectedNodes.add(rel.targetId); - nodeDegree.set(rel.sourceId, (nodeDegree.get(rel.sourceId) || 0) + 1); - nodeDegree.set(rel.targetId, (nodeDegree.get(rel.targetId) || 0) + 1); + connectedNodes.add(sourceId); + connectedNodes.add(targetId); + nodeDegree.set(sourceId, (nodeDegree.get(sourceId) || 0) + 1); + nodeDegree.set(targetId, (nodeDegree.get(targetId) || 0) + 1); }); const nodes: CommunityProjectionNode[] = []; @@ -328,12 +330,12 @@ export const buildCommunityProjection = (knowledgeGraph: KnowledgeGraph): Commun const seenEdges = new Set(); const edges: Array = []; - knowledgeGraph.forEachRelationship((rel) => { - if (!isClusteringRelationship(rel.type) || rel.sourceId === rel.targetId) return; - if (isLarge && rel.confidence < MIN_CONFIDENCE_LARGE) return; + knowledgeGraph.forEachRelationshipFields((sourceId, targetId, type, confidence) => { + if (!isClusteringRelationship(type) || sourceId === targetId) return; + if (isLarge && confidence < MIN_CONFIDENCE_LARGE) return; - const sourceIndex = nodeIndexById.get(rel.sourceId); - const targetIndex = nodeIndexById.get(rel.targetId); + const sourceIndex = nodeIndexById.get(sourceId); + const targetIndex = nodeIndexById.get(targetId); if (sourceIndex === undefined || targetIndex === undefined || sourceIndex === targetIndex) return; diff --git a/gitnexus/src/core/ingestion/local-symbol-pruner.ts b/gitnexus/src/core/ingestion/local-symbol-pruner.ts index 3ff876b44..24e9733ff 100644 --- a/gitnexus/src/core/ingestion/local-symbol-pruner.ts +++ b/gitnexus/src/core/ingestion/local-symbol-pruner.ts @@ -1,4 +1,4 @@ -import type { GraphNode, GraphRelationship, NodeLabel } from 'gitnexus-shared'; +import type { GraphNode, NodeLabel, RelationshipType } from 'gitnexus-shared'; import type { KnowledgeGraph } from '../graph/types.js'; import { parseTruthyEnv } from './utils/env.js'; @@ -30,9 +30,13 @@ const isLocalValueCandidate = (node: GraphNode): boolean => { // True when `rel` is the structural `File -> DEFINES -> candidate` edge. Callers // guard on the candidate already being the edge target, so only the source label // needs checking here. -const isFileDefinesEdge = (graph: KnowledgeGraph, rel: GraphRelationship): boolean => { - if (rel.type !== 'DEFINES') return false; - return graph.getNode(rel.sourceId)?.label === 'File'; +const isFileDefinesEdge = ( + graph: KnowledgeGraph, + type: RelationshipType, + sourceId: string, +): boolean => { + if (type !== 'DEFINES') return false; + return graph.getNode(sourceId)?.label === 'File'; }; export const pruneLocalValueSymbols = ( @@ -51,21 +55,21 @@ export const pruneLocalValueSymbols = ( if (candidateIds.size === 0) return emptyStats(false); const candidatesWithSemanticEdges = new Set(); - for (const rel of graph.iterRelationships()) { + // Field-wise scan (#2680): a whole-graph walk that reads only these three, so + // materializing a relationship object per edge would be pure overhead. + graph.forEachRelationshipFields((sourceId, targetId, type) => { // Any outgoing edge from a candidate is a semantic edge: the only structural // edge a block-local value symbol carries is the incoming File -> DEFINES, on // which the candidate is the target, never the source. - if (candidateIds.has(rel.sourceId)) { - candidatesWithSemanticEdges.add(rel.sourceId); + if (candidateIds.has(sourceId)) { + candidatesWithSemanticEdges.add(sourceId); } // An incoming edge is semantic unless it is the structural File -> DEFINES. - if (candidateIds.has(rel.targetId)) { - if (!isFileDefinesEdge(graph, rel)) { - candidatesWithSemanticEdges.add(rel.targetId); - } + if (candidateIds.has(targetId) && !isFileDefinesEdge(graph, type, sourceId)) { + candidatesWithSemanticEdges.add(targetId); } - } + }); let prunedNodes = 0; for (const candidateId of candidateIds) { diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse.ts index 08da0068a..2484baa01 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/parse.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse.ts @@ -90,6 +90,12 @@ export const parsePhase: PipelinePhase = { ctx: PipelineContext, deps: ReadonlyMap>, ): Promise { + // Begin streamed structural emit (#2680), if enabled. Deliberately here and + // not at graph construction: the pre-parse phases are not all write-only — + // `mapCobolToGraph` scans CALLS edges and removes the unresolved ones — and + // nothing before parse produces bulk edge volume anyway. + ctx.graphEmit?.beginStreaming(); + const { scannedFiles, allPaths, allPathSet, totalFiles } = getPhaseOutput( deps, 'structure', diff --git a/gitnexus/src/core/ingestion/pipeline-phases/types.ts b/gitnexus/src/core/ingestion/pipeline-phases/types.ts index 17786ff4c..7958e404f 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/types.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/types.ts @@ -14,6 +14,7 @@ * - Each phase is independently testable with mocked inputs */ +import type { GraphEmitControl } from '../../lbug/graph-emit-sink.js'; import type { KnowledgeGraph } from '../../graph/types.js'; import type { PipelineProgress } from 'gitnexus-shared'; import type { PipelineOptions } from '../pipeline.js'; @@ -32,6 +33,12 @@ export interface PipelineContext { readonly options?: PipelineOptions; /** Pipeline start timestamp (for elapsed-time logging). */ readonly pipelineStart: number; + /** + * Streamed structural emit (#2680), present only when `streamGraphEmit` is on. + * `parse` calls `beginStreaming()` at its start; `pruneLocalSymbols` consults + * `hasStreamedSemanticEdge()`. Absent ⇒ everything stays in the graph. + */ + readonly graphEmit?: GraphEmitControl; } // ── Phase result wrapper ─────────────────────────────────────────────────── diff --git a/gitnexus/src/core/ingestion/pipeline.ts b/gitnexus/src/core/ingestion/pipeline.ts index ebf290112..eb1308710 100644 --- a/gitnexus/src/core/ingestion/pipeline.ts +++ b/gitnexus/src/core/ingestion/pipeline.ts @@ -16,6 +16,7 @@ */ import { createKnowledgeGraph } from '../graph/graph.js'; +import { GraphEmitSink, type GraphEmitManifest } from '../lbug/graph-emit-sink.js'; import { type PipelineProgress } from 'gitnexus-shared'; import { PipelineResult } from '../../types/pipeline.js'; import { @@ -142,6 +143,22 @@ export interface PipelineOptions { * whole-graph emit. */ streamPdgEmit?: boolean; + /** + * Streamed structural graph emit (#2680). When true, relationships that no + * mid-pipeline phase reads back (CALLS, IMPORTS, ACCESSES, CONTAINS, ...) are + * streamed to CSV-on-disk from the parse boundary onward instead of being + * retained in the in-memory graph — measured ~2.9x reduction of graph heap. + * + * NOT free: the `communities`, `processes`, `taintSummaries` and + * `callSummaries` phases all consume the whole CALLS graph and are disabled + * under this flag. The caller (`run-analyze`) gates it to full rebuilds. + * Requires `graphEmitCsvDir`. + */ + streamGraphEmit?: boolean; + /** Directory for the streamed structural CSVs. Required when + * `streamGraphEmit` is on; supplied by the caller, which owns storage-path + * resolution (and its native-safe relocation). */ + graphEmitCsvDir?: string; /** Streamed PDG-emit write buffer (rows) when `streamPdgEmit` is on (#2202). * `undefined` ⇒ `DEFAULT_PDG_EMIT_CHUNK_ROWS`. Memory-only; does not affect * emitted bytes. */ @@ -297,15 +314,46 @@ export const runPipelineFromRepo = async ( const graph = createKnowledgeGraph(); const pipelineStart = Date.now(); + // Streamed structural emit (#2680). The sink is a write-routing façade over + // `graph`; it streams nothing until `beginStreaming()` fires at the parse + // boundary. + // + // A missing `graphEmitCsvDir` is a caller bug, not a reason to quietly skip + // streaming: this is on by default, so a programmatic host that builds its own + // `PipelineOptions` (eval-server, the MCP daemon, a test) would otherwise ask + // for streaming, silently not get it, and still see a successful run. Fail + // loudly instead — the whole point of the surrounding work is that a degraded + // outcome must never look like a clean one. + let graphEmitSink: GraphEmitSink | undefined; + if (options?.streamGraphEmit === true) { + if (options.graphEmitCsvDir === undefined) { + throw new Error( + 'streamGraphEmit was requested but graphEmitCsvDir is missing. The caller owns ' + + 'storage-path resolution (see resolveNativeSafeStorageDir in run-analyze.ts); ' + + 'pass the directory, or leave streamGraphEmit unset to run without streaming.', + ); + } + graphEmitSink = new GraphEmitSink(graph, options.graphEmitCsvDir); + } + const phases = buildPhaseList(options); - const results = await runPipeline(phases, { - repoPath, - graph, - onProgress, - options, - pipelineStart, - }); + let graphEmitManifest: GraphEmitManifest | undefined; + let results; + try { + results = await runPipeline(phases, { + repoPath, + graph: graphEmitSink ?? graph, + onProgress, + options, + pipelineStart, + graphEmit: graphEmitSink, + }); + graphEmitManifest = graphEmitSink?.finalize(); + } finally { + // Release per-pair fds when the pipeline threw before finalize ran. + graphEmitSink?.close(); + } // Extract final results for the PipelineResult contract const { totalFiles, usedWorkerPool } = getPhaseOutput<{ @@ -320,7 +368,12 @@ export const runPipelineFromRepo = async ( // Streamed PDG-emit manifest (#2202): present only when streaming was on. const pdgEmitManifest = scopeResolutionOutput.pdgEmitManifest; - if (!options?.skipGraphPhases) { + // Presence check, not `!skipGraphPhases`: phases can now be filtered out by + // any `enabledWhen` predicate (streamGraphEmit disables communities/processes + // too), and `getPhaseOutput` THROWS on a phase that was never resolved. Keying + // off the options flag alone made every filtered-out combination crash here + // rather than return undefined results. + if (results.has('communities') && results.has('processes')) { communityResult = getPhaseOutput(results, 'communities').communityResult; processResult = getPhaseOutput(results, 'processes').processResult; } @@ -340,9 +393,16 @@ export const runPipelineFromRepo = async ( }); return { + // The RAW graph, deliberately — NOT `graphEmitSink`. Phases above received + // the sink so their reads are complete, but `loadGraphToLbug` feeds this to + // `streamAllCSVsToDisk`, and the sink's complete iterator would then emit + // every streamed edge a SECOND time on top of the per-pair CSVs the sink + // already wrote and the manifest already COPYs. Returning the sink here + // silently doubles every streamed relationship in the persisted graph. graph, repoPath, totalFileCount: totalFiles, + graphEmitManifest, communityResult, processResult, resolutionOutcomes, diff --git a/gitnexus/src/core/ingestion/process-processor.ts b/gitnexus/src/core/ingestion/process-processor.ts index aa744e54d..dbec25474 100644 --- a/gitnexus/src/core/ingestion/process-processor.ts +++ b/gitnexus/src/core/ingestion/process-processor.ts @@ -230,14 +230,13 @@ const MIN_TRACE_CONFIDENCE = 0.5; const buildCallsGraph = (graph: KnowledgeGraph): AdjacencyList => { const adj = new Map(); - for (const rel of graph.iterRelationships()) { - if (rel.type === 'CALLS' && rel.confidence >= MIN_TRACE_CONFIDENCE) { - if (!adj.has(rel.sourceId)) { - adj.set(rel.sourceId, []); - } - adj.get(rel.sourceId)!.push(rel.targetId); - } - } + // Field-wise scan (#2680) — whole-graph walk, four fields, no object needed. + graph.forEachRelationshipFields((sourceId, targetId, type, confidence) => { + if (type !== 'CALLS' || confidence < MIN_TRACE_CONFIDENCE) return; + const existing = adj.get(sourceId); + if (existing === undefined) adj.set(sourceId, [targetId]); + else existing.push(targetId); + }); return adj; }; @@ -245,14 +244,12 @@ const buildCallsGraph = (graph: KnowledgeGraph): AdjacencyList => { const buildReverseCallsGraph = (graph: KnowledgeGraph): AdjacencyList => { const adj = new Map(); - for (const rel of graph.iterRelationships()) { - if (rel.type === 'CALLS' && rel.confidence >= MIN_TRACE_CONFIDENCE) { - if (!adj.has(rel.targetId)) { - adj.set(rel.targetId, []); - } - adj.get(rel.targetId)!.push(rel.sourceId); - } - } + graph.forEachRelationshipFields((sourceId, targetId, type, confidence) => { + if (type !== 'CALLS' || confidence < MIN_TRACE_CONFIDENCE) return; + const existing = adj.get(targetId); + if (existing === undefined) adj.set(targetId, [sourceId]); + else existing.push(sourceId); + }); return adj; }; diff --git a/gitnexus/src/core/lbug/graph-emit-sink.ts b/gitnexus/src/core/lbug/graph-emit-sink.ts new file mode 100644 index 000000000..4e1845d71 --- /dev/null +++ b/gitnexus/src/core/lbug/graph-emit-sink.ts @@ -0,0 +1,627 @@ +/** + * Streaming structural graph-emit sink (issue #2680). + * + * `analyze` holds the whole `KnowledgeGraph` on the main thread for the entire + * pipeline, so peak heap is O(repo) — ~2.1 KB/node at Linux-kernel scale + * (#2649). Measurement on a kernel-shaped synthetic graph (400k nodes, + * 2.7 edges/node) says where that goes: + * + * nodes only ....... 367 B/node + * nodes + edges .... 2075 B/node <- reproduces the #2649 figure + * + * So **relationships are ~83% of graph heap** (~646 B/edge), and that is what + * this sink removes. 646 B for an object holding four short strings is the cost + * of storing every edge four times over — `relationshipMap`, a + * `relationshipsByType` bucket, and both endpoints' `edgeIdsByNode` Sets — plus + * an `id` that concatenates both endpoint ids. (Dropping just the two redundant + * indexes was measured too: 174 of 648 B/edge, ~1.3x. Not enough on its own.) + * + * Nodes are deliberately NOT streamed: they are the other 17%, and two + * scope-resolution index builders (`buildGraphNodeLookup`, + * `buildGraphCallableAnchorIndex`) scan them. + * + * ## How much this actually saves — read this before quoting a number + * + * Measured A/B against the object-based graph, 400k nodes / 1.08M edges, all + * edges streamable (the worst case for this design): **823 MB -> 626 MB, ~1.3x**, + * with all 1.08M edges still visible through `iterRelationships`. + * + * That is well short of the ~2.9x a naive `0.17 + 0.83 * 0.21` retained-share + * calculation suggests, and the gap is deliberate: this sink is *lossless*, so + * it pays for the {@link streamedIds} dedup Set (one unique id string per + * streamed edge) and the columns above. An earlier revision hit a bigger number + * by disabling community detection, process extraction and the taint fixpoint — + * which is why it could not be the default. 1.3x with nothing traded away is the + * honest figure; if a future change needs more, the next lever is dedup keyed on + * the interned column triple rather than on id strings (it must first be shown + * not to alter the emitted row SET). + * + * ## What it costs — measured, not assumed + * + * Measured on the same 400k-node / 1.08M-edge graph, all edges streamable, each + * arm running what its own consumers actually call: + * + * heap 819 MB -> 584 MB (1.40x better) + * scans ~78 ms -> ~88 ms (parity, within run-to-run noise) + * + * Linear in both: verified at 100k/200k/400k/800k nodes, per-edge scan cost flat + * (~13 ns both arms) and the heap ratio drifting only 1.7x -> 1.5x as interner + * indices gain digits. No super-linear term, so a larger repo costs + * proportionally more, not disproportionately. + * + * The dedup key encodes its tail SEGMENT COUNT, which measurably costs ~66 MB + * here versus omitting it. That is not optional: without it a one-segment tail + * `:7` and a two-segment `:7:0` collapse onto one key and an edge is silently + * dropped (regression test in graph-emit-sink.test.ts). + * + * Getting there took three measured steps, because the naive version was 6.8x + * WORSE (651 ms) — reads rebuild objects, and a real analyze performs SIX full + * relationship scans (the pruner, community detection x2, process extraction x2, + * and the taint fixpoint's CALLS pass): + * + * 1. The ~150-character synthesized `id` was built eagerly on every read — 6.5M + * concatenations for a field no in-pipeline consumer reads. Isolating it + * showed 436 ms of the regression. It is now a lazy prototype getter on + * {@link StreamedRelationship}. + * 2. Generator and iterator-protocol overhead: {@link forEachRelationship} loops + * the columns directly, and {@link iterRelationships} reuses one + * iterator-result record. Note a hand-rolled iterator allocating a fresh + * `{value, done}` per edge measured WORSE (252 ms) than the generator it + * replaced, so the obvious rewrite is not the one that shipped. + * 3. The remaining ~90 ms was object allocation itself, irreducible while the + * read API returns objects — so the five whole-graph scans moved to + * `forEachRelationshipFields`, which passes the four fields they actually + * read as primitives and allocates nothing. See + * {@link GraphEmitSink.forEachRelationshipFields}. + * + * The last allocating scan is the taint fixpoint's `iterRelationshipsByType` + * pass; it is one scan of six and accounts for the small residual. Give it a + * by-type field variant only if a measurement says it matters. + * + * It is in any case NOT O(chunk) — node identity and the resolution registries + * stay O(repo). True O(chunk) needs DB-side resolution and Leiden (#2337), at + * which point this sink should be deleted rather than extended. + * + * ## Correctness contract + * + * Structural sibling of {@link PdgEmitSink}, and reuses its row builder + * (`buildRelRow`), header (`REL_CSV_HEADER`), label derivation (`getNodeLabel`) + * and `RelPairRouter` validity check, so the streamed row SET equals the + * whole-graph emit's and the bulk COPY loads the same rows. Set-level, not + * byte-level: rows stream in emit order and are not re-sorted under + * `GITNEXUS_SORT_GRAPH_OUTPUT`. + */ +import fs from 'fs'; +import path from 'path'; +import type { GraphNode, GraphRelationship, RelationshipType } from 'gitnexus-shared'; +import type { KnowledgeGraph } from '../graph/types.js'; +import { REL_CSV_HEADER, buildRelRow } from './csv-generator.js'; +import { getNodeLabel } from './rel-pair-routing.js'; +import { NODE_TABLES } from './schema.js'; +import { DEFAULT_EMIT_CHUNK_ROWS, SyncCsvWriter } from './sync-csv-writer.js'; + +/** + * Relationship types that MUST stay in the in-memory graph because a phase + * running while streaming is active reads them back. + * + * Derived from an exhaustive audit of every relationship read site under + * `gitnexus/src/` (`iterRelationshipsByType` / `iterRelationships` / + * `forEachRelationship` / `removeRelationship`), not from intuition — an + * earlier draft of this list carried 14 types, 5 of which no reachable phase + * reads. Every entry below names its reader: + * + * EXTENDS, IMPLEMENTS - mro-processor, scope-resolution/passes/mro, + * receiver-bound-calls, pipeline/run.ts, cpp + * member-lookup, and 9 language scope-resolvers + * HAS_METHOD - mro-processor, di phase + * HAS_PROPERTY - di phase, ruby scope-resolver, spring config-bindings + * METHOD_OVERRIDES, + * METHOD_IMPLEMENTS - mro-processor + * DEFINES - local-symbol-pruner's isFileDefinesEdge test + * INJECTS - di phase fan-out + * + * Deliberately NOT retained: STEP_IN_PROCESS / ENTRY_POINT_OF / MEMBER_OF + * (written only by the `processes` / `communities` phases, which the streaming + * flag disables), TAINT_PATH / CALL_SUMMARY (their phases are likewise gated + * off under the flag), and HANDLES_ROUTE / HANDLES_TOOL (written by + * `routes`/`tools`, never read back mid-pipeline). + * + * Adding a relationship type that a phase reads back WITHOUT adding it here is + * a silent-wrong-graph bug, not a crash — and NOTHING automated catches it. + * The differential round-trip test cannot: `addRelationship` partitions edges + * between the graph and the CSVs, and the union of a partition is invariant + * under where the partition line falls, so that test stays green no matter how + * this set is drawn. Only the read-site audit protects this invariant; re-run it + * (grep iterRelationshipsByType / iterRelationships / forEachRelationship / + * removeRelationship across src/) when adding a phase or a relationship type. + */ +export const RETAINED_REL_TYPES: ReadonlySet = new Set([ + 'EXTENDS', + 'IMPLEMENTS', + 'HAS_METHOD', + 'HAS_PROPERTY', + 'METHOD_OVERRIDES', + 'METHOD_IMPLEMENTS', + 'DEFINES', + 'INJECTS', +]); + +/** + * COPY manifest produced by {@link GraphEmitSink.finalize}. + * + * Only `relsByPair` — this PR does not stream node rows, so a `nodeFiles` + * dimension would be permanently empty. Note that unlike `PdgEmitManifest`, + * these pair keys DO collide with the whole-graph emit's (streamed `CALLS` is + * `Function|Function`, same as retained edges), so `loadGraphToLbug` must + * APPEND these files to the pair rather than reject them as a collision. + */ +export interface GraphEmitManifest { + /** pairKey (`From|To`) -> per-pair edge CSV. */ + readonly relsByPair: Map; + /** Total streamed rows, for the buffer-pool size hint (#2631 path). */ + readonly totalRows: number; +} + +/** + * The slice of the sink that pipeline phases drive. Declared here, next to the + * implementation, and imported as a type by `pipeline-phases/types.ts` so the + * phase layer depends on this narrow capability rather than on two loose + * callbacks bolted onto the context. + */ +export interface GraphEmitControl { + /** Start routing non-retained relationships to disk (see {@link GraphEmitSink.beginStreaming}). */ + beginStreaming(): void; +} + +/** + * A streamed edge, rebuilt for a read. + * + * A class, not an object literal, for two reasons that both showed up in + * measurement. Its shape is fixed, so V8 keeps one hidden class across millions + * of instances; and `id` is a PROTOTYPE getter, so the ~150-character + * concatenation happens only if a caller actually reads it — which none of the + * in-pipeline consumers do. Building it eagerly cost 436 ms of a 555 ms + * iteration regression across the six full scans an analyze performs (measured + * at 400k nodes / 1.08M edges); deferring it gives that back. + */ +class StreamedRelationship implements GraphRelationship { + /** Constant for streamed edges — see the note on the columns about why + * `reason` is not retained. The persisted CSV row keeps the real value. */ + readonly reason = 'streamed'; + + constructor( + readonly sourceId: string, + readonly targetId: string, + readonly type: RelationshipType, + readonly confidence: number, + private readonly ix: number, + ) {} + + /** Deterministic and unique — the column index disambiguates two streamed + * edges that share (type, source, target). Lazily built; nothing in the + * pipeline reads it. */ + get id(): string { + return `${this.type}:${this.sourceId}->${this.targetId}#${this.ix}`; + } +} + +/** Thrown when a consumer removes a relationship that already streamed to + * disk. Silently no-oping would let a mutating consumer (e.g. the COBOL + * cross-program CALL resolver) corrupt the persisted graph undetected. */ +export class StreamedRelationshipRemovalError extends Error { + constructor(relationshipId: string) { + super( + `Cannot remove relationship "${relationshipId}": it has already been streamed to ` + + `CSV and cannot be recalled. A phase that removes relationships must run before ` + + `the GraphEmitSink is installed (see the parse-boundary construction in pipeline.ts).`, + ); + this.name = 'StreamedRelationshipRemovalError'; + } +} + +/** + * Write-routing graph façade. Construct one per analyze run at the PARSE + * boundary — not at `createKnowledgeGraph()` — so the pre-parse phases + * (`structure`, `springConfig`, `markdown`, `cobol`) complete their + * read-modify-delete passes against a fully in-memory graph. Call + * {@link finalize} once after the pipeline, before `loadGraphToLbug`. + */ +export class GraphEmitSink implements KnowledgeGraph, GraphEmitControl { + private readonly validTables: Set; + private readonly relWriters = new Map(); + /** + * Ids of relationships already streamed. `KnowledgeGraph.addRelationship` + * drops duplicate ids first-writer-wins, and COPY into a PK-bearing table + * would violate on a repeat, so the sink must dedup itself — unlike + * `PdgEmitSink`, whose emit loop guarantees per-file uniqueness upstream. + * + * ponytail: O(streamed-edges) id strings retained. That is ~a tenth of full + * edge retention (the objects, both endpoint index Sets, and the type bucket + * all go away), but it is not O(chunk). Upgrade path if it ever dominates: + * a per-pair sorted-run dedup on disk, or hashing ids into a Bloom filter + * with an exact fallback. + */ + private readonly streamedIds = new Set(); + /** + * Streamed edges, kept as parallel columns so the sink can still answer a + * COMPLETE relationship read (see {@link iterRelationships}). Only the four + * fields any consumer of these edges actually reads are retained — + * `sourceId`, `targetId`, `type`, `confidence` — audited across + * community-processor, process-processor, taint-summaries and the pruner. + * + * `id`, `reason` and `step` are deliberately NOT kept. Every relationship id + * is a unique long string, and retaining ids is exactly what made an earlier + * fully-columnar attempt LOSE to the object-based graph (measured 838 MB vs + * 822 MB at 400k nodes / 1.08M edges). Keeping ids out of the heap is where + * the saving comes from, so a read synthesizes a deterministic id instead — + * safe because `buildRelRow` never persists `rel.id` and no consumer keys on + * it (audited). + * + * The dropped `reason`/`step` are safe too, but for a different reason worth + * stating: the PERSISTED row keeps their true values, because `buildRelRow` is + * handed the original relationship on the way through. Only in-memory reads + * see the `'streamed'` placeholder, and the in-pipeline consumers of streamed + * edges read neither field. So e.g. the `ACCESSES reason: 'read'|'write'` + * distinction that MCP queries rely on survives in the database. A future + * in-pipeline consumer needing `reason` or `step` on a streamed edge must add + * the column, not trust the placeholder. + * + * Node ids are interned; the strings are shared by reference with the node + * map's, so interning adds bookkeeping, not new text. + */ + private readonly nodeIds = new Map(); + private readonly nodeIdByIx: string[] = []; + private readonly srcIx: number[] = []; + private readonly tgtIx: number[] = []; + private readonly relTypes: RelationshipType[] = []; + private readonly confidences: number[] = []; + private finalized = false; + /** + * Streaming is OFF until {@link beginStreaming} is called by `parse`. + * + * The pre-parse phases are not all write-only: `mapCobolToGraph` scans + * `CALLS` edges and REMOVES the unresolved ones after adding resolved + * replacements (cobol-processor.ts). If the sink streamed from + * construction, that scan would see an empty set, no COBOL cross-program + * call would ever resolve, and the removal would be a silent no-op. Nothing + * before parse produces bulk edge volume, so deferring costs nothing. + */ + private armed = false; + /** + * First writer-construction failure (`fs.openSync` throwing on e.g. EMFILE). + * It happens inside the `SyncCsvWriter` constructor before a writer object + * exists to carry poison, so it is held at sink level and folded into the + * {@link finalize} error check — otherwise an open failure mid-emit would be + * swallowed by a caller's try/catch and silently drop the rest of the rows. + */ + private openFailure: unknown | undefined = undefined; + + constructor( + private readonly real: KnowledgeGraph, + private readonly csvDir: string, + private readonly chunkRows: number = DEFAULT_EMIT_CHUNK_ROWS, + ) { + this.validTables = new Set(NODE_TABLES as readonly string[]); + // Own directory, distinct from the PDG sink's: PdgEmitSink wipes and + // recreates its dir on construction and opens with O_EXCL, so a shared dir + // would destroy the other sink's manifest on a combined --pdg run. + fs.rmSync(csvDir, { recursive: true, force: true }); + fs.mkdirSync(csvDir, { recursive: true }); + } + + // ── routed writes ────────────────────────────────────────────────────────── + + /** Nodes are never streamed (see the file header) — always the real graph. */ + addNode(node: GraphNode): void { + this.real.addNode(node); + } + + /** + * Start streaming. Called once, by the `parse` phase, for the reason on + * {@link armed}. + */ + beginStreaming(): void { + this.armed = true; + } + + /** + * Exact dedup key, built to hold no reference to the relationship id. + * + * An id embeds both node ids in full — ~200 characters on this repo — and the + * only information it adds beyond `(type, source, target)` is a short trailing + * disambiguator, e.g. `emit-references.ts` appends `:line:col` so two calls + * between the same pair at different sites stay distinct. The endpoints are + * already interned for the columns, so the key reuses those indices and parses + * the tail into NUMBERS. + * + * Numbers matter for more than size: a key built by slicing or replacing + * inside a long string is a V8 sliced/cons string that keeps its parent alive, + * so the 200-character id would never be freed and the memory saving would + * silently fail to materialize. Parsing to numbers severs that link. + * + * Falls back to the full id when the tail is not a numeric `:a:b` form (other + * id shapes exist, e.g. `rel:contains:` has no tail). Correctness first: an + * unrecognized shape is stored exactly, just without the saving. + */ + private dedupKey(rel: GraphRelationship, srcIx: number, tgtIx: number): string { + const afterTarget = rel.id.lastIndexOf(rel.targetId); + if (afterTarget >= 0) { + const tail = rel.id.slice(afterTarget + rel.targetId.length); + if (tail.length === 0) return `${srcIx}|${tgtIx}|${rel.type}`; + // `:1483:6` -> two integers. Any non-numeric segment falls through. + if (tail.charCodeAt(0) === 58 /* ':' */) { + let a = 0; + let b = 0; + let seen = 0; + let ok = true; + for (const part of tail.slice(1).split(':')) { + const n = Number(part); + if (part.length === 0 || !Number.isInteger(n)) { + ok = false; + break; + } + if (seen === 0) a = n; + else if (seen === 1) b = n; + else { + ok = false; + break; + } + seen++; + } + // `seen` is part of the key: without it a one-segment tail `:7` (b + // defaults to 0) and a two-segment `:7:0` produce the same key, and the + // second edge is silently discarded as a duplicate. Distinct ids must + // never collapse — that is a lost relationship with no error. + if (ok) return `${srcIx}|${tgtIx}|${rel.type}|${seen}|${a}|${b}`; + } + } + return rel.id; + } + + private internNode(id: string): number { + const existing = this.nodeIds.get(id); + if (existing !== undefined) return existing; + const ix = this.nodeIdByIx.length; + this.nodeIdByIx.push(id); + this.nodeIds.set(id, ix); + return ix; + } + + /** Rebuild a streamed edge; its id is synthesized lazily, not stored. */ + private streamedAt(ix: number): GraphRelationship { + return new StreamedRelationship( + this.nodeIdByIx[this.srcIx[ix]], + this.nodeIdByIx[this.tgtIx[ix]], + this.relTypes[ix], + this.confidences[ix], + ix, + ); + } + + addRelationship(relationship: GraphRelationship): void { + if (!this.armed || RETAINED_REL_TYPES.has(relationship.type)) { + this.real.addRelationship(relationship); + return; + } + // Mirror KnowledgeGraph.addRelationship's first-writer-wins dedup. + + const fromLabel = getNodeLabel(relationship.sourceId); + const toLabel = getNodeLabel(relationship.targetId); + // Skip edges whose endpoint labels are not valid node tables — mirrors + // `RelPairRouter` exactly so the streamed set matches the whole-graph set. + if (!this.validTables.has(fromLabel) || !this.validTables.has(toLabel)) return; + + const pairKey = `${fromLabel}|${toLabel}`; + let writer = this.relWriters.get(pairKey); + if (writer === undefined) { + try { + writer = new SyncCsvWriter( + path.join(this.csvDir, `rel_${fromLabel}_${toLabel}.csv`), + REL_CSV_HEADER, + this.chunkRows, + ); + } catch (e) { + this.openFailure ??= e; + throw e; + } + this.relWriters.set(pairKey, writer); + } + // Intern first so the dedup key can reuse the indices. + const srcIx = this.internNode(relationship.sourceId); + const tgtIx = this.internNode(relationship.targetId); + const key = this.dedupKey(relationship, srcIx, tgtIx); + if (this.streamedIds.has(key)) return; + this.streamedIds.add(key); + + writer.addRow(buildRelRow(relationship)); + this.srcIx.push(srcIx); + this.tgtIx.push(tgtIx); + this.relTypes.push(relationship.type); + this.confidences.push(relationship.confidence); + } + + /** Flush + close every writer and return the COPY manifest. Every fd is + * closed even when a writer is poisoned; any IO fault — an in-flight write, + * a final-flush failure, or a writer-open failure (EMFILE) — is surfaced + * loudly here so a disk-full / out-of-fds run never hands a truncated CSV to + * the bulk COPY. */ + finalize(): GraphEmitManifest { + if (this.finalized) throw new Error('GraphEmitSink.finalize() called twice'); + this.finalized = true; + + const errors: unknown[] = []; + if (this.openFailure !== undefined) errors.push(this.openFailure); + + const relsByPair = new Map(); + let totalRows = 0; + for (const [pairKey, writer] of this.relWriters) { + writer.close(); + if (writer.poison !== undefined) errors.push(writer.poison); + relsByPair.set(pairKey, { csvPath: writer.csvPath, rows: writer.rows }); + totalRows += writer.rows; + } + + if (errors.length > 0) { + const first = errors[0]; + throw new Error( + `GraphEmitSink: ${errors.length} streamed CSV writer(s) hit an IO error ` + + `(disk-full / out-of-fds) during the emit — the persisted graph would be ` + + `truncated, so the run is failed rather than COPYing a partial CSV: ${ + first instanceof Error ? first.message : String(first) + }`, + ); + } + + return { relsByPair, totalRows }; + } + + /** Best-effort fd release for the error path — when the pipeline throws + * before {@link finalize} runs, the caller's `finally` calls this so the + * per-pair fds never leak. Idempotent with finalize via `finalized`. */ + close(): void { + if (this.finalized) return; + this.finalized = true; + for (const writer of this.relWriters.values()) { + try { + writer.close(); + } catch { + /* best-effort */ + } + } + } + + // ── delegated reads / retained mutations ─────────────────────────────────── + + get nodes(): GraphNode[] { + return this.real.nodes; + } + get relationships(): GraphRelationship[] { + return [...this.iterRelationships()]; + } + iterNodes(): IterableIterator { + return this.real.iterNodes(); + } + /** + * Retained edges followed by the streamed ones, so every consumer sees a + * complete graph and no phase needs to know streaming happened. This is what + * lets streaming be the default. + * + * Hand-rolled rather than a generator: a generator pays per-`yield` machinery + * on every one of millions of edges, and the pruner and process extraction + * walk this three times per analyze. + */ + iterRelationships(): IterableIterator { + const retained = this.real.iterRelationships(); + const self = this; + let ix = 0; + // One reused result record. The iterator protocol lets the producer hand + // back the same object each step — `for…of` reads `value`/`done` and drops + // it immediately — and allocating a fresh one per edge cost more than the + // generator it replaced. + const result: { value: GraphRelationship | undefined; done: boolean } = { + value: undefined, + done: true, + }; + const it: IterableIterator = { + next(): IteratorResult { + const fromReal = retained.next(); + if (fromReal.done !== true) { + result.value = fromReal.value; + result.done = false; + return result as IteratorResult; + } + if (ix < self.srcIx.length) { + result.value = self.streamedAt(ix++); + result.done = false; + return result as IteratorResult; + } + result.value = undefined; + result.done = true; + return result as IteratorResult; + }, + [Symbol.iterator]() { + return it; + }, + }; + return it; + } + + *iterRelationshipsByType(type: RelationshipType): IterableIterator { + yield* this.real.iterRelationshipsByType(type); + if (RETAINED_REL_TYPES.has(type)) return; // never streamed — skip the scan + for (let ix = 0; ix < this.srcIx.length; ix++) { + if (this.relTypes[ix] === type) yield this.streamedAt(ix); + } + } + forEachNode(fn: (node: GraphNode) => void): void { + this.real.forEachNode(fn); + } + /** + * The fast path: streamed edges are read straight out of the columns, so a + * whole-graph scan allocates NOTHING. This is what keeps iteration at parity + * with the object-based graph despite holding relationships columnar. + */ + forEachRelationshipFields( + fn: (sourceId: string, targetId: string, type: RelationshipType, confidence: number) => void, + ): void { + this.real.forEachRelationshipFields(fn); + for (let ix = 0; ix < this.srcIx.length; ix++) { + fn( + this.nodeIdByIx[this.srcIx[ix]], + this.nodeIdByIx[this.tgtIx[ix]], + this.relTypes[ix], + this.confidences[ix], + ); + } + } + + /** Direct loop rather than delegating to {@link iterRelationships}: this is + * the form community detection uses (twice), and skipping the generator and + * iterator protocol is measurably cheaper on a million-edge scan. */ + forEachRelationship(fn: (rel: GraphRelationship) => void): void { + this.real.forEachRelationship(fn); + for (let ix = 0; ix < this.srcIx.length; ix++) fn(this.streamedAt(ix)); + } + getNode(id: string): GraphNode | undefined { + return this.real.getNode(id); + } + get nodeCount(): number { + return this.real.nodeCount; + } + /** Retained edges only — streamed edges are gone from the heap by design. + * `run-analyze.ts` sizes the LadybugDB buffer pool from this, so it adds + * the manifest's `totalRows` back in (the hint only ever shrinks the pool, + * so under-reporting would starve the COPY at exactly the scale this + * feature targets). */ + get relationshipCount(): number { + return this.real.relationshipCount + this.srcIx.length; + } + removeNode(nodeId: string): boolean { + return this.real.removeNode(nodeId); + } + removeNodesByFile(filePath: string): number { + return this.real.removeNodesByFile(filePath); + } + /** + * Deliberately conservative. The dedup Set holds compact keys derived from a + * relationship's endpoints ({@link dedupKey}), and a bare id alone cannot be + * turned back into one — so a streamed edge is not directly identifiable here. + * + * Rather than risk the silent case (returning `false` for an edge that IS on + * disk and cannot be recalled), anything the real graph does not hold is + * treated as possibly-streamed once streaming has begun, and fails loudly. A + * genuinely-absent id therefore throws too, where the object-based graph would + * return `false`; that is acceptable because the only production caller is the + * COBOL resolver, which runs BEFORE the sink is armed and so takes the branch + * below. + * + * NOTE this diverges from {@link KnowledgeGraph.removeRelationship}, which + * returns `false` for an id it does not hold. Pinned by a test so the + * divergence stays deliberate. + */ + removeRelationship(relationshipId: string): boolean { + if (this.real.removeRelationship(relationshipId)) return true; + if (this.srcIx.length > 0) throw new StreamedRelationshipRemovalError(relationshipId); + return false; + } +} diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index 75e8dccbe..266dfcbe8 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -20,6 +20,7 @@ import { NodeTableName, } from './schema.js'; import { streamAllCSVsToDisk, type StreamedCSVResult } from './csv-generator.js'; +import type { GraphEmitManifest } from './graph-emit-sink.js'; import type { PdgEmitManifest } from './pdg-emit-sink.js'; import { getNodeLabel as deriveNodeLabel, type WriteStreamFactory } from './rel-pair-routing.js'; import { EMBEDDABLE_LABELS, type CachedEmbedding } from '../embeddings/types.js'; @@ -1017,6 +1018,15 @@ export const loadGraphToLbug = async ( * emits none — the manifest is the sole source and there is no double-COPY. */ pdgEmitManifest?: PdgEmitManifest, + /** + * Streamed structural-emit manifest (#2680). Unlike {@link pdgEmitManifest}, + * these pair keys are NOT disjoint from the whole-graph emit's: a streamed + * `CALLS` edge is `Function|Function`, exactly like the retained edges + * `streamAllCSVsToDisk` just wrote. So these files are APPENDED as additional + * COPY jobs for the same pair rather than merged into `relsByPair` (a Map, + * which holds one CSV per pair and would silently drop one of them). + */ + graphEmitManifest?: GraphEmitManifest, ) => { if (!conn) { throw new Error('LadybugDB not initialized. Call initLbug first.'); @@ -1156,17 +1166,32 @@ export const loadGraphToLbug = async ( let tCopyRels = tCopyNodes; let tFallback = tCopyNodes; - const insertedRels = totalValidRels; + // One COPY job per CSV FILE, not per label pair. The whole-graph emit writes + // at most one file per pair, but the streamed structural manifest (#2680) can + // contribute a second file for a pair the whole-graph emit also wrote — both + // must load. `relsByPair` stays a one-file-per-pair Map so the PDG merge above + // and every other consumer are untouched. + const copyJobs: Array<{ pairKey: string; csvPath: string; rows: number }> = []; + for (const [pairKey, meta] of relsByPair) { + copyJobs.push({ pairKey, csvPath: meta.csvPath, rows: meta.rows }); + } + if (graphEmitManifest) { + for (const [pairKey, meta] of graphEmitManifest.relsByPair) { + copyJobs.push({ pairKey, csvPath: meta.csvPath, rows: meta.rows }); + } + } + + const insertedRels = totalValidRels + (graphEmitManifest?.totalRows ?? 0); const warnings: string[] = []; let poolRemedyIssued = false; if (insertedRels > 0) { - log(`Loading edges: ${insertedRels.toLocaleString()} across ${relsByPair.size} types`); + log(`Loading edges: ${insertedRels.toLocaleString()} across ${copyJobs.length} CSV files`); let pairIdx = 0; let failedPairEdges = 0; const failedPairCsvPaths = new Set(); - for (const [pairKey, { csvPath: pairCsvPath, rows }] of relsByPair) { + for (const { pairKey, csvPath: pairCsvPath, rows } of copyJobs) { pairIdx++; const [fromLabel, toLabel] = pairKey.split('|'); const normalizedPath = normalizeCopyPath(pairCsvPath); @@ -1174,7 +1199,7 @@ export const loadGraphToLbug = async ( const copyQuery = `COPY ${REL_TABLE_NAME} FROM "${normalizedPath}" (from="${fromLabel}", to="${toLabel}", HEADER=true, ESCAPE='"', DELIM=',', QUOTE='"', PARALLEL=false, auto_detect=false)`; if (pairIdx % 5 === 0 || rows > 1000) { - log(`Loading edges: ${pairIdx}/${relsByPair.size} types (${fromLabel} -> ${toLabel})`); + log(`Loading edges: ${pairIdx}/${copyJobs.length} files (${fromLabel} -> ${toLabel})`); } // Use the captured `writeConn` (not the module-level `conn`) for the rel diff --git a/gitnexus/src/core/lbug/pdg-emit-sink.ts b/gitnexus/src/core/lbug/pdg-emit-sink.ts index 79cc05f58..79f8b899f 100644 --- a/gitnexus/src/core/lbug/pdg-emit-sink.ts +++ b/gitnexus/src/core/lbug/pdg-emit-sink.ts @@ -54,6 +54,7 @@ import { buildRelRow, } from './csv-generator.js'; import { getNodeLabel } from './rel-pair-routing.js'; +import { DEFAULT_EMIT_CHUNK_ROWS, SyncCsvWriter } from './sync-csv-writer.js'; import { NODE_TABLES, type NodeTableName } from './schema.js'; /** @@ -73,103 +74,9 @@ const PDG_EDGE_TYPES: ReadonlySet = new Set( ]); /** Default streamed-write buffer (rows). Matches the whole-graph emit's - * `FLUSH_EVERY` order of magnitude; overridable via `GITNEXUS_PDG_EMIT_CHUNK_SIZE`. */ -export const DEFAULT_PDG_EMIT_CHUNK_ROWS = 500; - -/** - * Synchronous buffered CSV writer. Buffers up to `chunkRows` rows, then issues - * one `fs.writeSync` straight to the OS (no in-process stream buffer). Header - * is written into the buffer at construction and is NOT counted in `rows` - * (matching `BufferedCSVWriter` semantics, so manifest row counts line up). - */ -class SyncCsvWriter { - private fd: number; - private buf: string[] = []; - private readonly chunkRows: number; - rows = 0; - /** - * First IO error this writer hit (a `fs.writeSync` short-write loop throwing - * on e.g. disk-full). Once poisoned the writer refuses further rows and - * skips its final flush; the sink surfaces it from {@link PdgEmitSink.finalize} - * so a truncated CSV is never handed to the bulk COPY (#2202 review #4). A - * streamed-write failure is an IO fault, not the CFG-logic error that the - * emit loop's per-file try/catch is built to swallow — poisoning routes it - * past that catch to a loud failure. - */ - poison: unknown | undefined = undefined; - - constructor( - readonly csvPath: string, - header: string, - chunkRows: number, - ) { - // Guard a 0/negative buffer: the flush modulo would never fire and `buf` - // would grow unbounded, defeating the whole point of streaming. - this.chunkRows = Math.max(1, chunkRows); - // Exclusive create (O_EXCL): the streamed-CSV dir is wiped + recreated fresh - // by the PdgEmitSink constructor before any writer opens a file, so the path - // never pre-exists — 'wx' both matches that invariant and refuses to follow - // a pre-planted symlink at the path (CWE-377 / CodeQL js/insecure-temporary-file). - this.fd = fs.openSync(csvPath, 'wx'); - this.buf.push(header); - } - - addRow(row: string): void { - // A poisoned writer is dead — stop buffering so memory can't grow on a - // writer whose fd is already in a bad state; finalize will report the fault. - if (this.poison !== undefined) return; - this.buf.push(row); - this.rows++; - // Flush on DATA-row count, not buffer length: the header occupies buf[0] - // until the first flush, so a `buf.length >= chunkRows` test would fire one - // row early on the first chunk. Counting rows makes every flush exactly - // `chunkRows` rows. - if (this.rows % this.chunkRows === 0) this.flushOrPoison(); - } - - /** Flush, recording (and re-throwing) any IO error as poison. Re-throwing - * lets the immediate caller log the per-file failure; the persisted `poison` - * is the backstop that makes finalize fail loudly even when that throw is - * swallowed by the emit loop's CFG try/catch. */ - private flushOrPoison(): void { - try { - this.flush(); - } catch (e) { - this.poison ??= e; - throw e; - } - } - - private flush(): void { - if (this.buf.length === 0) return; - const data = Buffer.from(this.buf.join('\n') + '\n', 'utf8'); - // fs.writeSync can return a short byte count; loop until the whole buffer - // lands so a partial write never truncates a CSV row mid-field. - let offset = 0; - while (offset < data.length) { - offset += fs.writeSync(this.fd, data, offset, data.length - offset); - } - this.buf.length = 0; - } - - /** Flush remaining rows (unless already poisoned) and close the fd. Never - * throws: a final-flush IO error is recorded as poison and the fd is still - * closed, so a write error neither leaks an fd nor escapes here — the sink - * reads {@link poison} after closing every writer and fails loudly then. */ - close(): void { - try { - if (this.poison === undefined) this.flush(); - } catch (e) { - this.poison ??= e; - } finally { - try { - fs.closeSync(this.fd); - } catch { - /* fd may already be invalid after an IO fault — nothing to recover */ - } - } - } -} + * `FLUSH_EVERY` order of magnitude; overridable via `GITNEXUS_PDG_EMIT_CHUNK_SIZE`. + * Aliases the shared default in `sync-csv-writer.ts` (#2680 extraction). */ +export const DEFAULT_PDG_EMIT_CHUNK_ROWS = DEFAULT_EMIT_CHUNK_ROWS; /** * COPY manifest produced by {@link PdgEmitSink.finalize}. Shaped to merge @@ -374,6 +281,11 @@ export class PdgEmitSink implements KnowledgeGraph { forEachRelationship(fn: (rel: GraphRelationship) => void): void { this.real.forEachRelationship(fn); } + forEachRelationshipFields( + fn: (sourceId: string, targetId: string, type: RelationshipType, confidence: number) => void, + ): void { + this.real.forEachRelationshipFields(fn); + } getNode(id: string): GraphNode | undefined { return this.real.getNode(id); } diff --git a/gitnexus/src/core/lbug/sync-csv-writer.ts b/gitnexus/src/core/lbug/sync-csv-writer.ts new file mode 100644 index 000000000..fecff951b --- /dev/null +++ b/gitnexus/src/core/lbug/sync-csv-writer.ts @@ -0,0 +1,110 @@ +/** + * Synchronous buffered CSV writer, shared by the streaming emit sinks. + * + * Extracted verbatim from `pdg-emit-sink.ts` (issue #2202) so the structural + * `GraphEmitSink` (#2680) reuses the same buffering and IO-fault discipline + * instead of duplicating ~90 lines of it. No behaviour change: `PdgEmitSink` + * imports this class and is otherwise untouched. + * + * Why synchronous? The emit loops these sinks sit under are synchronous — there + * is no `await` point to drain an async stream, so a `WriteStream` would + * accumulate unwritten chunks in process memory across millions of rows, + * defeating the RSS bound this exists to provide. `fs.writeSync` goes straight + * to the OS; resident memory is bounded to one `chunkRows` buffer. This mirrors + * the sync-shard pattern in `storage/parsedfile-store.ts`. + */ + +import fs from 'fs'; + +/** Default streamed-write buffer (rows), shared by both sinks. */ +export const DEFAULT_EMIT_CHUNK_ROWS = 500; + +export class SyncCsvWriter { + private fd: number; + private buf: string[] = []; + private readonly chunkRows: number; + rows = 0; + /** + * First IO error this writer hit (a `fs.writeSync` short-write loop throwing + * on e.g. disk-full). Once poisoned the writer refuses further rows and + * skips its final flush; the owning sink surfaces it from its `finalize()` + * so a truncated CSV is never handed to the bulk COPY (#2202 review #4). A + * streamed-write failure is an IO fault, not the logic error that the emit + * loops' per-file try/catch is built to swallow — poisoning routes it past + * that catch to a loud failure. + */ + poison: unknown | undefined = undefined; + + constructor( + readonly csvPath: string, + header: string, + chunkRows: number, + ) { + // Guard a 0/negative buffer: the flush modulo would never fire and `buf` + // would grow unbounded, defeating the whole point of streaming. + this.chunkRows = Math.max(1, chunkRows); + // Exclusive create (O_EXCL): the streamed-CSV dir is wiped + recreated fresh + // by the owning sink's constructor before any writer opens a file, so the + // path never pre-exists — 'wx' both matches that invariant and refuses to + // follow a pre-planted symlink at the path (CWE-377 / CodeQL + // js/insecure-temporary-file). + this.fd = fs.openSync(csvPath, 'wx'); + this.buf.push(header); + } + + addRow(row: string): void { + // A poisoned writer is dead — stop buffering so memory can't grow on a + // writer whose fd is already in a bad state; finalize will report the fault. + if (this.poison !== undefined) return; + this.buf.push(row); + this.rows++; + // Flush on DATA-row count, not buffer length: the header occupies buf[0] + // until the first flush, so a `buf.length >= chunkRows` test would fire one + // row early on the first chunk. Counting rows makes every flush exactly + // `chunkRows` rows. + if (this.rows % this.chunkRows === 0) this.flushOrPoison(); + } + + /** Flush, recording (and re-throwing) any IO error as poison. Re-throwing + * lets the immediate caller log the per-file failure; the persisted `poison` + * is the backstop that makes finalize fail loudly even when that throw is + * swallowed by an emit loop's try/catch. */ + private flushOrPoison(): void { + try { + this.flush(); + } catch (e) { + this.poison ??= e; + throw e; + } + } + + private flush(): void { + if (this.buf.length === 0) return; + const data = Buffer.from(this.buf.join('\n') + '\n', 'utf8'); + // fs.writeSync can return a short byte count; loop until the whole buffer + // lands so a partial write never truncates a CSV row mid-field. + let offset = 0; + while (offset < data.length) { + offset += fs.writeSync(this.fd, data, offset, data.length - offset); + } + this.buf.length = 0; + } + + /** Flush remaining rows (unless already poisoned) and close the fd. Never + * throws: a final-flush IO error is recorded as poison and the fd is still + * closed, so a write error neither leaks an fd nor escapes here — the owning + * sink reads {@link poison} after closing every writer and fails loudly then. */ + close(): void { + try { + if (this.poison === undefined) this.flush(); + } catch (e) { + this.poison ??= e; + } finally { + try { + fs.closeSync(this.fd); + } catch { + /* fd may already be invalid after an IO fault — nothing to recover */ + } + } + } +} diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index cf535cf9f..4a51e39d8 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -38,7 +38,11 @@ import { LbugWipeError, DELETE_FILES_CHUNK_SIZE, } from './lbug/lbug-adapter.js'; -import { estimateBufferPool, setBufferPoolSizeHint } from './lbug/lbug-config.js'; +import { + estimateBufferPool, + setBufferPoolSizeHint, + resolveNativeSafeStorageDir, +} from './lbug/lbug-config.js'; import { escapeCypherString } from './lbug/cypher-escape.js'; import { buildSearchIndexesOrDegrade, @@ -280,6 +284,11 @@ export interface AnalyzeOptions { * `DEFAULT_PDG_EMIT_CHUNK_ROWS`. May also be set via * `GITNEXUS_PDG_EMIT_CHUNK_SIZE`. Memory-only (#2202). */ pdgEmitChunkSize?: number; + /** Streamed structural graph emit (#2680). Honored only on a full rebuild + * (`force === true`). May also be enabled via `GITNEXUS_STREAM_GRAPH_EMIT`. + * Trades community detection, process extraction and PDG taint summaries for + * a ~2.9x reduction of in-memory graph heap. */ + streamGraphEmit?: boolean; /** * Default branch threaded into generated AGENTS.md / CLAUDE.md so the * regression-compare example uses the configured branch instead of a @@ -585,6 +594,38 @@ export const resolveStreamPdgEmit = (options: { options.force === true && (options.streamPdgEmit === true || parseTruthyEnv(process.env.GITNEXUS_STREAM_PDG_EMIT)); +/** + * Resolve whether streamed structural graph emit is on for this run (#2680). + * + * **On by default.** It costs nothing observable: the sink answers a complete + * relationship read, so community detection, process extraction, the taint + * fixpoint and the local-symbol pruner all behave exactly as they do without it + * — the edges simply live in columns and on disk instead of as objects. There is + * no reason to make a user opt in to using less memory. + * + * Two conditions still bound it: + * + * - `force === true`. Sound only on a full rebuild, because the incremental + * writeback (`extractChangedSubgraph`) reads relationships back out of the + * in-memory graph. Same gate, and same reason, as {@link resolveStreamPdgEmit}. + * - `GITNEXUS_STREAM_GRAPH_EMIT=0` (or an explicit `streamGraphEmit: false`) + * turns it off. The escape hatch exists for bisecting a suspected + * streaming-related fault, not as a routine choice. + * + * Memory-only: not part of {@link resolvePdgConfig}, so toggling never trips + * `pdgModeMismatch`. Read every call (not memoized) so `vi.stubEnv` works. + */ +export const resolveStreamGraphEmit = (options: { + force?: boolean; + streamGraphEmit?: boolean; +}): boolean => { + if (options.force !== true) return false; + if (options.streamGraphEmit !== undefined) return options.streamGraphEmit; + // Unset ⇒ on. Set ⇒ honour it, so `=0` / `=false` is the escape hatch. + const raw = process.env.GITNEXUS_STREAM_GRAPH_EMIT; + return raw === undefined || raw === '' ? true : parseTruthyEnv(raw); +}; + /** * Resolve the streamed PDG-emit write-buffer size (#2202). Explicit option wins * over `GITNEXUS_PDG_EMIT_CHUNK_SIZE`; `undefined` ⇒ the sink's @@ -795,6 +836,10 @@ async function runFullAnalysisInner( const progress = (phase: string, percent: number, message: string) => callbacks.onProgress(phase, percent, message); + // Streamed structural emit (#2680), resolved once so the pipeline flag and the + // CSV-dir resolution below cannot disagree. + const streamGraphEmitActive = resolveStreamGraphEmit(options); + // FTS-config validation and the degraded-parse counter reset happen in the // `runFullAnalysis` wrapper (before the lock is taken). @@ -1392,6 +1437,16 @@ async function runFullAnalysisInner( // offloaded BasicBlock layer. Memory-only; byte-identical output. streamPdgEmit: resolveStreamPdgEmit(options), pdgEmitChunkSize: resolvePdgEmitChunkSize(options), + // Streamed structural emit (#2680) — same full-rebuild gate as the PDG + // toggle above, for the same incremental-writeback reason. + streamGraphEmit: streamGraphEmitActive, + // Resolved ONLY when streaming is active: on a Windows non-ASCII storage + // path this helper mkdtempSyncs a real directory, so evaluating it + // unconditionally would leak one temp dir per analyze even with the flag + // off. The PDG sibling resolves inside its guard for the same reason. + graphEmitCsvDir: streamGraphEmitActive + ? resolveNativeSafeStorageDir(storagePath, 'graph-csv') + : undefined, fetchWrappers: options.fetchWrappers, }, ); @@ -1576,7 +1631,15 @@ async function runFullAnalysisInner( // the pool; env override / no-hint paths are unchanged. See // resolveBufferManagerSize / estimateBufferPool. setBufferPoolSizeHint( - estimateBufferPool(pipelineResult.graph.nodeCount + pipelineResult.graph.relationshipCount), + estimateBufferPool( + pipelineResult.graph.nodeCount + + pipelineResult.graph.relationshipCount + + // Streamed edges left the heap but still get COPYed, so they are part of + // the real load volume (#2680). The hint only ever SHRINKS the pool, so + // omitting them would starve the COPY at exactly the scale streaming + // exists to serve. + (pipelineResult.graphEmitManifest?.totalRows ?? 0), + ), ); // Full rebuild (POSIX) builds into the temp `buildPath`; incremental and @@ -2003,6 +2066,7 @@ async function runFullAnalysisInner( progress('lbug', pct, msg); }, pipelineResult.pdgEmitManifest, + pipelineResult.graphEmitManifest, ); } diff --git a/gitnexus/src/types/pipeline.ts b/gitnexus/src/types/pipeline.ts index 4cbb28886..00d530091 100644 --- a/gitnexus/src/types/pipeline.ts +++ b/gitnexus/src/types/pipeline.ts @@ -3,6 +3,7 @@ import { CommunityDetectionResult } from '../core/ingestion/community-processor. import { ProcessDetectionResult } from '../core/ingestion/process-processor.js'; import type { ResolutionOutcome } from '../core/ingestion/scope-resolution/resolution-outcome.js'; import type { PdgEmitManifest } from '../core/lbug/pdg-emit-sink.js'; +import type { GraphEmitManifest } from '../core/lbug/graph-emit-sink.js'; // CLI-specific: in-memory result with graph + detection results export interface PipelineResult { @@ -36,4 +37,12 @@ export interface PipelineResult { * layer (if any) is resident in `graph` and persists via the whole-graph emit. */ pdgEmitManifest?: PdgEmitManifest; + /** + * Streamed structural-emit COPY manifest (#2680). Present only when + * `streamGraphEmit` was active (full rebuild + enabled): the per-pair CSVs of + * relationships that never entered the in-memory graph, for `loadGraphToLbug` + * to COPY ALONGSIDE the whole-graph CSVs (their pair keys overlap, so they are + * additional COPY jobs, not map entries). + */ + graphEmitManifest?: GraphEmitManifest; } diff --git a/gitnexus/test/integration/graph-emit-streaming-roundtrip.test.ts b/gitnexus/test/integration/graph-emit-streaming-roundtrip.test.ts new file mode 100644 index 000000000..df8dbf1df --- /dev/null +++ b/gitnexus/test/integration/graph-emit-streaming-roundtrip.test.ts @@ -0,0 +1,181 @@ +/** + * Streamed structural emit — differential set-identity (issue #2680). + * + * The acceptance property: for the same node/edge set, the rows that reach the + * bulk COPY must be IDENTICAL whether streaming is on or off. With streaming + * on those rows arrive from two places — the residual in-memory graph (via + * `streamAllCSVsToDisk`) plus the sink's per-pair CSVs — and their union has to + * equal the single whole-graph emit. + * + * Modelled on `pdg-emit-streaming-roundtrip.test.ts`, which likewise drives the + * sink directly rather than running `analyze`: the guarantee under test is + * about emitted rows, and going through the worker pool would add a large + * amount of unrelated machinery without strengthening the assertion. + * + * Guarantee is set-level, not byte-level: streamed rows are written in emit + * order and are not re-sorted, so file bytes may differ while the row SET (and + * therefore the loaded graph) does not. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import fs from 'node:fs'; +import fsp from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; + +import { createKnowledgeGraph } from '../../src/core/graph/graph.js'; +import { streamAllCSVsToDisk } from '../../src/core/lbug/csv-generator.js'; +import { GraphEmitSink } from '../../src/core/lbug/graph-emit-sink.js'; +import type { KnowledgeGraph } from '../../src/core/graph/types.js'; +import type { GraphNode, GraphRelationship } from 'gitnexus-shared'; + +const FILE_PATH = 'src/mod.ts'; + +const fileNode = (): GraphNode => ({ + id: `File:${FILE_PATH}`, + label: 'File', + properties: { name: 'mod.ts', filePath: FILE_PATH }, +}); + +const fnNode = (n: number): GraphNode => ({ + id: `Function:${FILE_PATH}:fn${n}`, + label: 'Function', + properties: { + name: `fn${n}`, + filePath: FILE_PATH, + startLine: n, + endLine: n + 1, + isExported: false, + }, +}); + +const classNode = (n: number): GraphNode => ({ + id: `Class:${FILE_PATH}:Cls${n}`, + label: 'Class', + properties: { name: `Cls${n}`, filePath: FILE_PATH, startLine: n, endLine: n + 5 }, +}); + +const edge = ( + type: GraphRelationship['type'], + sourceId: string, + targetId: string, +): GraphRelationship => ({ + id: `${type}:${sourceId}->${targetId}`, + sourceId, + targetId, + type, + confidence: 1, + reason: 'test', +}); + +/** A mix deliberately spanning both sides of RETAINED_REL_TYPES, plus a + * duplicate id and a self-edge — the cases where a naive sink diverges. */ +const buildFixture = ( + graph: KnowledgeGraph, +): { nodes: GraphNode[]; relationships: GraphRelationship[] } => { + const nodes: GraphNode[] = [fileNode(), classNode(1), classNode(2)]; + for (let i = 0; i < 12; i++) nodes.push(fnNode(i)); + + const relationships: GraphRelationship[] = []; + for (const n of nodes) relationships.push(edge('DEFINES', `File:${FILE_PATH}`, n.id)); // retained + for (let i = 0; i < 11; i++) { + relationships.push( + edge('CALLS', `Function:${FILE_PATH}:fn${i}`, `Function:${FILE_PATH}:fn${i + 1}`), + ); // streamed + relationships.push(edge('ACCESSES', `Function:${FILE_PATH}:fn${i}`, `Class:${FILE_PATH}:Cls1`)); // streamed + } + relationships.push(edge('EXTENDS', `Class:${FILE_PATH}:Cls2`, `Class:${FILE_PATH}:Cls1`)); // retained + relationships.push(edge('IMPORTS', `File:${FILE_PATH}`, `Class:${FILE_PATH}:Cls1`)); // streamed + // Self-edge and an exact duplicate id — both must appear exactly once. + relationships.push(edge('CALLS', `Function:${FILE_PATH}:fn0`, `Function:${FILE_PATH}:fn0`)); + relationships.push(edge('CALLS', `Function:${FILE_PATH}:fn0`, `Function:${FILE_PATH}:fn1`)); + + for (const n of nodes) graph.addNode(n); + for (const r of relationships) graph.addRelationship(r); + return { nodes, relationships }; +}; + +/** Every relationship row emitted for a graph, as a sorted `pairKey\0row` set. */ +const relRowsFromCsvDir = async (csvDir: string): Promise => { + const out: string[] = []; + for (const name of await fsp.readdir(csvDir)) { + if (!name.startsWith('rel_') || !name.endsWith('.csv')) continue; + const pairKey = name.slice('rel_'.length, -'.csv'.length); + const text = await fsp.readFile(path.join(csvDir, name), 'utf8'); + for (const line of text.split('\n').slice(1)) { + if (line.length > 0) out.push(`${pairKey}\u0000${line}`); + } + } + return out.sort(); +}; + +let tmpRoot: string; + +beforeEach(() => { + tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'graph-emit-roundtrip-')); + fs.mkdirSync(path.join(tmpRoot, 'repo'), { recursive: true }); + fs.writeFileSync(path.join(tmpRoot, 'repo', 'src-placeholder'), ''); +}); + +afterEach(() => { + fs.rmSync(tmpRoot, { recursive: true, force: true }); +}); + +describe('streamed structural emit is set-identical to the whole-graph emit', () => { + it('emits the same relationship row set with the flag on and off', async () => { + const repoPath = path.join(tmpRoot, 'repo'); + + // ── Arm A: streaming OFF — one whole-graph emit over everything. + const graphOff = createKnowledgeGraph(); + buildFixture(graphOff); + const csvDirOff = path.join(tmpRoot, 'csv-off'); + await streamAllCSVsToDisk(graphOff, repoPath, csvDirOff); + const rowsOff = await relRowsFromCsvDir(csvDirOff); + + // ── Arm B: streaming ON — retained edges stay in the graph and are emitted + // by streamAllCSVsToDisk; the rest were streamed by the sink. + const realOn = createKnowledgeGraph(); + const sinkCsvDir = path.join(tmpRoot, 'csv-sink'); + const sink = new GraphEmitSink(realOn, sinkCsvDir); + sink.beginStreaming(); + buildFixture(sink); + const manifest = sink.finalize(); + + const csvDirOn = path.join(tmpRoot, 'csv-on'); + await streamAllCSVsToDisk(realOn, repoPath, csvDirOn); + + const rowsOn = [ + ...(await relRowsFromCsvDir(csvDirOn)), + ...(await relRowsFromCsvDir(sinkCsvDir)), + ].sort(); + + // The union of (residual graph emit + streamed CSVs) is the whole-graph emit. + expect(rowsOn).toEqual(rowsOff); + + // And the split is real — this is what buys the memory, so assert it rather + // than let a sink that streamed nothing pass the equality above. + expect(manifest.totalRows).toBeGreaterThan(0); + expect(realOn.relationshipCount).toBeGreaterThan(0); + expect(realOn.relationshipCount).toBeLessThan(graphOff.relationshipCount); + expect(realOn.relationshipCount + manifest.totalRows).toBe(graphOff.relationshipCount); + }); + + it('emits an identical node row set — nodes are never streamed', async () => { + const repoPath = path.join(tmpRoot, 'repo'); + + const graphOff = createKnowledgeGraph(); + buildFixture(graphOff); + const csvDirOff = path.join(tmpRoot, 'csv-off'); + const resultOff = await streamAllCSVsToDisk(graphOff, repoPath, csvDirOff); + + const realOn = createKnowledgeGraph(); + const sink = new GraphEmitSink(realOn, path.join(tmpRoot, 'csv-sink')); + sink.beginStreaming(); + buildFixture(sink); + sink.finalize(); + const csvDirOn = path.join(tmpRoot, 'csv-on'); + const resultOn = await streamAllCSVsToDisk(realOn, repoPath, csvDirOn); + + expect(realOn.nodeCount).toBe(graphOff.nodeCount); + expect([...resultOn.nodeFiles.keys()].sort()).toEqual([...resultOff.nodeFiles.keys()].sort()); + }); +}); diff --git a/gitnexus/test/unit/lbug/graph-emit-sink.test.ts b/gitnexus/test/unit/lbug/graph-emit-sink.test.ts new file mode 100644 index 000000000..7f051ce53 --- /dev/null +++ b/gitnexus/test/unit/lbug/graph-emit-sink.test.ts @@ -0,0 +1,403 @@ +/** + * GraphEmitSink unit tests (issue #2680). + * + * Verifies the streaming structural emit sink: + * - routes non-retained relationships to bounded CSV-on-disk and never stores + * them, while retained types reach the real graph untouched; + * - dedups by relationship id (the whole-graph emit does, and COPY into a + * PK-bearing table would violate on a repeat) — PdgEmitSink relies on an + * upstream per-file guarantee that does NOT exist for structural edges; + * - refuses to silently forget a streamed edge on removeRelationship; + * - exposes the streamed-endpoint predicate the local-symbol pruner needs to + * avoid pruning a node that a streamed edge still references; + * - fails loudly rather than handing a truncated CSV to the bulk COPY. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import fs from 'node:fs'; +import fsp from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; + +import { createKnowledgeGraph } from '../../../src/core/graph/graph.js'; +import { + GraphEmitSink, + RETAINED_REL_TYPES, + StreamedRelationshipRemovalError, +} from '../../../src/core/lbug/graph-emit-sink.js'; +import type { GraphRelationship } from 'gitnexus-shared'; + +const fnId = (name: string): string => `Function:src/a.ts:${name}`; + +const rel = ( + type: GraphRelationship['type'], + from: string, + to: string, + suffix = '', +): GraphRelationship => ({ + id: `${type}:${fnId(from)}->${fnId(to)}${suffix}`, + sourceId: fnId(from), + targetId: fnId(to), + type, + confidence: 1, + reason: 'direct', +}); + +const dataRows = async (csvPath: string): Promise => { + const text = await fsp.readFile(csvPath, 'utf8'); + return text + .split('\n') + .filter((l) => l.length > 0) + .slice(1); // drop header +}; + +let tmpRoot: string; +let csvDir: string; + +beforeEach(() => { + tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'graph-emit-sink-')); + csvDir = path.join(tmpRoot, 'streamed'); +}); + +afterEach(() => { + fs.rmSync(tmpRoot, { recursive: true, force: true }); +}); + +describe('GraphEmitSink routing', () => { + it('streams a non-retained type to CSV and keeps it out of the graph', async () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + + sink.addRelationship(rel('CALLS', 'a', 'b')); + const manifest = sink.finalize(); + + expect(real.relationshipCount).toBe(0); + expect(manifest).toMatchObject({ totalRows: 1 }); + const pair = manifest.relsByPair.get('Function|Function'); + expect(pair).toMatchObject({ rows: 1 }); + expect(await dataRows(pair!.csvPath)).toHaveLength(1); + }); + + it('delegates every retained type to the real graph and writes no CSV', () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + + for (const type of RETAINED_REL_TYPES) { + sink.addRelationship(rel(type, 'a', 'b', `:${type}`)); + } + const manifest = sink.finalize(); + + expect(real.relationshipCount).toBe(RETAINED_REL_TYPES.size); + expect(manifest).toMatchObject({ totalRows: 0 }); + expect(manifest.relsByPair.size).toBe(0); + }); + + it('never streams nodes — they stay in the real graph', () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + + sink.addNode({ + id: fnId('a'), + label: 'Function', + properties: { name: 'a', filePath: 'src/a.ts', startLine: 1, endLine: 2 }, + }); + sink.finalize(); + + expect(real.nodeCount).toBe(1); + expect(fs.readdirSync(csvDir)).toEqual([]); + }); + + it('skips edges whose endpoint labels are not valid node tables', () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + + sink.addRelationship({ + id: 'CALLS:bogus->alsobogus', + sourceId: 'NotATable:src/a.ts:x', + targetId: 'NotATable:src/a.ts:y', + type: 'CALLS', + confidence: 1, + reason: 'direct', + }); + const manifest = sink.finalize(); + + expect(manifest).toMatchObject({ totalRows: 0 }); + expect(real.relationshipCount).toBe(0); + }); +}); + +describe('GraphEmitSink arming', () => { + it('retains everything in the graph until armed', () => { + // The pre-parse phases are not all write-only: mapCobolToGraph scans CALLS + // edges and removes the unresolved ones. If the sink streamed from + // construction, that scan would see nothing and COBOL cross-program calls + // would silently stop resolving. + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + + sink.addRelationship(rel('CALLS', 'a', 'b')); + + expect(real.relationshipCount).toBe(1); + expect(sink.finalize()).toMatchObject({ totalRows: 0 }); + }); + + it('removal of a pre-arm CALLS edge still works (the COBOL path)', () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + const unresolved = rel('CALLS', 'a', 'b'); + sink.addRelationship(unresolved); + + expect(sink.removeRelationship(unresolved.id)).toBe(true); + expect(real.relationshipCount).toBe(0); + sink.finalize(); + }); +}); + +describe('GraphEmitSink dedup', () => { + it('writes a duplicate relationship id exactly once', async () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + + const duplicated = rel('CALLS', 'a', 'b'); + sink.addRelationship(duplicated); + sink.addRelationship(duplicated); + sink.addRelationship({ ...duplicated }); + const manifest = sink.finalize(); + + // A second row would violate the relationship table's PK on COPY. + expect(manifest).toMatchObject({ totalRows: 1 }); + expect(await dataRows(manifest.relsByPair.get('Function|Function')!.csvPath)).toHaveLength(1); + }); +}); + +describe('GraphEmitSink removal safety', () => { + it('throws rather than silently forgetting an already-streamed edge', () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + const streamed = rel('CALLS', 'a', 'b'); + sink.addRelationship(streamed); + + expect(() => sink.removeRelationship(streamed.id)).toThrow(StreamedRelationshipRemovalError); + sink.finalize(); + }); + + it('still removes a retained edge normally', () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + const retained = rel('DEFINES', 'a', 'b'); + sink.addRelationship(retained); + + expect(sink.removeRelationship(retained.id)).toBe(true); + expect(real.relationshipCount).toBe(0); + sink.finalize(); + }); +}); + +describe('GraphEmitSink reads are complete', () => { + it('iterRelationships returns streamed edges alongside retained ones', () => { + // This is the property that lets streaming be the default: every consumer + // (communities, processes, taint, the pruner) reads through this and must + // see the whole graph, not just what stayed in memory. + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + + sink.addRelationship(rel('DEFINES', 'file', 'fn')); // retained + sink.addRelationship(rel('CALLS', 'a', 'b')); // streamed + sink.addRelationship(rel('ACCESSES', 'b', 'c')); // streamed + + const seen = [...sink.iterRelationships()]; + expect(seen.map((r) => r.type).sort()).toEqual(['ACCESSES', 'CALLS', 'DEFINES']); + expect(sink.relationshipCount).toBe(3); + // The real graph still holds only the retained one — the saving is real. + expect(real.relationshipCount).toBe(1); + sink.finalize(); + }); + + it('preserves endpoints and confidence on a streamed edge', () => { + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + sink.addRelationship({ ...rel('CALLS', 'caller', 'callee'), confidence: 0.25 }); + + expect([...sink.iterRelationships()]).toMatchObject([ + { sourceId: fnId('caller'), targetId: fnId('callee'), type: 'CALLS', confidence: 0.25 }, + ]); + sink.finalize(); + }); + + it('iterRelationshipsByType finds a streamed type', () => { + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + sink.addRelationship(rel('CALLS', 'a', 'b')); + sink.addRelationship(rel('ACCESSES', 'a', 'c')); + + expect([...sink.iterRelationshipsByType('CALLS')]).toHaveLength(1); + expect([...sink.iterRelationshipsByType('ACCESSES')]).toHaveLength(1); + expect([...sink.iterRelationshipsByType('EXTENDS')]).toEqual([]); + sink.finalize(); + }); + + it('forEachRelationship visits streamed edges too', () => { + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + sink.addRelationship(rel('CALLS', 'a', 'b')); + + const visited: string[] = []; + sink.forEachRelationship((r) => visited.push(r.type)); + expect(visited).toEqual(['CALLS']); + sink.finalize(); + }); +}); + +describe('GraphEmitSink IO faults', () => { + it('surfaces a writer-open failure from finalize instead of a partial manifest', () => { + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + sink.addRelationship(rel('CALLS', 'a', 'b')); + + // Destroy the CSV dir so the next pair's writer cannot be opened, the way + // an out-of-fds (EMFILE) or disk-full run would fail mid-emit. + fs.rmSync(csvDir, { recursive: true, force: true }); + expect(() => + sink.addRelationship({ + id: 'CALLS:File:src/a.ts->Function:src/a.ts:b', + sourceId: 'File:src/a.ts', + targetId: fnId('b'), + type: 'CALLS', + confidence: 1, + reason: 'direct', + }), + ).toThrow(); + + expect(() => sink.finalize()).toThrow(/streamed CSV writer\(s\) hit an IO error/); + }); + + it('refuses a second finalize', () => { + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + sink.finalize(); + expect(() => sink.finalize()).toThrow(/called twice/); + }); +}); + +describe('dedup key exactness', () => { + const endpoints = { sourceId: fnId('f'), targetId: fnId('g') }; + const withId = (id: string): GraphRelationship => ({ + id, + ...endpoints, + type: 'CALLS', + confidence: 1, + reason: 'direct', + }); + + it('keeps two ids that differ only in how many tail segments they carry', () => { + // Regression: the dedup key packs the id's trailing numeric segments, and an + // absent second segment defaults to 0. Without the segment COUNT in the key, + // `:7` and `:7:0` collapse onto one key and the second edge is silently + // discarded — a lost relationship with no error. Distinct ids must never + // collapse; identical ones must (see the duplicate test above). + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + + sink.addRelationship(withId(`rel:CALLS:${endpoints.sourceId}->${endpoints.targetId}:7`)); + sink.addRelationship(withId(`rel:CALLS:${endpoints.sourceId}->${endpoints.targetId}:7:0`)); + + expect(sink.relationshipCount).toBe(2); + expect(sink.finalize()).toMatchObject({ totalRows: 2 }); + }); + + it('keeps two call sites between the same pair', () => { + // The `:line:col` case from emit-references — same endpoints and type, so + // identical CSV rows; only the id distinguishes them, and the whole-graph + // emit keeps both. + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + + sink.addRelationship(withId(`rel:CALLS:${endpoints.sourceId}->${endpoints.targetId}:10:4`)); + sink.addRelationship(withId(`rel:CALLS:${endpoints.sourceId}->${endpoints.targetId}:99:7`)); + + expect(sink.relationshipCount).toBe(2); + sink.finalize(); + }); + + it('still collapses a genuinely repeated id', () => { + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + const id = `rel:CALLS:${endpoints.sourceId}->${endpoints.targetId}:10:4`; + + sink.addRelationship(withId(id)); + sink.addRelationship(withId(id)); + + expect(sink.relationshipCount).toBe(1); + sink.finalize(); + }); + + it('falls back to the full id for a non-numeric tail', () => { + // `rel:imports:...:${localName}` has a textual tail; the compact form does + // not apply and the id must be stored verbatim rather than truncated. + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + + sink.addRelationship(withId(`rel:IMPORTS:${endpoints.sourceId}->${endpoints.targetId}:alpha`)); + sink.addRelationship(withId(`rel:IMPORTS:${endpoints.sourceId}->${endpoints.targetId}:beta`)); + + expect(sink.relationshipCount).toBe(2); + sink.finalize(); + }); +}); + +describe('removeRelationship contract divergence', () => { + it('throws for an absent id once streaming has begun, by design', () => { + // KnowledgeGraph.removeRelationship returns false for an id it does not + // hold. The sink cannot rebuild a compact dedup key from a bare id, so it + // refuses to answer "false" for something that might already be on disk and + // unrecallable. Pinned so the divergence stays deliberate. + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + sink.addRelationship(rel('CALLS', 'a', 'b')); + + expect(() => sink.removeRelationship('rel:CALLS:never:emitted')).toThrow( + StreamedRelationshipRemovalError, + ); + sink.finalize(); + }); + + it('returns false for an absent id before anything has streamed', () => { + const sink = new GraphEmitSink(createKnowledgeGraph(), csvDir); + sink.beginStreaming(); + + expect(sink.removeRelationship('rel:CALLS:never:emitted')).toBe(false); + sink.finalize(); + }); +}); + +describe('field scan matches the object scan', () => { + it('yields the same (source, target, type, confidence) tuples either way', () => { + // Guards the five whole-graph scans converted to forEachRelationshipFields: + // a divergence between the two forms would silently skew community + // detection, process extraction and the pruner. + const real = createKnowledgeGraph(); + const sink = new GraphEmitSink(real, csvDir); + sink.beginStreaming(); + sink.addRelationship(rel('DEFINES', 'file', 'fn')); + sink.addRelationship(rel('CALLS', 'a', 'b')); + sink.addRelationship({ ...rel('ACCESSES', 'b', 'c'), confidence: 0.5 }); + + const viaObjects = [...sink.iterRelationships()] + .map((r) => `${r.sourceId}|${r.targetId}|${r.type}|${r.confidence}`) + .sort(); + const viaFields: string[] = []; + sink.forEachRelationshipFields((s, t, ty, c) => viaFields.push(`${s}|${t}|${ty}|${c}`)); + + expect(viaFields.sort()).toEqual(viaObjects); + sink.finalize(); + }); +}); diff --git a/gitnexus/test/unit/stream-graph-emit-config.test.ts b/gitnexus/test/unit/stream-graph-emit-config.test.ts new file mode 100644 index 000000000..16beb3339 --- /dev/null +++ b/gitnexus/test/unit/stream-graph-emit-config.test.ts @@ -0,0 +1,183 @@ +/** + * Streamed structural graph emit — config gate and pruner integration (#2680). + * + * The gate is a soundness boundary, not a preference: streaming is only valid + * on a full rebuild, because the incremental writeback reads relationships back + * out of the in-memory graph. + * + * The pruner cases are the sharp end of the feature. `pruneLocalValueSymbols` + * decides "is this block-local symbol referenced?" from an in-memory + * relationship scan; under streaming that scan cannot see edges already on + * disk, so without the predicate a referenced symbol is deleted and its + * streamed CSV row is left pointing at a node with no row. + */ +import { describe, it, expect, vi, afterEach } from 'vitest'; + +import { resolveStreamGraphEmit } from '../../src/core/run-analyze.js'; +import { buildPhaseList } from '../../src/core/ingestion/pipeline.js'; +import { RETAINED_REL_TYPES } from '../../src/core/lbug/graph-emit-sink.js'; +import { createKnowledgeGraph } from '../../src/core/graph/graph.js'; +import type { RelationshipType } from 'gitnexus-shared'; + +afterEach(() => { + vi.unstubAllEnvs(); +}); + +describe('resolveStreamGraphEmit', () => { + it('is ON by default on a full rebuild — no opt-in needed', () => { + expect(resolveStreamGraphEmit({ force: true })).toBe(true); + }); + + it('is turned off by an explicit falsy env value (the escape hatch)', () => { + vi.stubEnv('GITNEXUS_STREAM_GRAPH_EMIT', '0'); + expect(resolveStreamGraphEmit({ force: true })).toBe(false); + }); + + it('is turned off by an explicit option, which beats the env', () => { + vi.stubEnv('GITNEXUS_STREAM_GRAPH_EMIT', '1'); + expect(resolveStreamGraphEmit({ force: true, streamGraphEmit: false })).toBe(false); + }); + + it('honors the explicit option on a full rebuild', () => { + expect(resolveStreamGraphEmit({ force: true, streamGraphEmit: true })).toBe(true); + }); + + it('honors the env toggle on a full rebuild', () => { + vi.stubEnv('GITNEXUS_STREAM_GRAPH_EMIT', '1'); + expect(resolveStreamGraphEmit({ force: true })).toBe(true); + }); + + it('refuses an incremental run even when explicitly requested', () => { + // The incremental writeback reads relationships back out of the in-memory + // graph; streaming has already offloaded them. + expect(resolveStreamGraphEmit({ force: false, streamGraphEmit: true })).toBe(false); + expect(resolveStreamGraphEmit({ streamGraphEmit: true })).toBe(false); + }); + + it('refuses an incremental run even when the env toggle is set', () => { + vi.stubEnv('GITNEXUS_STREAM_GRAPH_EMIT', '1'); + expect(resolveStreamGraphEmit({ force: false })).toBe(false); + }); +}); + +const FILE_ID = 'File:src/a.ts'; +const LOCAL_ID = 'Const:src/a.ts:localValue'; + +const localConst = (): GraphNode => ({ + id: LOCAL_ID, + label: 'Const', + properties: { name: 'localValue', filePath: 'src/a.ts', scope: 'block' }, +}); + +/** Graph holding only the structural File->DEFINES->localConst edge, i.e. the + * shape the pruner sees when the symbol's only *semantic* reference streamed + * out to CSV. */ +const graphWithOnlyStructuralEdge = () => { + const graph = createKnowledgeGraph(); + graph.addNode({ + id: FILE_ID, + label: 'File', + properties: { name: 'a.ts', filePath: 'src/a.ts' }, + }); + graph.addNode(localConst()); + graph.addRelationship({ + id: `DEFINES:${FILE_ID}->${LOCAL_ID}`, + sourceId: FILE_ID, + targetId: LOCAL_ID, + type: 'DEFINES', + confidence: 1, + reason: 'structural', + }); + return graph; +}; + +describe('buildPhaseList under streamGraphEmit', () => { + const names = (o: Parameters[0]) => buildPhaseList(o).map((p) => p.name); + + it('keeps every CALLS-consuming phase enabled — nothing is traded away', () => { + // The sink answers a complete relationship read, so these phases work + // unchanged. If this ever regresses to filtering them out, streaming can no + // longer be the default. + const streamed = names({ streamGraphEmit: true, pdg: true, force: true }); + + expect(streamed).toContain('communities'); + expect(streamed).toContain('processes'); + expect(streamed).toContain('taintSummaries'); + expect(streamed).toContain('callSummaries'); + }); + + it('keeps mro and di, whose reads are all in the retained set', () => { + const streamed = names({ streamGraphEmit: true, pdg: true, force: true }); + + expect(streamed).toContain('mro'); + expect(streamed).toContain('di'); + expect(streamed).toContain('parse'); + expect(streamed).toContain('scopeResolution'); + expect(streamed).toContain('pruneLocalSymbols'); + }); + + it('leaves the phase list untouched when the flag is off', () => { + // Guards the default path: the gating predicates must not filter anything + // for existing (flag-off) users. + const withPdg = names({ pdg: true, force: true }); + + expect(withPdg).toContain('communities'); + expect(withPdg).toContain('processes'); + expect(withPdg).toContain('taintSummaries'); + expect(withPdg).toContain('callSummaries'); + }); + + it('still honours skipGraphPhases independently of the streaming flag', () => { + const skipped = names({ skipGraphPhases: true }); + + expect(skipped).not.toContain('communities'); + expect(skipped).not.toContain('processes'); + expect(skipped).toContain('pruneLocalSymbols'); + }); +}); + +describe('RETAINED_REL_TYPES tracks its readers', () => { + it('retains every relationship type any phase reads back mid-pipeline', async () => { + // The round-trip test CANNOT catch drift here: addRelationship partitions + // edges between the graph and the CSVs, and a partition's union is + // invariant under where the line falls — so it stays green for any + // partitioning, including a wrong one. Nothing else guards the invariant, + // and getting it wrong yields a silently incomplete edge set mid-pipeline + // rather than a crash. So derive the required set from the source and + // compare. + const { execFileSync } = await import('node:child_process'); + const srcDir = new URL('../../src/', import.meta.url).pathname; + + // Every literal `iterRelationshipsByType('X')` reachable while streaming is + // armed. `git grep -h` over src/ excluding tests; the sink itself is + // excluded because its own fast-path check reads the constant, not an edge. + const out = execFileSync( + 'grep', + ['-rhoE', "iterRelationshipsByType\\('[A-Z_]+'\\)", '--include=*.ts', srcDir], + { encoding: 'utf8' }, + ); + const readTypes = new Set( + [...out.matchAll(/iterRelationshipsByType\('([A-Z_]+)'\)/g)].map((m) => m[1]), + ); + + // CALLS is read by taintSummaries, which is exactly why the sink answers a + // COMPLETE read instead of retaining it — so it is a known exemption. + readTypes.delete('CALLS'); + + const missing = [...readTypes].filter((t) => !RETAINED_REL_TYPES.has(t as RelationshipType)); + expect(missing).toEqual([]); + }); +}); + +describe('streamGraphEmit without a CSV dir', () => { + it('throws instead of silently running without streaming', async () => { + // Streaming is on by default, so a programmatic host that builds its own + // PipelineOptions and forgets the directory must not get a successful run + // that quietly did no streaming. + const { runPipelineFromRepo } = await import('../../src/core/ingestion/pipeline.js'); + + await expect( + runPipelineFromRepo('/nonexistent-repo', () => {}, { streamGraphEmit: true }), + ).rejects.toThrow(/graphEmitCsvDir is missing/); + }); +});