Cache validated source generation receipts

src/docforge/project.py

@@ -88,6 +88,17 @@ class _CapturedGeneration:
     directories: tuple[tuple[str, int, int, int, int, int], ...]
 
 
+@dataclass(frozen=True)
+class _ParsedGenerationReceipt:
+    signature: tuple[int, int, int, int, int]
+    source_hash: str
+    revision: str
+    files: tuple[tuple[object, ...], ...]
+    directories: tuple[tuple[object, ...], ...]
+    file_paths: tuple[Path, ...]
+    directory_paths: tuple[Path, ...]
+
+
 def project_root_fingerprint(root: Path) -> str:
     return hashlib.sha256(str(root).encode()).hexdigest()[:16]
 
@@ -183,6 +194,22 @@ def _receipt_paths(root: Path, value: object, *, width: int) -> tuple[Path, ...]
     return tuple(paths)
 
 
+def _receipt_signature(path: Path) -> tuple[int, int, int, int, int] | None:
+    try:
+        status = path.lstat()
+    except OSError:
+        return None
+    if not stat.S_ISREG(status.st_mode):
+        return None
+    return (
+        status.st_dev,
+        status.st_ino,
+        status.st_size,
+        status.st_mtime_ns,
+        status.st_ctime_ns,
+    )
+
+
 def _load_descriptor(root: Path) -> ProjectDescriptor:
     descriptor_path = root / ".docforge" / "project.toml"
     if not descriptor_path.is_file():
@@ -698,6 +725,7 @@ class Project:
     def __init__(self, descriptor: ProjectDescriptor) -> None:
         self.descriptor = descriptor
         self._captured_generation: _CapturedGeneration | None = None
+        self._generation_receipt_cache: _ParsedGenerationReceipt | None = None
 
     @classmethod
     def open(cls, project_root: str | Path) -> Project:
@@ -820,17 +848,54 @@ class Project:
 
     def _incremental_state(self) -> ProjectState | None:
         path = self.generation_path
-        if not path.is_file() or path.is_symlink():
+        signature = _receipt_signature(path)
+        if signature is None:
+            self._generation_receipt_cache = None
             return None
+        receipt = self._generation_receipt_cache
+        if receipt is None or receipt.signature != signature:
+            receipt = self._parse_generation_receipt(path, signature)
+            self._generation_receipt_cache = receipt
+        if receipt is None:
+            return None
+        try:
+            current_directories = _directory_generation(
+                self.descriptor.root,
+                receipt.directory_paths,
+            )
+        except DocForgeError:
+            return None
+        if receipt.directories != cast(tuple[tuple[object, ...], ...], current_directories):
+            return None
+        try:
+            current_files = _file_generation(self.descriptor.root, receipt.file_paths)
+        except DocForgeError:
+            return None
+        if receipt.files != cast(tuple[tuple[object, ...], ...], current_files):
+            return None
+        if _revision(self.descriptor.root) != receipt.revision:
+            return None
+        return ProjectState(
+            source_hash=receipt.source_hash,
+            revision=receipt.revision,
+        )
+
+    def _parse_generation_receipt(
+        self,
+        path: Path,
+        signature: tuple[int, int, int, int, int],
+    ) -> _ParsedGenerationReceipt | None:
         try:
             parsed: object = json.loads(path.read_text(encoding="utf-8"))
         except (OSError, UnicodeDecodeError, json.JSONDecodeError):
             return None
-        if not isinstance(parsed, dict):
+        if _receipt_signature(path) != signature or not isinstance(parsed, dict):
             return None
         payload = cast(dict[str, object], parsed)
         source_hash = payload.get("source_hash")
         revision = payload.get("revision")
+        files_value = payload.get("files")
+        directories_value = payload.get("directories")
         if (
             payload.get("schema_version") != SOURCE_GENERATION_SCHEMA_VERSION
             or payload.get("source_contract") != GENERIC_SOURCE_CONTRACT
@@ -841,35 +906,39 @@ class Project:
             or not isinstance(source_hash, str)
             or len(source_hash) != 64
             or not isinstance(revision, str)
+            or not isinstance(files_value, list)
+            or not isinstance(directories_value, list)
         ):
             return None
         directory_paths = _receipt_paths(
             self.descriptor.root,
-            payload.get("directories"),
+            cast(list[object], directories_value),
             width=6,
         )
         file_paths = _receipt_paths(
             self.descriptor.root,
-            payload.get("files"),
+            cast(list[object], files_value),
             width=7,
         )
         if directory_paths is None or file_paths is None:
             return None
-        try:
-            current_directories = _directory_generation(self.descriptor.root, directory_paths)
-        except DocForgeError:
-            return None
-        if payload.get("directories") != [list(identity) for identity in current_directories]:
-            return None
-        try:
-            current_files = _file_generation(self.descriptor.root, file_paths)
-        except DocForgeError:
-            return None
-        if payload.get("files") != [list(identity) for identity in current_files]:
-            return None
-        if _revision(self.descriptor.root) != revision:
-            return None
-        return ProjectState(source_hash=source_hash, revision=revision)
+        return _ParsedGenerationReceipt(
+            signature=signature,
+            source_hash=source_hash,
+            revision=revision,
+            files=tuple(
+                tuple(cast(list[object], item))
+                for item in cast(list[object], files_value)
+                if isinstance(item, list)
+            ),
+            directories=tuple(
+                tuple(cast(list[object], item))
+                for item in cast(list[object], directories_value)
+                if isinstance(item, list)
+            ),
+            file_paths=file_paths,
+            directory_paths=directory_paths,
+        )
 
     def record_generation(self, snapshot: ProjectSnapshot) -> None:
         """Persist a generation only after its complete derived index was verified."""