diff --git a/scripts/compiler_differential/_helpers/comparing/capture.py b/scripts/compiler_differential/_helpers/comparing/capture.py
new file mode 100644
index 0000000000..4f35198c9c
--- /dev/null
+++ b/scripts/compiler_differential/_helpers/comparing/capture.py
@@ -0,0 +1,53 @@
+"""Compare two stage captures on disk without loading either one whole."""
+
+from __future__ import annotations
+
+import itertools
+import json
+from collections.abc import Iterator
+from pathlib import Path
+
+from scripts.compiler_differential._helpers.comparing.compare import (
+    first_json_difference,
+    preview,
+)
+from scripts.compiler_differential._helpers.comparing.normalize import normalize_artifact_text
+from scripts.compiler_differential.classes.capture_file import CaptureFile
+from scripts.compiler_differential.constants import MISSING_VALUE
+from scripts.compiler_differential.models import Divergence
+
+
+def first_capture_difference(*, left: Path, right: Path) -> Divergence | None:
+    """Return the first logical JSON pointer, or line for invalid JSON, where captures differ."""
+
+    try:
+        left_capture: CaptureFile = CaptureFile(left)
+        right_capture: CaptureFile = CaptureFile(right)
+        return first_json_difference(
+            left=left_capture.root,
+            right=right_capture.root,
+            resolve=(left_capture.resolve, right_capture.resolve),
+        )
+    except (json.JSONDecodeError, UnicodeDecodeError):
+        return _first_line_difference(left=left, right=right)
+
+
+def _first_line_difference(*, left: Path, right: Path) -> Divergence | None:
+    with left.open("rb") as left_stream, right.open("rb") as right_stream:
+        lines: Iterator[tuple[bytes | None, bytes | None]] = itertools.zip_longest(
+            left_stream, right_stream
+        )
+        for number, (left_line, right_line) in enumerate(lines, start=1):
+            left_text: str = _line_text(left_line)
+            right_text: str = _line_text(right_line)
+            if left_text != right_text:
+                return Divergence(
+                    location=f"line {number}", left=preview(left_text), right=preview(right_text)
+                )
+    return None
+
+
+def _line_text(line: bytes | None) -> str:
+    if line is None:
+        return MISSING_VALUE
+    return normalize_artifact_text(line.decode("utf-8", "surrogateescape").rstrip("\n"))
diff --git a/scripts/compiler_differential/_helpers/comparing/compare.py b/scripts/compiler_differential/_helpers/comparing/compare.py
index 3ee796c4c3..4d6af02808 100644
--- a/scripts/compiler_differential/_helpers/comparing/compare.py
+++ b/scripts/compiler_differential/_helpers/comparing/compare.py
@@ -3,25 +3,37 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Callable
 from typing import cast
 
 from scripts.compiler_differential.constants import MISSING_VALUE, VALUE_PREVIEW_CHARACTERS
 from scripts.compiler_differential.models import Divergence
+from scripts.compiler_differential.types import NodeResolvers
+from sqlbuild.compiler.frontier.constants import STAGE_CAPTURE_SHARED_MARKER
 
 
-def first_json_difference(*, left: object, right: object, pointer: str = "") -> Divergence | None:
-    """Return the first JSON pointer whose value, type, or object key order differs."""
+def first_json_difference(
+    *, left: object, right: object, pointer: str = "", resolve: NodeResolvers | None = None
+) -> Divergence | None:
+    """Return the first JSON pointer whose value, type, or key order differs, via `resolve`."""
 
+    if resolve is not None:
+        left_digest: str | None = shared_digest(left)
+        if left_digest is not None and left_digest == shared_digest(right):
+            return None
+        left, right = resolve[0](left), resolve[1](right)
     if type(left) is not type(right):
         return _divergence(location=pointer or "/", left=left, right=right)
     left_object: dict[str, object] | None = as_json_object(left)
     right_object: dict[str, object] | None = as_json_object(right)
     if left_object is not None and right_object is not None:
-        return _first_object_difference(left=left_object, right=right_object, pointer=pointer)
+        return _first_object_difference(
+            left=left_object, right=right_object, pointer=pointer, resolve=resolve
+        )
     if isinstance(left, list) and isinstance(right, list):
         for index, (left_item, right_item) in enumerate(zip(left, right, strict=False)):
             found: Divergence | None = first_json_difference(
-                left=left_item, right=right_item, pointer=f"{pointer}/{index}"
+                left=left_item, right=right_item, pointer=f"{pointer}/{index}", resolve=resolve
             )
             if found is not None:
                 return found
@@ -29,8 +41,12 @@ def first_json_difference(*, left: object, right: object, pointer: str = "") ->
             index: int = min(len(left), len(right))
             return _divergence(
                 location=f"{pointer}/{index}",
-                left=left[index] if index < len(left) else MISSING_VALUE,
-                right=right[index] if index < len(right) else MISSING_VALUE,
+                left=_shown(value=left[index], resolve=resolve, side=0)
+                if index < len(left)
+                else MISSING_VALUE,
+                right=_shown(value=right[index], resolve=resolve, side=1)
+                if index < len(right)
+                else MISSING_VALUE,
             )
         return None
     if left != right:
@@ -50,20 +66,19 @@ def first_text_difference(*, left: str, right: str) -> Divergence | None:
         right_line: str = right_lines[index] if index < len(right_lines) else MISSING_VALUE
         if left_line != right_line:
             return Divergence(
-                location=f"line {index + 1}", left=_preview(left_line), right=_preview(right_line)
+                location=f"line {index + 1}", left=preview(left_line), right=preview(right_line)
             )
     return Divergence(location="end of file", left="<bytes differ>", right="<bytes differ>")
 
 
-def first_document_difference(*, left: str, right: str) -> Divergence | None:
-    """Compare two documents as JSON when both parse, otherwise line by line."""
+def shared_digest(value: object) -> str | None:
+    """Return the digest a stage-capture `{"__shared__": digest}` reference names."""
 
-    try:
-        left_payload: object = json.loads(left)
-        right_payload: object = json.loads(right)
-    except json.JSONDecodeError:
-        return first_text_difference(left=left, right=right)
-    return first_json_difference(left=left_payload, right=right_payload)
+    mapping: dict[str, object] | None = as_json_object(value)
+    if mapping is None or len(mapping) != 1:
+        return None
+    digest: object = mapping.get(STAGE_CAPTURE_SHARED_MARKER)
+    return digest if isinstance(digest, str) else None
 
 
 def as_json_object(value: object) -> dict[str, object] | None:
@@ -79,35 +94,70 @@ def json_pointer_token(key: str) -> str:
 
 
 def _first_object_difference(
-    *, left: dict[str, object], right: dict[str, object], pointer: str
+    *,
+    left: dict[str, object],
+    right: dict[str, object],
+    pointer: str,
+    resolve: NodeResolvers | None,
 ) -> Divergence | None:
     for key in (*left, *(key for key in right if key not in left)):
         location: str = f"{pointer}/{json_pointer_token(key)}"
         if key not in left or key not in right:
             return _divergence(
                 location=location,
-                left=left.get(key, MISSING_VALUE),
-                right=right.get(key, MISSING_VALUE),
+                left=_shown(value=left.get(key, MISSING_VALUE), resolve=resolve, side=0),
+                right=_shown(value=right.get(key, MISSING_VALUE), resolve=resolve, side=1),
             )
         found: Divergence | None = first_json_difference(
-            left=left[key], right=right[key], pointer=location
+            left=left[key], right=right[key], pointer=location, resolve=resolve
         )
         if found is not None:
             return found
     if list(left) != list(right):
         return Divergence(
             location=f"{pointer or '/'} (key order)",
-            left=_preview(list(left)),
-            right=_preview(list(right)),
+            left=preview(list(left)),
+            right=preview(list(right)),
         )
     return None
 
 
+def _shown(*, value: object, resolve: NodeResolvers | None, side: int) -> object:
+    if resolve is None:
+        return value
+    return _previewed(value=value, resolve=resolve[side], remaining=VALUE_PREVIEW_CHARACTERS)[0]
+
+
+def _previewed(
+    *, value: object, resolve: Callable[[object], object], remaining: int
+) -> tuple[object, int]:
+    if remaining <= 0:
+        return value, remaining
+    resolved: object = resolve(value)
+    mapping: dict[str, object] | None = as_json_object(resolved)
+    if mapping is not None:
+        shown: dict[str, object] = {}
+        for key, item in mapping.items():
+            shown[key], remaining = _previewed(
+                value=item, resolve=resolve, remaining=remaining - len(key)
+            )
+        return shown, remaining
+    if isinstance(resolved, list):
+        items: list[object] = []
+        for item in resolved:
+            shown_item, remaining = _previewed(value=item, resolve=resolve, remaining=remaining)
+            items.append(shown_item)
+        return items, remaining
+    return resolved, remaining - len(json.dumps(resolved, ensure_ascii=False))
+
+
 def _divergence(*, location: str, left: object, right: object) -> Divergence:
-    return Divergence(location=location, left=_preview(left), right=_preview(right))
+    return Divergence(location=location, left=preview(left), right=preview(right))
+
 
+def preview(value: object) -> str:
+    """Render a value for a report, shortened to the preview width."""
 
-def _preview(value: object) -> str:
     rendered: str = (
         value
         if isinstance(value, str) and value == MISSING_VALUE
diff --git a/scripts/compiler_differential/_helpers/comparing/comparison.py b/scripts/compiler_differential/_helpers/comparing/comparison.py
index 97dc2c2eea..f10225a405 100644
--- a/scripts/compiler_differential/_helpers/comparing/comparison.py
+++ b/scripts/compiler_differential/_helpers/comparing/comparison.py
@@ -4,10 +4,11 @@
 
 import json
 from collections.abc import Callable
+from pathlib import Path
 
+from scripts.compiler_differential._helpers.comparing.capture import first_capture_difference
 from scripts.compiler_differential._helpers.comparing.compare import (
     as_json_object,
-    first_document_difference,
     first_json_difference,
     first_text_difference,
 )
@@ -237,12 +238,12 @@ def _loads(text: str | None) -> object:
 
 
 def _capture_differences(
-    *, left: dict[str, dict[str, str]], right: dict[str, dict[str, str]]
+    *, left: dict[str, dict[str, Path]], right: dict[str, dict[str, Path]]
 ) -> list[tuple[str, Divergence | None]]:
     differences: list[tuple[str, Divergence | None]] = []
     for command in sorted(set(left) | set(right)):
-        left_files: dict[str, str] = left.get(command, {})
-        right_files: dict[str, str] = right.get(command, {})
+        left_files: dict[str, Path] = left.get(command, {})
+        right_files: dict[str, Path] = right.get(command, {})
         for name in sorted(set(left_files) | set(right_files)):
             artifact: str = f"stage capture {command}/{name}"
             if name not in left_files or name not in right_files:
@@ -258,10 +259,7 @@ def _capture_differences(
             differences.append(
                 (
                     artifact,
-                    first_document_difference(
-                        left=normalize_artifact_text(left_files[name]),
-                        right=normalize_artifact_text(right_files[name]),
-                    ),
+                    first_capture_difference(left=left_files[name], right=right_files[name]),
                 )
             )
     return differences
diff --git a/scripts/compiler_differential/_helpers/comparing/normalize.py b/scripts/compiler_differential/_helpers/comparing/normalize.py
index 728ed1c84f..d976cecb28 100644
--- a/scripts/compiler_differential/_helpers/comparing/normalize.py
+++ b/scripts/compiler_differential/_helpers/comparing/normalize.py
@@ -6,18 +6,16 @@
 from scripts.compiler_differential.constants import (
     ELAPSED_TIME_MASK,
     ELAPSED_TIME_PATTERN,
-    ENGINE_NAMESPACE_PATTERN,
-    INVOCATION_ID_MASK,
-    INVOCATION_ID_PATTERN,
     MANIFEST_VOLATILE_METADATA,
     STRIPPED_REPORT_FIELDS,
 )
+from sqlbuild.compiler.frontier.classes.stage_capture_encoder import mask_capture_noise
 
 
 def normalize_artifact_text(text: str) -> str:
     """Mask invocation ids and engine store suffixes, which differ by design."""
 
-    return ENGINE_NAMESPACE_PATTERN.sub("", INVOCATION_ID_PATTERN.sub(INVOCATION_ID_MASK, text))
+    return mask_capture_noise(text)
 
 
 def normalize_stderr(text: str) -> str:
diff --git a/scripts/compiler_differential/_helpers/comparing/project_comparison.py b/scripts/compiler_differential/_helpers/comparing/project_comparison.py
index c1e2f3af67..7e5c1397c4 100644
--- a/scripts/compiler_differential/_helpers/comparing/project_comparison.py
+++ b/scripts/compiler_differential/_helpers/comparing/project_comparison.py
@@ -13,7 +13,12 @@
 from scripts.compiler_differential._helpers.coverage.discovery import project_discovery_kinds
 from scripts.compiler_differential._helpers.coverage.render import project_render_kinds
 from scripts.compiler_differential._helpers.running.execution import run_engine
-from scripts.compiler_differential.constants import ERROR_SEVERITY, WARNING_SEVERITY
+from scripts.compiler_differential.constants import (
+    ERROR_SEVERITY,
+    LEFT_SIDE,
+    RIGHT_SIDE,
+    WARNING_SEVERITY,
+)
 from scripts.compiler_differential.models import (
     CommandOutcome,
     CorpusProject,
@@ -42,6 +47,7 @@ def compare_project(*, project: CorpusProject, options: DifferentialOptions) ->
         source_dir=source_dir,
         case_dir=case_dir,
         engine=left_engine,
+        side=LEFT_SIDE,
         options=options,
     )
     right: EngineRun = run_engine(
@@ -49,6 +55,7 @@ def compare_project(*, project: CorpusProject, options: DifferentialOptions) ->
         source_dir=source_dir,
         case_dir=case_dir,
         engine=right_engine,
+        side=RIGHT_SIDE,
         options=options,
     )
     counted: bool = project.seed_coverage and options.stage_captures
diff --git a/scripts/compiler_differential/_helpers/corpus/emitted_codes.py b/scripts/compiler_differential/_helpers/corpus/emitted_codes.py
index 810e3f1480..90b28c7d88 100644
--- a/scripts/compiler_differential/_helpers/corpus/emitted_codes.py
+++ b/scripts/compiler_differential/_helpers/corpus/emitted_codes.py
@@ -17,6 +17,7 @@
     DIAGNOSTIC_CODE_PATTERN,
     ERROR_SEVERITY,
     EXPECT_SUCCESS,
+    LEFT_SIDE,
     RENDER_CODE_SCAN_EXCLUDED,
     RENDER_RAISED_CODES,
     WARNING_SEVERITY,
@@ -139,6 +140,7 @@ def _emitted_codes(*, project: CorpusProject, options: DifferentialOptions) -> E
         source_dir=source_dir,
         case_dir=case_dir,
         engine=options.engines[0],
+        side=LEFT_SIDE,
         options=options,
     )
     return EmittedCodes(
diff --git a/scripts/compiler_differential/_helpers/coverage/capture_values.py b/scripts/compiler_differential/_helpers/coverage/capture_values.py
index 40e2d90287..16d29d8956 100644
--- a/scripts/compiler_differential/_helpers/coverage/capture_values.py
+++ b/scripts/compiler_differential/_helpers/coverage/capture_values.py
@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-import json
 import re
 from collections.abc import Iterator
 
 from scripts.compiler_differential._helpers.comparing.compare import as_json_object
+from scripts.compiler_differential.classes.capture_file import expand_capture_text
 from scripts.compiler_differential.constants import (
     CALLABLE_MARKER,
     COLLECTION_CAPTURE_MARKERS,
@@ -87,7 +87,7 @@ def capture_problems(
 ) -> tuple[str, ...]:
     """Return why a capture is incomplete or not canonical; empty when it is sound."""
 
-    capture: dict[str, object] = as_json_object(json.loads(capture_text)) or {}
+    capture: dict[str, object] = as_json_object(expand_capture_text(capture_text)) or {}
     problems: list[str] = []
     captured: tuple[str, ...] = tuple(key for key in capture if key != TYPE_MARKER)
     if captured != fields:
diff --git a/scripts/compiler_differential/_helpers/coverage/discovery.py b/scripts/compiler_differential/_helpers/coverage/discovery.py
index 37e800e26f..bb4d871983 100644
--- a/scripts/compiler_differential/_helpers/coverage/discovery.py
+++ b/scripts/compiler_differential/_helpers/coverage/discovery.py
@@ -17,6 +17,7 @@
     named_values,
     records,
 )
+from scripts.compiler_differential.classes.capture_file import expand_capture_text
 from scripts.compiler_differential.constants import (
     CALLABLE_CAPTURE_FIELDS,
     CRLF_BYTES,
@@ -59,7 +60,7 @@ def discovery_collection_kinds() -> tuple[str, ...]:
 
 
 def project_discovery_kinds(
-    *, captures: dict[str, dict[str, str]], source_dir: Path
+    *, captures: dict[str, dict[str, Path]], source_dir: Path
 ) -> frozenset[str]:
     """Return the kinds of every discovery capture plus CRLF, which only the bytes show."""
 
@@ -75,7 +76,7 @@ def project_discovery_kinds(
 def discovered_relative_paths(capture_text: str) -> frozenset[str]:
     """Return the project-relative path of every file a discovery capture says it read."""
 
-    capture: object = json.loads(capture_text)
+    capture: object = expand_capture_text(capture_text)
     return frozenset(
         str(path)
         for _, value in named_values(value=capture, names=RELATIVE_PATH_FIELDS)
@@ -94,17 +95,17 @@ def authored_byte_kinds(*, source_dir: Path, relative_paths: frozenset[str]) ->
     )
 
 
-def _discovery_capture_texts(captures: dict[str, dict[str, str]]) -> Iterator[str]:
+def _discovery_capture_texts(captures: dict[str, dict[str, Path]]) -> Iterator[str]:
     for files in captures.values():
-        for name, text in files.items():
+        for name, path in files.items():
             if name.endswith(DISCOVERY_STAGE_CAPTURE_SUFFIX):
-                yield text
+                yield path.read_text(encoding="utf-8")
 
 
 def discovered_input_kinds(capture_text: str) -> frozenset[str]:
     """Return the input kinds one canonical discovery capture proves were discovered."""
 
-    capture: dict[str, object] = as_json_object(json.loads(capture_text)) or {}
+    capture: dict[str, object] = as_json_object(expand_capture_text(capture_text)) or {}
     kinds: set[str] = {
         kind for kind in discovery_collection_kinds() if has_content(capture.get(kind))
     }
diff --git a/scripts/compiler_differential/_helpers/coverage/render.py b/scripts/compiler_differential/_helpers/coverage/render.py
index ffd99a14da..828659d4a4 100644
--- a/scripts/compiler_differential/_helpers/coverage/render.py
+++ b/scripts/compiler_differential/_helpers/coverage/render.py
@@ -6,7 +6,7 @@
 import json
 import re
 from collections.abc import Iterator
-from pathlib import PurePosixPath
+from pathlib import Path, PurePosixPath
 
 from scripts.compiler_differential._helpers.comparing.compare import as_json_object
 from scripts.compiler_differential._helpers.coverage.capture_values import (
@@ -20,6 +20,7 @@
     macro_function_source,
     takes_typed_reference,
 )
+from scripts.compiler_differential.classes.capture_file import expand_capture_text
 from scripts.compiler_differential.constants import (
     RENDER_CALLABLE_CAPTURE_FIELDS,
     RENDER_DETAIL_KINDS,
@@ -77,14 +78,14 @@ def render_collection_kinds() -> tuple[str, ...]:
     )
 
 
-def project_render_kinds(captures: dict[str, dict[str, str]]) -> frozenset[str]:
+def project_render_kinds(captures: dict[str, dict[str, Path]]) -> frozenset[str]:
     """Return the render kinds every `CompileProjectInputs` capture of one project proves."""
 
     kinds: set[str] = set()
     for files in captures.values():
-        for name, text in files.items():
+        for name, path in files.items():
             if name.endswith(RENDER_STAGE_CAPTURE_SUFFIX):
-                kinds.update(rendered_input_kinds(text))
+                kinds.update(rendered_input_kinds(path.read_text(encoding="utf-8")))
     return frozenset(kinds)
 
 
@@ -101,7 +102,7 @@ def render_capture_problems(capture_text: str) -> tuple[str, ...]:
 def rendered_input_kinds(capture_text: str) -> frozenset[str]:
     """Return the render input kinds one canonical `CompileProjectInputs` capture proves."""
 
-    capture: dict[str, object] = as_json_object(json.loads(capture_text)) or {}
+    capture: dict[str, object] = as_json_object(expand_capture_text(capture_text)) or {}
     kinds: set[str] = {kind for kind in render_collection_kinds() if has_content(capture.get(kind))}
     usages: list[tuple[str, dict[str, object]]] = list(_usages(capture))
     kinds.update(_macro_kinds(capture=capture, usages=usages))
diff --git a/scripts/compiler_differential/_helpers/running/execution.py b/scripts/compiler_differential/_helpers/running/execution.py
index 4835622891..5e1b13c5c4 100644
--- a/scripts/compiler_differential/_helpers/running/execution.py
+++ b/scripts/compiler_differential/_helpers/running/execution.py
@@ -45,9 +45,10 @@ def run_engine(
     source_dir: Path,
     case_dir: Path,
     engine: str,
+    side: str,
     options: DifferentialOptions,
 ) -> EngineRun:
-    """Run the project's commands under one engine in a fresh copy at a fixed path."""
+    """Run the project's commands under one engine in a fresh copy, keeping captures on disk."""
 
     workspace: Path = case_dir / PROJECT_DIRECTORY
     shutil.rmtree(workspace, ignore_errors=True)
@@ -57,7 +58,7 @@ def run_engine(
         if project.project_subdirectory is None
         else workspace / project.project_subdirectory
     )
-    capture_root: Path = case_dir / f"{engine}-{CAPTURES_DIRECTORY}"
+    capture_root: Path = case_dir / f"{side}-{engine}-{CAPTURES_DIRECTORY}"
     shutil.rmtree(capture_root, ignore_errors=True)
     outcomes: list[CommandOutcome] = []
     for index, command in enumerate(project.commands):
@@ -166,16 +167,10 @@ def _optional_text(path: Path) -> str | None:
     return path.read_text(encoding="utf-8", errors="surrogateescape") if path.is_file() else None
 
 
-def _captures(root: Path) -> dict[str, dict[str, str]]:
-    captures: dict[str, dict[str, str]] = {}
+def _captures(root: Path) -> dict[str, dict[str, Path]]:
+    captures: dict[str, dict[str, Path]] = {}
     for command_dir in sorted(root.iterdir()) if root.is_dir() else ():
-        captures[command_dir.name] = _capture_files(command_dir)
+        captures[command_dir.name] = {
+            capture.name: capture for capture in sorted(command_dir.iterdir()) if capture.is_file()
+        }
     return captures
-
-
-def _capture_files(command_dir: Path) -> dict[str, str]:
-    return {
-        capture.name: capture.read_text(encoding="utf-8")
-        for capture in sorted(command_dir.iterdir())
-        if capture.is_file()
-    }
diff --git a/scripts/compiler_differential/classes/capture_file.py b/scripts/compiler_differential/classes/capture_file.py
new file mode 100644
index 0000000000..bc73aaa489
--- /dev/null
+++ b/scripts/compiler_differential/classes/capture_file.py
@@ -0,0 +1,92 @@
+"""Read one stage capture from disk node by node instead of parsing it whole."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from scripts.compiler_differential._helpers.comparing.compare import as_json_object, shared_digest
+from scripts.compiler_differential._helpers.comparing.normalize import normalize_artifact_text
+from sqlbuild.compiler.frontier.constants import (
+    STAGE_CAPTURE_ROOT_KEY,
+    STAGE_CAPTURE_SHARED_NODES_KEY,
+)
+
+_SHARED_HEADER: bytes = f"{{{json.dumps(STAGE_CAPTURE_SHARED_NODES_KEY)}: {{\n".encode()
+_ROOT_PREFIX: bytes = f"{json.dumps(STAGE_CAPTURE_ROOT_KEY)}: ".encode()
+_SHARED_DOCUMENT_KEYS: frozenset[str] = frozenset(
+    {STAGE_CAPTURE_SHARED_NODES_KEY, STAGE_CAPTURE_ROOT_KEY}
+)
+_DIGEST_CHARACTERS: int = 64
+_NODE_PREFIX_BYTES: int = _DIGEST_CHARACTERS + len(b'"": ')
+
+
+class CaptureFile:
+    """A normalized stage capture whose shared nodes are parsed only when resolved."""
+
+    def __init__(self, path: Path) -> None:
+        self._path: Path = path
+        self._offsets: dict[str, int] = {}
+        self.root: object = self._index()
+
+    def resolve(self, value: object) -> object:
+        """Return the shared node a reference names, or the value itself."""
+
+        digest: str | None = shared_digest(value)
+        return value if digest is None else self._node(digest)
+
+    def _index(self) -> object:
+        with self._path.open("rb") as stream:
+            if stream.readline() != _SHARED_HEADER:
+                return _parse(self._path.read_bytes())
+            root: object = None
+            offset: int = stream.tell()
+            for line in iter(stream.readline, b""):
+                if line.startswith(_ROOT_PREFIX):
+                    root = _parse(line[len(_ROOT_PREFIX) :])
+                elif line.startswith(b'"') and len(line) > _NODE_PREFIX_BYTES:
+                    self._offsets[line[1 : _DIGEST_CHARACTERS + 1].decode("ascii")] = offset
+                offset += len(line)
+            return root
+
+    def _node(self, digest: str) -> object:
+        offset: int | None = self._offsets.get(digest)
+        if offset is None:
+            raise json.JSONDecodeError(f"missing shared node {digest}", str(self._path), 0)
+        with self._path.open("rb") as stream:
+            _ = stream.seek(offset)
+            line: bytes = stream.readline()
+        return _parse(line[_NODE_PREFIX_BYTES:])
+
+
+def expand_capture_text(text: str) -> object:
+    """Parse a whole capture document, replacing every shared reference by its node."""
+
+    document: object = json.loads(text)
+    mapping: dict[str, object] | None = as_json_object(document)
+    nodes: dict[str, object] | None = as_json_object(
+        None if mapping is None else mapping.get(STAGE_CAPTURE_SHARED_NODES_KEY)
+    )
+    if mapping is None or nodes is None or set(mapping) != _SHARED_DOCUMENT_KEYS:
+        return document
+    expanded: dict[str, object] = {}
+    for digest, node in nodes.items():
+        expanded[digest] = _expanded(value=node, nodes=expanded)
+    return _expanded(value=mapping[STAGE_CAPTURE_ROOT_KEY], nodes=expanded)
+
+
+def _expanded(*, value: object, nodes: dict[str, object]) -> object:
+    digest: str | None = shared_digest(value)
+    if digest is not None:
+        return nodes[digest]
+    items: dict[str, object] | None = as_json_object(value)
+    if items is not None:
+        return {key: _expanded(value=item, nodes=nodes) for key, item in items.items()}
+    if isinstance(value, list):
+        return [_expanded(value=item, nodes=nodes) for item in value]
+    return value
+
+
+def _parse(data: bytes) -> object:
+    text: str = data.decode("utf-8").rstrip("\n")
+    return json.loads(normalize_artifact_text(text.removesuffix(",")))
diff --git a/scripts/compiler_differential/constants.py b/scripts/compiler_differential/constants.py
index e9429d3521..8b72d0b5c7 100644
--- a/scripts/compiler_differential/constants.py
+++ b/scripts/compiler_differential/constants.py
@@ -17,15 +17,14 @@
 DUCKDB_ADAPTER: str = "duckdb"
 PROJECT_DIRECTORY: str = "project"
 CAPTURES_DIRECTORY: str = "captures"
+LEFT_SIDE: str = "left"
+RIGHT_SIDE: str = "right"
 TARGET_DIRECTORY: str = "target"
 COMPILED_DIRECTORY: str = "target/compiled"
 MANIFEST_FILE: str = "target/manifest.json"
 DAG_FILE: str = "target/sqlbuild_dag.json"
 STRIPPED_REPORT_FIELDS: frozenset[str] = frozenset({"compile_timings", "compiler_engine"})
 MANIFEST_VOLATILE_METADATA: frozenset[str] = frozenset({"generated_at", "invocation_id"})
-INVOCATION_ID_PATTERN: re.Pattern[str] = re.compile(r"\b\d{8}T\d{6}Z_[0-9a-f]{12}\b")
-INVOCATION_ID_MASK: str = "<invocation-id>"
-ENGINE_NAMESPACE_PATTERN: re.Pattern[str] = re.compile(r"-native-v\d+\b")
 ELAPSED_TIME_PATTERN: re.Pattern[str] = re.compile(r"\b\d+(?:\.\d+)?\s?m?s\b")
 ELAPSED_TIME_MASK: str = "<elapsed>"
 COMMAND_TIMEOUT_SECONDS: float = 900.0
diff --git a/scripts/compiler_differential/models.py b/scripts/compiler_differential/models.py
index 9237e10793..5a399c2e9c 100644
--- a/scripts/compiler_differential/models.py
+++ b/scripts/compiler_differential/models.py
@@ -57,7 +57,7 @@ class EngineRun:
     compiled: dict[str, bytes]
     manifest: str | None
     dag: str | None
-    captures: dict[str, dict[str, str]] = field(default_factory=dict)
+    captures: dict[str, dict[str, Path]] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
diff --git a/scripts/compiler_differential/types.py b/scripts/compiler_differential/types.py
new file mode 100644
index 0000000000..7ee809df54
--- /dev/null
+++ b/scripts/compiler_differential/types.py
@@ -0,0 +1,5 @@
+"""Type aliases for the compiler differential harness."""
+
+from collections.abc import Callable
+
+type NodeResolvers = tuple[Callable[[object], object], Callable[[object], object]]
diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py
index 54418cdf95..10a92d3e45 100644
--- a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py
+++ b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py
@@ -46,7 +46,7 @@
 from sqlbuild.compiler.sql_analysis.constants import BINDING_SEVERITIES, TYPE_CHECKED_DIALECTS
 from sqlbuild.compiler.sql_analysis.models import SqlBindingDiagnostic
 
-_ANALYSIS_CACHE_VERSION: int = 12
+_ANALYSIS_CACHE_VERSION: int = 13
 _ANALYSIS_ALGORITHM_FINGERPRINT: str = "model-sql-analysis-v21-completed-star-shapes"
 _LINEAGE_COLUMN_VALUE_COUNT: int = 4
 _LINEAGE_SOURCE_VALUE_COUNT: int = 3
diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py
index 3833027435..c2b40d58ef 100644
--- a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py
+++ b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py
@@ -1274,6 +1274,7 @@ def _compact_projected_analysis_result(
                 string_pool=string_pool,
                 rows=cached_facts.lineage_rows,
                 resource_name_indexes=request.resource_name_indexes,
+                resource_names=request.resource_names,
             ),
             has_star=bool(request.analysis and request.analysis.get("hasStar")),
             star_resolved=bool(request.analysis and request.analysis.get("starResolved")),
diff --git a/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py b/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py
index da8a53de2a..fc356661bd 100644
--- a/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py
+++ b/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py
@@ -1,11 +1,13 @@
 """Dump frontier objects as canonical JSON so two engines can be compared stage by stage."""
 
-import json
+import io
 import os
 from pathlib import Path
+from typing import TextIO
 
 from sqlbuild.compiler.frontier.classes.stage_capture_encoder import StageCaptureEncoder
 from sqlbuild.compiler.frontier.classes.stage_capture_sequence import StageCaptureSequence
+from sqlbuild.compiler.frontier.classes.stage_capture_writer import StageCaptureWriter
 from sqlbuild.compiler.frontier.constants import STAGE_CAPTURE_DIR_ENV_VAR, STAGE_CAPTURE_SUFFIX
 from sqlbuild.compiler.frontier.types import CompilerStage
 
@@ -20,18 +22,18 @@ def write_stage_capture(*, stage: CompilerStage, value: object) -> None:
         f"{StageCaptureSequence.next_value():03d}-{stage.value}{STAGE_CAPTURE_SUFFIX}"
     )
     path.parent.mkdir(parents=True, exist_ok=True)
-    _ = path.write_text(render_stage_capture(value), encoding="utf-8")
+    with path.open("w", encoding="utf-8", newline="\n") as stream:
+        _write_capture(value=value, stream=stream)
 
 
 def render_stage_capture(value: object) -> str:
-    """Render a value as indented JSON that keeps insertion order and sorts only sets."""
-
-    return (
-        json.dumps(
-            StageCaptureEncoder().encode(value),
-            indent=1,
-            ensure_ascii=False,
-            allow_nan=False,
-        )
-        + "\n"
-    )
+    """Render a value as the JSON document `write_stage_capture` writes."""
+
+    stream: io.StringIO = io.StringIO()
+    _write_capture(value=value, stream=stream)
+    return stream.getvalue()
+
+
+def _write_capture(*, value: object, stream: TextIO) -> None:
+    writer: StageCaptureWriter = StageCaptureWriter(stream)
+    writer.finish(StageCaptureEncoder(on_shared=writer.write_shared).encode(value))
diff --git a/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py b/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py
index b7d31c891a..4807ca8f74 100644
--- a/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py
+++ b/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py
@@ -12,12 +12,20 @@
 import math
 import re
 import types
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping, Sequence
 from collections.abc import Set as AbstractSet
 from pathlib import PurePath
 from typing import Any
 
 from sqlbuild.compiler.frontier.constants import (
+    STAGE_CAPTURE_DECODED_SEQUENCES,
+    STAGE_CAPTURE_ENGINE_NAMESPACE_PATTERN,
+    STAGE_CAPTURE_INVOCATION_ID_MASK,
+    STAGE_CAPTURE_INVOCATION_ID_PATTERN,
+    STAGE_CAPTURE_OMITTED_ATTRIBUTES,
+    STAGE_CAPTURE_RESERVED_KEYS,
+    STAGE_CAPTURE_SHARED_MARKER,
+    STAGE_CAPTURE_SHARED_MIN_BYTES,
     STAGE_CAPTURE_SKIPPED_SLOTS,
     STAGE_CAPTURE_UNORDERED_ATTRIBUTES,
 )
@@ -32,17 +40,29 @@
 
 
 class StageCaptureEncoder:
-    """Encode one object graph deterministically, replacing callables with qualified names."""
-
-    def __init__(self) -> None:
+    """Encode an object graph deterministically, sharing large subtrees by content digest."""
+
+    def __init__(
+        self,
+        *,
+        on_shared: Callable[..., None] | None = None,
+        share_min_bytes: int = STAGE_CAPTURE_SHARED_MIN_BYTES,
+    ) -> None:
         self._active: set[int] = set()
+        self._cycles: int = 0
+        self._encoded: dict[int, tuple[object, object]] = {}
+        self._on_shared: Callable[..., None] | None = on_shared
+        self._share_min_bytes: int = share_min_bytes
+        self._digests: set[str] = set()
 
     def encode(self, value: object) -> object:
         """Return a JSON-compatible value that keeps insertion order and sorts only sets."""
 
         if isinstance(value, enum.Enum):
             return {"__enum__": f"{qualified_name(type(value))}.{value.name}"}
-        if value is None or isinstance(value, (bool, int, str)):
+        if isinstance(value, str):
+            return value if len(value) < self._share_min_bytes else self._share(value)
+        if value is None or isinstance(value, (bool, int)):
             return value
         if isinstance(value, float):
             return value if math.isfinite(value) else {"__float__": repr(value)}
@@ -51,14 +71,37 @@ def encode(self, value: object) -> object:
             return scalar
         identity: int = id(value)
         if identity in self._active:
+            self._cycles += 1
             return {"__cycle__": qualified_name(type(value))}
+        known: tuple[object, object] | None = self._encoded.get(identity)
+        if known is not None:
+            return known[1]
+        cycles: int = self._cycles
         self._active.add(identity)
         try:
-            return self._encode_container(value)
+            encoded: object = self._share(self._encode_container(value))
         finally:
             self._active.discard(identity)
+        if self._cycles == cycles:
+            self._encoded[identity] = (value, encoded)
+        return encoded
+
+    def _share(self, encoded: object) -> object:
+        if self._on_shared is None or not isinstance(encoded, (dict, list, str)):
+            return encoded
+        text: str = json.dumps(encoded, ensure_ascii=False, allow_nan=False, separators=(",", ":"))
+        if len(text) < self._share_min_bytes:
+            return encoded
+        digest: str = hashlib.sha256(mask_capture_noise(text).encode("utf-8")).hexdigest()
+        if digest not in self._digests:
+            self._digests.add(digest)
+            self._on_shared(digest=digest, text=text)
+        return {STAGE_CAPTURE_SHARED_MARKER: digest}
 
     def _encode_container(self, value: object) -> object:
+        type_name: str = qualified_name(type(value))
+        if type_name in STAGE_CAPTURE_DECODED_SEQUENCES and isinstance(value, Sequence):
+            return [self.encode(item) for item in value]
         if dataclasses.is_dataclass(value) and not isinstance(value, type):
             return {
                 "__type__": qualified_name(type(value)),
@@ -80,15 +123,16 @@ def _encode_container(self, value: object) -> object:
                 "keywords": self.encode(value.keywords),
             }
         attributes: dict[str, object] | None = _instance_attributes(value)
-        type_name: str = qualified_name(type(value))
         if attributes is None:
             return {"__opaque__": type_name}
         unordered: frozenset[str] = STAGE_CAPTURE_UNORDERED_ATTRIBUTES.get(type_name, frozenset())
+        omitted: frozenset[str] = STAGE_CAPTURE_OMITTED_ATTRIBUTES.get(type_name, frozenset())
         return {
             "__type__": type_name,
             **{
                 name: self._encode_attribute(value=item, unordered=name in unordered)
                 for name, item in attributes.items()
+                if name not in omitted
             },
         }
 
@@ -98,7 +142,9 @@ def _encode_attribute(self, *, value: object, unordered: bool) -> object:
         return {"__unordered_mapping__": sorted(self._mapping_pairs(value), key=_sort_key)}
 
     def _encode_mapping(self, value: Mapping[Any, object]) -> object:
-        if all(isinstance(key, str) for key in value):
+        if all(isinstance(key, str) for key in value) and STAGE_CAPTURE_RESERVED_KEYS.isdisjoint(
+            value
+        ):
             return {str(key): self.encode(item) for key, item in value.items()}
         return {"__mapping__": self._mapping_pairs(value)}
 
@@ -135,8 +181,17 @@ def _encode_scalar(value: object) -> object | None:
     return None
 
 
+def mask_capture_noise(text: str) -> str:
+    """Mask invocation ids and native store suffixes, which differ between identical compiles."""
+
+    return STAGE_CAPTURE_ENGINE_NAMESPACE_PATTERN.sub(
+        "",
+        STAGE_CAPTURE_INVOCATION_ID_PATTERN.sub(STAGE_CAPTURE_INVOCATION_ID_MASK, text),
+    )
+
+
 def _sort_key(item: object) -> str:
-    return json.dumps(item, sort_keys=True, ensure_ascii=False)
+    return mask_capture_noise(json.dumps(item, sort_keys=True, ensure_ascii=False))
 
 
 def _instance_attributes(value: object) -> dict[str, object] | None:
diff --git a/src/sqlbuild/compiler/frontier/classes/stage_capture_writer.py b/src/sqlbuild/compiler/frontier/classes/stage_capture_writer.py
new file mode 100644
index 0000000000..1dd17b9ad1
--- /dev/null
+++ b/src/sqlbuild/compiler/frontier/classes/stage_capture_writer.py
@@ -0,0 +1,39 @@
+"""Stream one stage capture so shared subtrees reach the file as they are encoded."""
+
+from __future__ import annotations
+
+import json
+from typing import TextIO
+
+from sqlbuild.compiler.frontier.constants import (
+    STAGE_CAPTURE_ROOT_KEY,
+    STAGE_CAPTURE_SHARED_NODES_KEY,
+)
+
+
+class StageCaptureWriter:
+    """Write a capture as one JSON document with each shared node and the root on its own line."""
+
+    def __init__(self, stream: TextIO) -> None:
+        self._stream: TextIO = stream
+        self._shared: bool = False
+
+    def write_shared(self, *, digest: str, text: str) -> None:
+        """Append one shared node, given as compact JSON."""
+
+        prefix: str = (
+            ",\n" if self._shared else f"{{{json.dumps(STAGE_CAPTURE_SHARED_NODES_KEY)}: {{\n"
+        )
+        self._shared = True
+        _ = self._stream.write(f"{prefix}{json.dumps(digest)}: {text}")
+
+    def finish(self, root: object) -> None:
+        """Write the root and close the document."""
+
+        if not self._shared:
+            _ = self._stream.write(
+                json.dumps(root, indent=1, ensure_ascii=False, allow_nan=False) + "\n"
+            )
+            return
+        compact: str = json.dumps(root, ensure_ascii=False, allow_nan=False, separators=(",", ":"))
+        _ = self._stream.write(f"\n}},\n{json.dumps(STAGE_CAPTURE_ROOT_KEY)}: {compact}\n}}\n")
diff --git a/src/sqlbuild/compiler/frontier/constants.py b/src/sqlbuild/compiler/frontier/constants.py
index 68818cdc2a..09445fd0de 100644
--- a/src/sqlbuild/compiler/frontier/constants.py
+++ b/src/sqlbuild/compiler/frontier/constants.py
@@ -1,5 +1,7 @@
 """Compiler engine switch, cache namespace, and stage capture constants."""
 
+import re
+
 from sqlbuild.compiler.frontier.types import CompilerEngine
 
 COMPILER_ENGINE_ENV_VAR: str = "SQLBUILD_COMPILER_ENGINE"
@@ -12,9 +14,27 @@
 COMPILER_CACHE_DIRECTORY_NAME: str = "compiler"
 STAGE_CAPTURE_DIR_ENV_VAR: str = "SQLBUILD_COMPILER_STAGE_CAPTURE_DIR"
 STAGE_CAPTURE_SUFFIX: str = ".json"
+STAGE_CAPTURE_SHARED_MARKER: str = "__shared__"
+STAGE_CAPTURE_ROOT_KEY: str = "__capture__"
+STAGE_CAPTURE_SHARED_NODES_KEY: str = "__shared_nodes__"
+STAGE_CAPTURE_SHARED_MIN_BYTES: int = 1024
+STAGE_CAPTURE_RESERVED_KEYS: frozenset[str] = frozenset(
+    {STAGE_CAPTURE_SHARED_MARKER, STAGE_CAPTURE_ROOT_KEY, STAGE_CAPTURE_SHARED_NODES_KEY}
+)
+STAGE_CAPTURE_INVOCATION_ID_PATTERN: re.Pattern[str] = re.compile(r"\b\d{8}T\d{6}Z_[0-9a-f]{12}\b")
+STAGE_CAPTURE_INVOCATION_ID_MASK: str = "<invocation-id>"
+STAGE_CAPTURE_ENGINE_NAMESPACE_PATTERN: re.Pattern[str] = re.compile(r"-native-v\d+\b")
 STAGE_CAPTURE_SKIPPED_SLOTS: frozenset[str] = frozenset({"__dict__", "__weakref__"})
 STAGE_CAPTURE_UNORDERED_ATTRIBUTES: dict[str, frozenset[str]] = {
     "sqlbuild.compiler.sql_analysis.classes.binding_catalog:BindingCatalog": frozenset(
-        {"schemas", "analysis_shapes", "expression_shapes", "shared_analyses"}
+        {"schemas", "analysis_shapes", "expression_shapes"}
+    )
+}
+STAGE_CAPTURE_OMITTED_ATTRIBUTES: dict[str, frozenset[str]] = {
+    "sqlbuild.compiler.sql_analysis.classes.binding_catalog:BindingCatalog": frozenset(
+        {"shared_analyses"}
     )
 }
+STAGE_CAPTURE_DECODED_SEQUENCES: frozenset[str] = frozenset(
+    {"sqlbuild.compiler.compile.models:CompactLineageFacts"}
+)
diff --git a/tests/e2e/scripts/compiler_differential/test_compiler_differential.py b/tests/e2e/scripts/compiler_differential/test_compiler_differential.py
index ebf2ba845c..f66275611d 100644
--- a/tests/e2e/scripts/compiler_differential/test_compiler_differential.py
+++ b/tests/e2e/scripts/compiler_differential/test_compiler_differential.py
@@ -268,6 +268,47 @@ def test_given_required_coverage_missing_when_comparing_then_harness_fails_and_s
     assert not any(text in output for text in test_case.expected_absent), output
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        HarnessRunTestCase(
+            description="shared_analysis_seed_against_itself",
+            extra_arguments=(
+                "--corpus",
+                "seeds",
+                "--seed-start",
+                "29",
+                "--seeds",
+                "1",
+                "--engines",
+                "python",
+                "python",
+                "--stage-captures",
+            ),
+            expected_exit_code=0,
+            expected_lines=(
+                "OK   seed/29",
+                "Compiler differential passed: 1 projects identical (python vs python)",
+            ),
+            expected_patterns=(),
+            expected_absent=("DIFF",),
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_python_engine_twice_when_capturing_stages_then_captures_are_identical(
+    test_case: HarnessRunTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    exit_code: int = run_compiler_differential(
+        ["--jobs", "1", "--work-dir", str(tmp_path / "work"), *test_case.extra_arguments]
+    )
+
+    output: str = capsys.readouterr().out
+    assert exit_code == test_case.expected_exit_code, output
+    assert all(line in output for line in test_case.expected_lines), output
+    assert not any(text in output for text in test_case.expected_absent), output
+
+
 @pytest.mark.parametrize(
     "test_case",
     [
diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py
index 81a1f14d2d..0e6963f61c 100644
--- a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py
+++ b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py
@@ -440,6 +440,14 @@ class DataflowScheduleCase:
     expected_edit_exit_code: int = 1
 
 
+@dataclass(frozen=True)
+class DataflowCaptureCase:
+    description: str
+    write_project: Callable[[Path], object]
+    schedules: tuple[DataflowScheduleCase, ...]
+    expected_difference: str | None = None
+
+
 @dataclass(frozen=True)
 class DataflowFailureCase:
     description: str
diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py b/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py
index 66f7ada616..52cea2d482 100644
--- a/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py
+++ b/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py
@@ -41,6 +41,8 @@
 )
 from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
 from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+from sqlbuild.compiler.frontier.constants import STAGE_CAPTURE_DIR_ENV_VAR
+from sqlbuild.compiler.frontier.types import CompilerStage
 from sqlbuild.compiler.manifest.main.build import build_manifest
 from sqlbuild.compiler.pipeline.main.compile import run_compile_pipeline
 from sqlbuild.compiler.pipeline.main.project import compile_project
@@ -306,6 +308,22 @@ def timed_compile_outcome(
     ), timings
 
 
+def compiled_project_capture(
+    *,
+    project_dir: Path,
+    capture_dir: Path,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> Path:
+    """Cold-compile once with stage captures; return the compiled-project capture file."""
+
+    with monkeypatch.context() as capture_patch:
+        capture_patch.setenv(STAGE_CAPTURE_DIR_ENV_VAR, str(capture_dir))
+        _ = main(["--project-dir", str(project_dir), "compile", "--json", "--no-cache"])
+    _ = capsys.readouterr()
+    return next(capture_dir.glob(f"*-{CompilerStage.COMPILED_PROJECT.value}.json"))
+
+
 def reshape_models(*, project_dir: Path, names: tuple[str, ...], indexes: tuple[int, ...]) -> None:
     """Reshape the models at `indexes` of the generated topological name order."""
 
@@ -319,6 +337,14 @@ def use_wave_analysis(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(project_assembly, "analyze_binding_dataflow", analyze_binding_waves)
 
 
+def analyze_single_model_batches(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Analyze every model in its own native batch on one worker."""
+
+    monkeypatch.setattr(binding_waves, "_DATAFLOW_WORKERS", 1)
+    monkeypatch.setattr(binding_waves, "_DATAFLOW_BATCH_MIN", 1)
+    monkeypatch.setattr(binding_waves, "_DATAFLOW_BATCH_LIMIT", 1)
+
+
 def perturb_dataflow_schedule(
     *, monkeypatch: pytest.MonkeyPatch, case: DataflowScheduleCase, seed: int
 ) -> None:
diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py b/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py
index 4fb2220e19..6b951d7fa0 100644
--- a/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py
+++ b/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py
@@ -9,9 +9,13 @@
 
 from scripts.cold_compile_performance._helpers.random_dag_project import write_random_dag_project
 from scripts.cold_compile_performance.models import RandomDagProject
+from scripts.compiler_differential._helpers.comparing.capture import first_capture_difference
+from scripts.compiler_differential.main.generate_project import generate_project
+from scripts.compiler_differential.models import Divergence
 from sqlbuild.cli.commands.main.entrypoint.entry import main
 from tests.integration.src.sqlbuild.compiler.pipeline._test_types import (
     CompileOutcome,
+    DataflowCaptureCase,
     DataflowFailureCase,
     DataflowInterruptAfterFaultCase,
     DataflowInterruptCase,
@@ -22,17 +26,25 @@
 from tests.integration.src.sqlbuild.compiler.pipeline.helpers import (
     analysis_pool_thread_ids,
     compile_outcome,
+    compiled_project_capture,
     fail_model_analysis,
     fail_worker_start,
     interrupt_after_analysis_fault,
     interrupt_reacquiring_analysis,
     live_model_analysis_threads,
     perturb_dataflow_schedule,
+    random_dag_writer,
     reshape_models,
     use_wave_analysis,
 )
 
 _PROJECT: RandomDagProject = RandomDagProject(seed=97, model_count=40, errors=True)
+_SHARED_ANALYSIS_SEED: int = 29
+_CAPTURE_SCHEDULES: tuple[DataflowScheduleCase, ...] = (
+    DataflowScheduleCase("one worker with single models", 1, 1, 0.0),
+    DataflowScheduleCase("four workers with small batches", 4, 2, 0.004),
+    DataflowScheduleCase("four workers with wide batches", 4, 64, 0.002),
+)
 _EDITED: tuple[int, ...] = (10, 25)
 _REPEATS: tuple[int, ...] = (0, 1)
 _SETTLE_SECONDS: float = 0.3
@@ -81,6 +93,52 @@ def test_given_perturbed_schedule_when_compiling_repeatedly_then_output_is_ident
         assert actual_edit[0] == test_case.expected_edit_exit_code
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        DataflowCaptureCase(
+            "random dag",
+            random_dag_writer(_PROJECT),
+            _CAPTURE_SCHEDULES,
+        ),
+        DataflowCaptureCase(
+            "differential seed with shared analyses",
+            generate_project(seed=_SHARED_ANALYSIS_SEED).write,
+            _CAPTURE_SCHEDULES,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_perturbed_schedule_when_capturing_compiled_project_then_capture_is_identical(
+    test_case: DataflowCaptureCase,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    project_dir: Path = tmp_path / "project"
+    _ = test_case.write_project(project_dir)
+    with monkeypatch.context() as waves_patch:
+        use_wave_analysis(waves_patch)
+        reference: Path = compiled_project_capture(
+            project_dir=project_dir,
+            capture_dir=tmp_path / "reference",
+            capsys=capsys,
+            monkeypatch=monkeypatch,
+        )
+    for index, schedule in enumerate(test_case.schedules):
+        with monkeypatch.context() as schedule_patch:
+            perturb_dataflow_schedule(monkeypatch=schedule_patch, case=schedule, seed=index)
+            actual: Path = compiled_project_capture(
+                project_dir=project_dir,
+                capture_dir=tmp_path / f"schedule-{index}",
+                capsys=capsys,
+                monkeypatch=monkeypatch,
+            )
+
+        difference: Divergence | None = first_capture_difference(left=reference, right=actual)
+        assert getattr(difference, "location", None) == test_case.expected_difference, schedule
+
+
 @pytest.mark.parametrize(
     "test_case",
     [
diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py b/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py
index cd1d1b2cd1..6a07dbe703 100644
--- a/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py
+++ b/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py
@@ -10,6 +10,7 @@
 from sqlbuild.compiler.compile.models import CompactBatchPreparation
 from tests.integration.src.sqlbuild.compiler.pipeline._test_types import SharedBindingQueryCase
 from tests.integration.src.sqlbuild.compiler.pipeline.helpers import (
+    analyze_single_model_batches,
     trace_native_compact_batches,
     write_shared_binding_project,
 )
@@ -229,6 +230,57 @@ def test_given_later_wave_members_with_equal_keys_when_compiling_then_reuses_sha
         assert [step["source"]["resource_name"] for step in lineage["trace"]] == [upstream]
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SharedBindingQueryCase(
+            description="members analyzed in separate batches keep their own sources",
+            orders_summary_sql='MODEL (description "Test model.", materialized view); SELECT id, amount FROM __ref("orders")',
+            customers_summary_sql=(
+                'MODEL (description "Test model.", materialized view); SELECT id, amount FROM __ref("customers")'
+            ),
+            expected_shared_queries=0,
+            expected_lineage=(
+                ("orders_summary", "orders"),
+                ("customers_summary", "customers"),
+            ),
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_equal_keys_in_separate_batches_when_tracing_lineage_then_each_model_keeps_its_source(
+    test_case: SharedBindingQueryCase,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    write_shared_binding_project(project_dir=tmp_path, test_case=test_case)
+    analyze_single_model_batches(monkeypatch)
+
+    for model, upstream in test_case.expected_lineage:
+        assert (
+            main(
+                [
+                    "--project-dir",
+                    str(tmp_path),
+                    "lineage",
+                    f"{model}.amount",
+                    "--format",
+                    "json",
+                    "--direction",
+                    "upstream",
+                    "--depth",
+                    "1",
+                    "--mode",
+                    "fast",
+                ]
+            )
+            == 0
+        )
+        lineage: dict[str, Any] = json.loads(capsys.readouterr().out)
+        assert [step["source"]["resource_name"] for step in lineage["trace"]] == [upstream]
+
+
 @pytest.mark.parametrize(
     "test_case",
     [
diff --git a/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py b/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py
index f799bdcdf4..600124eabf 100644
--- a/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py
+++ b/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import dataclass
 
 
@@ -52,3 +53,34 @@ class PayloadStripTestCase:
     description: str
     payload: object
     expected_payload: object
+
+
+@dataclass(frozen=True)
+class CaptureDifferenceTestCase:
+    """Two frontier values captured to disk and where their first difference must be reported."""
+
+    description: str
+    left: Callable[[], object]
+    right: Callable[[], object]
+    expected_location: str | None
+
+
+@dataclass(frozen=True)
+class CaptureTextDifferenceTestCase:
+    """Two capture files that are not both JSON and the first differing line to report."""
+
+    description: str
+    left: str
+    right: str
+    expected_location: str | None
+
+
+@dataclass(frozen=True)
+class CapturePreviewTestCase:
+    """Two captures whose first difference is a missing value and the preview it must show."""
+
+    description: str
+    left: Callable[[], object]
+    right: Callable[[], object]
+    expected_location: str
+    expected_left_prefix: str
diff --git a/tests/unit/scripts/compiler_differential/_helpers/comparing/helpers.py b/tests/unit/scripts/compiler_differential/_helpers/comparing/helpers.py
new file mode 100644
index 0000000000..09c826ff02
--- /dev/null
+++ b/tests/unit/scripts/compiler_differential/_helpers/comparing/helpers.py
@@ -0,0 +1,80 @@
+"""Neutral frontier-like values whose captures share large repeated subtrees."""
+
+from __future__ import annotations
+
+from typing import cast
+
+_LINE_COUNT: int = 40
+_ORDER_COUNT: int = 6
+_RUN_ID: str = "20261006T172510Z_dee386e8ffa7"
+_OTHER_RUN_ID: str = "20261007T081500Z_0a1b2c3d4e5f"
+_CHANGED_QUANTITY: int = 99
+
+
+def orders_batch() -> dict[str, object]:
+    """Return orders that repeat two identical large orders."""
+
+    return {
+        "run_id": _RUN_ID,
+        "orders": [_order(order_id=index % 2) for index in range(_ORDER_COUNT)],
+    }
+
+
+def orders_batch_with_changed_line(*, order_index: int, line_index: int) -> dict[str, object]:
+    """Return the repeated orders with one quantity of one order changed."""
+
+    batch: dict[str, object] = orders_batch()
+    orders: list[dict[str, object]] = cast(list[dict[str, object]], batch["orders"])
+    lines: list[dict[str, object]] = cast(list[dict[str, object]], orders[order_index]["lines"])
+    lines[line_index]["quantity"] = _CHANGED_QUANTITY
+    return batch
+
+
+def orders_batch_for_run() -> dict[str, object]:
+    """Return the repeated orders, each recording the default run id."""
+
+    return _orders_recording(run_id=_RUN_ID)
+
+
+def orders_batch_from_other_run() -> dict[str, object]:
+    """Return the same orders, each recording a different run id."""
+
+    return _orders_recording(run_id=_OTHER_RUN_ID)
+
+
+def _orders_recording(*, run_id: str) -> dict[str, object]:
+    return {
+        "run_id": run_id,
+        "orders": [
+            {**_order(order_id=index % 2), "run_id": run_id} for index in range(_ORDER_COUNT)
+        ],
+    }
+
+
+def _order(*, order_id: int) -> dict[str, object]:
+    return {"order_id": order_id, "lines": _order_lines()}
+
+
+def _order_lines() -> list[dict[str, object]]:
+    return [{"sku": f"product-{line:04d}", "quantity": line} for line in range(_LINE_COUNT)]
+
+
+def store_paths(*, native_suffix: str) -> dict[str, object]:
+    """Return a set of two large store paths, one carrying an engine namespace suffix."""
+
+    return {"paths": frozenset({"a" * 1100 + f"/store{native_suffix}", "b" * 1100})}
+
+
+def marker_shaped_orders(*, name: str) -> dict[str, object]:
+    """Return authored data whose only key looks like a shared-node reference marker."""
+
+    return {"__shared__": name}
+
+
+def orders_batch_without_last() -> dict[str, object]:
+    """Return the repeated orders with the last one removed."""
+
+    batch: dict[str, object] = orders_batch()
+    orders: list[dict[str, object]] = cast(list[dict[str, object]], batch["orders"])
+    batch["orders"] = orders[:-1]
+    return batch
diff --git a/tests/unit/scripts/compiler_differential/_helpers/comparing/test_capture.py b/tests/unit/scripts/compiler_differential/_helpers/comparing/test_capture.py
new file mode 100644
index 0000000000..2a23951b6b
--- /dev/null
+++ b/tests/unit/scripts/compiler_differential/_helpers/comparing/test_capture.py
@@ -0,0 +1,154 @@
+"""Stage captures on disk are compared node by node with exact first-difference pointers."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from scripts.compiler_differential._helpers.comparing.capture import first_capture_difference
+from scripts.compiler_differential.models import Divergence
+from sqlbuild.compiler.frontier._helpers.stage_capture import render_stage_capture
+from tests.unit.scripts.compiler_differential._helpers.comparing._test_types import (
+    CaptureDifferenceTestCase,
+    CapturePreviewTestCase,
+    CaptureTextDifferenceTestCase,
+)
+from tests.unit.scripts.compiler_differential._helpers.comparing.helpers import (
+    marker_shaped_orders,
+    orders_batch,
+    orders_batch_for_run,
+    orders_batch_from_other_run,
+    orders_batch_with_changed_line,
+    orders_batch_without_last,
+    store_paths,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        CaptureDifferenceTestCase(
+            description="identical_shared_captures",
+            left=orders_batch,
+            right=orders_batch,
+            expected_location=None,
+        ),
+        CaptureDifferenceTestCase(
+            description="change_inside_a_shared_node",
+            left=orders_batch,
+            right=lambda: orders_batch_with_changed_line(order_index=3, line_index=7),
+            expected_location="/orders/3/lines/7/quantity",
+        ),
+        CaptureDifferenceTestCase(
+            description="change_to_one_copy_of_a_repeated_node",
+            left=orders_batch,
+            right=lambda: orders_batch_with_changed_line(order_index=4, line_index=0),
+            expected_location="/orders/4/lines/0/quantity",
+        ),
+        CaptureDifferenceTestCase(
+            description="run_identity_inside_shared_nodes_is_masked",
+            left=orders_batch_for_run,
+            right=orders_batch_from_other_run,
+            expected_location=None,
+        ),
+        CaptureDifferenceTestCase(
+            description="large_set_elements_differing_only_in_masked_noise",
+            left=lambda: store_paths(native_suffix="-native-v3"),
+            right=lambda: store_paths(native_suffix=""),
+            expected_location=None,
+        ),
+        CaptureDifferenceTestCase(
+            description="authored_marker_shaped_data_is_not_a_reference",
+            left=lambda: marker_shaped_orders(name="orders"),
+            right=lambda: marker_shaped_orders(name="customers"),
+            expected_location="/__mapping__/0/1",
+        ),
+        CaptureDifferenceTestCase(
+            description="identical_marker_shaped_data",
+            left=lambda: marker_shaped_orders(name="orders"),
+            right=lambda: marker_shaped_orders(name="orders"),
+            expected_location=None,
+        ),
+        CaptureDifferenceTestCase(
+            description="shared_node_against_inline_value",
+            left=orders_batch,
+            right=lambda: {**orders_batch(), "orders": [{"order_id": 0}]},
+            expected_location="/orders/0/lines",
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_two_captures_when_comparing_then_first_logical_difference_is_reported(
+    test_case: CaptureDifferenceTestCase, tmp_path: Path
+) -> None:
+    left: Path = tmp_path / "left.json"
+    right: Path = tmp_path / "right.json"
+    _ = left.write_text(render_stage_capture(test_case.left()), encoding="utf-8")
+    _ = right.write_text(render_stage_capture(test_case.right()), encoding="utf-8")
+
+    found: Divergence | None = first_capture_difference(left=left, right=right)
+
+    assert getattr(found, "location", None) == test_case.expected_location
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        CapturePreviewTestCase(
+            description="missing_list_item_previews_the_resolved_node",
+            left=orders_batch,
+            right=orders_batch_without_last,
+            expected_location="/orders/5",
+            expected_left_prefix='{"order_id": 1, "lines": [{"sku": "product-0000"',
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_missing_shared_value_when_comparing_then_preview_shows_its_content(
+    test_case: CapturePreviewTestCase, tmp_path: Path
+) -> None:
+    left: Path = tmp_path / "left.json"
+    right: Path = tmp_path / "right.json"
+    _ = left.write_text(render_stage_capture(test_case.left()), encoding="utf-8")
+    _ = right.write_text(render_stage_capture(test_case.right()), encoding="utf-8")
+
+    found: Divergence | None = first_capture_difference(left=left, right=right)
+
+    assert getattr(found, "location", None) == test_case.expected_location
+    assert getattr(found, "left", "").startswith(test_case.expected_left_prefix)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        CaptureTextDifferenceTestCase(
+            description="truncated_capture",
+            left='{"orders": [\n1,\n2\n]}\n',
+            right='{"orders": [\n1,\n3',
+            expected_location="line 3",
+        ),
+        CaptureTextDifferenceTestCase(
+            description="identical_invalid_captures",
+            left="not json\n",
+            right="not json\n",
+            expected_location=None,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_capture_that_is_not_json_when_comparing_then_first_differing_line_is_reported(
+    test_case: CaptureTextDifferenceTestCase, tmp_path: Path
+) -> None:
+    left: Path = tmp_path / "left.json"
+    right: Path = tmp_path / "right.json"
+    _ = left.write_text(test_case.left, encoding="utf-8")
+    _ = right.write_text(test_case.right, encoding="utf-8")
+
+    found: Divergence | None = first_capture_difference(left=left, right=right)
+
+    assert getattr(found, "location", None) == test_case.expected_location
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])
diff --git a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py
index 8acee8e92d..81d0289c37 100644
--- a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py
+++ b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py
@@ -236,14 +236,19 @@ def test_given_authored_bytes_when_classifying_then_crlf_counts_only_for_discove
         path: Path = tmp_path / relative_path
         path.parent.mkdir(parents=True, exist_ok=True)
         _ = path.write_bytes(contents)
-    capture: str = json.dumps(
-        empty_inputs_capture(
-            model_files=[{"relative_path": {"__path__": path}} for path in test_case.read_paths]
-        )
+    capture: Path = tmp_path / "captures" / "001-discovered_project_inputs.json"
+    capture.parent.mkdir()
+    _ = capture.write_text(
+        json.dumps(
+            empty_inputs_capture(
+                model_files=[{"relative_path": {"__path__": path}} for path in test_case.read_paths]
+            )
+        ),
+        encoding="utf-8",
     )
 
     kinds: frozenset[str] = project_discovery_kinds(
-        captures={"0-compile": {"001-discovered_project_inputs.json": capture}}, source_dir=tmp_path
+        captures={"0-compile": {capture.name: capture}}, source_dir=tmp_path
     )
 
     assert ("crlf" in kinds) is test_case.expected_crlf
diff --git a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py
index 5f84d87d99..4aebac3039 100644
--- a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py
+++ b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py
@@ -4,6 +4,7 @@
 
 import dataclasses
 import json
+from pathlib import Path
 
 import pytest
 
@@ -418,12 +419,14 @@ def test_given_render_capture_when_classifying_then_only_rendered_inputs_are_cre
     ids=lambda case: case.description,
 )
 def test_given_encoded_compile_inputs_when_classifying_then_encoded_empty_values_prove_nothing(
-    test_case: RenderEncodedInputsTestCase,
+    test_case: RenderEncodedInputsTestCase, tmp_path: Path
 ) -> None:
     capture: str = render_stage_capture(test_case.inputs)
+    capture_path: Path = tmp_path / "002-compile_project_inputs.json"
+    _ = capture_path.write_text(capture, encoding="utf-8")
 
     assert rendered_input_kinds(capture) == test_case.expected_kinds
-    assert project_render_kinds({"0-compile": {"002-compile_project_inputs.json": capture}}) == (
+    assert project_render_kinds({"0-compile": {capture_path.name: capture_path}}) == (
         test_case.expected_kinds
     )
 
diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py
index f96a30d800..ea1e0d943c 100644
--- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py
+++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py
@@ -194,7 +194,7 @@ def test_given_exact_compact_batch_when_entry_rows_are_absent_then_warm_compile_
     write_repo_files(tmp_path, _CACHE_REPO_FILES)
     monkeypatch.setattr(assembly_project, "_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT", 1)
     cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path)
-    cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3"
+    cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3"
     with sqlite3.connect(cache_path) as connection:
         batch_count: int = connection.execute(
             "SELECT COUNT(*) FROM model_analysis_compact_batch"
@@ -227,7 +227,7 @@ def test_given_one_changed_model_when_entry_rows_are_absent_then_compact_batch_r
     write_repo_files(tmp_path, _SELECTION_REPO_FILES)
     monkeypatch.setattr(assembly_project, "_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT", 1)
     cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path)
-    cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3"
+    cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3"
     with sqlite3.connect(cache_path) as connection:
         _ = connection.execute("DELETE FROM model_analysis")
     (tmp_path / "models" / "unrelated.sql").write_text(
@@ -265,7 +265,7 @@ def test_given_corrupt_compact_batch_when_entry_rows_are_valid_then_warm_compile
     write_repo_files(tmp_path, _CACHE_REPO_FILES)
     monkeypatch.setattr(assembly_project, "_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT", 1)
     cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path)
-    cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3"
+    cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3"
     with sqlite3.connect(cache_path) as connection:
         _ = connection.execute(
             "UPDATE model_analysis_compact_batch SET payload = ?",
@@ -316,7 +316,7 @@ def test_given_corrupt_analysis_when_compiling_then_reanalyzes_and_repairs_the_e
 ) -> None:
     write_repo_files(tmp_path, _CACHE_REPO_FILES)
     cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path)
-    cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3"
+    cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3"
     with sqlite3.connect(cache_path) as connection:
         persisted_contents: str = connection.execute(
             "SELECT payload FROM model_analysis"
@@ -339,7 +339,7 @@ def test_given_corrupt_analysis_when_compiling_then_reanalyzes_and_repairs_the_e
     _digest, _separator, serialized_payload = repaired_contents.partition("\n")
     repaired_payload: dict[str, object] = json.loads(serialized_payload)
     assert repaired_project.models == cold_project.models
-    assert repaired_payload["v"] == 12
+    assert repaired_payload["v"] == 13
     assert isinstance(repaired_payload["s"], str)
     assert analyzer.call_count == test_case.expected_count
 
@@ -357,7 +357,7 @@ def test_given_non_text_analysis_cache_when_compiling_then_reanalyzes_safely(
 ) -> None:
     write_repo_files(tmp_path, _CACHE_REPO_FILES)
     _ = compile_project_with_cache(project_dir=tmp_path)
-    cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3"
+    cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3"
     with sqlite3.connect(cache_path) as connection:
         _ = connection.execute(
             "UPDATE model_analysis SET payload = ?",
diff --git a/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py b/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py
index 7049921840..b35c5513e4 100644
--- a/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py
+++ b/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py
@@ -96,3 +96,12 @@ class FrontierCaptureTestCase:
     stages: tuple[CompilerStage, ...]
     expected_results: tuple[object, ...]
     expected_files: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class SharedCaptureTestCase:
+    """A value with repeated large subtrees and how many distinct shared nodes its capture holds."""
+
+    description: str
+    value: Callable[[], object]
+    expected_shared_nodes: int
diff --git a/tests/unit/src/sqlbuild/compiler/frontier/helpers.py b/tests/unit/src/sqlbuild/compiler/frontier/helpers.py
index 5259371578..0507124821 100644
--- a/tests/unit/src/sqlbuild/compiler/frontier/helpers.py
+++ b/tests/unit/src/sqlbuild/compiler/frontier/helpers.py
@@ -6,9 +6,11 @@
 from enum import StrEnum
 from pathlib import Path, PurePosixPath
 
+from sqlbuild.compiler.compile.models import CompactLineageFacts
 from sqlbuild.compiler.fact_cache.main._compile_cache_root import compile_cache_root
 from sqlbuild.compiler.frontier.main.compiler_cache_directory import compiler_cache_directory
 from sqlbuild.compiler.scopes.constants import SCOPE_CACHE_DIRECTORY_NAME
+from sqlbuild.compiler.sql_analysis.classes.binding_catalog import BindingCatalog
 from sqlbuild.rule_engine._helpers.run.cache_paths import rules_bulk_cache_path
 
 HELPERS_MODULE: str = "tests.unit.src.sqlbuild.compiler.frontier.helpers"
@@ -68,6 +70,55 @@ def memo_catalog(*, memo_keys: tuple[str, ...], ordered_keys: tuple[str, ...]) -
     )
 
 
+def orders_lineage(*, string_pool: tuple[str, ...], source_index: int) -> CompactLineageFacts:
+    """Return `order_id` read directly from a source, indexed into a batch-wide string pool."""
+
+    position: dict[str, int] = {value: index for index, value in enumerate(string_pool)}
+    return CompactLineageFacts(
+        string_pool=string_pool,
+        rows=(
+            (
+                position["order_id"],
+                0,
+                1,
+                ((position["source"], source_index, position["order_id"]),),
+            ),
+        ),
+    )
+
+
+def orders_binding_catalog(
+    *, shared_analyses: tuple[tuple[str, str], ...], relations: tuple[str, ...]
+) -> BindingCatalog:
+    """Return a catalog whose memo and relation map were filled in the given orders."""
+
+    catalog: BindingCatalog = BindingCatalog(
+        dialect="duckdb",
+        quoted_ignore_case=False,
+        known_functions=(),
+        known_types=(),
+        relations={name: {"order_id": "INTEGER"} for name in relations},
+    )
+    catalog.shared_analyses.update(shared_analyses)
+    return catalog
+
+
+def repeated_orders() -> dict[str, object]:
+    """Return equal large order lines and one long SQL text, each repeated several times."""
+
+    query_sql: str = "SELECT " + ", ".join(f"amount_{index:03d}" for index in range(120))
+    return {
+        "orders": [
+            {"order_id": index, "lines": _order_lines(), "query_sql": query_sql}
+            for index in range(3)
+        ],
+    }
+
+
+def _order_lines() -> list[dict[str, object]]:
+    return [{"sku": f"product-{line:03d}", "quantity": line} for line in range(40)]
+
+
 def order_line() -> OrderLine:
     """Return one neutral order line."""
 
diff --git a/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py b/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py
index 6b2c2384f9..e589d0e884 100644
--- a/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py
+++ b/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py
@@ -2,21 +2,26 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 from pathlib import Path
 
 import pytest
 
+from scripts.compiler_differential.classes.capture_file import expand_capture_text
 from sqlbuild.compiler.frontier._helpers.stage_capture import render_stage_capture
+from sqlbuild.compiler.frontier.classes.stage_capture_encoder import StageCaptureEncoder
 from sqlbuild.compiler.frontier.constants import (
     COMPILER_ENGINE_ENV_VAR,
     STAGE_CAPTURE_DIR_ENV_VAR,
+    STAGE_CAPTURE_SHARED_NODES_KEY,
     STAGE_CAPTURE_UNORDERED_ATTRIBUTES,
 )
 from sqlbuild.compiler.frontier.main._compile_frontier import compile_frontier
 from sqlbuild.compiler.frontier.types import CompilerEngine, CompilerStage
 from tests.unit.src.sqlbuild.compiler.frontier._test_types import (
     FrontierCaptureTestCase,
+    SharedCaptureTestCase,
     StageCaptureOrderTestCase,
     StageCaptureTestCase,
     UnorderedAttributeTestCase,
@@ -27,8 +32,15 @@
     memo_catalog,
     order_line,
     order_total,
+    orders_binding_catalog,
+    orders_lineage,
+    repeated_orders,
 )
 
+_COMPILE_MODELS: str = "sqlbuild.compiler.compile.models"
+_COMPILE_TYPES: str = "sqlbuild.compiler.compile.types"
+_LINEAGE_TYPES: str = "sqlbuild.compiler.lineage.types"
+
 
 @pytest.mark.parametrize(
     "test_case",
@@ -67,6 +79,35 @@
                 "peer": {"__cycle__": f"{HELPERS_MODULE}:LinkedCustomer"},
             },
         ),
+        StageCaptureTestCase(
+            description="compact_lineage_is_captured_as_decoded_facts",
+            value=lambda: orders_lineage(
+                string_pool=("customers", "order_id", "source", "orders"), source_index=3
+            ),
+            expected_capture=[
+                {
+                    "__type__": f"{_COMPILE_MODELS}:CompiledLineageColumnFact",
+                    "output_column": "order_id",
+                    "upstream_columns": [
+                        {
+                            "__type__": f"{_COMPILE_MODELS}:CompiledLineageSourceFact",
+                            "resource_type": {
+                                "__enum__": f"{_COMPILE_TYPES}:CompiledResourceType.SOURCE"
+                            },
+                            "resource_name": "orders",
+                            "column_name": "order_id",
+                        }
+                    ],
+                    "transform_kind": {"__enum__": f"{_LINEAGE_TYPES}:ColumnTransformKind.DIRECT"},
+                    "confidence": {"__enum__": f"{_LINEAGE_TYPES}:ColumnLineageConfidence.HIGH"},
+                }
+            ],
+        ),
+        StageCaptureTestCase(
+            description="mapping_with_a_reserved_key_is_encoded_as_pairs",
+            value=lambda: {"__shared__": "orders", "status": "placed"},
+            expected_capture={"__mapping__": [["__shared__", "orders"], ["status", "placed"]]},
+        ),
         StageCaptureTestCase(
             description="non_finite_float_and_bytes",
             value=lambda: (float("inf"), b"orders"),
@@ -114,6 +155,50 @@ def test_given_frontier_value_when_rendering_capture_then_json_is_canonical(
             second=lambda: frozenset(("products", "customers", "orders")),
             expected_identical=True,
         ),
+        StageCaptureOrderTestCase(
+            description="batch_string_pool_layout_is_not_visible",
+            first=lambda: orders_lineage(
+                string_pool=("order_id", "source", "orders"), source_index=2
+            ),
+            second=lambda: orders_lineage(
+                string_pool=("customers", "source", "orders", "amount", "order_id"),
+                source_index=2,
+            ),
+            expected_identical=True,
+        ),
+        StageCaptureOrderTestCase(
+            description="decoded_lineage_source_is_visible",
+            first=lambda: orders_lineage(
+                string_pool=("order_id", "source", "orders", "customers"), source_index=2
+            ),
+            second=lambda: orders_lineage(
+                string_pool=("order_id", "source", "orders", "customers"), source_index=3
+            ),
+            expected_identical=False,
+        ),
+        StageCaptureOrderTestCase(
+            description="shared_analysis_memo_fill_is_not_visible",
+            first=lambda: orders_binding_catalog(
+                shared_analyses=(("orders_key", "orders"), ("customers_key", "customers")),
+                relations=("orders", "customers"),
+            ),
+            second=lambda: orders_binding_catalog(
+                shared_analyses=(
+                    ("customers_key", "customers_copy"),
+                    ("orders_key", "orders_copy"),
+                ),
+                relations=("customers", "orders"),
+            ),
+            expected_identical=True,
+        ),
+        StageCaptureOrderTestCase(
+            description="binding_catalog_relations_stay_visible",
+            first=lambda: orders_binding_catalog(
+                shared_analyses=(), relations=("orders", "customers")
+            ),
+            second=lambda: orders_binding_catalog(shared_analyses=(), relations=("orders",)),
+            expected_identical=False,
+        ),
     ],
     ids=lambda case: case.description,
 )
@@ -165,6 +250,34 @@ def test_given_declared_memo_attribute_when_rendering_capture_then_only_it_is_so
     assert (first == second) is test_case.expected_identical
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SharedCaptureTestCase(
+            description="repeated_lines_and_sql_are_stored_once",
+            value=repeated_orders,
+            expected_shared_nodes=2,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_repeated_large_subtrees_when_rendering_capture_then_each_is_stored_once(
+    test_case: SharedCaptureTestCase,
+) -> None:
+    text: str = render_stage_capture(test_case.value())
+
+    shared: dict[str, object] = json.loads(text)[STAGE_CAPTURE_SHARED_NODES_KEY]
+    assert len(shared) == test_case.expected_shared_nodes
+    assert expand_capture_text(text) == StageCaptureEncoder().encode(test_case.value())
+    assert {
+        hashlib.sha256(
+            json.dumps(node, ensure_ascii=False, separators=(",", ":")).encode()
+        ).hexdigest()
+        for node in shared.values()
+    } == set(shared)
+    assert len(text.splitlines()) == test_case.expected_shared_nodes + 4
+
+
 @pytest.mark.parametrize(
     "test_case",
     [