perf(analyze): hold structural relationships out of the JS heap, on by default (#2680) (#2685)

* refactor(lbug): extract SyncCsvWriter into a shared module `PdgEmitSink` (#2202) declared `SyncCsvWriter` as a private, non-exported class. The structural streaming sink for #2680 needs the same buffered sync-write + poison/openFailure IO discipline, and importing it is not possible while it is module-private — so the alternative was copying ~90 lines of it. Extract the class (and the chunk-rows default it uses) into `sync-csv-writer.ts` and have `PdgEmitSink` import it. `DEFAULT_PDG_EMIT_CHUNK_ROWS` stays exported as an alias so no existing caller changes. Pure refactor: no behaviour change. pdg-emit-sink.ts 396 -> 302 lines; tsc clean; the 23 existing #2202 tests pass unchanged. Refs #2680 * feat(lbug): add GraphEmitSink for streaming structural relationship emit Structural sibling of PdgEmitSink (#2202): a KnowledgeGraph façade that routes relationships no mid-pipeline phase reads back to bounded CSV-on-disk and never stores them. Nothing constructs it yet. Measurement drove the design. On a kernel-shaped synthetic graph (400k nodes, 2.7 edges/node): nodes only ...... 367 B/node nodes + edges ... 2075 B/node <- reproduces the #2649 ~2.1 KB/node => the relationship layer is 83% of graph heap, ~646 B/edge so streaming *relationships* is where the memory is; nodes stay resident (they are 17%, and two scope-resolution index builders scan them). Dropping just the redundant relationshipsByType/edgeIdsByNode indexes was also measured — 174 of 648 B/edge, ~1.3x — and is not a substitute. RETAINED_REL_TYPES is derived from an exhaustive audit of every relationship read site under src/, and each entry names its reader. An earlier draft carried 14 types, 5 of which no reachable phase reads. Two deliberate departures from PdgEmitSink, both because its invariants do not hold here: - dedup by relationship id, since no upstream per-file uniqueness guarantee exists for structural edges and COPY would violate the PK; - removeRelationship on an already-streamed id throws instead of no-oping, so a mutating consumer cannot corrupt the graph undetected. Also exposes hasStreamedSemanticEdge for the local-symbol pruner: without it a block-local symbol referenced only by a streamed edge looks unreferenced and gets pruned, leaving a CSV row pointing at a node with no row. Refs #2680 * feat(analyze): stream structural relationships to CSV under GITNEXUS_STREAM_GRAPH_EMIT Wires GraphEmitSink into the pipeline behind a full-rebuild-only flag, so relationships that no mid-pipeline phase reads back never enter the JS heap. Measured ~2.9x reduction of graph heap: 0.17 (nodes) + 0.83 * 0.21 (retained edges) = 0.344 retained. This is a constant factor, NOT O(chunk) — node identity and the resolution registries stay O(repo). The sink is armed at the PARSE boundary, not at graph construction. An exhaustive audit of every relationship read site under src/ found four mid-pipeline CALLS consumers, not the two an earlier draft assumed: - local-symbol-pruner (full iterRelationships scan, then removeNode) - communities / processes (whole-graph forEachRelationship) - mapCobolToGraph, which scans CALLS and REMOVES the unresolved ones — and runs BEFORE parse, so streaming from construction would have silently stopped COBOL cross-program call resolution - taintSummaries, gated on `pdg` and NOT on `skipGraphPhases`, so it needs its own gate or --pdg + this flag yields an empty taint layer Accordingly communities, processes, taintSummaries and callSummaries are all disabled under the flag, and the run logs what it is giving up. Two fixes that are correct independently of the flag: - runPipelineFromRepo keyed its community/process extraction off `!skipGraphPhases` while getPhaseOutput THROWS on a phase filtered out by any enabledWhen predicate — now a presence check, so filtered combinations return undefined instead of crashing. - loadGraphToLbug COPYs one job per CSV FILE rather than per label pair. #2202's throw-on-collision merge is only sound because BasicBlock pairs are disjoint; a streamed CALLS edge is Function|Function and always collides with the whole-graph CSV for that pair, so the structural manifest appends instead. The buffer-pool hint adds the streamed row count back in: the hint only ever shrinks the pool, so sizing it from the post-streaming relationshipCount would starve the COPY at exactly the scale this targets. detect_changes: 18 symbols / 10 files / 9 processes, all within the planned scope. Full suite green with the flag off. Refs #2680 * fix(mcp): stop impact() under-reporting risk on a streamed index An index built with streamed structural emit has no Process or Community rows, and impact()'s risk scorer uses processCount >= 5 and moduleCount >= 5 as two of its four CRITICAL escalation criteria. The missing-table errors are swallowed as benign without raising `partial`, so nothing distinguished 'this repo has no processes' from 'this index was built without them' — the same change would report LOW off a streamed index and CRITICAL off a complete one, with no signal either way. That is the false-clean shape #2283 ruled out for detect_changes, and it matters more here because the repo's own workflow mandates impact() before every symbol edit. Stamp `graphPhases: 'complete' | 'skipped'` into RepoMeta and have impact() attach riskUnderstated + an explanatory riskNote when the index is stamped skipped, so the reported level is explicitly a lower bound. Unlike the rest of RepoMeta.capabilities this stamp has a real programmatic reader. Also documents GITNEXUS_STREAM_GRAPH_EMIT in the README env table, including everything the flag disables. Refs #2680 * test(lbug): differential set-identity gate for streamed structural emit The acceptance property for #2680: for the same node/edge set, the rows reaching the bulk COPY must be identical whether streaming is on or off. With streaming on they arrive from two places — the residual in-memory graph via streamAllCSVsToDisk, plus the sink's per-pair CSVs — so the test asserts their UNION equals the single whole-graph emit. Also asserts the split is real (retained + streamed == total, streamed > 0), so a sink that silently streamed nothing cannot pass the equality vacuously. Verified discriminating: with sink.arm() commented out the test fails ('expected 0 to be greater than 0'); restored, it passes. Fixture spans both sides of RETAINED_REL_TYPES and includes a self-edge and a duplicate relationship id — the cases where a naive sink diverges from the whole-graph emit. Drives the sink directly rather than running analyze, matching pdg-emit-streaming-roundtrip.test.ts: the guarantee is about emitted rows, and the worker pool would add unrelated machinery without strengthening the assertion. Refs #2680 * fix(test): remove literal NUL byte and cover streamGraphEmit phase gating Two review findings, both verified before accepting. 1. The round-trip test contained a literal NUL byte as a key separator, which made Git treat the whole .ts file as BINARY — `git show --numstat` reported `-\t-` for it, so the file would not diff or blame and CI text tooling would skip it. Replaced with the escaped \\u0000 sequence; behaviour is identical, the file is text again. (Found by the Codex swarm lane.) 2. buildPhaseList's four new streamGraphEmit gating predicates and the flag-off default path had no test that would fail on revert — two review lanes flagged this independently. Reversing any enabledWhen condition would have passed the suite silently, which matters because an ungated taintSummaries yields an empty taint layer rather than an error. Added four cases: the streamed run drops communities/processes/ taintSummaries/callSummaries; it keeps mro/di (their reads are all in RETAINED_REL_TYPES); the flag-off list is untouched; and skipGraphPhases still works independently. Refs #2680 * fix(analyze): don't leak a temp dir when streaming is off; correct two overclaims Three review findings, all verified before accepting. 1. `graphEmitCsvDir: resolveNativeSafeStorageDir(...)` was evaluated unconditionally inside the pipeline-options literal. On a Windows non-ASCII storage path that helper mkdtempSyncs a REAL directory, so every analyze leaked one temp dir even with the flag off. Now resolved only when streaming is active, matching how the PDG sibling resolves inside its own guard. This was the only finding affecting flag-off users. 2. The retain-set comment claimed 'the differential round-trip test is what catches drift'. It 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 RETAINED_REL_TYPES is drawn. Only the read-site audit protects the invariant, and the comment now says so and names the grep to re-run. 3. The ~2.9x figure assigned streamed edges a retained cost of zero, ignoring the sink's own streamedIds/streamedEndpoints Sets — and relationship ids are plain concatenations of both endpoint ids, not hashes. Review measured those Sets at ~35% of full per-edge retention, not the '~a tenth' assumed, putting the real figure nearer ~1.7-2.2x; a member-dense Java/C# repo lands lower still, since the retained structural spine is a larger share there than in the TypeScript census the 0.21 came from. Code comment and README now give a range and say plainly that no end-to-end measurement on a real repository exists yet. Refs #2680 * fix(mcp): disclose degraded risk in detect_changes; stop pinning the sink Two more review findings, both cross-lane corroborated. 1. detect_changes derives risk_level SOLELY from affected-process count, and a graphPhases:'skipped' index has zero Process rows by construction. The STEP_IN_PROCESS query then succeeds with zero rows, so queryDegraded stays false and the tool returns risk_level 'low', affected_count 0, with no partial marker — for every change, forever. That is a false-clean on the gate this repo mandates before every commit, and it is the same #2283 shape the previous commit fixed in impact() while leaving its sibling untouched. Now carries the same riskUnderstated + riskNote disclosure. 2. PipelineResult.graphEmitSink had zero readers — the pruner predicate and the manifest are both threaded elsewhere — but returning it kept the sink, and therefore its O(streamed-edges) id and endpoint Sets, reachable through the entire COPY/FTS/embedding phase. That is precisely the phase this feature exists to fit inside RAM, so the field actively worked against the change's purpose. Dropped. Refs #2680 * refactor(2680): one named capability, one risk helper, a shorter header Pure cleanup pass — no behaviour change, 66 tests across the six affected suites still green, and the round-trip test still fails when the sink is left un-started. Three things were untidy: 1. The phase layer reached the sink through TWO loose callbacks bolted onto PipelineContext (`armStreaming`, `hasStreamedSemanticEdge`) — two fields, two wiring lines, no name for the thing they belonged to. Replaced by one `graphEmit?: GraphEmitControl`, a two-method interface declared beside the sink. Phases now say what they mean: `ctx.graphEmit?.beginStreaming()`. Also renames `arm()` to `beginStreaming()`, which needs no comment to explain. 2. The degraded-index risk disclosure was copy-pasted into impact() and detect_changes() — two meta probes, two near-identical prose blocks, and two long comments restating the same reasoning. Now one `streamedIndexRiskDisclosure()` helper carrying the explanation once; each caller passes only the clause naming which count is structurally zero for it. Same file, 45 lines in / 45 out, with the duplication gone. 3. The sink's file header had grown into a changelog of my own review corrections ('this once assumed', 'review measured'). A reader does not care what an earlier draft believed. Rewritten to state the design argument once — relationships are ~83% of graph heap, so they are what streams; nodes are the other 17% and are scanned, so they stay — under headings, with the honest 'this is an estimate, ~1.7-2.2x, no real-repo measurement yet' caveat kept in full. Refs #2680 * feat(analyze): make streamed graph emit the default, with nothing traded away Streaming was opt-in because it disabled the four phases that consume the whole CALLS graph — communities, processes, taintSummaries, callSummaries. That made it unshippable as a default: query() is process-grouped and clusters/skill-gen are community-backed, so every index would have silently lost them. The sink now answers a COMPLETE relationship read. It keeps streamed edges as four parallel columns over an interned node table — sourceId, targetId, type, confidence — and iterRelationships/iterRelationshipsByType/ forEachRelationship/relationshipCount return the retained edges concatenated with those. Every consumer therefore sees the whole graph and no phase knows streaming happened. Four fields, not six, because an audit showed community-processor, process-processor, taint-summaries and the pruner read only those — none keys on rel.id. That matters: relationship ids are unique long strings, and retaining them is precisely what made a fully-columnar attempt LOSE to the object graph (measured 838 MB vs 822 MB). Ids stay out of the columns; a read synthesizes one, which is safe because buildRelRow never persists it. Consequently deleted, not merely disabled: - the four enabledWhen gates and the 'what you give up' warning; - the pruner's hasStreamedSemanticEdge predicate and its plumbing — a complete scan sees streamed edges, so the dangling-edge hazard is gone by construction rather than by compensation; - the whole degraded-index apparatus: the graphPhases RepoMeta stamp, streamedIndexRiskDisclosure, and the riskUnderstated markers on impact() and detect_changes(). Nothing degrades, so nothing needs disclosing. Default is ON for full rebuilds; GITNEXUS_STREAM_GRAPH_EMIT=0 (or an explicit option) is the escape hatch, for bisecting a suspected streaming fault rather than routine use. Incremental runs still refuse it — the writeback reads relationships back out of the in-memory graph. Measured A/B, 400k nodes / 1.08M edges, all edges streamable (worst case for this design): 823 MB -> 626 MB, ~1.3x, all 1.08M edges still visible. That is deliberately less than the ~2.9x the retained-share formula implies — losslessness costs the dedup Set and the columns. The earlier, bigger number was bought by disabling phases. README and the file header both state 1.3x measured; neither claims O(chunk). New coverage: reads are complete (proven discriminating — 3 tests fail when the streamed leg is removed), endpoints/confidence survive the round trip, per-type lookup finds streamed types, and every CALLS-consuming phase stays registered under the flag. Refs #2680 * docs(2680): pin the invariants the default-on change relies on Review follow-ups. No behaviour change except the id-uniqueness fix. - pipeline.ts returns the RAW graph, not the sink, and that is load-bearing: phases read the sink so their scans are complete, but loadGraphToLbug feeds this value to streamAllCSVsToDisk, whose iterator would then emit every streamed edge a SECOND time on top of the per-pair CSVs the sink already wrote. Returning the sink there silently doubles every streamed relationship in the persisted graph, so the reason is now written down at the return site. - Synthesized ids now carry the column index, making them unique even when two streamed edges share (type, source, target) and differ only in reason/step. Harmless today because no consumer keys on relationship id, but real ids are unique and the synthesized ones should match, so a future id-keyed consumer cannot silently collapse two edges. - Recorded WHY dropping reason/step is safe, which is not the same argument as for id: the persisted row keeps their true values because buildRelRow receives the original relationship on the way through, so only in-memory reads see the 'streamed' placeholder. The ACCESSES reason:'read'|'write' distinction that MCP queries depend on therefore survives in the database. A future in-pipeline consumer needing either field must add a column rather than trust the placeholder. Also verified while chasing a review lead: removeNodesByFile has no production callers and removeNode has exactly one (the pruner), which reads through the sink and so sees streamed edges. The dangling-edge hazard the deleted hasStreamedSemanticEdge predicate used to compensate for is closed by construction, not by luck. Refs #2680 * fix(2680): fail loudly on a missing CSV dir, and guard the retain set Resolves both findings from the review of this branch. MEDIUM — pipeline.ts silently skipped streaming when `streamGraphEmit` was true but `graphEmitCsvDir` was absent. The CLI always supplies the dir, but streaming is on by DEFAULT now, and the callers that build PipelineOptions themselves (eval-server, MCP daemon, tests) are exactly the ones that would omit it — so they would ask for streaming, not get it, and still see a successful run. That is the silent-degraded-outcome shape the rest of this work exists to prevent, so it now throws with the resolution hint. Covered by a test asserting the rejection. LOW — RETAINED_REL_TYPES had no automated guard, and the round-trip test structurally cannot be one: addRelationship PARTITIONS edges between the graph and the CSVs, and a partition's union is invariant under where the line falls, so that test stays green for any partitioning including a wrong one. Drift there yields a silently incomplete mid-pipeline edge set, not a crash. Added a test that derives the required set by grepping every literal iterRelationshipsByType('X') under src/ and asserts the constant covers it, with CALLS as the documented exemption (taintSummaries reads it, which is why the sink answers a complete read rather than retaining it). Proven discriminating: removing EXTENDS from the constant fails with "expected [ 'EXTENDS' ] to deeply equal []". 128 tests green across the eight affected suites, including the index-lock suite that arrived with the #2677 merge. Refs #2680 * docs(2680): record the measured CPU cost, not just the memory win I measured memory before shipping and never measured time, which was a gap: reads now allocate, rebuilding objects instead of returning stored ones, and a real analyze does SIX full relationship scans (pruner, communities x2, processes x2, the taint fixpoint's CALLS pass). Same 400k-node / 1.08M-edge graph: heap 820 MB -> 623 MB (1.32x better) scans 96 ms -> 651 ms (6.8x WORSE) 6.8x on iteration is worth knowing, but the absolute number decides it: ~0.5 s here, ~2 s extrapolated to kernel scale, against an analyze measured in minutes — under 1% of wall-clock. The ~26M short-lived objects at kernel scale are young-generation churn (the cheap case), and being ~800 MB further from the heap ceiling matters more than the churn costs: #2649's cascade came from GC thrash NEAR the limit, not from allocation volume as such. Also names the first lever if these scans ever go hot — a per-type index over the columns, so iterRelationshipsByType stops scanning all streamed edges — and notes that it trades memory back, so it needs a measurement first. Refs #2680 * perf(2680): cut the iteration regression from 6.8x to 1.8x The memory win came with an unmeasured CPU cost. Iteration went from returning stored objects to rebuilding them, across the SIX full relationship scans an analyze performs (pruner, communities x2, processes x2, taint's CALLS pass). First measurement: 90 ms -> 651 ms, 6.8x worse. Fixed properly rather than documented away. Two causes, each measured before and after: 1. The ~150-character synthesized `id` was built eagerly on every read — 6.5M concatenations per analyze, for a field NO in-pipeline consumer reads. Isolating it (constant id) showed 436 ms of the 555 ms regression. Now a lazy prototype getter on a fixed-shape `StreamedRelationship` class: the string is built only if someone asks, and V8 keeps one hidden class across millions of instances. 2. Generator and iterator-protocol overhead on million-edge walks. `forEachRelationship` (community detection's form, called twice) now loops the columns directly, skipping both. `iterRelationships` keeps an iterator but reuses one result record — a hand-rolled version allocating a fresh {value, done} per edge measured WORSE than the generator (252 ms), which is why the obvious rewrite is not the one that shipped. heap 821 MB -> 623 MB (1.32x better) scans 90 ms -> 180 ms (was 651 ms) The residual ~90 ms is object allocation, 6.5M instances across six scans, and it is irreducible while the read API returns objects at all. The remaining fix for true parity is a field-wise callback passing sourceId/targetId/type/ confidence as primitives — all four hot consumers read only those — but that changes the KnowledgeGraph interface and its consumers, so it belongs in its own measured change rather than bolted on here. Refs #2680 * perf(2680): zero-allocation field scan brings iteration back to parity Third and final step on the iteration cost. The memory win had come with a 6.8x iteration regression; the previous commit cut that to 1.8x by making the synthesized id lazy and removing generator overhead. The residual was object allocation itself — 6.5M instances across the six full relationship scans an analyze performs — which no amount of tuning removes while the read API hands back objects. So the hot consumers stop asking for objects. Adds `KnowledgeGraph.forEachRelationshipFields`, which passes (sourceId, targetId, type, confidence) as primitives — exactly and only what every whole-graph scan reads. On the sink those come straight out of the columns, allocating nothing; on the object-based graph they are read off the stored relationship, so the flag-off path is unaffected. Converted the five whole-graph scans: community detection (x2), process extraction (x2), and the local-symbol pruner. `isFileDefinesEdge` now takes (type, sourceId) rather than a relationship. The taint fixpoint's by-type pass is left alone — one scan of six, and converting it would turn an indexed bucket lookup into a full scan on the object-based graph. heap 820 MB -> 623 MB (1.32x better) scans ~82 ms -> ~90 ms (was 651 ms; now parity within noise) Also deletes the pruner's `hasStreamedSemanticEdge` option, which has had no caller since the sink's reads became complete — a dead knob is worse than no knob. Verified: 104 tests across the eight affected suites, including the pruner's pipeline integration test (which needs the raised worker-ready timeout on this host; it passes cleanly with it and its failures are the known 5s handshake). Refs #2680 * perf(2680): compact dedup keys — 1.32x -> 1.59x, speed unchanged An audit of where duplicate relationship ids actually come from, then the saving it unlocked. The audit (instrumented analyze of this repo): 25 duplicate-id hits across 63,412 streamed edges — 0.04%, all CALLS, every one the SAME call site re-emitted when a file is resolved in more than one language pass. Three things follow, and they rule out the cheap options: - dedup cannot be dropped (25 != 0, and a duplicate reaching COPY is a wrong graph); - it cannot move to row contents, because emit-references builds ids as `...->target:line:col`, so two calls between the same pair at different sites have byte-identical CSV rows that the whole-graph emit keeps; - it cannot move to a per-file source guard like `pdgEmittedFiles`, because a later language pass can resolve genuinely NEW edges for the same file. What was left was the key itself. An id embeds both node ids in full (~200 chars here) while the endpoints are ALREADY interned for the columns, so the Set was storing them twice. Keys are now built from the interner indices plus the id's trailing disambiguator parsed into NUMBERS. Numbers, not substrings, and that is load-bearing: a key built by slicing inside a long string is a V8 sliced/cons string that keeps its parent alive, so the id would never be freed and the saving would silently fail to appear. An earlier attempt at this measured no improvement for exactly that reason. Unrecognized id shapes (`rel:contains:` has no tail) fall back to storing the id verbatim — correctness first, saving second. heap 821 MB -> 518 MB (1.59x, was 1.32x) scans ~83 ms -> ~88 ms (parity, unchanged) Speed is untouched by construction: dedup is on the WRITE path, and none of the six full scans reads it. Also fixes removeRelationship, which the test suite caught: it looked up the raw id in a Set that now holds compact keys, so it silently stopped throwing on an already-streamed edge. It cannot recompute a key from a bare id, so it is now conservative — anything the real graph does not hold is treated as possibly-streamed once streaming has begun and fails loudly. A genuinely-absent id throws where main returns false; acceptable because the only production caller (the COBOL resolver) runs before the sink is armed. 89 tests green across the six affected suites. Refs #2680 * fix(2680): dedup key dropped edges when tail segment counts differed Both findings from the review of this branch, and the coverage gap named alongside them. HIGH — the compact dedup key packed the id's trailing numeric segments as `|${a}|${b}`, with `b` defaulting to 0 when only one segment was present and the segment COUNT absent from the key. So `:7` and `:7:0` produced the same key and the second edge was silently discarded as a duplicate: a lost relationship, no error, no warning. Found by probe, not by reading — two distinct ids for one (source, target, type) went in and one edge came out. The key now carries `seen`. Nothing existing caught it. The round-trip test compares the UNION of graph and CSV rows, and a dropped edge is missing from both, so it stayed green; the duplicate test only feeds a genuinely identical id, which is the case that SHOULD collapse. Four new cases pin the boundary instead: differing segment counts stay distinct, two call sites between one pair stay distinct (the `:line:col` shape from emit-references), a truly repeated id still collapses, and a non-numeric tail falls back to the full id. Proven discriminating — reverting the fix fails with "expected 1 to be 2". This costs ~66 MB at 400k nodes / 1.08M edges (584 MB, was 518 MB), so the heap win is 1.40x rather than 1.59x. Not a trade worth making the other way: a silently missing relationship is the exact failure class the rest of this work exists to prevent. I am not asserting a mechanism for why two extra characters per key cost that much — it is stable and reproducible across runs, and inventing a cause is how I got the earlier cons-string diagnosis wrong. LOW — removeRelationship throws for an absent id once streaming has begun, where KnowledgeGraph.removeRelationship returns false. The behaviour is deliberate (a bare id cannot be turned back into a compact key, and answering "false" for an edge already on disk is the worse failure) but it was undocumented and untested. Now stated on the interface itself and pinned by two cases: absent-id-while-streaming throws, absent-id-before-streaming returns false. Coverage gap — added a test asserting forEachRelationshipFields yields the same (source, target, type, confidence) tuples as iterRelationships. That guards the five whole-graph scans converted in 9fa18384, where a divergence would silently skew community detection, process extraction and the pruner. Also records the verified scaling in the file header: linear at 100k/200k/ 400k/800k nodes, per-edge scan cost flat at ~13 ns in both arms, heap ratio drifting only 1.7x -> 1.5x as interner indices gain digits. No super-linear term. 135 tests green across the eight affected suites. Refs #2680 --------- Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
2026-08-28 05:25:25 +00:00 · 2026-07-25 07:43:51 +01:00 · 2026-07-25 07:43:51 +01:00 · 7316503ebc
commit 7316503ebc
parent df0110b06f
18 changed files with 1767 additions and 151 deletions
--- 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.                                                                                                                                                                                                                                                                                                                                  |
--- 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
--- 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<GraphRelationship>;
  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;
 }
--- 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<string>();
  const nodeDegree = new Map<string, number>();

-  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<string>();
  const edges: Array<readonly [number, number]> = [];

-  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;

--- 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<string>();
-  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) {
--- 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<ParseOutput> = {
    ctx: PipelineContext,
    deps: ReadonlyMap<string, PhaseResult<unknown>>,
  ): Promise<ParseOutput> {
+    // 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<StructureOutput>(
      deps,
      'structure',
--- 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 ───────────────────────────────────────────────────
--- 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<CommunitiesOutput>(results, 'communities').communityResult;
    processResult = getPhaseOutput<ProcessesOutput>(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,
--- 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<string, string[]>();

-  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<string, string[]>();

-  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;
 };
--- a/gitnexus/src/core/lbug/graph-emit-sink.ts
+++ 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<RelationshipType> = new Set<RelationshipType>([
+  '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<string, { csvPath: string; rows: number }>;
+  /** 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<string>;
+  private readonly relWriters = new Map<string, SyncCsvWriter>();
+  /**
+   * 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<string>();
+  /**
+   * 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<string, number>();
+  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<string>(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<string, { csvPath: string; rows: number }>();
+    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<GraphNode> {
+    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<GraphRelationship> {
+    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<GraphRelationship> = {
+      next(): IteratorResult<GraphRelationship> {
+        const fromReal = retained.next();
+        if (fromReal.done !== true) {
+          result.value = fromReal.value;
+          result.done = false;
+          return result as IteratorResult<GraphRelationship>;
+        }
+        if (ix < self.srcIx.length) {
+          result.value = self.streamedAt(ix++);
+          result.done = false;
+          return result as IteratorResult<GraphRelationship>;
+        }
+        result.value = undefined;
+        result.done = true;
+        return result as IteratorResult<GraphRelationship>;
+      },
+      [Symbol.iterator]() {
+        return it;
+      },
+    };
+    return it;
+  }
+
+  *iterRelationshipsByType(type: RelationshipType): IterableIterator<GraphRelationship> {
+    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;
+  }
+}
--- 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<string>();

-    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
--- 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<RelationshipType> = new Set<RelationshipType>(
 ]);

 /** 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);
  }
--- a/gitnexus/src/core/lbug/sync-csv-writer.ts
+++ 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 */
+      }
+    }
+  }
+}
--- 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,
      );
    }

--- 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;
 }
--- a/gitnexus/test/integration/graph-emit-streaming-roundtrip.test.ts
+++ 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<string[]> => {
+  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());
+  });
+});
--- a/gitnexus/test/unit/lbug/graph-emit-sink.test.ts
+++ 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<string[]> => {
+  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();
+  });
+});
--- a/gitnexus/test/unit/stream-graph-emit-config.test.ts
+++ 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<typeof buildPhaseList>[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/);
+  });
+});