Add persistent source generations

ACTIVE_SLICE.md

@@ -1,13 +1,12 @@
-# Milestone status
+# Active milestone
 
 ```text
-Milestone: 0 — successor foundation and measured baseline
+Milestone: 1 — fast, observable core
-Goal: Seed DocForge2 from the most advanced local lineage without breaking DocForge v1 contracts.
+Goal: Make warm retrieval immediate by removing repeated whole-project work without changing graph meaning.
-In scope: Complete Git lineage; verified no-AST work; adapter lifecycle safeguards; compatibility guarantees; repository-native contract and quality gates; cold/warm, memory, rendering, and response-size baselines; public successor migration; fresh-clone verification.
+In scope: Structured profiling; immutable request snapshots; duplicate-check elimination; linear validation; indexed traversal; compact receipts and bounded pagination where measurements require them; receipt-based status; persistent source generations; cheap no-change detection.
-Out of scope: Storage redesign; compiler or renderer redesign; portable render plans; self-hosting; production MCP repointing; WorldForge or ScrapeStation changes; tags and releases.
+Out of scope: Speculative storage replacement; task-shaped agent retrieval; independent render-plan packages; adapter SDK expansion; self-hosting; WorldForge or ScrapeStation changes; production MCP repointing; tags and releases.
-Done when: administrator/DocForge2 is public and seeded from the advanced clean tree; administrator/DocForge remains intact; origin and legacy identify the successor and v1 remotes; all gates and a fresh-clone proof pass; the recorded baseline identifies measured bottlenecks without speculative optimization.
+Done when: Routine warm reads parse zero canonical sources; exact, search, traversal, context, synchronization, and status paths are bounded and measured; stale and corrupt state still fail closed or recover safely; legacy and no-AST adapters remain compatible; the complete repository gate passes.
-Status: Complete. DocForge2 is public and seeded from the complete advanced lineage. Compatibility and quality gates pass locally and from an anonymous fresh clone. The legacy repository and production MCP bindings remain unchanged.
+Status: Active. Repository audits and design reconciliation are in progress.
 ```
 
-No later milestone is active. `main` is the verified Milestone 0 state. `dev` begins at the same
+Milestones 2–5 remain directional context and are not active.
-commit and remains inactive until the next milestone is explicitly opened.

DEVELOPMENT_NOTES.md

@@ -0,0 +1,159 @@
+# DocForge2 development notes
+
+This is the running implementation record for DocForge2. It records what is active, what was
+measured, what changed, what failed, why architectural decisions were made, and which ideas were
+deferred. Stable user and compatibility contracts still belong in dedicated documentation.
+
+## Working rules
+
+- Only one milestone is active at a time.
+- `main` remains the last fully verified milestone.
+- Active implementation occurs on `dev`.
+- Every milestone begins from direct repository evidence and ends with focused tests, the complete
+  repository gate, updated measurements, documentation closeout, and a clean pushed state.
+- WorldForge, ScrapeStation, legacy DocForge, and production MCP bindings remain out of scope.
+- DocForge2 does not self-host during this program.
+- Release tags and Forgejo releases require Rob's explicit approval.
+
+## Milestone 0 — complete
+
+Milestone 0 established the public successor, preserved the complete lineage and v1 tag, integrated
+the no-AST and adapter-lifecycle work, froze compatibility guarantees, added repository-native
+quality and contract gates, and recorded cold/warm performance, memory, rendering, and response
+sizes.
+
+The central measurement was decisive: a 1,000-node warm exact lookup took about 286 ms while the
+generation-pinned SQLite query path took about 0.4–1.4 ms. Repeated whole-source loading and
+validation, not SQLite, is the first optimization target.
+
+## Milestone 1 — active: fast, observable core
+
+### Outcome
+
+Warm retrieval should disappear into normal tool overhead. Routine reads must not parse project
+sources. Status must not render or rebuild hidden work. Results must remain bounded independently
+of project size.
+
+### Starting evidence
+
+- Generic `Project.load()` walks, captures, parses, rereads, validates, hashes, and checks Git for
+  the complete source set.
+- Exact retrieval validates twice around one bounded SQLite query.
+- Context compilation performs three full project loads.
+- Render status recompiles the complete manual.
+- Incremental adapters already prove that manifest attestation can make no-change synchronization
+  and exact retrieval sub-millisecond on a tiny fixture.
+- Pinned viewer queries prove the current SQLite schema can serve bounded reads quickly.
+
+### Current work
+
+1. Audit request-scoped immutable snapshot and persistent-generation options.
+2. Audit result receipts, pagination, bounded response contracts, and side-effect-free status.
+3. Audit graph validation complexity, indexed traversal, profiling, and zero-source-parse proofs.
+4. Reconcile the audits into the smallest additive design that preserves v1 behavior.
+5. Implement and measure coherent slices, committing only after their gates pass.
+
+### Work log
+
+#### Linear dependency validation
+
+The inherited dependency-cycle preparation scanned every edge once for every node. The graph
+validator now constructs dependency adjacency in one edge pass and sorts each adjacency list before
+the existing deterministic depth-first cycle check.
+
+A 2,000-node regression test counts complete edge-collection iteration passes and caps them at
+four. The focused correctness and bounded-pass tests pass, and the configured strict source type
+gate is clean.
+
+One validation command initially included `tests/test_core.py` in a direct Pyright invocation.
+Repository Pyright intentionally covers `src` and `tools`, so that command reported existing
+untyped test-result indexing rather than a source defect. Rerunning the repository-configured type
+gate produced zero diagnostics.
+
+#### Persistent generic source generations
+
+Generic projects now persist a version-1 source-generation receipt only after complete source
+loading and fully verified index publication. The receipt binds the explicit generic source
+contract, project and root identity, adapter, revision, source hash, every canonical/authority/
+descriptor regular-file identity, and every source-membership directory identity.
+
+A normal warm check reads no canonical source bytes. It validates the known directories and files
+directly using device, inode, mode, size, nanosecond modification time, and nanosecond change time.
+Directory identities detect add, delete, and rename operations without an `rglob`. Any missing,
+malformed, incompatible, foreign, or dirty receipt becomes a cache miss and falls back to the full
+canonical load and row-verification oracle. Successful fallback verification repairs the disposable
+receipt.
+
+The receipt is deliberately generic-project behavior. Incremental adapter manifests retain
+authority over generated or specialist source identities. A one-method legacy adapter continues to
+work even when it cannot provide a cheap generation.
+
+#### Request-scoped immutable reads
+
+Index reads now use one read-only SQLite transaction pinned to one verified file signature and one
+source identity. Existence checks and queries share that connection. Before returning, the request
+rechecks the index signature and current cheap source generation. A concurrent source or index
+change fails closed.
+
+Context compilation hydrates nodes and edges from the pinned derived snapshot while retaining
+profiles from the immutable descriptor. It no longer loads or parses canonical sources. The public
+full `Project.load()` and deep `ProjectIndex.check()` behavior remains the recovery and equivalence
+oracle.
+
+Focused tests prove that fresh-process-style generic reads can run exact, search, filter,
+backlinks, dependency, impact, context, and no-change synchronization operations while
+`Project.load()` is forbidden. They also prove a final source-generation change is rejected before
+return and missing/corrupt receipts fall back and repair.
+
+On the maintained 1,000-file fixture, the current work-in-progress measurements are:
+
+| Operation | Milestone 0 median | Milestone 1 WIP median |
+|---|---:|---:|
+| Warm no-change synchronize | 142.479 ms | 20.007 ms |
+| Exact node | 286.306 ms | 40.277 ms |
+| Search, limit 20 | 288.793 ms | 41.455 ms |
+| Dependencies, depth 8 | 287.791 ms | 41.551 ms |
+| Context, 32k | 436.897 ms | 46.245 ms |
+| MCP exact node | 287.094 ms | 40.051 ms |
+| MCP context, 32k | 434.853 ms | 46.454 ms |
+
+The three-sample WIP run is directional, not the final Milestone 1 baseline. The final evidence run
+will use the maintained sample counts and committed clean-tree revision.
+
+#### Read-only audit reconciliation
+
+The three Milestone 1 audits agreed on the main architecture:
+
+- Keep complete loading and deep checking as independent truth oracles.
+- Trust only versioned, identity-bound disposable generation receipts.
+- Use one pinned read transaction and retain a final dirty check.
+- Hydrate context from the current index.
+- Replace full-edge traversal scans with bounded indexed frontier reads.
+- Add compact success receipts before allowing large mutations to report post-write size errors.
+- Replace hidden render-status rendering with a receipt comparison.
+- Add algorithmic counters and parse-count gates alongside wall-clock thresholds.
+
+One audit identified a correctness risk beyond latency: a large mutating MCP operation can commit
+successfully and then be replaced by `result_too_large`. This must be fixed in Milestone 1 so
+exactly-once operations never report a false failure after mutation.
+
+### Initial design constraints
+
+- Full rebuild remains the recovery and equivalence oracle.
+- Canonical content remains authoritative.
+- Existing one-method `load_projection()` adapters remain unchanged.
+- No-AST adapters remain first-class.
+- Indexes, source-generation receipts, and caches remain disposable.
+- Cheap reads may trust only identity-bound, versioned, corruption-checked receipts.
+- Any optimization must fail closed on source mutation and must preserve stale-read refusal.
+
+### Future ideas and suggestions
+
+These are notes, not commitments:
+
+- A stable source-generation provider may deserve a public adapter capability only after both the
+  generic project and one incremental adapter prove the same boundary.
+- Profiling receipts could eventually feed the human-facing project control panel, but Milestone 1
+  should expose structured data before adding UI.
+- Large context and changeset payloads may need cursor pagination or compact immutable receipts.
+  The choice should follow actual client workflows rather than generic pagination machinery.

src/docforge/context.py

@@ -35,92 +35,83 @@ def _profile(snapshot: ProjectSnapshot, profile_id: str) -> ContextProfile:
 def compile_context(
     index: ProjectIndex, profile_id: str, budget: int | None = None
 ) -> dict[str, object]:
-    checked = index.check()
+    def select(snapshot: ProjectSnapshot) -> dict[str, object]:
-    snapshot = index.project.load()
+        profile = _profile(snapshot, profile_id)
-    if snapshot.source_hash != checked["source_hash"] or snapshot.revision != checked["revision"]:
+        selected_budget = profile.token_budget if budget is None else budget
-        raise DocForgeError("source_changed", "Canonical source changed before context selection")
+        if (
-    profile = _profile(snapshot, profile_id)
+            isinstance(selected_budget, bool)
-    selected_budget = profile.token_budget if budget is None else budget
+            or selected_budget < 1
-    if (
+            or selected_budget > snapshot.descriptor.limits.max_context_tokens
-        isinstance(selected_budget, bool)
+        ):
-        or selected_budget < 1
+            raise DocForgeError("invalid_budget", "Context budget is outside the configured range")
-        or selected_budget > snapshot.descriptor.limits.max_context_tokens
-    ):
-        raise DocForgeError("invalid_budget", "Context budget is outside the configured range")
 
-    node_by_id = {node.node_id: node for node in snapshot.nodes}
+        node_by_id = {node.node_id: node for node in snapshot.nodes}
-    dependency_edges = {
+        dependency_lists: dict[str, list[str]] = {node_id: [] for node_id in node_by_id}
-        node_id: tuple(
+        for edge in snapshot.edges:
-            edge.target_id
+            if edge.relation == "depends_on":
-            for edge in snapshot.edges
+                dependency_lists[edge.source_id].append(edge.target_id)
-            if edge.source_id == node_id and edge.relation == "depends_on"
+        dependency_edges = {
-        )
+            node_id: tuple(targets) for node_id, targets in dependency_lists.items()
-        for node_id in node_by_id
+        }
-    }
+        reasons: dict[str, str] = {
-    reasons: dict[str, str] = {node_id: "required by profile" for node_id in profile.required_nodes}
+            node_id: "required by profile" for node_id in profile.required_nodes
-    queue = deque((node_id, 0) for node_id in profile.required_nodes)
+        }
-    while queue:
+        queue = deque((node_id, 0) for node_id in profile.required_nodes)
-        node_id, depth = queue.popleft()
+        while queue:
-        if depth >= profile.dependency_depth:
+            node_id, depth = queue.popleft()
-            continue
+            if depth >= profile.dependency_depth:
-        for dependency in dependency_edges[node_id]:
+                continue
-            if dependency not in reasons:
+            for dependency in dependency_edges[node_id]:
-                reasons[dependency] = f"dependency of {node_id}"
+                if dependency not in reasons:
-                queue.append((dependency, depth + 1))
+                    reasons[dependency] = f"dependency of {node_id}"
+                    queue.append((dependency, depth + 1))
 
-    eligible = [
+        eligible = [
-        node
+            node
-        for node in snapshot.nodes
+            for node in snapshot.nodes
-        if (not profile.families or node.family in profile.families)
+            if (not profile.families or node.family in profile.families)
-        and (not profile.statuses or node.status in profile.statuses)
+            and (not profile.statuses or node.status in profile.statuses)
-    ]
+        ]
-    ordered_ids = [*profile.required_nodes]
+        ordered_ids = [*profile.required_nodes]
-    ordered_ids.extend(sorted(set(reasons) - set(ordered_ids)))
+        ordered_ids.extend(sorted(set(reasons) - set(ordered_ids)))
-    ordered_ids.extend(node.node_id for node in eligible if node.node_id not in reasons)
+        ordered_ids.extend(node.node_id for node in eligible if node.node_id not in reasons)
 
-    entries: list[ContextEntry] = []
+        entries: list[ContextEntry] = []
-    omissions: list[dict[str, str]] = []
+        omissions: list[dict[str, str]] = []
-    used_tokens = 0
+        used_tokens = 0
-    required = set(profile.required_nodes)
+        required = set(profile.required_nodes)
-    for node_id in ordered_ids:
+        for node_id in ordered_ids:
-        node = node_by_id[node_id]
+            node = node_by_id[node_id]
-        text = _node_text(node)
+            text = _node_text(node)
-        tokens = _estimate_tokens(text)
+            tokens = _estimate_tokens(text)
-        if used_tokens + tokens > selected_budget:
+            if used_tokens + tokens > selected_budget:
-            if node_id in required:
+                if node_id in required:
-                raise DocForgeError(
+                    raise DocForgeError(
-                    "budget_too_small",
+                        "budget_too_small",
-                    "Context budget cannot contain every required node",
+                        "Context budget cannot contain every required node",
+                        node_id=node_id,
+                        required_tokens=used_tokens + tokens,
+                    )
+                omissions.append({"node_id": node_id, "reason": "token budget"})
+                continue
+            entries.append(
+                ContextEntry(
                     node_id=node_id,
-                    required_tokens=used_tokens + tokens,
+                    reason=reasons.get(node_id, "eligible profile node"),
+                    estimated_tokens=tokens,
+                    source_path=node.source_path,
+                    content_hash=node.content_hash,
+                    text=text,
                 )
-            omissions.append({"node_id": node_id, "reason": "token budget"})
-            continue
-        entries.append(
-            ContextEntry(
-                node_id=node_id,
-                reason=reasons.get(node_id, "eligible profile node"),
-                estimated_tokens=tokens,
-                source_path=node.source_path,
-                content_hash=node.content_hash,
-                text=text,
             )
-        )
+            used_tokens += tokens
-        used_tokens += tokens
 
-    after = index.check()
+        return {
-    if after["source_hash"] != checked["source_hash"] or after["revision"] != checked["revision"]:
+            "profile": profile.profile_id,
-        raise DocForgeError("source_changed", "Canonical source changed during context selection")
+            "budget": selected_budget,
-    return {
+            "estimated_tokens": used_tokens,
-        "status": "ok",
+            "entries": [entry.as_dict() for entry in entries],
-        "project_id": checked["project_id"],
+            "omissions": omissions,
-        "project_root_fingerprint": checked["project_root_fingerprint"],
+        }
-        "revision": checked["revision"],
+
-        "source_hash": checked["source_hash"],
+    return index.read_project_snapshot(select)
-        "adapter": checked["adapter"],
-        "profile": profile.profile_id,
-        "budget": selected_budget,
-        "estimated_tokens": used_tokens,
-        "entries": [entry.as_dict() for entry in entries],
-        "omissions": omissions,
-    }

src/docforge/index.py

@@ -10,8 +10,9 @@ import sqlite3
 import tempfile
 import time
 from collections import deque
-from collections.abc import Generator
+from collections.abc import Callable, Generator
 from contextlib import contextmanager
+from dataclasses import dataclass
 from pathlib import Path
 from typing import cast
 
@@ -19,12 +20,14 @@ from .errors import DocForgeError
 from .models import (
     BuildReportingProject,
     Edge,
+    GenerationRecordingProject,
     IncrementalStateProject,
     LogicEdge,
     LogicNode,
     LogicProject,
     LogicProjection,
     Node,
+    ProjectDescriptor,
     ProjectService,
     ProjectSnapshot,
     ProjectState,
@@ -104,6 +107,45 @@ def _status(
     }
 
 
+@dataclass(frozen=True)
+class _IndexReadSnapshot:
+    """One request-scoped read transaction over a verified immutable generation."""
+
+    connection: sqlite3.Connection
+    checked: dict[str, object]
+
+    def project_snapshot(self, descriptor: ProjectDescriptor) -> ProjectSnapshot:
+        nodes = tuple(
+            _row_to_node(row)
+            for row in self.connection.execute("SELECT * FROM nodes ORDER BY node_id")
+        )
+        edges = tuple(
+            Edge(*row)
+            for row in self.connection.execute(
+                "SELECT source_id, relation, target_id FROM edges "
+                "ORDER BY source_id, relation, target_id"
+            )
+        )
+        return ProjectSnapshot(
+            descriptor=descriptor,
+            nodes=nodes,
+            edges=edges,
+            source_hash=cast(str, self.checked["source_hash"]),
+            revision=cast(str, self.checked["revision"]),
+        )
+
+    def result(self, **payload: object) -> dict[str, object]:
+        return {
+            "status": "ok",
+            "project_id": self.checked["project_id"],
+            "project_root_fingerprint": self.checked["project_root_fingerprint"],
+            "revision": self.checked["revision"],
+            "source_hash": self.checked["source_hash"],
+            "adapter": self.checked["adapter"],
+            **payload,
+        }
+
+
 class ProjectIndex:
     """A disposable index that always checks current canonical source before queries."""
 
@@ -346,6 +388,8 @@ class ProjectIndex:
             os.replace(temporary, self.path)
             self._verified_index_signature = self._index_signature()
             self._write_attestation()
+            if isinstance(self.project, GenerationRecordingProject):
+                self.project.record_generation(current)
         except sqlite3.Error as error:
             temporary.unlink(missing_ok=True)
             raise DocForgeError("index_failure", "Could not build the derived index") from error
@@ -408,6 +452,81 @@ class ProjectIndex:
                 logic_projection_count=projection_count,
             )
 
+    @contextmanager
+    def _read_snapshot(self) -> Generator[_IndexReadSnapshot, None, None]:
+        """Pin one verified index and source generation for a complete read request."""
+
+        checked = self.check(verify_rows=False)
+        signature = self._verified_index_signature
+        if signature is None or self._index_signature() != signature:
+            raise DocForgeError(
+                "invalid_index",
+                "Derived index changed after validation",
+            )
+        with _read_connection(self.path) as connection:
+            connection.execute("PRAGMA query_only=ON")
+            connection.execute("BEGIN")
+            application_id = connection.execute("PRAGMA application_id").fetchone()[0]
+            schema_version = connection.execute("PRAGMA user_version").fetchone()[0]
+            if application_id != APPLICATION_ID or schema_version != INDEX_SCHEMA_VERSION:
+                raise DocForgeError("invalid_index", "Derived index has an unsupported schema")
+            metadata = dict(connection.execute("SELECT key, value FROM metadata"))
+            for key in (
+                "project_id",
+                "project_root_fingerprint",
+                "revision",
+                "source_hash",
+                "index_schema_version",
+                "adapter",
+            ):
+                if metadata.get(key) != str(checked[key]):
+                    raise DocForgeError(
+                        "invalid_index",
+                        "Derived index changed after validation",
+                        field=key,
+                    )
+            try:
+                logic_projection_count = int(metadata["logic_projection_count"])
+            except (KeyError, ValueError) as error:
+                raise DocForgeError(
+                    "invalid_index",
+                    "Derived index has invalid Logic metadata",
+                ) from error
+            self._require_logic_allowed(logic_projection_count)
+            snapshot = _IndexReadSnapshot(connection=connection, checked=checked)
+            try:
+                yield snapshot
+            except Exception:
+                raise
+            else:
+                self._confirm_read(snapshot.checked, signature)
+
+    def _confirm_read(
+        self,
+        checked: dict[str, object],
+        signature: tuple[int, int, int, int, int],
+    ) -> None:
+        if self._index_signature() != signature:
+            raise DocForgeError("invalid_index", "Derived index changed during the query")
+        state: ProjectState | None = None
+        if isinstance(self.project, IncrementalStateProject):
+            state = self.project.incremental_state()
+        if state is None:
+            current = self.project.load()
+            state = ProjectState(source_hash=current.source_hash, revision=current.revision)
+        if state.source_hash != checked["source_hash"] or state.revision != checked["revision"]:
+            raise DocForgeError("source_changed", "Canonical source changed during the query")
+
+    def read_project_snapshot(
+        self,
+        reader: Callable[[ProjectSnapshot], dict[str, object]],
+    ) -> dict[str, object]:
+        """Run one bounded reader against an immutable derived project snapshot."""
+
+        with self._read_snapshot() as snapshot:
+            payload = reader(snapshot.project_snapshot(self.project.descriptor))
+            return snapshot.result(**payload)
+
     def check(self, *, verify_rows: bool = True) -> dict[str, object]:
         if isinstance(self.project, IncrementalStateProject):
             state = self.project.incremental_state()
@@ -465,6 +584,9 @@ class ProjectIndex:
                 or fts_count != len(snapshot.nodes)
             ):
                 raise DocForgeError("invalid_index", "Derived index rows do not match source")
+        self._verified_index_signature = self._index_signature()
+        if isinstance(self.project, GenerationRecordingProject):
+            self.project.record_generation(snapshot)
         return {**expected, "database": str(self.path)}
 
     def _check_incremental_state(
@@ -654,39 +776,38 @@ class ProjectIndex:
         )
 
     def get_node(self, node_id: str) -> dict[str, object]:
-        checked = self.check(verify_rows=False)
+        with self._read_snapshot() as snapshot:
-        with _read_connection(self.path) as connection:
+            row = snapshot.connection.execute(
-            row = connection.execute("SELECT * FROM nodes WHERE node_id = ?", (node_id,)).fetchone()
+                "SELECT * FROM nodes WHERE node_id = ?",
-        if row is None:
+                (node_id,),
-            raise DocForgeError(
+            ).fetchone()
-                "missing_node", "No node has the requested stable ID", node_id=node_id
+            if row is None:
-            )
+                raise DocForgeError(
-        return self._result(checked, node=_row_to_node(row).as_dict())
+                    "missing_node", "No node has the requested stable ID", node_id=node_id
+                )
+            return snapshot.result(node=_row_to_node(row).as_dict())
 
     def get_logic(self, owner_node_id: str) -> dict[str, object]:
         """Return one function-scoped control-flow projection without expanding the graph."""
 
-        checked = self.check(verify_rows=False)
+        with self._read_snapshot() as snapshot:
-        with _read_connection(self.path) as connection:
+            owner = snapshot.connection.execute(
-            owner = connection.execute(
                 "SELECT * FROM nodes WHERE node_id = ?", (owner_node_id,)
             ).fetchone()
-            projection = _logic_projection_from_connection(connection, owner_node_id)
+            projection = _logic_projection_from_connection(snapshot.connection, owner_node_id)
-        if owner is None:
+            if owner is None:
-            raise DocForgeError(
+                raise DocForgeError(
-                "missing_node",
+                    "missing_node",
-                "No node has the requested stable ID",
+                    "No node has the requested stable ID",
-                node_id=owner_node_id,
+                    node_id=owner_node_id,
+                )
+            return snapshot.result(
+                owner=_row_to_node(owner).as_dict(include_content=False),
+                available=projection is not None,
+                projection=projection.as_dict() if projection is not None else None,
             )
-        return self._result(
-            checked,
-            owner=_row_to_node(owner).as_dict(include_content=False),
-            available=projection is not None,
-            projection=projection.as_dict() if projection is not None else None,
-        )
 
     def search(self, query: str, *, limit: int | None = None) -> dict[str, object]:
-        checked = self.check(verify_rows=False)
         limits = self.project.descriptor.limits
         if not query.strip() or len(query) > limits.max_query_chars:
             raise DocForgeError("invalid_query", "Search query is empty or exceeds its limit")
@@ -695,8 +816,8 @@ class ProjectIndex:
         if not terms:
             raise DocForgeError("invalid_query", "Search query contains no searchable text")
         expression = " AND ".join(f'"{term.replace(chr(34), chr(34) * 2)}"' for term in terms)
-        with _read_connection(self.path) as connection:
+        with self._read_snapshot() as snapshot:
-            rows = connection.execute(
+            rows = snapshot.connection.execute(
                 """
                 SELECT nodes.*, bm25(node_fts) AS rank,
                        snippet(node_fts, 3, '[', ']', ' … ', 18) AS snippet
@@ -707,12 +828,12 @@ class ProjectIndex:
                 """,
                 (expression, bounded),
             ).fetchall()
-        results: list[dict[str, object]] = []
+            results: list[dict[str, object]] = []
-        for row in rows:
+            for row in rows:
-            payload = _row_to_node(row).as_dict(include_content=False)
+                payload = _row_to_node(row).as_dict(include_content=False)
-            payload.update({"rank": row["rank"], "snippet": row["snippet"]})
+                payload.update({"rank": row["rank"], "snippet": row["snippet"]})
-            results.append(payload)
+                results.append(payload)
-        return self._result(checked, query=query, count=len(results), results=results)
+            return snapshot.result(query=query, count=len(results), results=results)
 
     def filter_nodes(
         self,
@@ -723,7 +844,6 @@ class ProjectIndex:
         tag: str | None = None,
         limit: int | None = None,
     ) -> dict[str, object]:
-        checked = self.check(verify_rows=False)
         bounded = _bounded_limit(limit, self.project.descriptor.limits.max_results, default=100)
         clauses: list[str] = []
         values: list[object] = []
@@ -735,12 +855,12 @@ class ProjectIndex:
             clauses.append("EXISTS (SELECT 1 FROM json_each(tags_json) WHERE value = ?)")
             values.append(tag)
         where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
-        with _read_connection(self.path) as connection:
+        with self._read_snapshot() as snapshot:
-            rows = connection.execute(
+            rows = snapshot.connection.execute(
                 f"SELECT * FROM nodes {where} ORDER BY node_id LIMIT ?", (*values, bounded)
             ).fetchall()
-        results = [_row_to_node(row).as_dict(include_content=False) for row in rows]
+            results = [_row_to_node(row).as_dict(include_content=False) for row in rows]
-        return self._result(checked, count=len(results), results=results)
+            return snapshot.result(count=len(results), results=results)
 
     def backlinks(self, node_id: str, *, relation: str | None = None) -> dict[str, object]:
         return self._edges(node_id, incoming=True, relation=relation)
@@ -752,93 +872,80 @@ class ProjectIndex:
         return self._traverse(node_id, incoming=True, depth=depth, relation=None)
 
     def _edges(self, node_id: str, *, incoming: bool, relation: str | None) -> dict[str, object]:
-        checked = self.check(verify_rows=False)
-        self._require_node(node_id)
         source_column = "target_id" if incoming else "source_id"
         relation_clause = " AND relation = ?" if relation is not None else ""
         values: tuple[object, ...] = (node_id, relation) if relation is not None else (node_id,)
-        with _read_connection(self.path) as connection:
+        with self._read_snapshot() as snapshot:
-            rows = connection.execute(
+            self._require_node(snapshot.connection, node_id)
+            rows = snapshot.connection.execute(
                 f"SELECT source_id, relation, target_id FROM edges "
                 f"WHERE {source_column} = ?{relation_clause} "
                 "ORDER BY source_id, relation, target_id",
                 values,
             ).fetchall()
-        return self._result(checked, edges=[Edge(*row).as_dict() for row in rows])
+            return snapshot.result(edges=[Edge(*row).as_dict() for row in rows])
 
     def _traverse(
         self, node_id: str, *, incoming: bool, depth: int, relation: str | None
     ) -> dict[str, object]:
-        checked = self.check(verify_rows=False)
-        self._require_node(node_id)
         maximum = self.project.descriptor.limits.max_traversal_depth
         if type(depth) is not int or depth < 0 or depth > maximum:
             raise DocForgeError("invalid_depth", "Traversal depth is outside the configured limit")
-        with _read_connection(self.path) as connection:
+        with self._read_snapshot() as snapshot:
+            self._require_node(snapshot.connection, node_id)
             edges = tuple(
                 Edge(*row)
-                for row in connection.execute(
+                for row in snapshot.connection.execute(
                     "SELECT source_id, relation, target_id FROM edges "
                     "ORDER BY source_id, relation, target_id"
                 )
             )
-        queue: deque[tuple[str, int, tuple[str, ...]]] = deque([(node_id, 0, (node_id,))])
+            queue: deque[tuple[str, int, tuple[str, ...]]] = deque([(node_id, 0, (node_id,))])
-        seen = {node_id}
+            seen = {node_id}
-        results: list[dict[str, object]] = []
+            results: list[dict[str, object]] = []
-        while queue:
+            while queue:
-            current, current_depth, path = queue.popleft()
+                current, current_depth, path = queue.popleft()
-            if current_depth >= depth:
+                if current_depth >= depth:
-                continue
-            candidates = [
-                edge
-                for edge in edges
-                if (relation is None or edge.relation == relation)
-                and ((edge.target_id if incoming else edge.source_id) == current)
-            ]
-            for edge in candidates:
-                target = edge.source_id if incoming else edge.target_id
-                if target in seen:
                     continue
-                seen.add(target)
+                candidates = [
-                target_path = (*path, target)
+                    edge
-                results.append(
+                    for edge in edges
-                    {
+                    if (relation is None or edge.relation == relation)
-                        "node_id": target,
+                    and ((edge.target_id if incoming else edge.source_id) == current)
-                        "depth": current_depth + 1,
+                ]
-                        "relation": edge.relation,
+                for edge in candidates:
-                        "path": target_path,
+                    target = edge.source_id if incoming else edge.target_id
-                    }
+                    if target in seen:
-                )
+                        continue
-                queue.append((target, current_depth + 1, target_path))
+                    seen.add(target)
-        return self._result(checked, root=node_id, depth=depth, count=len(results), results=results)
+                    target_path = (*path, target)
+                    results.append(
+                        {
+                            "node_id": target,
+                            "depth": current_depth + 1,
+                            "relation": edge.relation,
+                            "path": target_path,
+                        }
+                    )
+                    queue.append((target, current_depth + 1, target_path))
+            return snapshot.result(
+                root=node_id,
+                depth=depth,
+                count=len(results),
+                results=results,
+            )
 
-    def _require_node(self, node_id: str) -> None:
+    @staticmethod
-        with _read_connection(self.path) as connection:
+    def _require_node(connection: sqlite3.Connection, node_id: str) -> None:
-            exists = connection.execute(
+        exists = connection.execute(
-                "SELECT 1 FROM nodes WHERE node_id = ?", (node_id,)
+            "SELECT 1 FROM nodes WHERE node_id = ?",
-            ).fetchone()
+            (node_id,),
+        ).fetchone()
         if exists is None:
             raise DocForgeError(
                 "missing_node", "No node has the requested stable ID", node_id=node_id
             )
 
-    def _result(self, checked: dict[str, object], **payload: object) -> dict[str, object]:
-        after = self.check(verify_rows=False)
-        if (
-            after["source_hash"] != checked["source_hash"]
-            or after["revision"] != checked["revision"]
-        ):
-            raise DocForgeError("source_changed", "Canonical source changed during the query")
-        return {
-            "status": "ok",
-            "project_id": checked["project_id"],
-            "project_root_fingerprint": checked["project_root_fingerprint"],
-            "revision": checked["revision"],
-            "source_hash": checked["source_hash"],
-            "adapter": checked["adapter"],
-            **payload,
-        }
-
 
 def _row_to_node(row: sqlite3.Row) -> Node:
     return Node(

src/docforge/models.py

@@ -204,6 +204,13 @@ class IncrementalStateProject(ProjectService, Protocol):
     def incremental_state(self) -> ProjectState | None: ...
 
 
+@runtime_checkable
+class GenerationRecordingProject(IncrementalStateProject, Protocol):
+    """Optional project boundary that can persist a verified cheap source generation."""
+
+    def record_generation(self, snapshot: ProjectSnapshot) -> None: ...
+
+
 @runtime_checkable
 class RuntimeValidatedProject(ProjectService, Protocol):
     """Optional project boundary that proves its loaded implementation is current."""

src/docforge/project.py

@@ -4,12 +4,15 @@ from __future__ import annotations
 
 import hashlib
 import json
+import os
+import stat
 import subprocess
+import tempfile
 import tomllib
 from collections import Counter
 from collections.abc import Mapping
-from dataclasses import replace
+from dataclasses import dataclass, replace
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from typing import Any, cast
 
 from .config_validation import (
@@ -28,10 +31,14 @@ from .models import (
     Node,
     ProjectDescriptor,
     ProjectSnapshot,
+    ProjectState,
     ProposalWriter,
 )
 from .render_config import load_render_config
 
+SOURCE_GENERATION_SCHEMA_VERSION = 1
+GENERIC_SOURCE_CONTRACT = "docforge-core:0.7.1:index:1"
+
 _CORE_METADATA = frozenset(
     {
         "schema_version",
@@ -72,10 +79,109 @@ _PROFILE_KEYS = frozenset(
 _OPERATIONS = frozenset({"create", "update", "move", "delete"})
 
 
+@dataclass(frozen=True)
+class _CapturedGeneration:
+    source_hash: str
+    revision: str
+    files: tuple[tuple[str, int, int, int, int, int, int], ...]
+    directories: tuple[tuple[str, int, int, int, int, int], ...]
+
+
 def project_root_fingerprint(root: Path) -> str:
     return hashlib.sha256(str(root).encode()).hexdigest()[:16]
 
 
+def _file_generation(
+    root: Path,
+    paths: tuple[Path, ...],
+) -> tuple[tuple[str, int, int, int, int, int, int], ...]:
+    """Capture cheap identities that change on ordinary source or metadata mutation."""
+
+    identities: list[tuple[str, int, int, int, int, int, int]] = []
+    for path in paths:
+        try:
+            status = path.lstat()
+        except OSError as error:
+            raise DocForgeError(
+                "source_changed",
+                "Canonical source disappeared during generation capture",
+                source=path.relative_to(root).as_posix(),
+            ) from error
+        if not stat.S_ISREG(status.st_mode):
+            raise DocForgeError(
+                "source_changed",
+                "Canonical generation inputs must remain regular files",
+                source=path.relative_to(root).as_posix(),
+            )
+        identities.append(
+            (
+                path.relative_to(root).as_posix(),
+                status.st_dev,
+                status.st_ino,
+                status.st_mode,
+                status.st_size,
+                status.st_mtime_ns,
+                status.st_ctime_ns,
+            )
+        )
+    return tuple(identities)
+
+
+def _directory_generation(
+    root: Path,
+    paths: tuple[Path, ...],
+) -> tuple[tuple[str, int, int, int, int, int], ...]:
+    """Capture directory identities so source membership changes invalidate a receipt."""
+
+    identities: list[tuple[str, int, int, int, int, int]] = []
+    for path in paths:
+        try:
+            status = path.lstat()
+        except OSError as error:
+            raise DocForgeError(
+                "source_changed",
+                "Canonical source directory disappeared during generation capture",
+                source=path.relative_to(root).as_posix(),
+            ) from error
+        if not stat.S_ISDIR(status.st_mode):
+            raise DocForgeError(
+                "source_changed",
+                "Canonical source directories must remain directories",
+                source=path.relative_to(root).as_posix(),
+            )
+        identities.append(
+            (
+                path.relative_to(root).as_posix(),
+                status.st_dev,
+                status.st_ino,
+                status.st_mode,
+                status.st_mtime_ns,
+                status.st_ctime_ns,
+            )
+        )
+    return tuple(identities)
+
+
+def _receipt_paths(root: Path, value: object, *, width: int) -> tuple[Path, ...] | None:
+    if not isinstance(value, list):
+        return None
+    paths: list[Path] = []
+    for raw_item in cast(list[object], value):
+        if not isinstance(raw_item, list):
+            return None
+        item = cast(list[object], raw_item)
+        if len(item) != width or not isinstance(item[0], str):
+            return None
+        relative = PurePosixPath(item[0])
+        if relative.is_absolute() or not relative.parts or ".." in relative.parts:
+            return None
+        path = root.joinpath(*relative.parts)
+        if not path.is_relative_to(root):
+            return None
+        paths.append(path)
+    return tuple(paths)
+
+
 def _load_descriptor(root: Path) -> ProjectDescriptor:
     descriptor_path = root / ".docforge" / "project.toml"
     if not descriptor_path.is_file():
@@ -472,41 +578,60 @@ def validate_graph(nodes: tuple[Node, ...], edges: tuple[Edge, ...]) -> None:
         counts = Counter(node.node_id for node in nodes)
         duplicates = sorted(node_id for node_id, count in counts.items() if count > 1)
         raise DocForgeError("duplicate_node", "Stable node IDs must be unique", ids=duplicates)
-    edge_keys = {(edge.source_id, edge.relation, edge.target_id) for edge in edges}
-    if len(edge_keys) != len(edges):
-        raise DocForgeError("duplicate_edge", "Relationships must be unique")
-    missing = sorted({edge.target_id for edge in edges if edge.target_id not in node_ids})
-    if missing:
-        raise DocForgeError("broken_edge", "Relationships target missing nodes", targets=missing)
 
-    dependencies = {
+    dependencies: dict[str, list[str]] = {node_id: [] for node_id in node_ids}
-        node_id: sorted(
+    edge_keys: set[tuple[str, str, str]] = set()
-            edge.target_id
+    missing_sources: set[str] = set()
-            for edge in edges
+    missing_targets: set[str] = set()
-            if edge.source_id == node_id and edge.relation == "depends_on"
+    for edge in edges:
+        key = (edge.source_id, edge.relation, edge.target_id)
+        if key in edge_keys:
+            raise DocForgeError("duplicate_edge", "Relationships must be unique")
+        edge_keys.add(key)
+        if edge.source_id not in node_ids:
+            missing_sources.add(edge.source_id)
+        if edge.target_id not in node_ids:
+            missing_targets.add(edge.target_id)
+        if edge.relation == "depends_on" and edge.source_id in dependencies:
+            dependencies[edge.source_id].append(edge.target_id)
+    if missing_sources or missing_targets:
+        raise DocForgeError(
+            "broken_edge",
+            "Relationships reference missing nodes",
+            sources=sorted(missing_sources),
+            targets=sorted(missing_targets),
         )
-        for node_id in sorted(node_ids)
+    for targets in dependencies.values():
-    }
+        targets.sort()
-    visiting: set[str] = set()
-    visited: set[str] = set()
 
-    def visit(node_id: str, trail: tuple[str, ...]) -> None:
+    states: dict[str, int] = {}
-        if node_id in visiting:
+    for root in sorted(node_ids):
-            raise DocForgeError(
+        if states.get(root) == 2:
-                "dependency_cycle",
+            continue
-                "depends_on relationships contain a cycle",
+        path: list[str] = []
-                path=(*trail, node_id),
+        stack: list[tuple[str, int]] = [(root, 0)]
-            )
+        while stack:
-        if node_id in visited:
+            node_id, child_index = stack[-1]
-            return
+            if states.get(node_id, 0) == 0:
-        visiting.add(node_id)
+                states[node_id] = 1
-        for target in dependencies[node_id]:
+                path.append(node_id)
-            visit(target, (*trail, node_id))
+            targets = dependencies[node_id]
-        visiting.remove(node_id)
+            if child_index < len(targets):
-        visited.add(node_id)
+                target = targets[child_index]
-
+                stack[-1] = (node_id, child_index + 1)
-    for node_id in sorted(node_ids):
+                state = states.get(target, 0)
-        visit(node_id, ())
+                if state == 1:
+                    raise DocForgeError(
+                        "dependency_cycle",
+                        "depends_on relationships contain a cycle",
+                        path=(*path, target),
+                    )
+                if state == 0:
+                    stack.append((target, 0))
+                continue
+            stack.pop()
+            path.pop()
+            states[node_id] = 2
 
 
 def validate_source_layout(nodes: tuple[Node, ...]) -> None:
@@ -571,6 +696,7 @@ class Project:
 
     def __init__(self, descriptor: ProjectDescriptor) -> None:
         self.descriptor = descriptor
+        self._captured_generation: _CapturedGeneration | None = None
 
     @classmethod
     def open(cls, project_root: str | Path) -> Project:
@@ -588,15 +714,18 @@ class Project:
             raise DocForgeError(
                 "source_changed", "Project descriptor changed after the project was opened"
             )
-        ordered_sources = self.canonical_source_paths()
+        ordered_sources, ordered_directories = self._canonical_inventory()
-        captured = {
+        generation_paths = (
-            path: path.read_bytes()
+            self.descriptor.descriptor_path,
-            for path in (
+            *self.descriptor.authority_files,
-                self.descriptor.descriptor_path,
+            *ordered_sources,
-                *self.descriptor.authority_files,
+        )
-                *ordered_sources,
+        before_generation = _file_generation(self.descriptor.root, generation_paths)
-            )
+        before_directories = _directory_generation(
-        }
+            self.descriptor.root,
+            ordered_directories,
+        )
+        captured = {path: path.read_bytes() for path in generation_paths}
 
         nodes: list[Node] = []
         edges: list[Edge] = []
@@ -619,7 +748,8 @@ class Project:
                     "invalid_config", "Context profile requires missing nodes", nodes=missing
                 )
 
-        if self.canonical_source_paths() != ordered_sources:
+        current_sources, current_directories = self._canonical_inventory()
+        if current_sources != ordered_sources or current_directories != ordered_directories:
             raise DocForgeError("source_changed", "Canonical source set changed during loading")
         for path, raw in captured.items():
             if not path.is_file() or path.read_bytes() != raw:
@@ -628,6 +758,16 @@ class Project:
                     "Canonical source changed during loading",
                     source=path.relative_to(self.descriptor.root).as_posix(),
                 )
+        after_generation = _file_generation(self.descriptor.root, generation_paths)
+        after_directories = _directory_generation(
+            self.descriptor.root,
+            ordered_directories,
+        )
+        if after_generation != before_generation or after_directories != before_directories:
+            raise DocForgeError(
+                "source_changed",
+                "Canonical source metadata changed during loading",
+            )
 
         digest = hashlib.sha256()
         for path in sorted(
@@ -637,21 +777,157 @@ class Project:
             digest.update(relative.encode())
             digest.update(b"\0")
             digest.update(hashlib.sha256(captured[path]).digest())
-        digest.update(b"docforge-core:0.7.1:index:1")
+        digest.update(GENERIC_SOURCE_CONTRACT.encode("ascii"))
-        return ProjectSnapshot(
+        source_hash = digest.hexdigest()
+        revision = _revision(self.descriptor.root)
+        snapshot = ProjectSnapshot(
             descriptor=self.descriptor,
             nodes=ordered_nodes,
             edges=ordered_edges,
-            source_hash=digest.hexdigest(),
+            source_hash=source_hash,
-            revision=_revision(self.descriptor.root),
+            revision=revision,
         )
+        self._captured_generation = _CapturedGeneration(
+            source_hash=source_hash,
+            revision=revision,
+            files=after_generation,
+            directories=after_directories,
+        )
+        return snapshot
+
+    @property
+    def generation_path(self) -> Path:
+        """Return the confined disposable receipt for one verified source generation."""
+
+        return self.descriptor.cache_root / "source-generation.json"
+
+    def incremental_state(self) -> ProjectState | None:
+        """Return current source identity without reading or parsing canonical source bytes."""
+
+        path = self.generation_path
+        if not path.is_file() or path.is_symlink():
+            return None
+        try:
+            parsed: object = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError):
+            return None
+        if not isinstance(parsed, dict):
+            return None
+        payload = cast(dict[str, object], parsed)
+        source_hash = payload.get("source_hash")
+        revision = payload.get("revision")
+        if (
+            payload.get("schema_version") != SOURCE_GENERATION_SCHEMA_VERSION
+            or payload.get("source_contract") != GENERIC_SOURCE_CONTRACT
+            or payload.get("project_id") != self.descriptor.project_id
+            or payload.get("project_root_fingerprint")
+            != project_root_fingerprint(self.descriptor.root)
+            or payload.get("adapter") != self.descriptor.adapter
+            or not isinstance(source_hash, str)
+            or len(source_hash) != 64
+            or not isinstance(revision, str)
+        ):
+            return None
+        directory_paths = _receipt_paths(
+            self.descriptor.root,
+            payload.get("directories"),
+            width=6,
+        )
+        file_paths = _receipt_paths(
+            self.descriptor.root,
+            payload.get("files"),
+            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)
+
+    def record_generation(self, snapshot: ProjectSnapshot) -> None:
+        """Persist a generation only after its complete derived index was verified."""
+
+        captured = self._captured_generation
+        if (
+            captured is None
+            or captured.source_hash != snapshot.source_hash
+            or captured.revision != snapshot.revision
+        ):
+            raise DocForgeError(
+                "source_changed",
+                "Cannot record a source generation without a matching complete load",
+            )
+        root = self.descriptor.cache_root
+        path = self.generation_path
+        if path.parent != root or path.is_symlink() or root.resolve(strict=False) != root:
+            raise DocForgeError("path_escape", "Source generation receipt path is not safe")
+        root.mkdir(parents=True, exist_ok=True)
+        if not root.is_dir() or root.resolve(strict=False) != root:
+            raise DocForgeError("path_escape", "Source generation receipt directory is not safe")
+        payload = {
+            "schema_version": SOURCE_GENERATION_SCHEMA_VERSION,
+            "source_contract": GENERIC_SOURCE_CONTRACT,
+            "project_id": self.descriptor.project_id,
+            "project_root_fingerprint": project_root_fingerprint(self.descriptor.root),
+            "adapter": self.descriptor.adapter,
+            "source_hash": captured.source_hash,
+            "revision": captured.revision,
+            "files": [list(identity) for identity in captured.files],
+            "directories": [list(identity) for identity in captured.directories],
+        }
+        raw = json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") + b"\n"
+        descriptor, temporary_name = tempfile.mkstemp(prefix=".source-generation-", dir=root)
+        temporary = Path(temporary_name)
+        try:
+            with os.fdopen(descriptor, "wb") as handle:
+                handle.write(raw)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temporary, path)
+            directory_descriptor = os.open(root, os.O_RDONLY)
+            try:
+                os.fsync(directory_descriptor)
+            finally:
+                os.close(directory_descriptor)
+        except Exception:
+            temporary.unlink(missing_ok=True)
+            raise
 
     def canonical_source_paths(self) -> tuple[Path, ...]:
         """Return the deterministic confined canonical source set."""
 
+        sources, _ = self._canonical_inventory()
+        return sources
+
+    def _canonical_inventory(self) -> tuple[tuple[Path, ...], tuple[Path, ...]]:
+        """Return deterministic canonical files and membership-bearing directories."""
+
         source_paths: set[Path] = set()
+        directories: set[Path] = set()
         for content_root in self.descriptor.content_roots:
+            directories.add(content_root)
             for path in content_root.rglob("*"):
+                if path.is_dir():
+                    resolved_directory = path.resolve()
+                    if not resolved_directory.is_relative_to(self.descriptor.root):
+                        raise DocForgeError(
+                            "path_escape",
+                            "Canonical source directory resolves outside project root",
+                        )
+                    directories.add(resolved_directory)
+                    continue
                 if path.suffix not in {".md", ".toml"} or not path.is_file():
                     continue
                 resolved = path.resolve()
@@ -665,7 +941,11 @@ class Project:
         )
         if not ordered_sources:
             raise DocForgeError("empty_project", "No canonical Markdown or TOML sources were found")
-        return tuple(ordered_sources)
+        ordered_directories = sorted(
+            directories,
+            key=lambda path: path.relative_to(self.descriptor.root).as_posix(),
+        )
+        return tuple(ordered_sources), tuple(ordered_directories)
 
     def validate_proposal(
         self,

tests/test_changesets.py

@@ -761,7 +761,7 @@ operations = ["create", "update", "move", "delete"]
                 mock.patch.object(
                     race_project,
                     "canonical_source_paths",
-                    side_effect=[sources, sources, sources, (*sources, invented)],
+                    side_effect=[sources, (*sources, invented)],
                 ),
                 self.assertRaisesRegex(DocForgeError, "changed during changeset storage"),
             ):

tests/test_core.py

@@ -8,7 +8,10 @@ import sqlite3
 import sys
 import tempfile
 import unittest
+from collections.abc import Iterator
+from dataclasses import replace
 from pathlib import Path
+from typing import TypeVar, cast
 from unittest import mock
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -18,9 +21,26 @@ from docforge.cli import main  # noqa: E402
 from docforge.context import compile_context  # noqa: E402
 from docforge.errors import DocForgeError  # noqa: E402
 from docforge.index import ProjectIndex  # noqa: E402
-from docforge.project import Project  # noqa: E402
+from docforge.models import Edge, ProjectState  # noqa: E402
+from docforge.project import Project, validate_graph  # noqa: E402
 
 FIXTURES = ROOT / "tests" / "fixtures"
+T = TypeVar("T")
+
+
+class CountingTuple(tuple[T, ...]):
+    """Count complete iteration passes without changing tuple behavior."""
+
+    iterations: int
+
+    def __new__(cls, values: tuple[T, ...]) -> CountingTuple[T]:
+        instance = super().__new__(cls, values)
+        instance.iterations = 0
+        return instance
+
+    def __iter__(self) -> Iterator[T]:
+        self.iterations += 1
+        return super().__iter__()
 
 
 class DocForgeCoreTests(unittest.TestCase):
@@ -118,6 +138,7 @@ class DocForgeCoreTests(unittest.TestCase):
     def test_duplicate_nodes_broken_edges_and_dependency_cycles_fail(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             root = self.copy_fixture("alpha", Path(directory))
+            snapshot = Project.open(root).load()
             content = root / "docs" / "content"
             duplicate = content / "duplicate.md"
             duplicate.write_text((content / "foundation.md").read_text(), encoding="utf-8")
@@ -149,6 +170,48 @@ class DocForgeCoreTests(unittest.TestCase):
             with self.assertRaisesRegex(DocForgeError, "cycle"):
                 Project.open(root).load()
 
+            with self.assertRaises(DocForgeError) as missing_source:
+                validate_graph(
+                    snapshot.nodes,
+                    (
+                        Edge(
+                            "missing.source",
+                            "depends_on",
+                            snapshot.nodes[0].node_id,
+                        ),
+                    ),
+                )
+            self.assertEqual("broken_edge", missing_source.exception.code)
+            self.assertEqual(
+                ["missing.source"],
+                missing_source.exception.details["sources"],
+            )
+
+    def test_graph_validation_uses_a_bounded_number_of_edge_passes(self) -> None:
+        snapshot = Project.open(FIXTURES / "alpha").load()
+        nodes = tuple(
+            replace(
+                snapshot.nodes[0],
+                node_id=f"linear.node-{index:05d}",
+                source_path=f"docs/node-{index:05d}.md",
+            )
+            for index in range(10_000)
+        )
+        edges = CountingTuple(
+            tuple(
+                Edge(
+                    f"linear.node-{index:05d}",
+                    "depends_on",
+                    f"linear.node-{index - 1:05d}",
+                )
+                for index in range(1, len(nodes))
+            )
+        )
+
+        validate_graph(nodes, cast(tuple[Edge, ...], edges))
+
+        self.assertLessEqual(edges.iterations, 4)
+
     def test_index_build_is_repeatable_and_validates_every_row(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             root = self.copy_fixture("alpha", Path(directory))
@@ -163,6 +226,58 @@ class DocForgeCoreTests(unittest.TestCase):
             self.assertEqual(3, validated["node_count"])
             self.assertEqual(2, validated["edge_count"])
 
+    def test_persisted_generic_generation_avoids_warm_source_loading(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = self.copy_fixture("alpha", Path(directory))
+            ProjectIndex(Project.open(root)).build()
+            fresh_project = Project.open(root)
+            fresh_index = ProjectIndex(fresh_project)
+
+            with mock.patch.object(
+                fresh_project,
+                "load",
+                side_effect=AssertionError("warm reads must not load canonical sources"),
+            ):
+                self.assertEqual(
+                    "Editing workflow",
+                    fresh_index.get_node("guide.workflow")["node"]["title"],
+                )
+                self.assertEqual(1, fresh_index.search("canonical nodes")["count"])
+                self.assertEqual(1, fresh_index.filter_nodes(family="proof")["count"])
+                self.assertEqual(1, len(fresh_index.backlinks("guide.workflow")["edges"]))
+                self.assertEqual(1, fresh_index.dependencies("guide.workflow")["count"])
+                self.assertEqual(2, fresh_index.impact("guide.foundation")["count"])
+                self.assertEqual("active", compile_context(fresh_index, "active")["profile"])
+                self.assertEqual("current", fresh_index.synchronize()["synchronization"]["action"])
+
+            workflow = root / "docs" / "content" / "workflow.md"
+            workflow.write_text(
+                workflow.read_text(encoding="utf-8") + "\nChanged after generation.\n",
+                encoding="utf-8",
+            )
+            with self.assertRaisesRegex(DocForgeError, "does not match"):
+                fresh_index.get_node("guide.workflow")
+
+    def test_missing_or_corrupt_generation_falls_back_and_repairs(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = self.copy_fixture("alpha", Path(directory))
+            project = Project.open(root)
+            index = ProjectIndex(project)
+            index.build()
+            generation_path = project.generation_path
+
+            for raw in (None, "{not-json"):
+                with self.subTest(raw=raw):
+                    if raw is None:
+                        generation_path.unlink(missing_ok=True)
+                    else:
+                        generation_path.write_text(raw, encoding="utf-8")
+                    with mock.patch.object(project, "load", wraps=project.load) as load:
+                        checked = index.check(verify_rows=False)
+                    self.assertEqual("ok", checked["status"])
+                    self.assertGreaterEqual(load.call_count, 1)
+                    self.assertIsNotNone(project.incremental_state())
+
     def test_index_rejects_tampered_rows_and_another_project_cache(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             parent = Path(directory)
@@ -235,23 +350,38 @@ class DocForgeCoreTests(unittest.TestCase):
     def test_query_rechecks_source_identity_before_returning(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             root = self.copy_fixture("alpha", Path(directory))
-            index = ProjectIndex(Project.open(root))
+            project = Project.open(root)
-            checked = index.build()
+            index = ProjectIndex(project)
-            changed = {**checked, "source_hash": "0" * 64}
+            index.build()
+            current = project.incremental_state()
+            self.assertIsNotNone(current)
+            changed = ProjectState(
+                source_hash="0" * 64,
+                revision=current.revision,
+            )
 
             with (
-                mock.patch.object(index, "check", side_effect=[checked, changed]),
+                mock.patch.object(
+                    project,
+                    "incremental_state",
+                    side_effect=[current, changed],
+                ),
                 self.assertRaisesRegex(DocForgeError, "changed during the query"),
             ):
                 index.get_node("guide.workflow")
 
     def test_source_set_change_during_load_fails_closed(self) -> None:
         project = Project.open(FIXTURES / "alpha")
-        sources = project.canonical_source_paths()
+        sources, directories = project._canonical_inventory()
         invented = project.descriptor.root / "docs" / "content" / "invented.md"
         with (
             mock.patch.object(
-                project, "canonical_source_paths", side_effect=[sources, (*sources, invented)]
+                project,
+                "_canonical_inventory",
+                side_effect=[
+                    (sources, directories),
+                    ((*sources, invented), directories),
+                ],
             ),
             self.assertRaisesRegex(DocForgeError, "source set changed"),
         ):