diff --git a/src/mobius/integrations/gguf/_builder.py b/src/mobius/integrations/gguf/_builder.py
index c761efcfe..3d8cb21fe 100644
--- a/src/mobius/integrations/gguf/_builder.py
+++ b/src/mobius/integrations/gguf/_builder.py
@@ -5274,22 +5274,13 @@ def _resolve_mmproj_companion_path(
     return _resolve_gguf_path_impl(gguf_path, allow_mmproj_companion=True)
 
 
-def _hub_cache_identity_paths(paths: Collection[Path]) -> list[Path] | None:
-    """Resolve trusted Hub snapshot links to regular blob paths for hashing."""
-    from huggingface_hub.constants import HF_HUB_CACHE
-
-    cache_root = Path(HF_HUB_CACHE).expanduser().absolute()
+def _regular_file_identity_paths(paths: Collection[Path]) -> list[Path] | None:
+    """Resolve opened shard paths to regular-file targets for identity hashing."""
     resolved_paths: list[Path] = []
     for path in paths:
-        absolute = path.expanduser().absolute()
-        try:
-            absolute.relative_to(cache_root)
-        except ValueError:
-            return None
-        resolved = absolute.resolve(strict=True)
         try:
-            resolved.relative_to(cache_root)
-        except ValueError:
+            resolved = path.expanduser().resolve(strict=True)
+        except OSError:
             return None
         if not resolved.is_file() or resolved.is_symlink():
             return None
@@ -5504,7 +5495,7 @@ def build_from_gguf(
             shard_open_kwargs["expected_sizes"] = expected_sizes
         gguf_model = open_gguf_model(gguf_path, **shard_open_kwargs)
     if isinstance(gguf_model, GgufShardSet):
-        identity_paths = _hub_cache_identity_paths(gguf_model.shard_paths)
+        identity_paths = _regular_file_identity_paths(gguf_model.shard_paths)
         if identity_paths is not None:
             gguf_model._set_identity_paths(identity_paths)
     _validate_gguf_model(
@@ -6049,7 +6040,7 @@ def combine_mtp_report(mtp_report) -> None:
             for name, tensor in state_dict.items()
             if id(tensor) in reuse_candidates_by_id
         }
-        attach_reused_initializers(pkg, gguf_path, final_candidates)
+        attach_reused_initializers(pkg, gguf_path, final_candidates, gguf_model)
 
     # 9b. Sparse-MoE fusion + honesty gate (final graph state).
     # Now that every native block carries its real packed bytes, collapse the
@@ -8175,11 +8166,11 @@ def _load_quantized_state_dict(
             n_repacked += num_experts
         elif should_repack:
             if is_tencent_q1_0_tensor:
-                gguf_path, data_section_offset, reader_tensor = gguf_model._tensor_source(
-                    gguf_name
+                read_source_range, data_section_offset, reader_tensor = (
+                    gguf_model._tensor_source(gguf_name)
                 )
                 repacked = parse_tencent_q1_0_tensor(
-                    str(gguf_path),
+                    read_source_range,
                     data_section_offset,
                     reader_tensor,
                 )
diff --git a/src/mobius/integrations/gguf/_builder_test.py b/src/mobius/integrations/gguf/_builder_test.py
index f34471414..04d814ca0 100644
--- a/src/mobius/integrations/gguf/_builder_test.py
+++ b/src/mobius/integrations/gguf/_builder_test.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import errno
 import json
 import os
 import re
@@ -47,6 +48,18 @@ def _gguf_header_prefix(*architectures: str) -> bytes:
     )
 
 
+def _symlink_or_skip(link: Path, target: Path) -> None:
+    try:
+        link.symlink_to(target)
+    except OSError as error:
+        if os.name == "nt" and (
+            getattr(error, "winerror", None) in {1, 50, 1314}
+            or error.errno in {errno.EPERM, errno.EACCES, errno.ENOSYS}
+        ):
+            pytest.skip(f"Windows runner cannot create test symlinks: {error}")
+        raise
+
+
 def _run_gather_block_quantized(
     tmp_path: Path,
     *,
@@ -2467,7 +2480,6 @@ def tracking_open(self, *args, **kwargs):
     def test_mixed_save_preserves_ranges_and_runs(self, tmp_path: Path):
         from mobius._model_package import ModelPackage
         from mobius.integrations.gguf import (
-            _reuse,
             build_from_gguf,
             verify_gguf_reuse_manifest,
         )
@@ -2475,7 +2487,14 @@ def test_mixed_save_preserves_ranges_and_runs(self, tmp_path: Path):
         gguf_path = tmp_path / "model.gguf"
         _write_quantized_gguf(gguf_path, projection_quantization="f32")
         package = build_from_gguf(gguf_path, reuse_gguf_weights=True)
-        with mock.patch.object(_reuse, "_sha256", wraps=_reuse._sha256) as sha256:
+        from mobius.integrations.gguf._reader import GGUFModel
+
+        with mock.patch.object(
+            GGUFModel,
+            "source_sha256",
+            autospec=True,
+            side_effect=GGUFModel.source_sha256,
+        ) as sha256:
             package.save(str(tmp_path), progress_bar=False)
         assert sha256.call_count == 1
 
@@ -2551,9 +2570,13 @@ def test_save_rehashes_source_immediately_before_publish(
         def mutate_source_before_verification(path, payload):
             nonlocal mutated
             if path.name.startswith(".gguf-reuse.json.") and path.name.endswith(".tmp"):
-                source = bytearray(gguf_path.read_bytes())
-                source[-1] ^= 1
-                gguf_path.write_bytes(source)
+                with gguf_path.open("r+b") as stream:
+                    stream.seek(-1, os.SEEK_END)
+                    value = stream.read(1)
+                    stream.seek(-1, os.SEEK_END)
+                    stream.write(bytes([value[0] ^ 0xFF]))
+                    stream.flush()
+                    os.fsync(stream.fileno())
                 mutated = True
             return real_write_json(path, payload)
 
@@ -2565,6 +2588,211 @@ def mutate_source_before_verification(path, payload):
         assert not (tmp_path / "model.onnx").exists()
         assert not (tmp_path / "gguf-reuse.json").exists()
 
+    def test_reuse_plan_hashes_the_model_source_used_for_tensor_parsing(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        from mobius.integrations.gguf import build_from_gguf
+        from mobius.integrations.gguf._reader import GGUFModel
+
+        gguf_path = tmp_path / "model.gguf"
+        _write_quantized_gguf(gguf_path, projection_quantization="f32")
+        source_sha256 = GGUFModel.source_sha256
+        mutated = False
+
+        def mutate_before_plan_hash(model, **kwargs):
+            nonlocal mutated
+            if not mutated and Path(model._path) == gguf_path:
+                with gguf_path.open("r+b") as stream:
+                    stream.seek(-1, os.SEEK_END)
+                    value = stream.read(1)
+                    stream.seek(-1, os.SEEK_END)
+                    stream.write(bytes([value[0] ^ 0xFF]))
+                mutated = True
+            return source_sha256(model, **kwargs)
+
+        monkeypatch.setattr(GGUFModel, "source_sha256", mutate_before_plan_hash)
+
+        with pytest.raises(ValueError, match="source changed after its reader was opened"):
+            build_from_gguf(gguf_path, reuse_gguf_weights=True)
+        assert mutated
+
+    def test_reuse_plan_with_relative_source_survives_cwd_change(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        from mobius.integrations.gguf import build_from_gguf, verify_gguf_reuse_manifest
+
+        gguf_path = tmp_path / "model.gguf"
+        _write_quantized_gguf(gguf_path, projection_quantization="f32")
+        monkeypatch.chdir(tmp_path)
+        package = build_from_gguf(Path("model.gguf"), reuse_gguf_weights=True)
+        other = tmp_path / "other"
+        other.mkdir()
+        monkeypatch.chdir(other)
+
+        package.save(str(tmp_path), progress_bar=False)
+        verify_gguf_reuse_manifest(tmp_path)
+
+    def test_reuse_verifier_detects_mutation_between_hash_and_tensor_validation(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        from mobius.integrations.gguf import build_from_gguf, verify_gguf_reuse_manifest
+        from mobius.integrations.gguf._reader import GGUFModel
+
+        gguf_path = tmp_path / "model.gguf"
+        _write_quantized_gguf(gguf_path, projection_quantization="f32")
+        build_from_gguf(gguf_path, reuse_gguf_weights=True).save(
+            str(tmp_path), progress_bar=False
+        )
+        source_sha256 = GGUFModel.source_sha256
+        mutated = False
+
+        def mutate_after_verifier_hash(model, **kwargs):
+            nonlocal mutated
+            digest = source_sha256(model, **kwargs)
+            if not mutated and Path(model._path) == gguf_path:
+                with gguf_path.open("r+b") as stream:
+                    stream.seek(-1, os.SEEK_END)
+                    value = stream.read(1)
+                    stream.seek(-1, os.SEEK_END)
+                    stream.write(bytes([value[0] ^ 0xFF]))
+                mutated = True
+            return digest
+
+        monkeypatch.setattr(GGUFModel, "source_sha256", mutate_after_verifier_hash)
+
+        with pytest.raises(ValueError, match="reuse manifest was verified"):
+            verify_gguf_reuse_manifest(tmp_path)
+        assert mutated
+
+    def test_reuse_verifier_rejects_source_replaced_by_symlink_before_open(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        from mobius.integrations.gguf import build_from_gguf, verify_gguf_reuse_manifest
+        from mobius.integrations.gguf._reader import GGUFModel
+
+        gguf_path = tmp_path / "model.gguf"
+        _write_quantized_gguf(gguf_path, projection_quantization="f32")
+        build_from_gguf(gguf_path, reuse_gguf_weights=True).save(
+            str(tmp_path), progress_bar=False
+        )
+        moved_source = tmp_path / "moved-source.gguf"
+        original_init = GGUFModel.__init__
+        raced = False
+
+        def replace_before_verifier_open(model, path, **kwargs):
+            nonlocal raced
+            if not raced and Path(path) == gguf_path:
+                gguf_path.replace(moved_source)
+                _symlink_or_skip(gguf_path, moved_source)
+                raced = True
+            return original_init(model, path, **kwargs)
+
+        monkeypatch.setattr(GGUFModel, "__init__", replace_before_verifier_open)
+
+        with pytest.raises(ValueError, match="source is missing or unsafe"):
+            verify_gguf_reuse_manifest(tmp_path)
+        assert raced
+
+    def test_reuse_verifier_rechecks_symlink_after_final_handle_comparison(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        from mobius.integrations.gguf import build_from_gguf, verify_gguf_reuse_manifest
+        from mobius.integrations.gguf._reader import GGUFModel
+
+        gguf_path = tmp_path / "model.gguf"
+        _write_quantized_gguf(gguf_path, projection_quantization="f32")
+        build_from_gguf(gguf_path, reuse_gguf_weights=True).save(
+            str(tmp_path), progress_bar=False
+        )
+        source_matches_path = GGUFModel.source_matches_path
+        path_is_symlink = Path.is_symlink
+        raced = False
+        symlink_recheck_observed = False
+
+        def replace_after_handle_comparison(model, path=None):
+            nonlocal raced
+            matches = source_matches_path(model, path)
+            if not raced and Path(model._path) == gguf_path:
+                assert matches
+                raced = True
+            return matches
+
+        def report_post_comparison_symlink(path):
+            nonlocal symlink_recheck_observed
+            if path == gguf_path and raced:
+                symlink_recheck_observed = True
+                return True
+            return path_is_symlink(path)
+
+        monkeypatch.setattr(GGUFModel, "source_matches_path", replace_after_handle_comparison)
+        monkeypatch.setattr(Path, "is_symlink", report_post_comparison_symlink)
+
+        with pytest.raises(ValueError, match="reuse manifest was verified"):
+            verify_gguf_reuse_manifest(tmp_path)
+        assert raced
+        assert symlink_recheck_observed
+
+    def test_reuse_verifier_rechecks_source_after_onnx_validation(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        from mobius.integrations.gguf import _reuse, build_from_gguf
+
+        gguf_path = tmp_path / "model.gguf"
+        _write_quantized_gguf(gguf_path, projection_quantization="f32")
+        build_from_gguf(gguf_path, reuse_gguf_weights=True).save(
+            str(tmp_path), progress_bar=False
+        )
+        load_model = _reuse.ir.load
+        mutated = False
+
+        def mutate_after_onnx_load(path):
+            nonlocal mutated
+            model = load_model(path)
+            if not mutated:
+                with gguf_path.open("r+b") as stream:
+                    stream.seek(-1, os.SEEK_END)
+                    value = stream.read(1)
+                    stream.seek(-1, os.SEEK_END)
+                    stream.write(bytes([value[0] ^ 0xFF]))
+                mutated = True
+            return model
+
+        monkeypatch.setattr(_reuse.ir, "load", mutate_after_onnx_load)
+
+        with pytest.raises(ValueError, match="reuse manifest was verified"):
+            _reuse.verify_gguf_reuse_manifest(tmp_path)
+        assert mutated
+
+    def test_reuse_save_rechecks_source_immediately_before_publication(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        from mobius.integrations.gguf import _reuse, build_from_gguf
+
+        gguf_path = tmp_path / "model.gguf"
+        _write_quantized_gguf(gguf_path, projection_quantization="f32")
+        package = build_from_gguf(gguf_path, reuse_gguf_weights=True)
+        verify_manifest = _reuse.verify_gguf_reuse_manifest
+        mutated = False
+
+        def mutate_after_verification(*args, **kwargs):
+            nonlocal mutated
+            result = verify_manifest(*args, **kwargs)
+            with gguf_path.open("r+b") as stream:
+                stream.seek(-1, os.SEEK_END)
+                value = stream.read(1)
+                stream.seek(-1, os.SEEK_END)
+                stream.write(bytes([value[0] ^ 0xFF]))
+            mutated = True
+            return result
+
+        monkeypatch.setattr(_reuse, "verify_gguf_reuse_manifest", mutate_after_verification)
+
+        with pytest.raises(ValueError, match="package was being prepared"):
+            package.save(str(tmp_path), progress_bar=False)
+        assert mutated
+        assert not (tmp_path / "model.onnx").exists()
+        assert not (tmp_path / "gguf-reuse.json").exists()
+
     def test_native_projection_bytes_are_not_copied_to_sidecar(self, tmp_path: Path):
         from mobius.integrations.gguf import build_from_gguf
         from mobius.integrations.gguf._reader import GGUFModel
@@ -4240,6 +4468,50 @@ def test_tencent_q1_0_preflight_uses_layout_and_exact_payload_bytes(
         }
         assert any(node.op_type == "MatMulNBits" for node in package["model"].graph)
 
+    def test_tencent_q1_0_reads_pinned_bytes_during_symlink_aba(self, tmp_path: Path):
+        from mobius.integrations.gguf._reader import GGUFModel
+        from mobius.integrations.gguf._tencent_q1_0 import parse_tencent_q1_0_tensor
+
+        source_a = tmp_path / "source-a.gguf"
+        source_b = tmp_path / "source-b.gguf"
+        _write_tencent_q1_0_gguf(source_a)
+        _write_tencent_q1_0_gguf(source_b)
+        inspect_b = GGUFModel(source_b)
+        _read_b, data_offset_b, tensor_b = inspect_b._tensor_source("blk.0.attn_q.weight")
+        tensor_offset_b = int(tensor_b.field.parts[tensor_b.field.data[-1]][0])
+        inspect_b.close()
+        with source_b.open("r+b") as stream:
+            stream.seek(data_offset_b + tensor_offset_b + 2)
+            value = stream.read(1)
+            stream.seek(data_offset_b + tensor_offset_b + 2)
+            stream.write(bytes([value[0] ^ 0xFF]))
+
+        logical = tmp_path / "logical.gguf"
+        _symlink_or_skip(logical, source_a)
+        model_a = GGUFModel(logical)
+        read_a, data_offset_a, tensor_a = model_a._tensor_source("blk.0.attn_q.weight")
+        expected = parse_tencent_q1_0_tensor(read_a, data_offset_a, tensor_a)
+
+        logical.unlink()
+        _symlink_or_skip(logical, source_b)
+        injected = parse_tencent_q1_0_tensor(read_a, data_offset_a, tensor_a)
+        logical.unlink()
+        _symlink_or_skip(logical, source_a)
+
+        model_b = GGUFModel(source_b)
+        read_b, replacement_offset, replacement_tensor = model_b._tensor_source(
+            "blk.0.attn_q.weight"
+        )
+        replacement = parse_tencent_q1_0_tensor(
+            read_b,
+            replacement_offset,
+            replacement_tensor,
+        )
+
+        np.testing.assert_array_equal(injected.weight, expected.weight)
+        assert not np.array_equal(injected.weight, replacement.weight)
+        assert model_a.source_matches_path()
+
     def test_preflight_records_mapped_bias_and_scale_parameters(self) -> None:
         from gguf import GGMLQuantizationType
         from onnxscript import nn
diff --git a/src/mobius/integrations/gguf/_preflight.py b/src/mobius/integrations/gguf/_preflight.py
index fe87222eb..89a9999d6 100644
--- a/src/mobius/integrations/gguf/_preflight.py
+++ b/src/mobius/integrations/gguf/_preflight.py
@@ -491,13 +491,9 @@ def preflight_local_gguf(
         is_sharded = False
         split_count = 1
         total_tensors = model.num_tensors
-        size_bytes = path.stat().st_size
+        size_bytes = model.source_size
         total_bytes = size_bytes
-        sha = None
-        if verify_checksums:
-            from mobius.integrations.gguf._shard_set import _sha256_of
-
-            sha = _sha256_of(path)
+        sha = model.source_sha256() if verify_checksums else None
         files.append(
             GgufFileMeta(
                 filename=path.name,
@@ -525,6 +521,8 @@ def preflight_local_gguf(
     if unsupported:
         blockers.append(_classification_blocker(unsupported))
     vram_weights_bytes = output_bytes
+    if not model.source_matches_path():
+        raise ValueError("GGUF source changed while its preflight report was being built.")
 
     return GgufPreflightReport(
         source=str(path),
diff --git a/src/mobius/integrations/gguf/_preflight_test.py b/src/mobius/integrations/gguf/_preflight_test.py
index 63bd63f8f..14a6d071a 100644
--- a/src/mobius/integrations/gguf/_preflight_test.py
+++ b/src/mobius/integrations/gguf/_preflight_test.py
@@ -12,7 +12,9 @@
 from __future__ import annotations
 
 import argparse
+import errno
 import json
+import os
 import re
 from pathlib import Path
 
@@ -30,6 +32,18 @@
 )
 
 
+def _symlink_or_skip(link: Path, target: Path) -> None:
+    try:
+        link.symlink_to(target)
+    except OSError as error:
+        if os.name == "nt" and (
+            getattr(error, "winerror", None) in {1, 50, 1314}
+            or error.errno in {errno.EPERM, errno.EACCES, errno.ENOSYS}
+        ):
+            pytest.skip(f"Windows runner cannot create test symlinks: {error}")
+        raise
+
+
 def _write_sharded_gguf(
     directory: Path,
     *,
@@ -69,6 +83,25 @@ def _write_sharded_gguf(
     return sorted(directory.glob(f"{stem}-*.gguf"))
 
 
+def _write_single_gguf(path: Path, *, seed: int = 0) -> Path:
+    from gguf import GGUFWriter
+
+    writer = GGUFWriter(str(path), "llama")
+    writer.add_context_length(128)
+    writer.add_embedding_length(16)
+    writer.add_block_count(1)
+    writer.add_head_count(4)
+    writer.add_head_count_kv(2)
+    writer.add_vocab_size(32)
+    tensor = np.random.default_rng(seed).standard_normal((8, 16)).astype(np.float32)
+    writer.add_tensor("token_embd.weight", tensor)
+    writer.write_header_to_file()
+    writer.write_kv_data_to_file()
+    writer.write_tensors_to_file()
+    writer.close()
+    return path
+
+
 def _raw_quant_rows(qtype, n_rows: int, k: int) -> np.ndarray:
     """Zeroed raw-block bytes shaped ``(n_rows, bytes_per_row)`` for *qtype*."""
     from gguf import GGML_QUANT_SIZES
@@ -252,21 +285,7 @@ def test_local_preflight_checksums_optional(tmp_path):
 
 
 def test_local_preflight_single_file(tmp_path):
-    from gguf import GGUFWriter
-
-    path = tmp_path / "plain.gguf"
-    writer = GGUFWriter(str(path), "llama")
-    writer.add_context_length(128)
-    writer.add_embedding_length(16)
-    writer.add_block_count(1)
-    writer.add_head_count(4)
-    writer.add_head_count_kv(2)
-    writer.add_vocab_size(32)
-    writer.add_tensor("token_embd.weight", np.zeros((8, 16), np.float32))
-    writer.write_header_to_file()
-    writer.write_kv_data_to_file()
-    writer.write_tensors_to_file()
-    writer.close()
+    path = _write_single_gguf(tmp_path / "plain.gguf")
 
     report = preflight_local_gguf(path)
     assert not report.is_sharded
@@ -274,6 +293,29 @@ def test_local_preflight_single_file(tmp_path):
     assert report.split_count == 1
 
 
+def test_local_preflight_checksum_rejects_path_replacement(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+):
+    from mobius.integrations.gguf._reader import GGUFModel
+
+    source = _write_single_gguf(tmp_path / "source.gguf")
+    replacement = _write_single_gguf(tmp_path / "replacement.gguf", seed=1)
+    path = tmp_path / "logical.gguf"
+    _symlink_or_skip(path, source)
+    source_sha256 = GGUFModel.source_sha256
+
+    def replace_path_before_hash(model, **kwargs):
+        digest = source_sha256(model, **kwargs)
+        path.unlink()
+        _symlink_or_skip(path, replacement)
+        return digest
+
+    monkeypatch.setattr(GGUFModel, "source_sha256", replace_path_before_hash)
+
+    with pytest.raises(ValueError, match="preflight report was being built"):
+        preflight_local_gguf(path, verify_checksums=True)
+
+
 def test_report_json_roundtrip_is_resumable(tmp_path):
     shards = _write_sharded_gguf(tmp_path, split_max_tensors=3)
     cache = tmp_path / "pf.json"
diff --git a/src/mobius/integrations/gguf/_reader.py b/src/mobius/integrations/gguf/_reader.py
index 5f28ec606..6866162e6 100644
--- a/src/mobius/integrations/gguf/_reader.py
+++ b/src/mobius/integrations/gguf/_reader.py
@@ -24,12 +24,17 @@
 
 __all__ = ["GGUFModel"]
 
+import hashlib
 import logging
 import mmap
+import os
+import stat
+import threading
 from array import array
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 import numpy as np
 
@@ -38,9 +43,79 @@
 logger = logging.getLogger(__name__)
 
 
-def _stat_identity(path: Path) -> tuple[int, int, int, int]:
-    stat = path.stat()
-    return (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns)
+def _descriptor_change_time(descriptor: int, source_stat: os.stat_result) -> int:
+    if os.name != "nt":
+        return source_stat.st_ctime_ns
+
+    import ctypes
+    import msvcrt
+
+    class _FileBasicInfo(ctypes.Structure):
+        _fields_ = [
+            ("creation_time", ctypes.c_int64),
+            ("last_access_time", ctypes.c_int64),
+            ("last_write_time", ctypes.c_int64),
+            ("change_time", ctypes.c_int64),
+            ("file_attributes", ctypes.c_uint32),
+        ]
+
+    basic_info = _FileBasicInfo()
+    get_file_information = ctypes.windll.kernel32.GetFileInformationByHandleEx
+    get_file_information.argtypes = [
+        ctypes.c_void_p,
+        ctypes.c_int,
+        ctypes.c_void_p,
+        ctypes.c_uint32,
+    ]
+    get_file_information.restype = ctypes.c_int
+    handle = msvcrt.get_osfhandle(descriptor)
+    if not get_file_information(
+        handle,
+        0,  # FileBasicInfo
+        ctypes.byref(basic_info),
+        ctypes.sizeof(basic_info),
+    ):
+        raise ctypes.WinError()
+    return basic_info.change_time
+
+
+def _descriptor_identity(descriptor: int) -> tuple[int, int, int, int, int]:
+    source_stat = os.fstat(descriptor)
+    return (
+        source_stat.st_dev,
+        source_stat.st_ino,
+        source_stat.st_size,
+        source_stat.st_mtime_ns,
+        _descriptor_change_time(descriptor, source_stat),
+    )
+
+
+def _path_matches_source_identity(
+    path: Path,
+    expected_identity: tuple[int, int, int, int, int],
+    *,
+    follow_symlinks: bool,
+) -> bool:
+    try:
+        with _open_regular_descriptor(path, follow_symlinks=follow_symlinks) as descriptor:
+            return _descriptor_identity(descriptor) == expected_identity
+    except OSError:
+        return False
+
+
+@contextmanager
+def _open_regular_descriptor(path: Path, *, follow_symlinks: bool = False) -> Iterator[int]:
+    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0)
+    if not follow_symlinks:
+        flags |= getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except FileNotFoundError:
+        raise FileNotFoundError(f"GGUF file not found: {path}") from None
+    try:
+        yield descriptor
+    finally:
+        os.close(descriptor)
 
 
 def _parse_field_value(field) -> Any:
@@ -148,7 +223,7 @@ class GGUFModel:
         path: Path to the ``.gguf`` file.
     """
 
-    def __init__(self, path: str | Path) -> None:
+    def __init__(self, path: str | Path, *, follow_symlinks: bool = True) -> None:
         try:
             from gguf import GGUFReader
         except ImportError as e:
@@ -158,53 +233,154 @@ def __init__(self, path: str | Path) -> None:
             ) from e
 
         self._path = Path(path)
-        if not self._path.is_file():
-            raise FileNotFoundError(f"GGUF file not found: {self._path}")
-
-        source_identity = _stat_identity(self._path)
-        if source_identity[2] < 24:
-            raise ValueError(f"{str(self._path)!r} does not begin with a valid GGUF header.")
-        with (
-            self._path.open("rb") as stream,
-            mmap.mmap(stream.fileno(), length=0, access=mmap.ACCESS_READ) as mapped,
-        ):
-            _gguf_architecture_from_header(
-                mapped,
-                source=str(self._path),
-                require_architecture=False,
+        self._follow_source_symlinks = follow_symlinks
+        with _open_regular_descriptor(
+            self._path, follow_symlinks=follow_symlinks
+        ) as descriptor:
+            source_stat = os.fstat(descriptor)
+            if not stat.S_ISREG(source_stat.st_mode):
+                raise FileNotFoundError(f"GGUF file not found: {self._path}")
+            source_identity = _descriptor_identity(descriptor)
+            if source_stat.st_size < 24:
+                raise ValueError(
+                    f"{str(self._path)!r} does not begin with a valid GGUF header."
+                )
+            with os.fdopen(os.dup(descriptor), "rb") as stream:
+                with mmap.mmap(stream.fileno(), length=0, access=mmap.ACCESS_READ) as mapped:
+                    _gguf_architecture_from_header(
+                        mapped,
+                        source=str(self._path),
+                        require_architecture=False,
+                    )
+                stream.seek(0)
+                self._reader = GGUFReader(cast(Any, stream))
+                stream.seek(0)
+                header = stream.read(8)
+            if _descriptor_identity(descriptor) != source_identity:
+                raise ValueError("GGUF source changed while the reader was opening it")
+            with _open_regular_descriptor(
+                self._path,
+                follow_symlinks=follow_symlinks,
+            ) as path_descriptor:
+                if _descriptor_identity(path_descriptor) != source_identity:
+                    raise ValueError(
+                        "GGUF source path changed while the reader was opening it"
+                    )
+            if len(header) != 8 or header[:4] != b"GGUF":
+                raise ValueError(f"Invalid GGUF header in {self._path}")
+            little_version = int.from_bytes(header[4:], byteorder="little")
+            format_version = (
+                little_version
+                if little_version <= 0xFFFF
+                else int.from_bytes(header[4:], byteorder="big")
             )
-        self._reader = GGUFReader(str(self._path))
-        if _stat_identity(self._path) != source_identity:
-            raise ValueError("GGUF source changed while the reader was opening it")
+            # Build tensor name → index map for O(1) lookup
+            tensor_index = {t.name: i for i, t in enumerate(self._reader.tensors)}
+            source_descriptor = os.dup(descriptor)
+        self._source_descriptor: int | None = source_descriptor
+        self._source_lock = threading.Lock()
         self._source_identity = source_identity
-        with self._path.open("rb") as stream:
-            header = stream.read(8)
-        if len(header) != 8 or header[:4] != b"GGUF":
-            raise ValueError(f"Invalid GGUF header in {self._path}")
-        little_version = int.from_bytes(header[4:], byteorder="little")
-        self._format_version = (
-            little_version
-            if little_version <= 0xFFFF
-            else int.from_bytes(header[4:], byteorder="big")
-        )
+        self._format_version = format_version
         self._metadata: dict[str, Any] | None = None
-        # Build tensor name → index map for O(1) lookup
-        self._tensor_index: dict[str, int] = {
-            t.name: i for i, t in enumerate(self._reader.tensors)
-        }
+        self._tensor_index: dict[str, int] = tensor_index
 
     @property
     def format_version(self) -> int:
         """GGUF container version parsed from the file header."""
         return self._format_version
 
-    def source_matches_path(self) -> bool:
+    def source_matches_path(self, path: str | Path | None = None) -> bool:
         """Return whether the path still names the exact file opened by this reader."""
         try:
-            return _stat_identity(self._path) == self._source_identity
+            descriptor = self._source_descriptor
+            if descriptor is None or _descriptor_identity(descriptor) != self._source_identity:
+                return False
+            logical_path = self._path if path is None else Path(path)
+            with _open_regular_descriptor(
+                logical_path,
+                follow_symlinks=self._follow_source_symlinks,
+            ) as path_descriptor:
+                return _descriptor_identity(path_descriptor) == self._source_identity
         except OSError:
             return False
 
+    @property
+    def source_identity(self) -> tuple[int, int, int, int, int]:
+        """Filesystem identity captured while this reader was opened."""
+        return self._source_identity
+
+    @contextmanager
+    def open_source_descriptor(self) -> Iterator[int]:
+        """Yield a serialized descriptor duplicate pinned to this reader."""
+        with self._source_lock:
+            descriptor = self._source_descriptor
+            if descriptor is None:
+                raise ValueError("GGUF source is already closed")
+            duplicate = os.dup(descriptor)
+            try:
+                yield duplicate
+            finally:
+                os.close(duplicate)
+
+    @property
+    def source_size(self) -> int:
+        """Size of the unchanged descriptor backing this reader."""
+        descriptor = self._source_descriptor
+        if descriptor is None or _descriptor_identity(descriptor) != self._source_identity:
+            raise ValueError("GGUF source changed after its reader was opened")
+        return os.fstat(descriptor).st_size
+
+    def source_sha256(self, *, chunk_size: int = 1 << 20) -> str:
+        """Hash the unchanged descriptor backing this reader."""
+        with self.open_source_descriptor() as descriptor:
+            if _descriptor_identity(descriptor) != self._source_identity:
+                raise ValueError("GGUF source changed after its reader was opened")
+            os.lseek(descriptor, 0, os.SEEK_SET)
+            digest = hashlib.sha256()
+            while chunk := os.read(descriptor, chunk_size):
+                digest.update(chunk)
+            if _descriptor_identity(descriptor) != self._source_identity:
+                raise ValueError("GGUF source changed while its checksum was computed")
+            return digest.hexdigest()
+
+    def read_source_range(self, offset: int, length: int) -> bytes:
+        """Read one exact byte range from the unchanged pinned source."""
+        if offset < 0 or length < 0:
+            raise ValueError("GGUF source range offset and length must be non-negative")
+        with self.open_source_descriptor() as descriptor:
+            if _descriptor_identity(descriptor) != self._source_identity:
+                raise ValueError("GGUF source changed after its reader was opened")
+            os.lseek(descriptor, offset, os.SEEK_SET)
+            chunks: list[bytes] = []
+            remaining = length
+            while remaining:
+                chunk = os.read(descriptor, min(remaining, 1 << 20))
+                if not chunk:
+                    break
+                chunks.append(chunk)
+                remaining -= len(chunk)
+            if _descriptor_identity(descriptor) != self._source_identity:
+                raise ValueError("GGUF source changed while a tensor range was read")
+            return b"".join(chunks)
+
+    def close(self) -> None:
+        """Release the retained source descriptor."""
+        lock = getattr(self, "_source_lock", None)
+        if lock is None:
+            descriptor = getattr(self, "_source_descriptor", None)
+            if descriptor is not None:
+                os.close(descriptor)
+                self._source_descriptor = None
+            return
+        with lock:
+            descriptor = self._source_descriptor
+            if descriptor is not None:
+                os.close(descriptor)
+                self._source_descriptor = None
+
+    def __del__(self) -> None:
+        self.close()
+
     @property
     def architecture(self) -> str:
         """Model architecture name (e.g. ``'llama'``, ``'qwen2'``)."""
@@ -274,12 +450,12 @@ def reader_tensors(self):
         """
         return list(self._reader.tensors)
 
-    def _tensor_source(self, name: str) -> tuple[Path, int, Any]:
-        """Return the owning path, data-section offset, and reader record."""
+    def _tensor_source(self, name: str) -> tuple[Callable[[int, int], bytes], int, Any]:
+        """Return the pinned range reader, data-section offset, and tensor record."""
         if name not in self._tensor_index:
             raise KeyError(f"Tensor '{name}' not found in GGUF file.")
         return (
-            self._path,
+            self.read_source_range,
             int(self._reader.data_offset),
             self._reader.get_tensor(self._tensor_index[name]),
         )
diff --git a/src/mobius/integrations/gguf/_reader_test.py b/src/mobius/integrations/gguf/_reader_test.py
index 48ed95056..18b350bf2 100644
--- a/src/mobius/integrations/gguf/_reader_test.py
+++ b/src/mobius/integrations/gguf/_reader_test.py
@@ -9,7 +9,11 @@
 
 from __future__ import annotations
 
+import errno
+import os
 import struct
+import threading
+from concurrent.futures import ThreadPoolExecutor, TimeoutError
 from pathlib import Path
 
 import numpy as np
@@ -26,6 +30,19 @@
 )
 from mobius.integrations.gguf._header import _gguf_architecture_from_header
 from mobius.integrations.gguf._reader import GGUFModel
+from mobius.integrations.gguf._runtime_evidence import gguf_artifact_identity
+
+
+def _symlink_or_skip(link: Path, target: Path) -> None:
+    try:
+        link.symlink_to(target)
+    except OSError as error:
+        if os.name == "nt" and (
+            getattr(error, "winerror", None) in {1, 50, 1314}
+            or error.errno in {errno.EPERM, errno.EACCES, errno.ENOSYS}
+        ):
+            pytest.skip(f"Windows runner cannot create test symlinks: {error}")
+        raise
 
 
 def _raw_gguf_string(value: bytes) -> bytes:
@@ -410,6 +427,101 @@ def test_file_not_found(self, tmp_path: Path):
         with pytest.raises(FileNotFoundError, match="not found"):
             GGUFModel(tmp_path / "nonexistent.gguf")
 
+    @pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFOs are unavailable")
+    def test_rejects_fifo_without_waiting_for_a_writer(self, tmp_path: Path):
+        path = tmp_path / "source.gguf"
+        os.mkfifo(path)
+
+        with pytest.raises(FileNotFoundError, match="not found"):
+            GGUFModel(path)
+
+    def test_reader_uses_pinned_source_during_opening_time_symlink_aba(
+        self,
+        llama_gguf: Path,
+        gemma4_gguf: Path,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        import gguf
+
+        original_reader = gguf.GGUFReader
+        logical = tmp_path / "source.gguf"
+        _symlink_or_skip(logical, llama_gguf)
+
+        def replace_path_before_reader_opens(source):
+            logical.unlink()
+            _symlink_or_skip(logical, gemma4_gguf)
+            logical.unlink()
+            _symlink_or_skip(logical, llama_gguf)
+            return original_reader(source)
+
+        monkeypatch.setattr(gguf, "GGUFReader", replace_path_before_reader_opens)
+
+        model = GGUFModel(logical)
+        assert model.architecture == "llama"
+        assert model.source_matches_path()
+
+    def test_reader_fails_closed_when_symlink_is_retargeted_during_open(
+        self,
+        llama_gguf: Path,
+        gemma4_gguf: Path,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        import gguf
+
+        path = tmp_path / "source.gguf"
+        _symlink_or_skip(path, llama_gguf)
+        original_reader = gguf.GGUFReader
+
+        def retarget_symlink_before_reader_opens(source):
+            path.unlink()
+            _symlink_or_skip(path, gemma4_gguf)
+            return original_reader(source)
+
+        monkeypatch.setattr(gguf, "GGUFReader", retarget_symlink_before_reader_opens)
+
+        with pytest.raises(
+            ValueError, match="source path changed while the reader was opening"
+        ):
+            GGUFModel(path)
+
+    def test_artifact_hash_waits_for_exclusive_pinned_descriptor_access(
+        self, llama_gguf: Path
+    ):
+        model = GGUFModel(llama_gguf)
+        descriptor_held = threading.Event()
+        release_descriptor = threading.Event()
+        hash_started = threading.Event()
+
+        def hold_descriptor():
+            with model.open_source_descriptor():
+                descriptor_held.set()
+                release_descriptor.wait()
+
+        def hash_artifact():
+            hash_started.set()
+            return gguf_artifact_identity(
+                llama_gguf,
+                model,
+                architecture=model.architecture,
+            )
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            holder = executor.submit(hold_descriptor)
+            assert descriptor_held.wait(timeout=1)
+            hasher = executor.submit(hash_artifact)
+            assert hash_started.wait(timeout=1)
+            try:
+                with pytest.raises(TimeoutError):
+                    hasher.result(timeout=0.05)
+            finally:
+                release_descriptor.set()
+            holder.result(timeout=1)
+            identity = hasher.result(timeout=1)
+
+        assert identity.sha256 == model.source_sha256()
+
     def test_rejects_huge_metadata_array_before_constructing_upstream_reader(
         self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
     ):
diff --git a/src/mobius/integrations/gguf/_reuse.py b/src/mobius/integrations/gguf/_reuse.py
index b97fa761b..2332597af 100644
--- a/src/mobius/integrations/gguf/_reuse.py
+++ b/src/mobius/integrations/gguf/_reuse.py
@@ -7,7 +7,6 @@
 
 __all__ = ["GGUFReuseCandidate", "GGUFReusePlan", "verify_gguf_reuse_manifest"]
 
-import hashlib
 import json
 import os
 import re
@@ -15,7 +14,7 @@
 import uuid
 from collections.abc import Iterator, Mapping
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path, PurePosixPath, PureWindowsPath
 from typing import TYPE_CHECKING, TypedDict
 
@@ -24,6 +23,7 @@
 
 if TYPE_CHECKING:
     from mobius._model_package import ModelPackage
+    from mobius.integrations.gguf._reader import GGUFModel
 
 if os.name == "nt":
     import msvcrt
@@ -92,20 +92,14 @@ class GGUFReusePlan:
     size: int
     sha256: str
     tensors: tuple[GGUFReuseTensor, ...]
-
-
-def _sha256(path: Path) -> str:
-    digest = hashlib.sha256()
-    with path.open("rb") as stream:
-        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
-            digest.update(chunk)
-    return digest.hexdigest()
+    source_model: GGUFModel = field(compare=False, repr=False)
 
 
 def attach_reused_initializers(
     package: ModelPackage,
     source_path: str | Path,
     candidates: dict[str, GGUFReuseCandidate],
+    source_model: GGUFModel,
 ) -> None:
     """Replace eligible in-memory initializers with GGUF ExternalTensors."""
     if len(package) != 1:
@@ -166,11 +160,19 @@ def attach_reused_initializers(
             "weights use the ONNX sidecar."
         )
 
+    source_size = source_model.source_size
+    source_sha256 = source_model.source_sha256()
+    if not source_model.source_matches_path(source):
+        raise ValueError(
+            "The GGUF source changed while its reuse plan was being created. "
+            "Retry the build with an unchanged source file."
+        )
     package.gguf_reuse_plan = GGUFReusePlan(
         source_path=source,
-        size=source.stat().st_size,
-        sha256=_sha256(source),
+        size=source_size,
+        sha256=source_sha256,
         tensors=tuple(sorted(reused, key=lambda tensor: tensor.initializer)),
+        source_model=source_model,
     )
 
 
@@ -338,21 +340,10 @@ def _insert_external_transform(
     graph.insert_before(earliest_consumer, nodes)
 
 
-def _source_identity(path: Path) -> tuple[int, int, int, int, int]:
-    source_stat = path.stat()
-    return (
-        source_stat.st_dev,
-        source_stat.st_ino,
-        source_stat.st_size,
-        source_stat.st_mtime_ns,
-        source_stat.st_ctime_ns,
-    )
-
-
 def _validate_source(
     plan: GGUFReusePlan,
     output_directory: Path,
-) -> tuple[int, int, int, int, int]:
+) -> None:
     source = plan.source_path
     if source.is_symlink():
         raise ValueError("The GGUF source became a symlink; refusing an unsafe external path.")
@@ -374,21 +365,21 @@ def _validate_source(
                 f"The GGUF source is hard-linked to generated artifact "
                 f"{generated_name!r}. Use an independent real file."
             )
-    identity = _source_identity(source)
-    if identity[2] != plan.size:
+    if plan.source_model.source_size != plan.size:
         raise ValueError(
             "The GGUF source no longer matches the file used to build this package "
             "(size changed). Rebuild from the intended GGUF."
         )
-    return identity
+    if not plan.source_model.source_matches_path(source):
+        raise ValueError(
+            "The GGUF source changed while this package was being prepared. "
+            "Retry the save with an unchanged source file."
+        )
 
 
-def _require_source_identity(
-    plan: GGUFReusePlan,
-    expected: tuple[int, int, int, int, int],
-) -> None:
+def _require_source_identity(plan: GGUFReusePlan) -> None:
     source = plan.source_path
-    if source.is_symlink() or _source_identity(source) != expected:
+    if source.is_symlink() or not plan.source_model.source_matches_path(source):
         raise ValueError(
             "The GGUF source changed while this package was being prepared. "
             "Retry the save with an unchanged source file."
@@ -828,10 +819,10 @@ def save_reuse_package(
     path = Path(path)
     # Pin cheap file identity before taking the package-wide writer lock. The
     # final verifier hashes the source exactly once immediately before publish.
-    source_identity = _validate_source(plan, path.parent)
+    _validate_source(plan, path.parent)
     with _package_lock(path.parent):
         _recover_transaction_locked(path.parent)
-        _require_source_identity(plan, source_identity)
+        _require_source_identity(plan)
         token = uuid.uuid4().hex
         staged_model = path.with_name(f".{path.name}.{token}.tmp")
         staged_sidecar = path.with_name(f".{_SIDECAR_NAME}.{token}.tmp")
@@ -891,6 +882,7 @@ def save_reuse_package(
             }
             if memory_initializers:
                 replacements[final_sidecar] = staged_sidecar
+            _require_source_identity(plan)
             _replace_artifacts_locked(
                 replacements,
                 (path, final_sidecar, final_manifest),
@@ -966,42 +958,56 @@ def verify_gguf_reuse_manifest(
     source = root / location
     if source.is_symlink() or not source.is_file():
         raise ValueError(f"GGUF manifest source is missing or unsafe: {source}")
-    if (
-        source.stat().st_size != source_info["size"]
-        or _sha256(source) != source_info["sha256"]
-    ):
-        raise ValueError("GGUF source identity mismatch (size or SHA-256).")
-    reused_entries = manifest["reused_tensors"]
-    converted_names = manifest["converted_tensors"]
-    if len({entry["initializer"] for entry in reused_entries}) != len(reused_entries):
-        raise ValueError("GGUF reuse manifest contains duplicate initializer routes.")
-    if len(set(converted_names)) != len(converted_names):
-        raise ValueError("GGUF reuse manifest contains duplicate converted tensors.")
-    reused_names = {entry["initializer"] for entry in reused_entries}
-    if reused_names.intersection(converted_names):
-        raise ValueError("GGUF reused and converted initializer routes must be disjoint.")
+    from mobius.integrations.gguf._reader import GGUFModel, _path_matches_source_identity
 
-    from mobius.integrations.gguf._reader import GGUFModel
-
-    gguf_model = GGUFModel(source)
-    reused_by_name = {entry["initializer"]: entry for entry in reused_entries}
-    for tensor in reused_entries:
-        if tensor["offset"] < 0 or tensor["length"] <= 0:
-            raise ValueError(f"Invalid GGUF range for {tensor['initializer']!r}.")
-        if tensor["offset"] + tensor["length"] > source_info["size"]:
-            raise ValueError(f"GGUF range exceeds the source for {tensor['initializer']!r}.")
-        actual_offset, actual_length, actual_qtype = gguf_model.tensor_storage_range(
-            tensor["source_tensor"]
-        )
-        if (actual_offset, actual_length, actual_qtype) != (
-            tensor["offset"],
-            tensor["length"],
-            tensor["qtype"],
+    try:
+        gguf_model = GGUFModel(source, follow_symlinks=False)
+    except OSError as error:
+        raise ValueError(f"GGUF manifest source is missing or unsafe: {source}") from error
+    try:
+        if source.is_symlink():
+            raise ValueError(f"GGUF manifest source is missing or unsafe: {source}")
+        if (
+            gguf_model.source_size != source_info["size"]
+            or gguf_model.source_sha256() != source_info["sha256"]
         ):
-            raise ValueError(
-                f"Manifest GGUF route does not match source tensor "
-                f"{tensor['source_tensor']!r}."
+            raise ValueError("GGUF source identity mismatch (size or SHA-256).")
+        reused_entries = manifest["reused_tensors"]
+        converted_names = manifest["converted_tensors"]
+        if len({entry["initializer"] for entry in reused_entries}) != len(reused_entries):
+            raise ValueError("GGUF reuse manifest contains duplicate initializer routes.")
+        if len(set(converted_names)) != len(converted_names):
+            raise ValueError("GGUF reuse manifest contains duplicate converted tensors.")
+        reused_names = {entry["initializer"] for entry in reused_entries}
+        if reused_names.intersection(converted_names):
+            raise ValueError("GGUF reused and converted initializer routes must be disjoint.")
+
+        reused_by_name = {entry["initializer"]: entry for entry in reused_entries}
+        for tensor in reused_entries:
+            if tensor["offset"] < 0 or tensor["length"] <= 0:
+                raise ValueError(f"Invalid GGUF range for {tensor['initializer']!r}.")
+            if tensor["offset"] + tensor["length"] > source_info["size"]:
+                raise ValueError(
+                    f"GGUF range exceeds the source for {tensor['initializer']!r}."
+                )
+            actual_offset, actual_length, actual_qtype = gguf_model.tensor_storage_range(
+                tensor["source_tensor"]
             )
+            if (actual_offset, actual_length, actual_qtype) != (
+                tensor["offset"],
+                tensor["length"],
+                tensor["qtype"],
+            ):
+                raise ValueError(
+                    f"Manifest GGUF route does not match source tensor "
+                    f"{tensor['source_tensor']!r}."
+                )
+        source_matches = gguf_model.source_matches_path()
+        if not source_matches or source.is_symlink():
+            raise ValueError("GGUF source changed while its reuse manifest was verified.")
+        source_identity = gguf_model.source_identity
+    finally:
+        gguf_model.close()
 
     model_file = Path(model_path) if model_path is not None else root / "model.onnx"
     _require_regular_or_missing(model_file, artifact="model")
@@ -1107,6 +1113,13 @@ def verify_gguf_reuse_manifest(
         if start < previous_end:
             raise ValueError(f"Overlapping sidecar range for initializer {name!r}.")
         previous_end = end
+    source_matches = _path_matches_source_identity(
+        source,
+        source_identity,
+        follow_symlinks=False,
+    )
+    if not source_matches or source.is_symlink():
+        raise ValueError("GGUF source changed while its reuse manifest was verified.")
 
 
 def _verify_transform_graph(
diff --git a/src/mobius/integrations/gguf/_runtime_evidence.py b/src/mobius/integrations/gguf/_runtime_evidence.py
index 15b8012ae..69027ede6 100644
--- a/src/mobius/integrations/gguf/_runtime_evidence.py
+++ b/src/mobius/integrations/gguf/_runtime_evidence.py
@@ -28,6 +28,8 @@
 from types import MappingProxyType
 from typing import Any
 
+from mobius.integrations.gguf._reader import _descriptor_identity
+
 
 @dataclasses.dataclass(frozen=True, slots=True)
 class GGUFArtifactIdentity:
@@ -668,21 +670,49 @@ def gguf_artifact_identity(
     """Fingerprint source bytes and parsed tensor census under a canonical architecture."""
     shard_paths = getattr(gguf_model, "shard_paths", None)
     if shard_paths is None:
-        stat, sha256 = _hash_regular_file(source_path)
+        open_descriptor = getattr(gguf_model, "open_source_descriptor", None)
+        if callable(open_descriptor):
+            with open_descriptor() as descriptor:
+                stat, sha256 = _hash_regular_descriptor(
+                    descriptor,
+                    path=source_path,
+                    expected_identity=gguf_model.source_identity,
+                )
+        else:
+            stat, sha256 = _hash_regular_file(source_path)
         size = stat.st_size
     else:
         paths = tuple(Path(path) for path in shard_paths)
         identity_paths = tuple(
             Path(path) for path in getattr(gguf_model, "identity_paths", paths)
         )
+        source_identities = getattr(gguf_model, "source_identities", None)
         if not paths:
             raise ValueError("A GGUF shard set must contain at least one source file.")
         if len(identity_paths) != len(paths):
             raise ValueError("GGUF shard identity paths must match the shard set length.")
+        if source_identities is None or len(source_identities) != len(paths):
+            raise ValueError(
+                "GGUF shard source identities must be captured for the complete shard set."
+            )
         digest = hashlib.sha256()
         size = 0
-        for path, identity_path in zip(paths, identity_paths):
-            stat, file_sha256 = _hash_regular_file(identity_path)
+        open_descriptor = getattr(gguf_model, "open_source_descriptor", None)
+        for index, (path, identity_path, source_identity) in enumerate(
+            zip(paths, identity_paths, source_identities)
+        ):
+            if callable(open_descriptor):
+                with open_descriptor(index) as descriptor:
+                    stat, file_sha256 = _hash_regular_descriptor(
+                        descriptor,
+                        path=identity_path,
+                        expected_identity=source_identity,
+                    )
+            else:
+                stat, file_sha256 = _hash_regular_file(
+                    identity_path,
+                    expected_identity=source_identity,
+                )
             encoded = path.name.encode("utf-8")
             digest.update(len(encoded).to_bytes(8, "big"))
             digest.update(encoded)
@@ -728,24 +758,56 @@ def gguf_graph_package_identity(package_dir: Path) -> GGUFGraphPackageIdentity:
     return GGUFGraphPackageIdentity(files=tuple(names), sha256=digest.hexdigest())
 
 
-def _hash_regular_file(path: Path) -> tuple[os.stat_result, str]:
+def _hash_regular_file(
+    path: Path,
+    *,
+    expected_identity: tuple[int, int, int, int, int] | None = None,
+) -> tuple[os.stat_result, str]:
     """Hash one non-symlink regular file through the descriptor being validated."""
-    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    flags = (
+        os.O_RDONLY
+        | getattr(os, "O_BINARY", 0)
+        | getattr(os, "O_NOFOLLOW", 0)
+        | getattr(os, "O_NONBLOCK", 0)
+    )
     descriptor = os.open(path, flags)
     try:
-        before = os.fstat(descriptor)
-        if not stat.S_ISREG(before.st_mode) or path.is_symlink():
+        stat_result, sha256 = _hash_regular_descriptor(
+            descriptor,
+            path=path,
+            expected_identity=expected_identity,
+        )
+        if path.is_symlink():
             raise ValueError(f"Expected a non-symlink regular file: {path}")
-        digest = hashlib.sha256()
-        while chunk := os.read(descriptor, 1024 * 1024):
-            digest.update(chunk)
-        after = os.fstat(descriptor)
-
-        def identity(value: os.stat_result) -> tuple[int, int, int, int]:
-            return (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns)
-
-        if identity(before) != identity(after) or identity(after) != identity(path.stat()):
-            raise ValueError(f"File changed while its immutable identity was computed: {path}")
-        return after, digest.hexdigest()
+        path_descriptor = os.open(path, flags)
+        try:
+            if _descriptor_identity(descriptor) != _descriptor_identity(path_descriptor):
+                raise ValueError(f"File changed while its identity was computed: {path}")
+        finally:
+            os.close(path_descriptor)
+        return stat_result, sha256
     finally:
         os.close(descriptor)
+
+
+def _hash_regular_descriptor(
+    descriptor: int,
+    *,
+    path: Path,
+    expected_identity: tuple[int, int, int, int, int] | None,
+) -> tuple[os.stat_result, str]:
+    """Hash the exact descriptor retained by a GGUF reader."""
+    os.lseek(descriptor, 0, os.SEEK_SET)
+    before = os.fstat(descriptor)
+    before_identity = _descriptor_identity(descriptor)
+    if not stat.S_ISREG(before.st_mode):
+        raise ValueError(f"Expected a regular GGUF source file: {path}")
+    if expected_identity is not None and before_identity != expected_identity:
+        raise ValueError(f"File no longer matches the opened GGUF source identity: {path}")
+    digest = hashlib.sha256()
+    while chunk := os.read(descriptor, 1024 * 1024):
+        digest.update(chunk)
+    after = os.fstat(descriptor)
+    if before_identity != _descriptor_identity(descriptor):
+        raise ValueError(f"File changed while its immutable identity was computed: {path}")
+    return after, digest.hexdigest()
diff --git a/src/mobius/integrations/gguf/_runtime_evidence_test.py b/src/mobius/integrations/gguf/_runtime_evidence_test.py
index b9c3bcc86..db39a1df8 100644
--- a/src/mobius/integrations/gguf/_runtime_evidence_test.py
+++ b/src/mobius/integrations/gguf/_runtime_evidence_test.py
@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import hashlib
+import os
 from collections import Counter
 from dataclasses import replace
 from types import MappingProxyType, SimpleNamespace
@@ -16,6 +17,7 @@
 from mobius._builder import build_from_module
 from mobius._configs import NemotronHConfig
 from mobius.integrations.gguf import _runtime_evidence
+from mobius.integrations.gguf._reader import _descriptor_identity
 from mobius.integrations.gguf._runtime_blocker_evidence import (
     iter_runtime_blocker_evidence,
     runtime_blocker_evidence,
@@ -32,6 +34,14 @@
 from mobius.tasks import HybridCausalLMTask
 
 
+def _file_identity(path) -> tuple[int, int, int, int, int]:
+    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_BINARY", 0))
+    try:
+        return _descriptor_identity(descriptor)
+    finally:
+        os.close(descriptor)
+
+
 def _pinned_nemotron_h_config() -> NemotronHConfig:
     pattern = "MEMEM*EMEMEM*EMEMEM*EMEMEM*EMEMEM*EMEMEMEM*EMEMEMEME"
     layer_types = {
@@ -303,6 +313,10 @@ def test_sharded_artifact_identity_frames_every_shard_and_tensor(tmp_path) -> No
     ]
     model = SimpleNamespace(
         shard_paths=[first, second],
+        source_identities=[
+            _file_identity(first),
+            _file_identity(second),
+        ],
         reader_tensors=lambda: tensors,
     )
 
@@ -319,13 +333,13 @@ def test_sharded_artifact_identity_frames_every_shard_and_tensor(tmp_path) -> No
     assert identity.tensor_qtypes == (("F32", 1), ("Q4_K", 1))
 
     second.write_bytes(b"change")
-    changed = gguf_artifact_identity(
-        first,
-        model,
-        architecture="llama",
-        filename=first.name,
-    )
-    assert changed.sha256 != identity.sha256
+    with pytest.raises(ValueError, match="no longer matches the opened GGUF source identity"):
+        gguf_artifact_identity(
+            first,
+            model,
+            architecture="llama",
+            filename=first.name,
+        )
 
 
 def test_sharded_artifact_identity_hashes_regular_aliases_for_snapshot_links(
@@ -347,6 +361,10 @@ def test_sharded_artifact_identity_hashes_regular_aliases_for_snapshot_links(
     model = SimpleNamespace(
         shard_paths=[first, second],
         identity_paths=[first_blob, second_blob],
+        source_identities=[
+            _file_identity(first),
+            _file_identity(second),
+        ],
         reader_tensors=lambda: tensors,
     )
 
diff --git a/src/mobius/integrations/gguf/_shard_set.py b/src/mobius/integrations/gguf/_shard_set.py
index 2c9d2a5e7..602dde969 100644
--- a/src/mobius/integrations/gguf/_shard_set.py
+++ b/src/mobius/integrations/gguf/_shard_set.py
@@ -47,10 +47,10 @@
     "SHARD_FILENAME_RE",
 ]
 
-import hashlib
 import logging
 import re
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -336,6 +336,10 @@ def __init__(
             expected_sizes=expected_sizes,
         )
         _validate_shard_set(self._infos, shards)
+        if not all(shard.source_matches_path() for shard in shards):
+            raise GgufShardError(
+                "GGUF shard source changed while its manifest was being built."
+            )
 
         # Order shards by their authoritative ``split.no`` (order independence:
         # the caller may pass them in any order). Primary shard is split.no == 0.
@@ -472,8 +476,8 @@ def tensor_storage_range(self, name: str) -> tuple[int, int, str]:
             raise KeyError(f"Tensor {name!r} not found in split set.")
         return shard.tensor_storage_range(name)
 
-    def _tensor_source(self, name: str) -> tuple[Path, int, Any]:
-        """Return the owning path, data-section offset, and reader record."""
+    def _tensor_source(self, name: str) -> tuple[Callable[[int, int], bytes], int, Any]:
+        """Return the owning pinned range reader, data offset, and tensor record."""
         shard = self._owner.get(name)
         if shard is None:
             raise KeyError(f"Tensor {name!r} not found in split set.")
@@ -503,6 +507,17 @@ def identity_paths(self) -> list[Path]:
         """Regular files used for immutable identity hashing, in shard order."""
         return list(self._identity_paths)
 
+    @property
+    def source_identities(self) -> list[tuple[int, int, int, int, int]]:
+        """Filesystem identities captured by the opened shard readers."""
+        return [shard.source_identity for shard in self._shards]
+
+    @contextmanager
+    def open_source_descriptor(self, index: int) -> Iterator[int]:
+        """Yield serialized access to one pinned shard descriptor."""
+        with self._shards[index].open_source_descriptor() as descriptor:
+            yield descriptor
+
     def _set_identity_paths(self, paths: list[Path]) -> None:
         """Bind trusted regular-file aliases of the opened shard sources."""
         if len(paths) != len(self._paths):
@@ -537,14 +552,6 @@ def __repr__(self) -> str:
         )
 
 
-def _sha256_of(path: Path, *, chunk_size: int = 1 << 20) -> str:
-    digest = hashlib.sha256()
-    with path.open("rb") as handle:
-        for block in iter(lambda: handle.read(chunk_size), b""):
-            digest.update(block)
-    return digest.hexdigest()
-
-
 def _build_shard_infos(
     paths: list[Path],
     shards: list[GGUFModel],
@@ -567,7 +574,7 @@ def _build_shard_infos(
             index, count = split_no + 1, split_count
         else:
             _prefix, index, count = parsed
-        size_bytes = path.stat().st_size
+        size_bytes = shard.source_size
 
         expected_size = (expected_sizes or {}).get(path.name)
         if expected_size is not None and expected_size != size_bytes:
@@ -580,7 +587,7 @@ def _build_shard_infos(
         sha256: str | None = None
         want_sha = (expected_sha256 or {}).get(path.name)
         if verify_checksums:
-            sha256 = _sha256_of(path)
+            sha256 = shard.source_sha256()
             if want_sha is not None and sha256.lower() != want_sha.lower():
                 raise GgufShardError(
                     f"Shard {path.name} SHA-256 mismatch: manifest expected "
@@ -612,6 +619,10 @@ def _int_or_none(value: Any) -> int | None:
         return None
 
 
+def _format_filenames(names: list[str]) -> str:
+    return ", ".join(repr(name) for name in sorted(names))
+
+
 def _validate_shard_set(infos: list[ShardInfo], shards: list[GGUFModel]) -> None:
     """Fail closed on any structural inconsistency in the split set."""
     n = len(infos)
@@ -641,7 +652,8 @@ def _validate_shard_set(infos: list[ShardInfo], shards: list[GGUFModel]) -> None
     missing_counts = [info.path.name for info in infos if info.split_count is None]
     if missing_counts:
         raise GgufShardError(
-            f"Every shard must declare split.count; missing from {missing_counts}."
+            "Every shard must declare split.count; missing from "
+            f"{_format_filenames(missing_counts)}."
         )
     declared_counts = {info.split_count for info in infos}
     if declared_counts != {filename_count}:
@@ -653,7 +665,9 @@ def _validate_shard_set(infos: list[ShardInfo], shards: list[GGUFModel]) -> None
     # split.no must be a contiguous 0..count-1 permutation when present.
     missing_nos = [info.path.name for info in infos if info.split_no is None]
     if missing_nos:
-        raise GgufShardError(f"Every shard must declare split.no; missing from {missing_nos}.")
+        raise GgufShardError(
+            f"Every shard must declare split.no; missing from {_format_filenames(missing_nos)}."
+        )
     split_nos = [info.split_no for info in infos]
     if sorted(split_nos) != list(range(n)):
         raise GgufShardError(
@@ -678,7 +692,7 @@ def _validate_shard_set(infos: list[ShardInfo], shards: list[GGUFModel]) -> None
     if missing_tensor_totals:
         raise GgufShardError(
             "Every shard must declare split.tensors.count; missing from "
-            f"{missing_tensor_totals}."
+            f"{_format_filenames(missing_tensor_totals)}."
         )
     declared_tensor_totals = {info.split_tensors_count for info in infos}
     observed_total = sum(info.tensor_count for info in infos)
diff --git a/src/mobius/integrations/gguf/_shard_set_test.py b/src/mobius/integrations/gguf/_shard_set_test.py
index b316900c3..a161aa4b9 100644
--- a/src/mobius/integrations/gguf/_shard_set_test.py
+++ b/src/mobius/integrations/gguf/_shard_set_test.py
@@ -10,7 +10,9 @@
 
 from __future__ import annotations
 
+import errno
 import hashlib
+import os
 import tracemalloc
 from pathlib import Path, PurePosixPath
 from types import SimpleNamespace
@@ -25,12 +27,25 @@
     GgufShardManifest,
     GgufShardSet,
     _merge_metadata,
+    _validate_shard_set,
     discover_gguf_shards,
     open_gguf_model,
     parse_shard_filename,
 )
 
 
+def _symlink_or_skip(link: Path, target: Path) -> None:
+    try:
+        link.symlink_to(target)
+    except OSError as error:
+        if os.name == "nt" and (
+            getattr(error, "winerror", None) in {1, 50, 1314}
+            or error.errno in {errno.EPERM, errno.EACCES, errno.ENOSYS}
+        ):
+            pytest.skip(f"Windows runner cannot create test symlinks: {error}")
+        raise
+
+
 def _sha256(path: Path) -> str:
     return hashlib.sha256(path.read_bytes()).hexdigest()
 
@@ -380,6 +395,37 @@ def test_checksum_mismatch_rejected(tmp_path):
         open_gguf_model(shards[0], verify_checksums=True, expected_sha256=bad)
 
 
+def test_checksum_rejects_shard_path_replacement_after_reader_open(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+):
+    shards = _write_sharded_gguf(tmp_path / "sources", split_max_tensors=3)
+    replacements = _write_sharded_gguf(
+        tmp_path / "replacements",
+        split_max_tensors=3,
+        seed=1,
+    )
+    links_dir = tmp_path / "links"
+    links_dir.mkdir()
+    links = []
+    for shard in shards:
+        link = links_dir / shard.name
+        _symlink_or_skip(link, shard)
+        links.append(link)
+    source_sha256 = GGUFModel.source_sha256
+
+    def replace_path_before_hash(model, **kwargs):
+        digest = source_sha256(model, **kwargs)
+        if Path(model._path) == links[0]:
+            links[0].unlink()
+            _symlink_or_skip(links[0], replacements[0])
+        return digest
+
+    monkeypatch.setattr(GGUFModel, "source_sha256", replace_path_before_hash)
+
+    with pytest.raises(GgufShardError, match="source changed while its manifest"):
+        open_gguf_model(links[0], verify_checksums=True)
+
+
 def test_manifest_rejects_swapped_continuation_shard(tmp_path):
     """A same-model mixed-quant continuation-shard swap is caught by the manifest.
 
@@ -489,6 +535,47 @@ def test_continuation_may_repeat_primary_semantics_but_cannot_override_them(tmp_
         )
 
 
+@pytest.mark.parametrize(
+    ("missing_field", "metadata_name"),
+    [
+        ("split_count", "split.count"),
+        ("split_no", "split.no"),
+        ("split_tensors_count", "split.tensors.count"),
+    ],
+)
+def test_missing_split_metadata_errors_sort_filenames(missing_field, metadata_name):
+    infos = [
+        SimpleNamespace(
+            path=Path("tiny-00002-of-00002.gguf"),
+            filename_index=2,
+            filename_count=2,
+            split_no=1,
+            split_count=2,
+            split_tensors_count=2,
+            tensor_count=1,
+        ),
+        SimpleNamespace(
+            path=Path("tiny-00001-of-00002.gguf"),
+            filename_index=1,
+            filename_count=2,
+            split_no=0,
+            split_count=2,
+            split_tensors_count=2,
+            tensor_count=1,
+        ),
+    ]
+    for info in infos:
+        setattr(info, missing_field, None)
+
+    with pytest.raises(GgufShardError) as caught:
+        _validate_shard_set(infos, [mock.Mock(), mock.Mock()])
+
+    assert str(caught.value) == (
+        f"Every shard must declare {metadata_name}; missing from "
+        "'tiny-00001-of-00002.gguf', 'tiny-00002-of-00002.gguf'."
+    )
+
+
 def test_open_validates_primary_metadata_authority_eagerly(tmp_path):
     from mobius.integrations.gguf import _shard_set
 
@@ -1094,22 +1181,147 @@ def test_hub_download_space_counts_only_uncached_shards(tmp_path):
     assert preflight.call_args.args == (missing.stat().st_size,)
 
 
-def test_hub_cache_identity_paths_resolve_only_inside_cache(tmp_path):
+def test_regular_file_identity_paths_resolve_local_symlinks(tmp_path):
     from mobius.integrations.gguf import _builder as builder
 
-    cache = tmp_path / "hub"
-    blobs = cache / "blobs"
-    snapshot = cache / "snapshots" / ("a" * 40)
-    blobs.mkdir(parents=True)
-    snapshot.mkdir(parents=True)
-    blob = blobs / "abc"
+    blob = tmp_path / "blob"
     blob.write_bytes(b"GGUF")
-    shard = snapshot / "model-00001-of-00002.gguf"
-    shard.symlink_to(blob)
+    shard = tmp_path / "model-00001-of-00002.gguf"
+    _symlink_or_skip(shard, blob)
+
+    assert builder._regular_file_identity_paths([shard]) == [blob]
+    assert builder._regular_file_identity_paths([tmp_path / "missing.gguf"]) is None
+
+
+def test_build_binds_local_symlinked_shards_to_regular_identity_paths(tmp_path):
+    from mobius.integrations.gguf import _builder as builder
+    from mobius.integrations.gguf._runtime_evidence import gguf_artifact_identity
+
+    targets = _write_sharded_gguf(tmp_path / "targets")
+    links_dir = tmp_path / "links"
+    links_dir.mkdir()
+    links = []
+    for target in targets:
+        link = links_dir / target.name
+        _symlink_or_skip(link, target)
+        links.append(link)
+
+    class _ValidationReachedError(Exception):
+        pass
+
+    def validate(model, *, source, **_kwargs):
+        assert model.identity_paths == [target.resolve() for target in targets]
+        identity = gguf_artifact_identity(
+            Path(source),
+            model,
+            architecture=model.architecture,
+        )
+        assert identity.tensor_count == sum(
+            info.tensor_count for info in model.manifest.shards
+        )
+        raise _ValidationReachedError
+
+    with (
+        mock.patch.object(builder, "_validate_gguf_model", side_effect=validate),
+        pytest.raises(_ValidationReachedError),
+    ):
+        builder.build_from_gguf(links[0])
+
+
+def test_artifact_identity_hashes_pinned_source_after_logical_symlink_retarget(tmp_path):
+    from mobius.integrations.gguf import _builder as builder
+    from mobius.integrations.gguf._runtime_evidence import gguf_artifact_identity
+
+    targets = _write_sharded_gguf(tmp_path / "targets")
+    replacements = _write_sharded_gguf(tmp_path / "replacements", seed=1)
+    links_dir = tmp_path / "links"
+    links_dir.mkdir()
+    links = []
+    for target in targets:
+        link = links_dir / target.name
+        _symlink_or_skip(link, target)
+        links.append(link)
+
+    captured = {}
+
+    class _ValidationReachedError(Exception):
+        pass
+
+    def capture(model, **_kwargs):
+        captured["model"] = model
+        raise _ValidationReachedError
 
-    with mock.patch("huggingface_hub.constants.HF_HUB_CACHE", str(cache)):
-        assert builder._hub_cache_identity_paths([shard]) == [blob]
-        assert builder._hub_cache_identity_paths([tmp_path / "outside.gguf"]) is None
+    with (
+        mock.patch.object(builder, "_validate_gguf_model", side_effect=capture),
+        pytest.raises(_ValidationReachedError),
+    ):
+        builder.build_from_gguf(links[0])
+
+    model = captured["model"]
+    bound_identity_path = model.identity_paths[0]
+    expected_identity = gguf_artifact_identity(
+        links[0],
+        model,
+        architecture=model.architecture,
+    )
+    links[0].unlink()
+    _symlink_or_skip(links[0], replacements[0])
+
+    assert bound_identity_path == targets[0]
+    assert not model.source_matches_path()
+    assert (
+        gguf_artifact_identity(
+            links[0],
+            model,
+            architecture=model.architecture,
+        )
+        == expected_identity
+    )
+
+
+def test_artifact_identity_rejects_restored_in_place_mutation_after_binding(tmp_path):
+    from mobius.integrations.gguf import _builder as builder
+    from mobius.integrations.gguf._runtime_evidence import gguf_artifact_identity
+
+    targets = _write_sharded_gguf(tmp_path / "targets")
+
+    captured = {}
+
+    class _ValidationReachedError(Exception):
+        pass
+
+    def capture(model, **_kwargs):
+        captured["model"] = model
+        raise _ValidationReachedError
+
+    with (
+        mock.patch.object(builder, "_validate_gguf_model", side_effect=capture),
+        pytest.raises(_ValidationReachedError),
+    ):
+        builder.build_from_gguf(targets[0])
+
+    model = captured["model"]
+    source_stat = targets[0].stat()
+    original_bytes = targets[0].read_bytes()
+    with targets[0].open("r+b") as stream:
+        stream.seek(-1, os.SEEK_END)
+        value = stream.read(1)
+        stream.seek(-1, os.SEEK_END)
+        stream.write(bytes([value[0] ^ 0xFF]))
+        stream.flush()
+        os.fsync(stream.fileno())
+        stream.seek(0)
+        stream.write(original_bytes)
+        stream.flush()
+        os.fsync(stream.fileno())
+    os.utime(targets[0], ns=(source_stat.st_atime_ns, source_stat.st_mtime_ns))
+
+    with pytest.raises(ValueError, match="no longer matches the opened GGUF source identity"):
+        gguf_artifact_identity(
+            targets[0],
+            model,
+            architecture=model.architecture,
+        )
 
 
 # --------------------------------------------------------------------------- #
diff --git a/src/mobius/integrations/gguf/_tencent_q1_0.py b/src/mobius/integrations/gguf/_tencent_q1_0.py
index c16ca31c7..97c1040b6 100644
--- a/src/mobius/integrations/gguf/_tencent_q1_0.py
+++ b/src/mobius/integrations/gguf/_tencent_q1_0.py
@@ -67,7 +67,7 @@
 ]
 
 import math
-import os
+from collections.abc import Callable
 
 import numpy as np
 
@@ -158,7 +158,7 @@ def tencent_q1_0_target_bits() -> int:
 
 
 def parse_tencent_q1_0_tensor(
-    file_path: str | os.PathLike,
+    read_source_range: Callable[[int, int], bytes],
     data_section_offset: int,
     tensor,
 ) -> RepackedTensor:
@@ -174,7 +174,7 @@ def parse_tencent_q1_0_tensor(
       float ``zp=1.5``. Native 2 bpw but slow on CPU EP today.
 
     Args:
-        file_path: Path to the source ``.gguf`` file.
+        read_source_range: Pinned source reader accepting byte offset and length.
         data_section_offset: Absolute byte offset where the GGUF data
             section begins (``GGUFReader.data_offset``).
         tensor: ``gguf.ReaderTensor`` for the target weight. Must have
@@ -185,7 +185,7 @@ def parse_tencent_q1_0_tensor(
         The ``bits`` field is 2 or 4 depending on the flag.
     """
     native_scales, codes_2bit, ne1, n_native = _read_tencent_blocks(
-        file_path, data_section_offset, tensor
+        read_source_range, data_section_offset, tensor
     )
     if flags.tencent_q1_0_use_native_2bit:
         return _pack_native_2bit(native_scales, codes_2bit, ne1, n_native)
@@ -193,7 +193,7 @@ def parse_tencent_q1_0_tensor(
 
 
 def _read_tencent_blocks(
-    file_path: str | os.PathLike,
+    read_source_range: Callable[[int, int], bytes],
     data_section_offset: int,
     tensor,
 ) -> tuple[np.ndarray, np.ndarray, int, int]:
@@ -219,9 +219,7 @@ def _read_tencent_blocks(
     total_bytes = ne1 * bytes_per_row
 
     abs_offset = data_section_offset + _tensor_data_offset(tensor)
-    with open(file_path, "rb") as f:
-        f.seek(abs_offset)
-        blob = f.read(total_bytes)
+    blob = read_source_range(abs_offset, total_bytes)
     if len(blob) != total_bytes:
         raise OSError(
             f"Short read for {tensor.name!r}: got {len(blob)} bytes, expected {total_bytes}"
diff --git a/src/mobius/integrations/gguf/_tencent_q1_0_test.py b/src/mobius/integrations/gguf/_tencent_q1_0_test.py
index cb126f94a..d4ab01c2e 100644
--- a/src/mobius/integrations/gguf/_tencent_q1_0_test.py
+++ b/src/mobius/integrations/gguf/_tencent_q1_0_test.py
@@ -71,7 +71,17 @@ def _round_trip(file_path: Path, ne0: int, ne1: int, codes: np.ndarray, scales:
                     _make_tencent_q1_0_block(float(scales[n, b]), codes[n, start:end].tolist())
                 )
     tensor = _FakeTensor("w", (ne0, ne1), offset=0)
-    return parse_tencent_q1_0_tensor(file_path, data_section_offset=0, tensor=tensor)
+
+    def read_source_range(offset: int, length: int) -> bytes:
+        with file_path.open("rb") as stream:
+            stream.seek(offset)
+            return stream.read(length)
+
+    return parse_tencent_q1_0_tensor(
+        read_source_range,
+        data_section_offset=0,
+        tensor=tensor,
+    )
 
 
 class TestTencentQ10DefaultInflated4Bit:
@@ -212,4 +222,4 @@ def test_rejects_unaligned_k(self, tmp_path: Path):
         """K not divisible by 512 raises in both modes."""
         tensor = _FakeTensor("w", (256, 1), offset=0)
         with pytest.raises(ValueError, match="not divisible"):
-            parse_tencent_q1_0_tensor(tmp_path / "t.bin", 0, tensor)
+            parse_tencent_q1_0_tensor(lambda _offset, _length: b"", 0, tensor)