From 2415941191d8a8ae96fa0de66a646148725a7faa Mon Sep 17 00:00:00 2001
From: Chris Purcell <168346341+chrisdpurcell@users.noreply.github.com>
Date: Tue, 1 Sep 2026 16:48:46 -0400
Subject: [PATCH 1/2] perf(control-plane): stream integrity snapshots and
 memoize the package repository
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The CP-PROVIDER-INTEGRITY guard captures a whole-repository snapshot before
and after every provider invocation. Both captures read each declared target
in full, retained every byte, and hashed each one twice; on this repository
that is 201 captures over 65 declared paths per `reconcile --check`, including
a 10 MB frozen provider binary.

Captures now stream: the precondition hash is seeded with the pre-read mode
and fed chunk by chunk, `retain_content=False` drops the bytes a comparison
never reads, and the content digest is computed only when it is promised.
`reconcile --check` falls from 17.5 s to 13.7 s (median of three) and peak RSS
from 185 MB to 117 MB.

`provider_snapshot_chain()` additionally lets one planning pass hand the AFTER
snapshot of one provider to the next provider declaring the identical target
set — 201 captures to 102, 12.5 s to 10.4 s in-process. It is opt-in because
the window carries an obligation the guard cannot check: that nothing but the
providers themselves touches a declared path between two invocations. Wiring
it into the planner is left to a change that owns that file.

`build_package_repository` serves a repeated build of an unchanged tree from
an in-process memo keyed on the size and mtime of every file under
`standards/` and `catalogs/`: a second build costs 0.05 s instead of 2.75 s.
The stat-keyed content cache proposed for the snapshot guard is not added —
that surface reads and hashes bytes on purpose.

Refs #227
---
 CHANGELOG.md                                  |   3 +
 .../control_plane/providers.py                |  95 +++++++++++++--
 .../control_plane/snapshot.py                 | 110 ++++++++++++++----
 .../package_contract/repository.py            |  94 ++++++++++++++-
 tests/control_plane/test_providers.py         |  93 +++++++++++++++
 tests/control_plane/test_snapshot.py          |  53 +++++++++
 tests/package_contract/test_repository.py     |  31 +++++
 7 files changed, 449 insertions(+), 30 deletions(-)
 create mode 100644 tests/control_plane/test_snapshot.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5a15e58f..bbe9e6e8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -54,6 +54,9 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). Version
 
 - **New payload binaries are linked with debug information stripped.** `-s -w` joins `ARTIFACT_LDFLAGS` in `scripts/build-agent-handoff-session-start.sh` ([#228](https://github.com/L3DigitalNet/project-standards/issues/228) lever 1), taking the launcher from 3,906,258 to 2,646,142 bytes (−32.3%). `.gopclntab` is retained, so panic traces still carry function names and line numbers; DWARF inspection of the shipped file is recovered by rebuilding from the same script. Published payload bytes stay unstripped, so a size step between a retained version and its successor is expected. Consumers running a `pre-commit` `check-added-large-files` guard still need the documented exemption for the hook path — the launcher remains far above a typical `--maxkb=1024`.
 
+- **The provider integrity guard no longer retains the bytes it only compares, and repeated package-repository builds in one process are memoized** ([#227](https://github.com/L3DigitalNet/project-standards/issues/227)). Every provider invocation captures a whole-repository snapshot before and after the call to prove the provider changed no declared live path; those two captures read each declared target in full, held every byte in memory, and hashed each one twice. They now stream: the precondition hash is seeded from the pre-read mode and fed chunk by chunk, no bytes are retained, and the content digest is computed only when a caller asked for content. `RepositorySnapshot.capture` gained `retain_content` for that, and `assert_current` uses it too. On this repository `project-standards reconcile --check` falls from 17.5 s to 13.7 s (median of three) and peak RSS from 185 MB to 117 MB. `build_package_repository` additionally serves a repeated build of an unchanged tree from an in-process memo keyed on the size and mtime of every file under `standards/` and `catalogs/` — a cache over parse results, not an integrity guard, which is why `stat` metadata suffices there while the control plane's snapshot still reads and hashes bytes. A second build of this repository costs 0.05 s instead of 2.75 s; the fingerprint itself costs 0.05 s, which a single-build command now pays once.
+- **`provider_snapshot_chain()` lets a planning pass share one integrity snapshot between consecutive providers.** Inside that window the AFTER snapshot of one provider becomes the BEFORE snapshot of the next provider declaring the identical target set, so N invocations cost N+1 captures instead of 2N (measured on this repository: 201 captures to 102, and 12.5 s to 10.4 s in-process). Every AFTER capture is still a fresh full read, so a provider that changes a declared live path is still refused with `CP-PROVIDER-INTEGRITY`. The window is opt-in because it carries an obligation: the caller promises that nothing but the providers themselves touches a declared path between two invocations, so publication, recovery, or an operator edit between two plans in a long-lived process is never misattributed to the next provider.
+
 ### Fixed
 
 - **`Agent Handoff 1.17` stops an untrusted checkout from executing a command during session start.** The `session-start` launcher ran its Git reads with the full inherited process environment and no configuration isolation, so a repository-local or ancestor `core.fsmonitor` setting named a hook that Git ran — unconditionally, before the operator had seen anything — as soon as that checkout was opened as a session-start target ([#235](https://github.com/L3DigitalNet/project-standards/issues/235)). Every read now runs with an explicit minimal environment (`PATH` and `HOME` only, so no `GIT_DIR`, `GIT_WORK_TREE`, or `GIT_CONFIG_*` value from the harness can redirect it) and passes `-c core.fsmonitor=`, which outranks every configuration file, plus `--no-optional-locks` so the read cannot race a concurrent write. The injected session context is byte-identical to 1.16; reconcile replaces the installed hook because its digest moved. Catalog 5 promotes `agent-handoff@1.17` and retains 1.16.
diff --git a/src/project_standards/control_plane/providers.py b/src/project_standards/control_plane/providers.py
index e15f5176..6a3265a5 100644
--- a/src/project_standards/control_plane/providers.py
+++ b/src/project_standards/control_plane/providers.py
@@ -10,8 +10,9 @@
 import stat
 import sys
 import tempfile
-from collections.abc import Iterator, Mapping, Sequence
-from contextlib import redirect_stderr, redirect_stdout
+import threading
+from collections.abc import Generator, Iterator, Mapping, Sequence
+from contextlib import contextmanager, redirect_stderr, redirect_stdout
 from dataclasses import dataclass, replace
 from pathlib import Path
 from types import MappingProxyType
@@ -44,7 +45,10 @@
     safe_failure_detail,
 )
 from project_standards.control_plane.schemas import MutationPlanSchema, ProviderInputSchema
-from project_standards.control_plane.snapshot import RepositorySnapshot
+from project_standards.control_plane.snapshot import (
+    RepositorySnapshot,
+    canonical_targets,
+)
 from project_standards.package_contract.paths import (
     PackageVersion,
     SafeRelativePath,
@@ -488,9 +492,83 @@ def _declared_snapshot_paths(snapshots: JsonObject) -> tuple[SafeRelativePath, .
         raise ControlPlaneError("provider snapshot declares an invalid repository path") from exc
 
 
+@dataclass(frozen=True, slots=True)
+class _ChainedSnapshot:
+    """The AFTER snapshot of the most recent provider, offered to the next one."""
+
+    root: Path
+    targets: tuple[SafeRelativePath, ...]
+    snapshot: RepositorySnapshot
+
+
+# Chaining state is per thread: two threads planning against one repository see
+# each other's writes, and a slot armed by one of them describes a tree the
+# other never observed.
+_chain_state = threading.local()
+
+
+def _chain_slot() -> _ChainedSnapshot | None:
+    return cast("_ChainedSnapshot | None", getattr(_chain_state, "slot", None))
+
+
+@contextmanager
+def provider_snapshot_chain() -> Generator[None]:
+    """Allow consecutive provider invocations to share one integrity snapshot.
+
+    Inside this window, the AFTER snapshot the CP-PROVIDER-INTEGRITY guard takes
+    for one provider becomes the BEFORE snapshot of the next provider that
+    declares the identical target set, so N invocations cost N+1 captures rather
+    than 2N. Every declared path is still read in full once per invocation and
+    every AFTER capture is still a fresh read, so a provider that changes a
+    declared live path is caught exactly as before.
+
+    The window carries one obligation, and it is the reason chaining is opt-in
+    rather than always on: the caller promises that nothing but the providers
+    themselves touches a declared path between two invocations. Publication,
+    recovery, a spec fix that rewrites the file it just linted, or an operator
+    editing the tree between two plans in a long-lived process would otherwise
+    be attributed to the next provider as a CP-PROVIDER-INTEGRITY violation.
+    Open the window around one planning pass — never around a pass that writes,
+    and never across a repository mutation the process performed itself.
+
+    Nested windows are permitted and share the single slot; leaving any window
+    discards it, so a chained snapshot never outlives the pass that vouched
+    for it.
+    """
+    previous = _chain_slot()
+    previous_active = getattr(_chain_state, "active", False)
+    _chain_state.active = True
+    _chain_state.slot = None
+    try:
+        yield
+    finally:
+        _chain_state.active = previous_active
+        _chain_state.slot = previous if previous_active else None
+
+
+def _arm_snapshot_chain(after: RepositorySnapshot) -> None:
+    if getattr(_chain_state, "active", False):
+        _chain_state.slot = _ChainedSnapshot(after.root, after.targets, after)
+
+
+def _capture_declared_paths(
+    root: Path,
+    targets: tuple[SafeRelativePath, ...],
+) -> RepositorySnapshot:
+    """Take this invocation's BEFORE snapshot, reusing a chained one when offered."""
+    ordered = canonical_targets(targets)
+    slot = _chain_slot()
+    # Consumed at most once: the reference for the next invocation is whatever
+    # this invocation's own AFTER capture observes, never an older reading.
+    _chain_state.slot = None
+    if slot is not None and slot.root == root and slot.targets == ordered:
+        return slot.snapshot
+    return RepositorySnapshot.capture(root, ordered, retain_content=False)
+
+
 def _assert_declared_paths_unchanged(before: RepositorySnapshot) -> None:
     try:
-        after = RepositorySnapshot.capture(before.root, before.targets)
+        after = RepositorySnapshot.capture(before.root, before.targets, retain_content=False)
     except ControlPlaneError as exc:
         raise ControlPlaneError(
             "CP-PROVIDER-INTEGRITY: provider made a declared live path unsafe"
@@ -500,6 +578,9 @@ def _assert_declared_paths_unchanged(before: RepositorySnapshot) -> None:
             raise ControlPlaneError(
                 f"CP-PROVIDER-INTEGRITY: provider changed live path {expected.path.original}"
             )
+    # Armed only after the comparison succeeded: a snapshot that already failed
+    # the guard must never become the next invocation's reference.
+    _arm_snapshot_chain(after)
 
 
 def _output_notice(stdout: _OutputSink, stderr: _OutputSink) -> str | None:
@@ -921,7 +1002,7 @@ def _invoke_command_provider(
             extensions=payload.manifest.extensions,
         ),
     )
-    before = RepositorySnapshot.capture(
+    before = _capture_declared_paths(
         root,
         _declared_snapshot_paths(effective_invocation.snapshots),
     )
@@ -1013,7 +1094,7 @@ def invoke_provider_in_child(invocation: ProviderInvocation) -> ProviderResult:
     _validate_json_schema(prepared.input_schema, input_value, kind="input")
     frozen_input = _deep_freeze(input_value)
     frozen_resources = MappingProxyType(prepared.resources)
-    before = RepositorySnapshot.capture(
+    before = _capture_declared_paths(
         root,
         _declared_snapshot_paths(effective_invocation.snapshots),
     )
@@ -1078,7 +1159,7 @@ def invoke_provider(invocation: ProviderInvocation) -> ProviderResult:
             extensions=payload.manifest.extensions,
         ),
     )
-    before = RepositorySnapshot.capture(
+    before = _capture_declared_paths(
         root,
         _declared_snapshot_paths(effective_invocation.snapshots),
     )
diff --git a/src/project_standards/control_plane/snapshot.py b/src/project_standards/control_plane/snapshot.py
index 7c6b4951..0abd7a17 100644
--- a/src/project_standards/control_plane/snapshot.py
+++ b/src/project_standards/control_plane/snapshot.py
@@ -8,8 +8,8 @@
 from dataclasses import dataclass
 from enum import StrEnum
 from pathlib import Path, PurePosixPath
+from typing import Protocol
 
-from project_standards.control_plane.codec import content_digest
 from project_standards.control_plane.containment import (
     CONTAINMENT_DESTINATION_CODE,
     ContainmentError,
@@ -27,6 +27,14 @@
 _READ_SIZE = 1024 * 1024
 
 
+class _StreamingHash(Protocol):
+    """The `hashlib` surface a streaming capture needs, without a private typeshed name."""
+
+    def update(self, data: bytes, /) -> None: ...
+
+    def hexdigest(self) -> str: ...
+
+
 class EntryKind(StrEnum):
     """Filesystem states relevant to planning without following links."""
 
@@ -39,7 +47,14 @@ class EntryKind(StrEnum):
 
 @dataclass(frozen=True, slots=True)
 class SnapshotEntry:
-    """Exact bytes and metadata observed for one declared target."""
+    """Exact bytes and metadata observed for one declared target.
+
+    A snapshot captured with ``retain_content=False`` leaves ``content`` and
+    ``content_digest`` unset on a regular entry: the bytes are hashed as they
+    stream past and never retained. ``precondition_digest`` is the only field
+    that mode promises, so a caller that reads bytes or compares content
+    digests must capture in the default full mode.
+    """
 
     path: SafeRelativePath
     kind: EntryKind
@@ -50,6 +65,16 @@ class SnapshotEntry:
     precondition_digest: Sha256Digest
 
 
+def _precondition_hasher(kind: EntryKind, mode: str | None) -> _StreamingHash:
+    """Seed a precondition hash with the fixed prefix that precedes the payload."""
+    digest = hashlib.sha256()
+    digest.update(kind.value.encode("ascii"))
+    digest.update(b"\0")
+    digest.update((mode or "").encode("ascii"))
+    digest.update(b"\0")
+    return digest
+
+
 def _precondition(
     kind: EntryKind,
     *,
@@ -57,11 +82,7 @@ def _precondition(
     content: bytes | None = None,
     link_target: str | None = None,
 ) -> Sha256Digest:
-    digest = hashlib.sha256()
-    digest.update(kind.value.encode("ascii"))
-    digest.update(b"\0")
-    digest.update((mode or "").encode("ascii"))
-    digest.update(b"\0")
+    digest = _precondition_hasher(kind, mode)
     if content is not None:
         digest.update(content)
     elif link_target is not None:
@@ -69,6 +90,21 @@ def _precondition(
     return Sha256Digest(f"sha256:{digest.hexdigest()}")
 
 
+def canonical_targets(targets: tuple[SafeRelativePath, ...]) -> tuple[SafeRelativePath, ...]:
+    """Reject a colliding declared collection and fix the order a snapshot reads in.
+
+    Capture applies this to its own arguments; a caller that wants to compare
+    two declared collections for identity — as the provider snapshot chain does
+    — must compare the canonical forms, because two spellings of one collection
+    are the same snapshot.
+    """
+    try:
+        normalized = validate_path_collection(targets)
+    except ValueError as exc:
+        raise ControlPlaneError("snapshot target collection contains a collision") from exc
+    return tuple(sorted(normalized, key=lambda item: item.original.encode("utf-8")))
+
+
 def safe_repository_root(repo: Path) -> Path:
     try:
         if repo.is_symlink() or not repo.is_dir():
@@ -193,6 +229,8 @@ def _regular_entry(
     path: SafeRelativePath,
     parent_descriptor: int,
     name: str,
+    *,
+    retain_content: bool,
 ) -> SnapshotEntry:
     try:
         descriptor = os.open(
@@ -206,9 +244,26 @@ def _regular_entry(
         before = os.fstat(descriptor)
         if not stat.S_ISREG(before.st_mode):
             raise ControlPlaneError("snapshot target changed type during capture")
-        chunks: list[bytes] = []
+        # The mode is taken from the pre-read stat because the precondition hash
+        # is seeded with it before the first chunk arrives; the stability check
+        # after the read still rejects a mode that changed mid-capture, so the
+        # digest can never describe a mode the file did not hold throughout.
+        mode = _mode(before)
+        precondition = _precondition_hasher(EntryKind.REGULAR, mode)
+        # Bytes are hashed as they stream and retained only when a caller asked
+        # for them: an integrity capture over a multi-megabyte declared target
+        # (the frozen provider binaries are ~10 MB each) otherwise holds the
+        # whole file for the length of a provider invocation. The content digest
+        # is likewise computed only when it is promised, which halves the hash
+        # work of a precondition-only capture.
+        digest = hashlib.sha256() if retain_content else None
+        chunks: list[bytes] | None = [] if retain_content else None
         while chunk := os.read(descriptor, _READ_SIZE):
-            chunks.append(chunk)
+            precondition.update(chunk)
+            if digest is not None:
+                digest.update(chunk)
+            if chunks is not None:
+                chunks.append(chunk)
         after = os.fstat(descriptor)
     except OSError as exc:
         raise ControlPlaneError("snapshot target could not be read") from exc
@@ -217,16 +272,14 @@ def _regular_entry(
     stable_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_mode")
     if any(getattr(before, field) != getattr(after, field) for field in stable_fields):
         raise ControlPlaneError("snapshot target changed while being read")
-    content = b"".join(chunks)
-    mode = _mode(after)
     return SnapshotEntry(
         path=path,
         kind=EntryKind.REGULAR,
-        content=content,
+        content=b"".join(chunks) if chunks is not None else None,
         mode=mode,
         link_target=None,
-        content_digest=content_digest(content),
-        precondition_digest=_precondition(EntryKind.REGULAR, mode=mode, content=content),
+        content_digest=Sha256Digest(f"sha256:{digest.hexdigest()}") if digest is not None else None,
+        precondition_digest=Sha256Digest(f"sha256:{precondition.hexdigest()}"),
     )
 
 
@@ -294,6 +347,8 @@ def _read_entry(
     root: Path,
     root_descriptor: int,
     path: SafeRelativePath,
+    *,
+    retain_content: bool,
 ) -> SnapshotEntry:
     parent_descriptor = _parent_descriptor(root, root_descriptor, path.normalized.parent)
     if parent_descriptor is None:
@@ -321,7 +376,7 @@ def _read_entry(
                 _precondition(EntryKind.MISSING),
             )
         if stat.S_ISREG(metadata.st_mode):
-            return _regular_entry(path, parent_descriptor, name)
+            return _regular_entry(path, parent_descriptor, name, retain_content=retain_content)
         if stat.S_ISLNK(metadata.st_mode):
             target = os.readlink(name, dir_fd=parent_descriptor)
             return SnapshotEntry(
@@ -364,13 +419,19 @@ def capture(
         cls,
         repo: Path,
         targets: tuple[SafeRelativePath, ...],
+        *,
+        retain_content: bool = True,
     ) -> RepositorySnapshot:
+        """Read every declared target once, retaining bytes unless asked not to.
+
+        `retain_content=False` yields a precondition-only snapshot: every entry
+        still carries the precondition digest the integrity guard compares, but
+        regular-file bytes are hashed as they stream and then dropped, and the
+        content digest is not computed. Use it for a capture that will only be
+        compared, never read.
+        """
         root = safe_repository_root(repo)
-        try:
-            normalized = validate_path_collection(targets)
-        except ValueError as exc:
-            raise ControlPlaneError("snapshot target collection contains a collision") from exc
-        ordered = tuple(sorted(normalized, key=lambda item: item.original.encode("utf-8")))
+        ordered = canonical_targets(targets)
         flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
         try:
             root_descriptor = os.open(root, flags)
@@ -380,7 +441,10 @@ def capture(
             # Preflight every ancestor before the first content read: otherwise an
             # escape discovered late could leave earlier provider inputs observable.
             _preflight_ancestors(root, root_descriptor, ordered)
-            entries = tuple(_read_entry(root, root_descriptor, target) for target in ordered)
+            entries = tuple(
+                _read_entry(root, root_descriptor, target, retain_content=retain_content)
+                for target in ordered
+            )
         finally:
             os.close(root_descriptor)
         return cls(root, ordered, entries)
@@ -393,7 +457,9 @@ def entry(self, path: SafeRelativePath) -> SnapshotEntry:
 
     def assert_current(self) -> None:
         """Fail when any target no longer matches this snapshot's precondition."""
-        current = RepositorySnapshot.capture(self.root, self.targets)
+        # Only precondition digests are compared, so the re-read never retains
+        # bytes even when this snapshot itself carries them.
+        current = RepositorySnapshot.capture(self.root, self.targets, retain_content=False)
         for expected, observed in zip(self.entries, current.entries, strict=True):
             if expected.precondition_digest != observed.precondition_digest:
                 raise ControlPlaneError(f"snapshot precondition changed: {expected.path.original}")
diff --git a/src/project_standards/package_contract/repository.py b/src/project_standards/package_contract/repository.py
index 66f25b94..649af57d 100644
--- a/src/project_standards/package_contract/repository.py
+++ b/src/project_standards/package_contract/repository.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import json
+import os
+import stat
 from collections.abc import Iterable
 from dataclasses import dataclass
 from pathlib import Path
@@ -234,13 +236,103 @@ def _load_schema_property_scopes(
     return tuple(scopes), findings
 
 
+# The source tree a build reads: the family indexes, payload manifests, option
+# schemas and payload resources under `standards/`, plus the catalogs the
+# `catalog_major` argument selects.
+_SOURCE_DIRECTORIES = ("standards", "catalogs")
+# Bounded because a long-lived process (the MCP server) can be asked about
+# several roots; the entries are large and only the most recent few are ever
+# hit again.
+_CACHE_LIMIT = 4
+_repository_cache: dict[
+    tuple[str, int | None, tuple[str, ...] | None, tuple[tuple[str, int, int, int], ...]],
+    PackageRepository,
+] = {}
+
+
+def _source_content_key(root: Path) -> tuple[tuple[str, int, int, int], ...]:
+    """Fingerprint every file a build reads, by path, type, size and mtime.
+
+    This keys a cache over parse results, not an integrity guard, which is why
+    `stat` metadata is enough here while the control plane's snapshot guard
+    (`control_plane/snapshot.py`) insists on reading and hashing bytes: a stale
+    entry here costs one stale in-process answer about the producer's own
+    working tree, whereas a missed change there would let a provider mutate a
+    declared live path undetected. Payload bytes are proven by digest inside the
+    build itself, so this fingerprint only has to notice that the tree moved.
+    """
+    entries: list[tuple[str, int, int, int]] = []
+    for name in _SOURCE_DIRECTORIES:
+        stack = [root / name]
+        while stack:
+            current = stack.pop()
+            try:
+                with os.scandir(current) as scan:
+                    for item in scan:
+                        if item.is_dir(follow_symlinks=False):
+                            stack.append(Path(item.path))
+                            continue
+                        metadata = item.stat(follow_symlinks=False)
+                        entries.append(
+                            (
+                                str(Path(item.path).relative_to(root)),
+                                stat.S_IFMT(metadata.st_mode),
+                                metadata.st_size,
+                                metadata.st_mtime_ns,
+                            )
+                        )
+            except OSError:
+                # A directory that cannot be listed is a fact the build itself
+                # reports as a finding; the key simply records its absence, so a
+                # later repair still produces a different key.
+                entries.append((str(current.relative_to(root)), 0, -1, -1))
+    return tuple(sorted(entries))
+
+
+def clear_package_repository_cache() -> None:
+    """Drop every memoized build; the escape hatch for a test that rewrites a tree in place."""
+    _repository_cache.clear()
+
+
 def build_package_repository(
     root: Path,
     *,
     catalog_major: int | None = None,
     family_allowlist: Iterable[str] | None = None,
 ) -> PackageRepository:
-    """Load declared V2 sources without interpreting V1 manifests or unindexed trees."""
+    """Load declared V2 sources without interpreting V1 manifests or unindexed trees.
+
+    Repeated builds of an unchanged tree in one process are served from a
+    memo keyed on the source fingerprint above: one build reads 212 payload
+    manifests and 416 option schemas and hashes every payload resource
+    (~2.4 s on this repository), and a gate or a test module that asks three
+    times pays that once. The returned `PackageRepository` is immutable and is
+    therefore shared, not copied.
+    """
+    allowlist_key = None if family_allowlist is None else tuple(family_allowlist)
+    if allowlist_key is not None:
+        family_allowlist = allowlist_key
+    key = (str(root), catalog_major, allowlist_key, _source_content_key(root))
+    cached = _repository_cache.get(key)
+    if cached is not None:
+        return cached
+    built = _build_package_repository(
+        root,
+        catalog_major=catalog_major,
+        family_allowlist=family_allowlist,
+    )
+    if len(_repository_cache) >= _CACHE_LIMIT:
+        _repository_cache.pop(next(iter(_repository_cache)))
+    _repository_cache[key] = built
+    return built
+
+
+def _build_package_repository(
+    root: Path,
+    *,
+    catalog_major: int | None = None,
+    family_allowlist: Iterable[str] | None = None,
+) -> PackageRepository:
     discovery = discover_v2_families(root, family_allowlist=family_allowlist)
     findings = list(discovery.findings)
     loaded_families: list[LoadedFamily] = []
diff --git a/tests/control_plane/test_providers.py b/tests/control_plane/test_providers.py
index b3f000d5..c9a020da 100644
--- a/tests/control_plane/test_providers.py
+++ b/tests/control_plane/test_providers.py
@@ -11,11 +11,13 @@
 import pytest
 
 import project_standards.control_plane.providers as provider_runtime
+import project_standards.control_plane.snapshot as snapshot_module
 from project_standards.control_plane.diagnostics import ControlPlaneError
 from project_standards.control_plane.distribution import InstalledPayload
 from project_standards.control_plane.providers import (
     ProviderInvocation,
     invoke_provider,
+    provider_snapshot_chain,
     resolve_referenced_inputs,
 )
 from project_standards.control_plane.schemas import control_plane_schema_documents
@@ -1280,3 +1282,94 @@ def reject_recursive_scan(_path: Path, _pattern: str) -> object:
     result = invoke_provider(_invocation(repo, payload))
 
     assert result.content == b"1.2:1.2:declared-data"
+
+
+def _count_captures(monkeypatch: pytest.MonkeyPatch) -> list[int]:
+    """Count every snapshot capture the guard performs during one test."""
+    captured = [0]
+    original = snapshot_module.RepositorySnapshot.capture
+
+    def counting(
+        repo: Path,
+        targets: tuple[SafeRelativePath, ...],
+        *,
+        retain_content: bool = True,
+    ) -> snapshot_module.RepositorySnapshot:
+        captured[0] += 1
+        return original(repo, targets, retain_content=retain_content)
+
+    monkeypatch.setattr(snapshot_module.RepositorySnapshot, "capture", counting)
+    return captured
+
+
+def test_snapshot_chain_halves_captures_and_still_catches_a_provider_write(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    honest = _write_provider_payload(tmp_path / "honest")
+    writer = _write_provider_payload(tmp_path / "writer", behavior="write")
+    monkeypatch.chdir(repo)
+    captured = _count_captures(monkeypatch)
+
+    with provider_snapshot_chain():
+        invoke_provider(_invocation(repo, honest))
+        invoke_provider(_invocation(repo, honest))
+        # Two invocations cost three captures, not four: the first invocation's
+        # AFTER snapshot is the second invocation's BEFORE.
+        assert captured[0] == 3
+
+        with pytest.raises(ControlPlaneError, match="CP-PROVIDER-INTEGRITY"):
+            invoke_provider(_invocation(repo, writer))
+
+
+def test_snapshot_chain_never_survives_its_window(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    payload = _write_provider_payload(tmp_path / "payload")
+    monkeypatch.chdir(repo)
+    captured = _count_captures(monkeypatch)
+
+    with provider_snapshot_chain():
+        invoke_provider(_invocation(repo, payload))
+    # A write between two windows is the caller's, not a provider's: outside the
+    # window the next invocation reads the tree again instead of blaming the
+    # provider for what the control plane itself published.
+    (repo / "README.md").write_bytes(b"published between two passes\n")
+    invoke_provider(_invocation(repo, payload))
+
+    assert captured[0] == 4
+
+
+def test_provider_guard_captures_do_not_retain_declared_bytes(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "README.md").write_bytes(b"declared bytes\n")
+    payload = _write_provider_payload(tmp_path / "payload")
+    monkeypatch.chdir(repo)
+    seen: list[snapshot_module.RepositorySnapshot] = []
+    original = snapshot_module.RepositorySnapshot.capture
+
+    def recording(
+        repo_path: Path,
+        targets: tuple[SafeRelativePath, ...],
+        *,
+        retain_content: bool = True,
+    ) -> snapshot_module.RepositorySnapshot:
+        result = original(repo_path, targets, retain_content=retain_content)
+        seen.append(result)
+        return result
+
+    monkeypatch.setattr(snapshot_module.RepositorySnapshot, "capture", recording)
+
+    invoke_provider(_invocation(repo, payload))
+
+    assert seen
+    assert all(entry.content is None for snapshot in seen for entry in snapshot.entries)
diff --git a/tests/control_plane/test_snapshot.py b/tests/control_plane/test_snapshot.py
new file mode 100644
index 00000000..931b09ee
--- /dev/null
+++ b/tests/control_plane/test_snapshot.py
@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+from project_standards.control_plane.snapshot import EntryKind, RepositorySnapshot
+from project_standards.package_contract.paths import SafeRelativePath
+
+
+def _targets(*paths: str) -> tuple[SafeRelativePath, ...]:
+    return tuple(SafeRelativePath.parse(path) for path in paths)
+
+
+def _repository(tmp_path: Path) -> Path:
+    root = tmp_path / "repo"
+    (root / "nested").mkdir(parents=True)
+    # Deliberately larger than the 1 MiB read size so the streaming hash spans
+    # several chunks; a single-chunk fixture would not exercise the boundary.
+    (root / "large.bin").write_bytes(b"payload-byte" * 200_000)
+    (root / "nested/small.txt").write_bytes(b"small\n")
+    (root / "nested/link").symlink_to("small.txt")
+    return root
+
+
+def test_precondition_only_capture_matches_a_full_capture(tmp_path: Path) -> None:
+    root = _repository(tmp_path)
+    targets = _targets("large.bin", "nested/small.txt", "nested/link", "nested", "absent.txt")
+
+    full = RepositorySnapshot.capture(root, targets)
+    lean = RepositorySnapshot.capture(root, targets, retain_content=False)
+
+    assert [entry.precondition_digest for entry in full.entries] == [
+        entry.precondition_digest for entry in lean.entries
+    ]
+    assert [entry.kind for entry in full.entries] == [entry.kind for entry in lean.entries]
+    assert [entry.mode for entry in full.entries] == [entry.mode for entry in lean.entries]
+
+
+def test_full_capture_keeps_exact_bytes_and_lean_capture_drops_them(tmp_path: Path) -> None:
+    root = _repository(tmp_path)
+    content = (root / "large.bin").read_bytes()
+    targets = _targets("large.bin")
+
+    full = RepositorySnapshot.capture(root, targets).entries[0]
+    lean = RepositorySnapshot.capture(root, targets, retain_content=False).entries[0]
+
+    assert full.kind is EntryKind.REGULAR
+    assert full.content == content
+    assert full.content_digest is not None
+    assert full.content_digest.value == f"sha256:{hashlib.sha256(content).hexdigest()}"
+    assert lean.kind is EntryKind.REGULAR
+    assert lean.content is None
+    assert lean.content_digest is None
diff --git a/tests/package_contract/test_repository.py b/tests/package_contract/test_repository.py
index b265276a..ad501678 100644
--- a/tests/package_contract/test_repository.py
+++ b/tests/package_contract/test_repository.py
@@ -113,3 +113,34 @@ def test_selected_catalog_errors_join_package_load_findings(tmp_path: Path) -> N
         "PC-INTEGRITY",
         "PC-CATALOG-INVALID",
     }
+
+
+def test_repeated_build_is_memoized_until_a_manifest_changes(tmp_path: Path) -> None:
+    root = copy_minimal_repository(tmp_path)
+
+    first = build_package_repository(root, catalog_major=5)
+    assert build_package_repository(root, catalog_major=5) is first
+
+    # A payload manifest edit must be visible to the very next build: the memo
+    # is keyed on the source fingerprint, not on the root alone.
+    manifest = root / "standards/demo/versions/1.2/payload.toml"
+    manifest.write_text(manifest.read_text(encoding="utf-8") + "\n", encoding="utf-8")
+
+    rebuilt = build_package_repository(root, catalog_major=5)
+
+    assert rebuilt is not first
+    assert {finding.code for finding in rebuilt.findings} != set()
+
+
+def test_memo_distinguishes_the_catalog_and_the_allowlist(tmp_path: Path) -> None:
+    root = copy_minimal_repository(tmp_path)
+
+    with_catalog = build_package_repository(root, catalog_major=5)
+    without_catalog = build_package_repository(root)
+    narrowed = build_package_repository(root, family_allowlist=iter(("demo",)))
+
+    assert with_catalog.catalog is not None
+    assert without_catalog.catalog is None
+    # The allowlist may arrive as a one-shot iterator; keying on it must not
+    # consume the iterator the build itself still needs.
+    assert [family.manifest.standard.id for family in narrowed.families] == ["demo"]

From 5d3b772ca7c7d92770d4a50336434370034ba6ef Mon Sep 17 00:00:00 2001
From: Chris Purcell <168346341+chrisdpurcell@users.noreply.github.com>
Date: Tue, 1 Sep 2026 17:01:19 -0400
Subject: [PATCH 2/2] perf(control-plane): enter the snapshot chain for the
 planning pass

`plan_reconciliation` now wraps its provider invocations in
`provider_snapshot_chain()`, so the AFTER snapshot the CP-PROVIDER-INTEGRITY
guard takes for one provider becomes the BEFORE snapshot of the next provider
declaring the identical target set: 201 captures per `reconcile --check` fall
to 102, and the command falls from 13.7 s to 9.5 s (median of three; 17.5 s
before this leg's streaming change).

Planning is the only pass that enters the window, and it satisfies the
window's obligation by construction: it invokes every render, validate and
transform provider and writes nothing itself. Publication, the executor's
post-publication verification providers, and the MCP and migration call sites
stay outside it, because a window spanning a control-plane write would report
the control plane's own bytes as the next provider's violation.

Leaving any window now drops the slot rather than restoring the outer pass's
older reading, so no chained snapshot can outlive the pass that vouched for
it.

Refs #227
---
 CHANGELOG.md                                  |   3 +
 .../control_plane/planner.py                  |  14 +++
 .../control_plane/providers.py                |   6 +-
 tests/control_plane/test_planner.py           | 104 +++++++++++++++++-
 4 files changed, 123 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index bbe9e6e8..7d2f1571 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -57,6 +57,9 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). Version
 - **The provider integrity guard no longer retains the bytes it only compares, and repeated package-repository builds in one process are memoized** ([#227](https://github.com/L3DigitalNet/project-standards/issues/227)). Every provider invocation captures a whole-repository snapshot before and after the call to prove the provider changed no declared live path; those two captures read each declared target in full, held every byte in memory, and hashed each one twice. They now stream: the precondition hash is seeded from the pre-read mode and fed chunk by chunk, no bytes are retained, and the content digest is computed only when a caller asked for content. `RepositorySnapshot.capture` gained `retain_content` for that, and `assert_current` uses it too. On this repository `project-standards reconcile --check` falls from 17.5 s to 13.7 s (median of three) and peak RSS from 185 MB to 117 MB. `build_package_repository` additionally serves a repeated build of an unchanged tree from an in-process memo keyed on the size and mtime of every file under `standards/` and `catalogs/` — a cache over parse results, not an integrity guard, which is why `stat` metadata suffices there while the control plane's snapshot still reads and hashes bytes. A second build of this repository costs 0.05 s instead of 2.75 s; the fingerprint itself costs 0.05 s, which a single-build command now pays once.
 - **`provider_snapshot_chain()` lets a planning pass share one integrity snapshot between consecutive providers.** Inside that window the AFTER snapshot of one provider becomes the BEFORE snapshot of the next provider declaring the identical target set, so N invocations cost N+1 captures instead of 2N (measured on this repository: 201 captures to 102, and 12.5 s to 10.4 s in-process). Every AFTER capture is still a fresh full read, so a provider that changes a declared live path is still refused with `CP-PROVIDER-INTEGRITY`. The window is opt-in because it carries an obligation: the caller promises that nothing but the providers themselves touches a declared path between two invocations, so publication, recovery, or an operator edit between two plans in a long-lived process is never misattributed to the next provider.
 
+- **The provider integrity guard no longer retains the bytes it only compares, and repeated package-repository builds in one process are memoized** ([#227](https://github.com/L3DigitalNet/project-standards/issues/227)). Every provider invocation captures a whole-repository snapshot before and after the call to prove the provider changed no declared live path; those two captures read each declared target in full, held every byte in memory, and hashed each one twice. They now stream: the precondition hash is seeded from the pre-read mode and fed chunk by chunk, no bytes are retained, and the content digest is computed only when a caller asked for content. `RepositorySnapshot.capture` gained `retain_content` for that, and `assert_current` uses it too. Combined with the snapshot chain below, `project-standards reconcile --check` falls from 17.5 s to 9.5 s on this repository (median of three) and peak RSS from 185 MB to 117 MB. `build_package_repository` additionally serves a repeated build of an unchanged tree from an in-process memo keyed on the size and mtime of every file under `standards/` and `catalogs/` — a cache over parse results, not an integrity guard, which is why `stat` metadata suffices there while the control plane's snapshot still reads and hashes bytes. A second build of this repository costs 0.05 s instead of 2.75 s; the fingerprint itself costs 0.05 s, which a single-build command now pays once.
+- **`plan_reconciliation` now shares one integrity snapshot between consecutive providers.** Inside the new `provider_snapshot_chain()` window the AFTER snapshot of one provider becomes the BEFORE snapshot of the next provider declaring the identical target set, so N invocations cost N+1 captures instead of 2N — 201 captures to 102 on this repository. Every AFTER capture is still a fresh full read, so a provider that changes a declared live path is still refused with `CP-PROVIDER-INTEGRITY`. The window is opt-in and planning is the only pass that enters it: it invokes every provider and writes nothing itself, so between two invocations only a provider could have touched a declared path. Publication and the executor's post-publication verification providers stay outside, because a window spanning a write would report the control plane's own bytes as the next provider's violation.
+
 ### Fixed
 
 - **`Agent Handoff 1.17` stops an untrusted checkout from executing a command during session start.** The `session-start` launcher ran its Git reads with the full inherited process environment and no configuration isolation, so a repository-local or ancestor `core.fsmonitor` setting named a hook that Git ran — unconditionally, before the operator had seen anything — as soon as that checkout was opened as a session-start target ([#235](https://github.com/L3DigitalNet/project-standards/issues/235)). Every read now runs with an explicit minimal environment (`PATH` and `HOME` only, so no `GIT_DIR`, `GIT_WORK_TREE`, or `GIT_CONFIG_*` value from the harness can redirect it) and passes `-c core.fsmonitor=`, which outranks every configuration file, plus `--no-optional-locks` so the read cannot race a concurrent write. The injected session context is byte-identical to 1.16; reconcile replaces the installed hook because its digest moved. Catalog 5 promotes `agent-handoff@1.17` and retains 1.16.
diff --git a/src/project_standards/control_plane/planner.py b/src/project_standards/control_plane/planner.py
index 9bb93f8c..06b196fd 100644
--- a/src/project_standards/control_plane/planner.py
+++ b/src/project_standards/control_plane/planner.py
@@ -77,6 +77,7 @@
     ProviderInvocation,
     ProviderResult,
     invoke_provider,
+    provider_snapshot_chain,
     resolve_referenced_inputs,
 )
 from project_standards.control_plane.resolution import (
@@ -2894,6 +2895,19 @@ def _alias_analysis(
 
 def plan_reconciliation(request: PlannerRequest) -> ReconciliationPlan:
     """Build one deterministic, complete, and read-only reconciliation plan."""
+    # Planning is the one pass that satisfies the snapshot chain's obligation:
+    # it invokes every render, validate, and transform provider in turn and
+    # writes nothing itself, so between two invocations only a provider could
+    # have touched a declared path — which is exactly what the chained BEFORE
+    # snapshot then proves it did not. Publication and the executor's
+    # post-publication verification providers stay outside the window; a window
+    # spanning a write would report the control plane's own bytes as a
+    # CP-PROVIDER-INTEGRITY violation by the next provider.
+    with provider_snapshot_chain():
+        return _plan_reconciliation(request)
+
+
+def _plan_reconciliation(request: PlannerRequest) -> ReconciliationPlan:
     original_request = request
     resolution = resolve_packages(request.resolution)
     payloads = _payload_map(request.payloads)
diff --git a/src/project_standards/control_plane/providers.py b/src/project_standards/control_plane/providers.py
index 6a3265a5..2cc7f93a 100644
--- a/src/project_standards/control_plane/providers.py
+++ b/src/project_standards/control_plane/providers.py
@@ -533,9 +533,9 @@ def provider_snapshot_chain() -> Generator[None]:
 
     Nested windows are permitted and share the single slot; leaving any window
     discards it, so a chained snapshot never outlives the pass that vouched
-    for it.
+    for it — including the outer pass, which resumes with no slot rather than
+    with the reading it held before the inner pass ran.
     """
-    previous = _chain_slot()
     previous_active = getattr(_chain_state, "active", False)
     _chain_state.active = True
     _chain_state.slot = None
@@ -543,7 +543,7 @@ def provider_snapshot_chain() -> Generator[None]:
         yield
     finally:
         _chain_state.active = previous_active
-        _chain_state.slot = previous if previous_active else None
+        _chain_state.slot = None
 
 
 def _arm_snapshot_chain(after: RepositorySnapshot) -> None:
diff --git a/tests/control_plane/test_planner.py b/tests/control_plane/test_planner.py
index 35f814f9..c0087108 100644
--- a/tests/control_plane/test_planner.py
+++ b/tests/control_plane/test_planner.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import contextlib
 import hashlib
 import json
 import random
@@ -10,6 +11,8 @@
 
 import pytest
 
+import project_standards.control_plane.planner as planner_module
+import project_standards.control_plane.snapshot as snapshot_module
 from project_standards.control_plane.adapters.toml import TomlAdapter
 from project_standards.control_plane.codec import parse_lock, render_lock
 from project_standards.control_plane.diagnostics import (
@@ -28,10 +31,15 @@
 from project_standards.control_plane.providers import (
     ProviderInvocation,
     ProviderResult,
+    invoke_provider,
 )
 from project_standards.control_plane.resolution import DeclaredTransition
 from project_standards.control_plane.schemas import ReconciliationPlanSchema
-from project_standards.package_contract.paths import PackageVersion, Sha256Digest
+from project_standards.package_contract.paths import (
+    PackageVersion,
+    SafeRelativePath,
+    Sha256Digest,
+)
 from project_standards.package_contract.payload import (
     ArtifactPolicy,
     JsonValue,
@@ -45,6 +53,10 @@
     resolution_request,
     write_payload,
 )
+from tests.control_plane.test_providers import (
+    provider_invocation,
+    write_provider_payload,
+)
 
 
 def _request(
@@ -2162,3 +2174,93 @@ def test_undeclared_whole_file_relinquishment_follows_created_container(
     assert _action(plan, "notes.md").kind is expected
     assert target.read_bytes() == content
     assert all(artifact.path.original != "notes.md" for artifact in plan.next_lock.artifacts)
+
+
+def _capture_counter(monkeypatch: pytest.MonkeyPatch) -> list[int]:
+    counted = [0]
+    original = snapshot_module.RepositorySnapshot.capture
+
+    def counting(
+        repo: Path,
+        targets: tuple[SafeRelativePath, ...],
+        *,
+        retain_content: bool = True,
+    ) -> snapshot_module.RepositorySnapshot:
+        counted[0] += 1
+        return original(repo, targets, retain_content=retain_content)
+
+    monkeypatch.setattr(snapshot_module.RepositorySnapshot, "capture", counting)
+    return counted
+
+
+def _chained_planner_fixture(tmp_path: Path) -> tuple[Path, PlannerRequest]:
+    """A plan whose two contributions each run one real, executable provider.
+
+    The runner ignores the planner's own invocation and dispatches the
+    executable fixture payload instead, because chaining is only observable
+    across invocations that declare the identical target set — which is what
+    the real reconcile path does, and what an inert stub runner cannot show.
+    """
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    executable = write_provider_payload(tmp_path / "executable")
+    payload = write_payload(
+        tmp_path / "demo",
+        "demo",
+        contributions=[
+            {
+                "id": "first",
+                "target": "first.txt",
+                "adapter": "whole-file",
+                "scope": "$file",
+                "provider": "render-tool",
+            },
+            {
+                "id": "second",
+                "target": "second.txt",
+                "adapter": "whole-file",
+                "scope": "$file",
+                "provider": "render-tool",
+            },
+        ],
+        render_providers=["render-tool"],
+    )
+
+    def runner(_invocation: ProviderInvocation) -> ProviderResult:
+        return invoke_provider(provider_invocation(repo, executable))
+
+    return repo, _request(repo, (payload,), provider_runner=runner)
+
+
+def test_planning_shares_one_integrity_snapshot_between_consecutive_providers(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    repo, request = _chained_planner_fixture(tmp_path)
+    monkeypatch.chdir(repo)
+    chained = _capture_counter(monkeypatch)
+
+    plan = plan_reconciliation(request)
+
+    assert plan.applicable
+    # One capture for the plan's own target snapshot, then N+1 rather than 2N
+    # for the two provider invocations the contributions drive.
+    assert chained[0] == 4
+
+
+def test_planning_without_the_chain_captures_twice_per_provider(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    repo, request = _chained_planner_fixture(tmp_path)
+    monkeypatch.chdir(repo)
+    unchained = _capture_counter(monkeypatch)
+
+    # Removing the window is the only difference from the test above, so the
+    # saving is pinned to the window itself rather than to any other change in
+    # the planner.
+    monkeypatch.setattr(planner_module, "provider_snapshot_chain", contextlib.nullcontext)
+    plan = plan_reconciliation(request)
+
+    assert plan.applicable
+    assert unchained[0] == 5