diff --git a/DEVELOPMENT_NOTES.md b/DEVELOPMENT_NOTES.md index b56d143..0b50857 100644 --- a/DEVELOPMENT_NOTES.md +++ b/DEVELOPMENT_NOTES.md @@ -70,6 +70,73 @@ Repository Pyright intentionally covers `src` and `tools`, so that command repor 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. diff --git a/src/docforge/context.py b/src/docforge/context.py index 1e4cb7a..36f1975 100644 --- a/src/docforge/context.py +++ b/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() - snapshot = index.project.load() - if snapshot.source_hash != checked["source_hash"] or snapshot.revision != checked["revision"]: - raise DocForgeError("source_changed", "Canonical source changed before context selection") - profile = _profile(snapshot, profile_id) - selected_budget = profile.token_budget if budget is None else budget - if ( - isinstance(selected_budget, bool) - or selected_budget < 1 - or selected_budget > snapshot.descriptor.limits.max_context_tokens - ): - raise DocForgeError("invalid_budget", "Context budget is outside the configured range") + def select(snapshot: ProjectSnapshot) -> dict[str, object]: + profile = _profile(snapshot, profile_id) + selected_budget = profile.token_budget if budget is None else budget + if ( + isinstance(selected_budget, bool) + or selected_budget < 1 + 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} - dependency_edges = { - node_id: tuple( - edge.target_id - for edge in snapshot.edges - if edge.source_id == node_id and edge.relation == "depends_on" - ) - for node_id in node_by_id - } - reasons: dict[str, str] = {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: - node_id, depth = queue.popleft() - if depth >= profile.dependency_depth: - continue - for dependency in dependency_edges[node_id]: - if dependency not in reasons: - reasons[dependency] = f"dependency of {node_id}" - queue.append((dependency, depth + 1)) + node_by_id = {node.node_id: node for node in snapshot.nodes} + dependency_lists: dict[str, list[str]] = {node_id: [] for node_id in node_by_id} + for edge in snapshot.edges: + if edge.relation == "depends_on": + dependency_lists[edge.source_id].append(edge.target_id) + dependency_edges = { + node_id: tuple(targets) for node_id, targets in dependency_lists.items() + } + reasons: dict[str, str] = { + 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: + node_id, depth = queue.popleft() + if depth >= profile.dependency_depth: + continue + for dependency in dependency_edges[node_id]: + if dependency not in reasons: + reasons[dependency] = f"dependency of {node_id}" + queue.append((dependency, depth + 1)) - eligible = [ - node - for node in snapshot.nodes - if (not profile.families or node.family in profile.families) - and (not profile.statuses or node.status in profile.statuses) - ] - ordered_ids = [*profile.required_nodes] - 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) + eligible = [ + node + for node in snapshot.nodes + if (not profile.families or node.family in profile.families) + and (not profile.statuses or node.status in profile.statuses) + ] + ordered_ids = [*profile.required_nodes] + 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) - entries: list[ContextEntry] = [] - omissions: list[dict[str, str]] = [] - used_tokens = 0 - required = set(profile.required_nodes) - for node_id in ordered_ids: - node = node_by_id[node_id] - text = _node_text(node) - tokens = _estimate_tokens(text) - if used_tokens + tokens > selected_budget: - if node_id in required: - raise DocForgeError( - "budget_too_small", - "Context budget cannot contain every required node", + entries: list[ContextEntry] = [] + omissions: list[dict[str, str]] = [] + used_tokens = 0 + required = set(profile.required_nodes) + for node_id in ordered_ids: + node = node_by_id[node_id] + text = _node_text(node) + tokens = _estimate_tokens(text) + if used_tokens + tokens > selected_budget: + if node_id in required: + raise DocForgeError( + "budget_too_small", + "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() - if after["source_hash"] != checked["source_hash"] or after["revision"] != checked["revision"]: - raise DocForgeError("source_changed", "Canonical source changed during context selection") - 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"], - "profile": profile.profile_id, - "budget": selected_budget, - "estimated_tokens": used_tokens, - "entries": [entry.as_dict() for entry in entries], - "omissions": omissions, - } + return { + "profile": profile.profile_id, + "budget": selected_budget, + "estimated_tokens": used_tokens, + "entries": [entry.as_dict() for entry in entries], + "omissions": omissions, + } + + return index.read_project_snapshot(select) diff --git a/src/docforge/index.py b/src/docforge/index.py index 1fe6e5f..f753a1d 100644 --- a/src/docforge/index.py +++ b/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 _read_connection(self.path) as connection: - row = connection.execute("SELECT * FROM nodes WHERE node_id = ?", (node_id,)).fetchone() - if row is None: - raise DocForgeError( - "missing_node", "No node has the requested stable ID", node_id=node_id - ) - return self._result(checked, node=_row_to_node(row).as_dict()) + with self._read_snapshot() as snapshot: + row = snapshot.connection.execute( + "SELECT * FROM nodes WHERE node_id = ?", + (node_id,), + ).fetchone() + if row is None: + raise DocForgeError( + "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 _read_connection(self.path) as connection: - owner = connection.execute( + with self._read_snapshot() as snapshot: + owner = snapshot.connection.execute( "SELECT * FROM nodes WHERE node_id = ?", (owner_node_id,) ).fetchone() - projection = _logic_projection_from_connection(connection, owner_node_id) - if owner is None: - raise DocForgeError( - "missing_node", - "No node has the requested stable ID", - node_id=owner_node_id, + projection = _logic_projection_from_connection(snapshot.connection, owner_node_id) + if owner is None: + raise DocForgeError( + "missing_node", + "No node has the requested stable 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: - rows = connection.execute( + with self._read_snapshot() as snapshot: + 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]] = [] - for row in rows: - payload = _row_to_node(row).as_dict(include_content=False) - payload.update({"rank": row["rank"], "snippet": row["snippet"]}) - results.append(payload) - return self._result(checked, query=query, count=len(results), results=results) + results: list[dict[str, object]] = [] + for row in rows: + payload = _row_to_node(row).as_dict(include_content=False) + payload.update({"rank": row["rank"], "snippet": row["snippet"]}) + results.append(payload) + 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: - rows = connection.execute( + with self._read_snapshot() as snapshot: + 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] - return self._result(checked, count=len(results), results=results) + results = [_row_to_node(row).as_dict(include_content=False) for row in rows] + 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: - rows = connection.execute( + with self._read_snapshot() as snapshot: + 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,))]) - seen = {node_id} - results: list[dict[str, object]] = [] - while queue: - current, current_depth, path = queue.popleft() - 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: + queue: deque[tuple[str, int, tuple[str, ...]]] = deque([(node_id, 0, (node_id,))]) + seen = {node_id} + results: list[dict[str, object]] = [] + while queue: + current, current_depth, path = queue.popleft() + if current_depth >= depth: continue - seen.add(target) - 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 self._result(checked, root=node_id, depth=depth, count=len(results), results=results) + 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) + 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: - with _read_connection(self.path) as connection: - exists = connection.execute( - "SELECT 1 FROM nodes WHERE node_id = ?", (node_id,) - ).fetchone() + @staticmethod + def _require_node(connection: sqlite3.Connection, node_id: str) -> None: + exists = connection.execute( + "SELECT 1 FROM nodes WHERE node_id = ?", + (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( diff --git a/src/docforge/models.py b/src/docforge/models.py index 10a6b67..47d5339 100644 --- a/src/docforge/models.py +++ b/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.""" diff --git a/src/docforge/project.py b/src/docforge/project.py index c91c9e1..487bc06 100644 --- a/src/docforge/project.py +++ b/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 pathlib import Path +from dataclasses import dataclass, replace +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,39 +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: dict[str, list[str]] = {node_id: [] for node_id in node_ids} + edge_keys: set[tuple[str, str, str]] = set() + missing_sources: set[str] = set() + missing_targets: set[str] = set() for edge in edges: - if edge.relation == "depends_on": + 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 targets in dependencies.values(): targets.sort() - visiting: set[str] = set() - visited: set[str] = set() - def visit(node_id: str, trail: tuple[str, ...]) -> None: - if node_id in visiting: - raise DocForgeError( - "dependency_cycle", - "depends_on relationships contain a cycle", - path=(*trail, node_id), - ) - if node_id in visited: - return - visiting.add(node_id) - for target in dependencies[node_id]: - visit(target, (*trail, node_id)) - visiting.remove(node_id) - visited.add(node_id) - - for node_id in sorted(node_ids): - visit(node_id, ()) + states: dict[str, int] = {} + for root in sorted(node_ids): + if states.get(root) == 2: + continue + path: list[str] = [] + stack: list[tuple[str, int]] = [(root, 0)] + while stack: + node_id, child_index = stack[-1] + if states.get(node_id, 0) == 0: + states[node_id] = 1 + path.append(node_id) + targets = dependencies[node_id] + if child_index < len(targets): + target = targets[child_index] + stack[-1] = (node_id, child_index + 1) + state = states.get(target, 0) + 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: @@ -569,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: @@ -586,15 +714,18 @@ class Project: raise DocForgeError( "source_changed", "Project descriptor changed after the project was opened" ) - ordered_sources = self.canonical_source_paths() - captured = { - path: path.read_bytes() - for path in ( - self.descriptor.descriptor_path, - *self.descriptor.authority_files, - *ordered_sources, - ) - } + ordered_sources, ordered_directories = self._canonical_inventory() + generation_paths = ( + self.descriptor.descriptor_path, + *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] = [] @@ -617,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: @@ -626,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( @@ -635,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") - return ProjectSnapshot( + digest.update(GENERIC_SOURCE_CONTRACT.encode("ascii")) + source_hash = digest.hexdigest() + revision = _revision(self.descriptor.root) + snapshot = ProjectSnapshot( descriptor=self.descriptor, nodes=ordered_nodes, edges=ordered_edges, - source_hash=digest.hexdigest(), - revision=_revision(self.descriptor.root), + source_hash=source_hash, + 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() @@ -663,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, diff --git a/tests/test_changesets.py b/tests/test_changesets.py index bc61750..208591c 100644 --- a/tests/test_changesets.py +++ b/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"), ): diff --git a/tests/test_core.py b/tests/test_core.py index 444a164..e8075cc 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -21,7 +21,7 @@ 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.models import Edge # noqa: E402 +from docforge.models import Edge, ProjectState # noqa: E402 from docforge.project import Project, validate_graph # noqa: E402 FIXTURES = ROOT / "tests" / "fixtures" @@ -138,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") @@ -169,6 +170,23 @@ 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( @@ -177,7 +195,7 @@ class DocForgeCoreTests(unittest.TestCase): node_id=f"linear.node-{index:05d}", source_path=f"docs/node-{index:05d}.md", ) - for index in range(2_000) + for index in range(10_000) ) edges = CountingTuple( tuple( @@ -208,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) @@ -280,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)) - checked = index.build() - changed = {**checked, "source_hash": "0" * 64} + project = Project.open(root) + index = ProjectIndex(project) + 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"), ):