From 9b3d36dee7700b5944c8b8050e783cd255cb5418 Mon Sep 17 00:00:00 2001 From: gunbc-ci-auto-heal Date: Fri, 25 Sep 2026 16:29:33 +0000 Subject: [PATCH 1/4] V4.1 Engram file-backed: release verify page cache (fail-closed on residue); batched fetch_rows gather Co-Authored-By: Claude Opus 5.5 (1M context) --- .../d2d649e6/engram_file_backed.py.patch | 201 ++++++++++++++++-- dag/gunbc/spark/v41_engram_file_backed.dag | 77 +++++++ .../v41_engram_file_backed_witness_test.dag | 106 +++++++++ 3 files changed, 366 insertions(+), 18 deletions(-) create mode 100644 dag/gunbc/spark/v41_engram_file_backed.dag create mode 100644 dag/test/claim/spark/v41_engram_file_backed_witness_test.dag diff --git a/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch b/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch index d97323426b0..ea2ca06190c 100644 --- a/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch +++ b/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch @@ -1,9 +1,9 @@ diff --git a/vllm/models/deepseek_v4_1/nvidia/engram_file_backed.py b/vllm/models/deepseek_v4_1/nvidia/engram_file_backed.py new file mode 100644 -index 0000000..b5c332c +index 0000000..16abb9b --- /dev/null +++ b/vllm/models/deepseek_v4_1/nvidia/engram_file_backed.py -@@ -0,0 +1,115 @@ +@@ -0,0 +1,280 @@ +# File-backed Engram storage for vLLM d2d649e674c75425d2d6975c87eb89fd4d55fff8. +# +# THIS IS A BACKING SWAP, NOT NEW PREFETCH MACHINERY. Upstream @@ -16,7 +16,7 @@ index 0000000..b5c332c +# +# This module keeps the staging buffer, the stream, and the prefetch protocol. +# It changes where the rows come from: a rank-local file whose identity is -+# verified at open, mmap'd MAP_PRIVATE without registering the whole table ++# verified at open, mmap'd read-only without registering the whole table +# with CUDA. Token-time gathers copy requested rows into the existing bounded +# pinned staging buffer. A missing local path refuses; there is no NFS arm. +# @@ -24,14 +24,24 @@ index 0000000..b5c332c +# v41_row_store_magic then weight||scale records; v41_row_address offsets follow; +# the store's SHA-256 is over those prefixed bytes. Open checks expected.magic +# (supplied from that symbol, not minted here) and whole-file digest against -+# rank_digest_hex. encoding_digest / rank / format_revision are not on this ++# rank_digest_hex. expected.record_bytes is v41_row_bytes of the admitted spec, ++# supplied likewise. encoding_digest / rank / format_revision are not on this +# type until a modeled header carries them. ++# ++# The open and gather contracts are gunbc.spark.v41_engram_file_backed: ++# v41_open_residency_verdict (verification's page-cache residue is released and ++# read back as zero, else open refuses) and v41_gather_rows (one batched gather ++# at record-aligned offsets, in request order; a misaligned offset refuses). + +from __future__ import annotations + ++import ctypes ++import ctypes.util +import hashlib +import mmap +import os ++import time ++import warnings +from dataclasses import dataclass +from typing import Tuple + @@ -50,18 +60,75 @@ index 0000000..b5c332c + pass + + ++class EngramPageCacheResident(RuntimeError): ++ pass ++ ++ ++class EngramRowOffsetMisaligned(RuntimeError): ++ pass ++ ++ +@dataclass(frozen=True) +class ExpectedRowStoreIdentity: + rank_digest_hex: str + magic: bytes ++ record_bytes: int ++ ++ ++@dataclass(frozen=True) ++class EngramStoreOpenReceipt: ++ """What open() observed: the verified range and its page-cache residency ++ before and after the release (gunbc.spark.v41_engram_file_backed ++ V41StoreOpenReceipt).""" ++ ++ verified_bytes: int ++ resident_bytes_after_verify: int ++ resident_bytes_after_release: int ++ ++ ++_libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True) ++_libc.mmap.restype = ctypes.c_void_p ++_libc.mmap.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_long] ++_libc.munmap.argtypes = [ctypes.c_void_p, ctypes.c_size_t] ++_libc.mincore.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_void_p] ++_MAP_FAILED = ctypes.c_void_p(-1).value ++_RESIDENCY_WINDOW = 1 << 30 ++ + ++def resident_bytes(fd: int, size: int) -> int: ++ """Page-cache resident bytes of [0, size) of fd, by mincore over a ++ transient mapping that is never touched (so reading does not fault).""" ++ page = os.sysconf("SC_PAGE_SIZE") ++ resident_pages = 0 ++ off = 0 ++ while off < size: ++ length = min(_RESIDENCY_WINDOW, size - off) ++ addr = _libc.mmap(None, length, mmap.PROT_READ, mmap.MAP_SHARED, fd, off) ++ if addr in (None, _MAP_FAILED): ++ raise OSError(ctypes.get_errno(), "mmap for residency readback failed") ++ try: ++ pages = (length + page - 1) // page ++ vec = (ctypes.c_ubyte * pages)() ++ if _libc.mincore(addr, length, vec) != 0: ++ raise OSError(ctypes.get_errno(), "mincore failed") ++ resident_pages += sum(b & 1 for b in bytes(vec)) ++ finally: ++ _libc.munmap(addr, length) ++ off += length ++ return min(resident_pages * page, size) + -+def _sha256_file(path: str) -> str: ++ ++def _sha256_fd(fd: int, magic_len: int) -> Tuple[str, bytes, int]: + h = hashlib.sha256() -+ with open(path, "rb") as f: ++ head = b"" ++ total = 0 ++ with os.fdopen(os.dup(fd), "rb", closefd=True) as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): ++ if len(head) < magic_len: ++ head += chunk[: magic_len - len(head)] + h.update(chunk) -+ return h.hexdigest() ++ total += len(chunk) ++ return h.hexdigest(), head, total + + +class FileBackedEngramStorage: @@ -73,31 +140,90 @@ index 0000000..b5c332c + f"rank-local Engram row store missing at {path}; " + "token-time NFS fallback is refused" + ) -+ opened = _sha256_file(path) -+ if opened != expected.rank_digest_hex: -+ raise EngramRowStoreIdentityMismatch( -+ f"opened {path} digest {opened}, expected rank digest " -+ f"{expected.rank_digest_hex}" -+ ) + self.path = os.path.abspath(path) + self.expected = expected + self._fd = os.open(self.path, os.O_RDONLY) -+ self._mmap = mmap.mmap(self._fd, 0, access=mmap.ACCESS_READ) -+ if self._mmap[: len(expected.magic)] != expected.magic: -+ raise EngramRowStoreIdentityMismatch("row-store MAGIC mismatch") ++ try: ++ self.receipt = self._verify_and_release() ++ self._mmap = mmap.mmap(self._fd, 0, access=mmap.ACCESS_READ) ++ except BaseException: ++ os.close(self._fd) ++ raise ++ with warnings.catch_warnings(): ++ # The mmap is read-only; the table is only ever read through index. ++ warnings.simplefilter("ignore", UserWarning) ++ self._table = torch.frombuffer(self._mmap, dtype=torch.uint8) ++ self._span = torch.arange(expected.record_bytes, dtype=torch.int64) + self._header_ok = True + ++ def _verify_and_release(self) -> EngramStoreOpenReceipt: ++ # Verification stays fail-closed and whole-file; only its cache residue goes. ++ opened, head, verified = _sha256_fd(self._fd, len(self.expected.magic)) ++ if opened != self.expected.rank_digest_hex: ++ raise EngramRowStoreIdentityMismatch( ++ f"opened {self.path} digest {opened}, expected rank digest " ++ f"{self.expected.rank_digest_hex}" ++ ) ++ if head != self.expected.magic: ++ raise EngramRowStoreIdentityMismatch("row-store MAGIC mismatch") ++ after_verify = resident_bytes(self._fd, verified) ++ os.posix_fadvise(self._fd, 0, verified, os.POSIX_FADV_DONTNEED) ++ after_release = resident_bytes(self._fd, verified) ++ receipt = EngramStoreOpenReceipt( ++ verified_bytes=verified, ++ resident_bytes_after_verify=after_verify, ++ resident_bytes_after_release=after_release, ++ ) ++ if after_release != 0: ++ raise EngramPageCacheResident( ++ f"{self.path}: {after_release} of {verified} verified bytes remain " ++ f"page-cache resident after POSIX_FADV_DONTNEED; receipt {receipt}" ++ ) ++ return receipt ++ + def close(self) -> None: ++ del self._table + self._mmap.close() + os.close(self._fd) + -+ def fetch_rows(self, offsets: list[int], nbytes: int, dest: torch.Tensor) -> None: -+ """Copy selected rows into an already-allocated bounded staging tensor.""" ++ def _check_dest(self, dest: torch.Tensor) -> None: + if dest.is_pinned() and dest.numel() * dest.element_size() >= self._mmap.size(): + raise EngramFullTablePinned( + "staging destination covers the whole table; this runtime refuses " + "full-Engram pinned allocation" + ) ++ ++ def _record_index(self, offsets: list[int], nbytes: int) -> torch.Tensor: ++ record = self.expected.record_bytes ++ if nbytes != record: ++ raise EngramRowOffsetMisaligned( ++ f"fetch of {nbytes} bytes per row; the store's record is {record} bytes" ++ ) ++ magic = len(self.expected.magic) ++ size = self._mmap.size() ++ for off in offsets: ++ if off < magic or (off - magic) % record != 0 or off + record > size: ++ raise EngramRowOffsetMisaligned( ++ f"offset {off} is not magic + k * {record} inside a {size}-byte store" ++ ) ++ return (torch.as_tensor(offsets, dtype=torch.int64).unsqueeze(1) + self._span).view(-1) ++ ++ def fetch_rows(self, offsets: list[int], nbytes: int, dest: torch.Tensor) -> None: ++ """Copy selected rows, in request order, into an already-allocated ++ bounded staging tensor with one batched gather.""" ++ self._check_dest(dest) ++ index = self._record_index(offsets, nbytes) ++ rows = dest.view(torch.uint8).view(dest.shape[0], -1)[: len(offsets), :nbytes] ++ if rows.is_contiguous(): ++ torch.index_select(self._table, 0, index, out=rows.view(-1)) ++ else: ++ rows.copy_(torch.index_select(self._table, 0, index).view(len(offsets), nbytes)) ++ ++ def _fetch_rows_per_row(self, offsets: list[int], nbytes: int, dest: torch.Tensor) -> None: ++ # The replaced per-row copy, kept ONLY as the differential oracle for ++ # measure_fetch_rows. It is never a fallback for fetch_rows. ++ self._check_dest(dest) ++ self._record_index(offsets, nbytes) + view = memoryview(self._mmap) + for i, off in enumerate(offsets): + dest[i].view(torch.uint8).view(-1)[:nbytes].copy_( @@ -105,6 +231,26 @@ index 0000000..b5c332c + ) + + ++def measure_fetch_rows(storage: FileBackedEngramStorage, rows: int, repeats: int, seed: int = 0) -> dict: ++ """THE INSTRUMENT: seconds per fetch_rows call of `rows` random records, ++ batched gather versus the per-row oracle, and whether their bytes agree.""" ++ record = storage.expected.record_bytes ++ count = (storage._mmap.size() - len(storage.expected.magic)) // record ++ g = torch.Generator().manual_seed(seed) ++ local = torch.randint(0, count, (rows,), generator=g) ++ offsets = (len(storage.expected.magic) + local * record).tolist() ++ a = torch.zeros((rows, record), dtype=torch.uint8) ++ b = torch.zeros((rows, record), dtype=torch.uint8) ++ timings = {} ++ for name, fn, dest in (("batched", storage.fetch_rows, a), ("per_row", storage._fetch_rows_per_row, b)): ++ fn(offsets, record, dest) ++ start = time.perf_counter() ++ for _ in range(repeats): ++ fn(offsets, record, dest) ++ timings[name] = (time.perf_counter() - start) / repeats ++ return {"rows": rows, "repeats": repeats, "seconds_per_call": timings, "bytes_agree": bool(torch.equal(a, b))} ++ ++ +def allocate_file_backed( + path: str, + expected: ExpectedRowStoreIdentity, @@ -119,3 +265,22 @@ index 0000000..b5c332c + "pinned staging is not smaller than the row store; refusing full-table pin" + ) + return storage, staging ++ ++ ++if __name__ == "__main__": ++ import argparse ++ import json ++ ++ p = argparse.ArgumentParser(description="measure fetch_rows over a rank-local Engram row store") ++ p.add_argument("--store", required=True) ++ p.add_argument("--rank-digest-hex", required=True) ++ p.add_argument("--magic-hex", required=True) ++ p.add_argument("--record-bytes", type=int, required=True) ++ p.add_argument("--rows", type=int, required=True) ++ p.add_argument("--repeats", type=int, default=100) ++ a = p.parse_args() ++ s = FileBackedEngramStorage(a.store, ExpectedRowStoreIdentity(a.rank_digest_hex, bytes.fromhex(a.magic_hex), a.record_bytes)) ++ try: ++ print(json.dumps({"open_receipt": s.receipt.__dict__, "fetch_rows": measure_fetch_rows(s, a.rows, a.repeats)})) ++ finally: ++ s.close() diff --git a/dag/gunbc/spark/v41_engram_file_backed.dag b/dag/gunbc/spark/v41_engram_file_backed.dag new file mode 100644 index 00000000000..a9da7cc311e --- /dev/null +++ b/dag/gunbc/spark/v41_engram_file_backed.dag @@ -0,0 +1,77 @@ +module gunbc.spark.v41_engram_file_backed + +import std.types { String, Bool, List, NonEmptyStr } +import std.nat { Nat } +import std.integer { UInt8 } +import std.measure { ByteSize, byte_size_count } +import v2.std.optional { Present, Absent } +import gunbc.spark.v41_engram_row_store { + V41RowStoreEncodingSpec, V41CompleteStore, + v41_row_bytes, v41_row_store_magic_size, +} + +// ── THE FILE-BACKED OVERLAY'S OPEN AND GATHER CONTRACTS ───────────────────────────────────────── +// +// dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch FileBackedEngramStorage realizes +// what this module declares; the patch is the realization and these are its facts. +// +// OPEN. Whole-file SHA-256 verification stays, and it stays fail-closed. What goes is its residue: +// on a DGX Spark the CPU and GPU share one LPDDR5x pool, so the page cache the sequential verify +// read leaves behind is ordinary memory taken from weights and KV. After verifying, open releases +// the verified range (POSIX_FADV_DONTNEED) and reads its residency back (mincore). The receipt +// carries both readings, and open is admitted only when the readback is zero over the whole file. +// +// GATHER. fetch_rows is one batched gather: every requested offset must be a record boundary of +// this store (magic, then k whole records, with the record inside the file), and the rows come back +// in request order, each exactly the record v41_place_row put at that address. A shifted offset +// would read the tail of one record and the head of the next, a plausible row of the wrong bytes, +// so it refuses rather than being read. +// +// FRONTIER. No live receipt exists yet: the wiring patch that constructs FileBackedEngramStorage in +// ParallelEngramEmbedding._allocate_weights (gunbc.spark.v41_engram_row_store names that trigger) +// is the first producer of a real EngramStoreOpenReceipt on a Spark rank. Until then the verdicts +// here are exercised by test.claim.spark.v41_engram_file_backed_witness over supplied receipts and +// fixture stores. The patch's `measure_fetch_rows` / `__main__` is the named instrument for +// fetch_rows time; its readings are not transcribed here. + +type V41StoreOpenReceipt sole_constructor { + store_bytes: ByteSize + verified_bytes: ByteSize + resident_after_verify: ByteSize + resident_after_release: ByteSize +} + +type V41OpenResidency + = V41OpenReleased { receipt: V41StoreOpenReceipt } + | V41OpenResidueRefused { defect: NonEmptyStr } + +fn v41_open_residency_verdict(r: V41StoreOpenReceipt) -> V41OpenResidency { + if byte_size_count(b: r.verified_bytes) != byte_size_count(b: r.store_bytes) { + V41OpenResidueRefused { defect: join(["verified ", to_string(byte_size_count(b: r.verified_bytes)), " of ", to_string(byte_size_count(b: r.store_bytes)), " store bytes; verification is over the whole file"], "") as NonEmptyStr } + } else if byte_size_count(b: r.resident_after_release) != 0 { + V41OpenResidueRefused { defect: join([to_string(byte_size_count(b: r.resident_after_release)), " verified bytes remain page-cache resident after release (", to_string(byte_size_count(b: r.resident_after_verify)), " after verify)"], "") as NonEmptyStr } + } else { + V41OpenReleased { receipt: r } + } +} + +type V41Gather + = V41Gathered { rows: List> } + | V41GatherRefused { defect: NonEmptyStr } + +fn v41_is_record_offset(spec: V41RowStoreEncodingSpec, store_len: Nat, offset: ByteSize) -> Bool { + let magic = byte_size_count(b: v41_row_store_magic_size()) + let record = byte_size_count(b: v41_row_bytes(g: spec.geometry)) + let o = byte_size_count(b: offset) + o >= magic && (o - magic) % record == 0 && o + record <= store_len +} + +fn v41_gather_rows(store: V41CompleteStore, spec: V41RowStoreEncodingSpec, offsets: List) -> V41Gather { + let record = byte_size_count(b: v41_row_bytes(g: spec.geometry)) + let n = length(store.bytes) + let misaligned = filter(offsets, o => !v41_is_record_offset(spec: spec, store_len: n, offset: o)) + match first(misaligned) { + Present { value: o } => V41GatherRefused { defect: join(["offset ", to_string(byte_size_count(b: o)), " is not magic + k * ", to_string(record), " inside a ", to_string(n), "-byte store"], "") as NonEmptyStr } + Absent => V41Gathered { rows: map(offsets, o => store.bytes |> skip(n: byte_size_count(b: o)) |> take(n: record)) } + } +} diff --git a/dag/test/claim/spark/v41_engram_file_backed_witness_test.dag b/dag/test/claim/spark/v41_engram_file_backed_witness_test.dag new file mode 100644 index 00000000000..6fae561577e --- /dev/null +++ b/dag/test/claim/spark/v41_engram_file_backed_witness_test.dag @@ -0,0 +1,106 @@ +module test.claim.spark.v41_engram_file_backed_witness + +import std.types { Bool, List, NonEmptyStr } +import std.nat { Nat } +import std.integer { UInt8 } +import std.measure { ByteSize, byte_size, byte_size_count } +import v2.std.optional { Present, Absent } +import gunbc.spark.v41_engram_row_store { + V41CompleteStore, V41Placed, V41PlacementRefused, V41FixtureCorrespondenceHolds, V41PublishedTransformUnestablished, V41TransformUnfaithful, + v41_place_row, v41_row_address, v41_transform_fixture, v41_row_store_magic_size, +} +import gunbc.spark.v41_engram_file_backed { + V41StoreOpenReceipt, V41OpenReleased, V41OpenResidueRefused, V41Gathered, V41GatherRefused, + v41_open_residency_verdict, v41_gather_rows, +} +import test.claim.spark.v41_engram_row_store_witness { w_tiny_spec, w_source, w_row } + +// ── OPEN: VERIFIED AND RELEASED, OR REFUSED ───────────────────────────────────────────────────── + +fn w_receipt(verified: Nat, after_release: Nat) -> V41StoreOpenReceipt { + V41StoreOpenReceipt { store_bytes: byte_size(count: 1000), verified_bytes: byte_size(count: verified), resident_after_verify: byte_size(count: 1000), resident_after_release: byte_size(count: after_release) } +} + +test fn a_verified_store_whose_cache_was_released_opens() -> Bool { + match v41_open_residency_verdict(r: w_receipt(verified: 1000, after_release: 0)) { + V41OpenReleased { receipt: r } => byte_size_count(b: r.resident_after_verify) == 1000 + V41OpenResidueRefused { defect: _ } => false + } +} + +// RED (a): verification that leaves its page cache resident fails the residency claim, by any +// residue, including a single page; and a partial verification is no verification. +test fn verification_that_leaves_the_cache_resident_refuses() -> Bool { + let resident = match v41_open_residency_verdict(r: w_receipt(verified: 1000, after_release: 1000)) { + V41OpenReleased { receipt: _ } => false + V41OpenResidueRefused { defect: d } => (d as String).contains("remain page-cache resident") + } + let one_page = match v41_open_residency_verdict(r: w_receipt(verified: 1000, after_release: 4096)) { + V41OpenReleased { receipt: _ } => false + V41OpenResidueRefused { defect: _ } => true + } + let partial = match v41_open_residency_verdict(r: w_receipt(verified: 999, after_release: 0)) { + V41OpenReleased { receipt: _ } => false + V41OpenResidueRefused { defect: d } => (d as String).contains("whole file") + } + resident && one_page && partial +} + +// ── GATHER: THE BATCH IS THE PER-ROW RECORDS, IN ORDER; A SHIFTED OFFSET REFUSES ───────────────── + +fn w_rank1_store() -> V41CompleteStore? { + match v41_transform_fixture(source: w_source(), spec: w_tiny_spec()) { + V41FixtureCorrespondenceHolds { subject: _, placed: _, stores: st } => first(filter(st, c => c.rank == 1)) + V41PublishedTransformUnestablished { obligation: _ } => none + V41TransformUnfaithful { defects: _ } => none + } +} + +// The per-row reference: the record v41_place_row put for this global row, and its address. +fn w_record(i: Nat) -> List { + match v41_place_row(row: w_row(i: i), spec: w_tiny_spec()) { + V41Placed { row: p } => p.record + V41PlacementRefused { defect: _ } => [] as List + } +} + +fn w_offset(i: Nat) -> ByteSize { + match v41_row_address(spec: w_tiny_spec(), layer: 1, global_row: i) { + Present { value: a } => a.offset + Absent => byte_size(count: 0) + } +} + +// RED (b), agreement arm: rank 1 owns global rows 2 and 3, its first and last local rows. A batch +// over both boundaries, out of order and with a repeat, returns exactly the per-row records in +// request order. +test fn the_batched_gather_is_the_per_row_records_in_request_order_over_boundary_rows() -> Bool { + match w_rank1_store() { + Absent => false + Present { value: s } => match v41_gather_rows(store: s, spec: w_tiny_spec(), offsets: [w_offset(i: 3), w_offset(i: 2), w_offset(i: 3)]) { + V41GatherRefused { defect: _ } => false + V41Gathered { rows: rs } => length(w_record(i: 2)) == 3 && rs == [w_record(i: 3), w_record(i: 2), w_record(i: 3)] + } + } +} + +fn w_refused(s: V41CompleteStore, offsets: List) -> Bool { + match v41_gather_rows(store: s, spec: w_tiny_spec(), offsets: offsets) { + V41Gathered { rows: _ } => false + V41GatherRefused { defect: d } => (d as String).contains("is not magic + k * 3") + } +} + +// RED (b), refusal arm: an offset shifted by one byte, one inside the magic, and one whose record +// would run past the file each refuse the WHOLE batch, even beside a valid offset. +test fn a_shifted_offset_batch_refuses() -> Bool { + match w_rank1_store() { + Absent => false + Present { value: s } => { + let end = length(s.bytes) + w_refused(s: s, offsets: [w_offset(i: 2), byte_size(count: byte_size_count(b: w_offset(i: 2)) + 1)]) + && w_refused(s: s, offsets: [byte_size(count: byte_size_count(b: v41_row_store_magic_size()) - 1)]) + && w_refused(s: s, offsets: [byte_size(count: end)]) + } + } +} From 2c553680342cdd0e10272558df17a316805261c8 Mon Sep 17 00:00:00 2001 From: gunbc-ci-auto-heal Date: Fri, 25 Sep 2026 18:38:26 +0000 Subject: [PATCH 2/4] Engram file-backed: realization refuses a partial verify like the model; receipt carries store_bytes; name the verdict's consumer frontier Co-Authored-By: Claude Opus 5.5 (1M context) --- .../d2d649e6/engram_file_backed.py.patch | 15 +++++++++++--- dag/gunbc/spark/v41_engram_file_backed.dag | 20 ++++++++++++++----- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch b/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch index ea2ca06190c..c18cf45a3f3 100644 --- a/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch +++ b/dag/gunbc/spark/patches/d2d649e6/engram_file_backed.py.patch @@ -1,9 +1,9 @@ diff --git a/vllm/models/deepseek_v4_1/nvidia/engram_file_backed.py b/vllm/models/deepseek_v4_1/nvidia/engram_file_backed.py new file mode 100644 -index 0000000..16abb9b +index 0000000..4026664 --- /dev/null +++ b/vllm/models/deepseek_v4_1/nvidia/engram_file_backed.py -@@ -0,0 +1,280 @@ +@@ -0,0 +1,289 @@ +# File-backed Engram storage for vLLM d2d649e674c75425d2d6975c87eb89fd4d55fff8. +# +# THIS IS A BACKING SWAP, NOT NEW PREFETCH MACHINERY. Upstream @@ -30,7 +30,9 @@ index 0000000..16abb9b +# +# The open and gather contracts are gunbc.spark.v41_engram_file_backed: +# v41_open_residency_verdict (verification's page-cache residue is released and -+# read back as zero, else open refuses) and v41_gather_rows (one batched gather ++# read back as zero over the whole file, else open refuses; its two refusal ++# branches are the two raises in _verify_and_release, and the receipt's fields ++# are V41StoreOpenReceipt's) and v41_gather_rows (one batched gather +# at record-aligned offsets, in request order; a misaligned offset refuses). + +from __future__ import annotations @@ -81,6 +83,7 @@ index 0000000..16abb9b + before and after the release (gunbc.spark.v41_engram_file_backed + V41StoreOpenReceipt).""" + ++ store_bytes: int + verified_bytes: int + resident_bytes_after_verify: int + resident_bytes_after_release: int @@ -158,7 +161,12 @@ index 0000000..16abb9b + + def _verify_and_release(self) -> EngramStoreOpenReceipt: + # Verification stays fail-closed and whole-file; only its cache residue goes. ++ store = os.fstat(self._fd).st_size + opened, head, verified = _sha256_fd(self._fd, len(self.expected.magic)) ++ if verified != store: ++ raise EngramRowStoreIdentityMismatch( ++ f"verified {verified} of {store} store bytes; verification is over the whole file" ++ ) + if opened != self.expected.rank_digest_hex: + raise EngramRowStoreIdentityMismatch( + f"opened {self.path} digest {opened}, expected rank digest " @@ -170,6 +178,7 @@ index 0000000..16abb9b + os.posix_fadvise(self._fd, 0, verified, os.POSIX_FADV_DONTNEED) + after_release = resident_bytes(self._fd, verified) + receipt = EngramStoreOpenReceipt( ++ store_bytes=store, + verified_bytes=verified, + resident_bytes_after_verify=after_verify, + resident_bytes_after_release=after_release, diff --git a/dag/gunbc/spark/v41_engram_file_backed.dag b/dag/gunbc/spark/v41_engram_file_backed.dag index a9da7cc311e..65505f6d01c 100644 --- a/dag/gunbc/spark/v41_engram_file_backed.dag +++ b/dag/gunbc/spark/v41_engram_file_backed.dag @@ -27,11 +27,21 @@ import gunbc.spark.v41_engram_row_store { // would read the tail of one record and the head of the next, a plausible row of the wrong bytes, // so it refuses rather than being read. // -// FRONTIER. No live receipt exists yet: the wiring patch that constructs FileBackedEngramStorage in -// ParallelEngramEmbedding._allocate_weights (gunbc.spark.v41_engram_row_store names that trigger) -// is the first producer of a real EngramStoreOpenReceipt on a Spark rank. Until then the verdicts -// here are exercised by test.claim.spark.v41_engram_file_backed_witness over supplied receipts and -// fixture stores. The patch's `measure_fetch_rows` / `__main__` is the named instrument for +// THE REALIZATION MIRRORS THIS, BRANCH FOR BRANCH. FileBackedEngramStorage._verify_and_release raises +// on exactly the two refusals of v41_open_residency_verdict (verified != fstat size; any residue after +// release), and EngramStoreOpenReceipt carries V41StoreOpenReceipt's four fields. _record_index raises +// on exactly the non-record offsets v41_is_record_offset rejects. The Python refuses at runtime because +// a Spark rank cannot call this fold; the fold is the declaration those raises realize. +// +// CONSUMER: A DECLARED FRONTIER, NOT YET IN THIS CLOSURE. The consumer of v41_open_residency_verdict is +// the (layer, rank) open-receipt reading the wiring patch introduces: when +// ParallelEngramEmbedding._allocate_weights constructs FileBackedEngramStorage (the trigger +// gunbc.spark.v41_engram_row_store already names), each rank's EngramStoreOpenReceipt is read back into +// gunbc.spark.v41_runtime_candidate beside v41_published_store_readback_readings and folded through this +// verdict, so a rank that verified but stayed resident refuses the candidate, not only its own open. +// v41_gather_rows' consumer is the same change's differential over boundary rows of a published store. +// Until then both are exercised by test.claim.spark.v41_engram_file_backed_witness over supplied +// receipts and fixture stores. The patch's measure_fetch_rows / __main__ is the named instrument for // fetch_rows time; its readings are not transcribed here. type V41StoreOpenReceipt sole_constructor { From 193d9c6a3a20dd5fc4e1bfd046a4e8566cb3e0d9 Mon Sep 17 00:00:00 2001 From: gunbc-ci-auto-heal Date: Fri, 25 Sep 2026 18:39:11 +0000 Subject: [PATCH 3/4] v41_engram_file_backed: import filter/length/skip from v2.std.algebra (UnimportedBareProvider) Co-Authored-By: Claude Opus 5.5 (1M context) --- dag/gunbc/spark/v41_engram_file_backed.dag | 1 + 1 file changed, 1 insertion(+) diff --git a/dag/gunbc/spark/v41_engram_file_backed.dag b/dag/gunbc/spark/v41_engram_file_backed.dag index 65505f6d01c..78efaaa4c31 100644 --- a/dag/gunbc/spark/v41_engram_file_backed.dag +++ b/dag/gunbc/spark/v41_engram_file_backed.dag @@ -5,6 +5,7 @@ import std.nat { Nat } import std.integer { UInt8 } import std.measure { ByteSize, byte_size_count } import v2.std.optional { Present, Absent } +import v2.std.algebra { filter, length, skip } import gunbc.spark.v41_engram_row_store { V41RowStoreEncodingSpec, V41CompleteStore, v41_row_bytes, v41_row_store_magic_size, From 5339445e21b4485a34807b349ea57beb51d98c29 Mon Sep 17 00:00:00 2001 From: gunbc-ci-auto-heal Date: Fri, 25 Sep 2026 19:14:56 +0000 Subject: [PATCH 4/4] v41_engram_file_backed: v41_store_open_receipt constructor (sole_constructor type built outside its module) Co-Authored-By: Claude Opus 5.5 (1M context) --- dag/gunbc/spark/v41_engram_file_backed.dag | 4 ++++ dag/test/claim/spark/v41_engram_file_backed_witness_test.dag | 4 ++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/dag/gunbc/spark/v41_engram_file_backed.dag b/dag/gunbc/spark/v41_engram_file_backed.dag index 78efaaa4c31..af93da42a11 100644 --- a/dag/gunbc/spark/v41_engram_file_backed.dag +++ b/dag/gunbc/spark/v41_engram_file_backed.dag @@ -52,6 +52,10 @@ type V41StoreOpenReceipt sole_constructor { resident_after_release: ByteSize } +fn v41_store_open_receipt(store_bytes: ByteSize, verified_bytes: ByteSize, resident_after_verify: ByteSize, resident_after_release: ByteSize) -> V41StoreOpenReceipt { + V41StoreOpenReceipt { store_bytes: store_bytes, verified_bytes: verified_bytes, resident_after_verify: resident_after_verify, resident_after_release: resident_after_release } +} + type V41OpenResidency = V41OpenReleased { receipt: V41StoreOpenReceipt } | V41OpenResidueRefused { defect: NonEmptyStr } diff --git a/dag/test/claim/spark/v41_engram_file_backed_witness_test.dag b/dag/test/claim/spark/v41_engram_file_backed_witness_test.dag index 6fae561577e..e70ef53e1ac 100644 --- a/dag/test/claim/spark/v41_engram_file_backed_witness_test.dag +++ b/dag/test/claim/spark/v41_engram_file_backed_witness_test.dag @@ -11,14 +11,14 @@ import gunbc.spark.v41_engram_row_store { } import gunbc.spark.v41_engram_file_backed { V41StoreOpenReceipt, V41OpenReleased, V41OpenResidueRefused, V41Gathered, V41GatherRefused, - v41_open_residency_verdict, v41_gather_rows, + v41_store_open_receipt, v41_open_residency_verdict, v41_gather_rows, } import test.claim.spark.v41_engram_row_store_witness { w_tiny_spec, w_source, w_row } // ── OPEN: VERIFIED AND RELEASED, OR REFUSED ───────────────────────────────────────────────────── fn w_receipt(verified: Nat, after_release: Nat) -> V41StoreOpenReceipt { - V41StoreOpenReceipt { store_bytes: byte_size(count: 1000), verified_bytes: byte_size(count: verified), resident_after_verify: byte_size(count: 1000), resident_after_release: byte_size(count: after_release) } + v41_store_open_receipt(store_bytes: byte_size(count: 1000), verified_bytes: byte_size(count: verified), resident_after_verify: byte_size(count: 1000), resident_after_release: byte_size(count: after_release)) } test fn a_verified_store_whose_cache_was_released_opens() -> Bool {