administrator/DocForge2
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages

Guard long-running adapter implementations

Browse source
This commit is contained in:
Andraxion 2026-07-28 19:44:25 -04:00
parent bb13258861
commit 1ef76f0271
11 changed files with 541 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

248
src/docforge/adapter_contract.py
View file

@ -3,9 +3,10 @@
from __future__ import annotations
import hashlib
import inspect
import json
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Protocol, runtime_checkable
@ -179,6 +180,18 @@ ProposalValidator = Callable[
],
None,
]
MAX_IMPLEMENTATION_DIFF_PATHS = 50
MAX_IMPLEMENTATION_FILES = 4_096
MAX_IMPLEMENTATION_BYTES = 64_000_000
@dataclass(frozen=True)
class AdapterImplementation:
"""One confined implementation boundary that must remain stable for a process."""
roots: tuple[Path, ...] = ()
files: tuple[Path, ...] = ()
suffixes: tuple[str, ...] = ()
@dataclass(frozen=True)
@ -194,6 +207,13 @@ class AdapterProjectSettings:
render: RenderConfig | None = None
limits: Limits | None = None
proposal_validator: ProposalValidator | None = None
implementation: AdapterImplementation | None = None
@dataclass(frozen=True)
class _ImplementationSnapshot:
fingerprint: str
files: tuple[tuple[str, str], ...]
class AdapterProject:
@ -262,6 +282,12 @@ class AdapterProject:
adapter_version,
root,
)
self._implementation = self._validate_implementation(
root,
loader,
self.settings.implementation,
descriptor_path=self.settings.descriptor_path,
)
self._cache_path = resolved_cache / "extractions.json"
self._last_build_report: dict[str, object] = {
"mode": "full",
@ -341,8 +367,10 @@ class AdapterProject:
limits=limits,
)
self._canonical_sources = canonical_sources
self._implementation_snapshot = self._capture_implementation(initial=True)
def load(self) -> ProjectSnapshot:
self.validate_runtime()
canonical_sources = self.canonical_source_paths()
captured = {path: path.read_bytes() for path in canonical_sources}
projection = (
@ -359,6 +387,7 @@ class AdapterProject:
)
if identity != self._identity:
raise DocForgeError("adapter_changed", "Adapter identity changed during the operation")
self.validate_runtime()
if self.canonical_source_paths() != canonical_sources or any(
not path.is_file() or path.read_bytes() != raw for path, raw in captured.items()
):
@ -382,6 +411,7 @@ class AdapterProject:
def incremental_state(self) -> ProjectState | None:
"""Return current source identity without reconstructing the complete projection."""
self.validate_runtime()
loader = self._incremental_loader
if loader is None:
return None
@ -391,6 +421,7 @@ class AdapterProject:
validate_manifest(manifest)
if manifest.identity() != self._identity:
raise DocForgeError("adapter_changed", "Adapter identity changed during the operation")
self.validate_runtime()
if self.canonical_source_paths() != canonical_sources or any(
not path.is_file() or path.read_bytes() != raw for path, raw in captured.items()
):
@ -412,6 +443,45 @@ class AdapterProject:
return dict(self._last_build_report)
def validate_runtime(self) -> None:
"""Reject use after declared adapter implementation files change."""
expected = self._implementation_snapshot
if expected is None:
return
current = self._capture_implementation()
if current is None:
raise DocForgeError(
"adapter_restart_required",
"Adapter implementation policy changed after the project server started",
)
if current == expected:
return
expected_files = dict(expected.files)
current_files = dict(current.files)
added = sorted(set(current_files) - set(expected_files))
deleted = sorted(set(expected_files) - set(current_files))
changed = sorted(
path
for path in set(expected_files) & set(current_files)
if expected_files[path] != current_files[path]
)
raise DocForgeError(
"adapter_restart_required",
"Adapter implementation changed after the project server started",
started_fingerprint=expected.fingerprint,
current_fingerprint=current.fingerprint,
added_count=len(added),
deleted_count=len(deleted),
changed_count=len(changed),
added=added[:MAX_IMPLEMENTATION_DIFF_PATHS],
deleted=deleted[:MAX_IMPLEMENTATION_DIFF_PATHS],
changed=changed[:MAX_IMPLEMENTATION_DIFF_PATHS],
paths_truncated=any(
len(paths) > MAX_IMPLEMENTATION_DIFF_PATHS for paths in (added, deleted, changed)
),
)
def logic_projection(self, owner_node_id: str) -> LogicProjection | None:
"""Load one lazily stored function-scoped logic projection."""
@ -433,6 +503,7 @@ class AdapterProject:
def verify_incremental_equivalence(self) -> dict[str, object]:
"""Prove the incremental and full loader contracts produce the same graph."""
self.validate_runtime()
if self._incremental_loader is None:
raise DocForgeError(
"incremental_disabled", "Adapter does not implement incremental extraction"
@ -456,6 +527,7 @@ class AdapterProject:
"Incremental extraction does not match a full adapter projection",
fields=mismatches,
)
self.validate_runtime()
return {
"status": "ok",
"project_id": incremental.project_id,
@ -656,6 +728,7 @@ class AdapterProject:
projected: ProjectSnapshot,
operations: tuple[Mapping[str, object], ...],
) -> None:
self.validate_runtime()
validator = self.settings.proposal_validator
if validator is None:
if operations:
@ -665,6 +738,179 @@ class AdapterProject:
)
return
validator(base, projected, operations)
self.validate_runtime()
@classmethod
def _validate_implementation(
cls,
root: Path,
loader: AdapterLoader,
implementation: AdapterImplementation | None,
*,
descriptor_path: Path | None,
) -> AdapterImplementation | None:
if implementation is None:
implementation = cls._infer_implementation(root, loader)
if descriptor_path is not None and (
implementation is None
or descriptor_path.resolve(strict=False)
not in {path.resolve(strict=False) for path in implementation.files}
):
implementation = (
AdapterImplementation(files=(descriptor_path,))
if implementation is None
else replace(
implementation,
files=(*implementation.files, descriptor_path),
)
)
if implementation is None:
return None
roots = cls._resolved_directories(
root,
implementation.roots,
label="implementation root",
)
files = cls._resolved_files(
root,
implementation.files,
label="implementation file",
)
suffixes = tuple(sorted(implementation.suffixes))
if (
not roots
and not files
or len(suffixes) != len(set(suffixes))
or any(
not suffix or not suffix.startswith(".") or "/" in suffix or "\\" in suffix
for suffix in suffixes
)
):
raise DocForgeError(
"invalid_adapter",
"Adapter implementation policy is invalid",
)
return AdapterImplementation(roots=roots, files=files, suffixes=suffixes)
@staticmethod
def _infer_implementation(
root: Path,
loader: AdapterLoader,
) -> AdapterImplementation | None:
source = inspect.getsourcefile(type(loader))
if source is None:
return None
try:
source_path = Path(source).resolve(strict=True)
except OSError:
return None
if not source_path.is_file() or not source_path.is_relative_to(root):
return None
module = inspect.getmodule(type(loader))
package = module.__package__.strip() if module and module.__package__ else ""
if package:
package_root = source_path.parent
for _ in package.split(".")[1:]:
package_root = package_root.parent
if package_root != root and (package_root / "__init__.py").is_file():
return AdapterImplementation(roots=(package_root,), suffixes=(".py",))
return AdapterImplementation(files=(source_path,))
def _capture_implementation(
self,
*,
initial: bool = False,
) -> _ImplementationSnapshot | None:
implementation = self._implementation
if implementation is None:
return None
root = self.descriptor.root
candidates = set(implementation.files)
if len(candidates) > MAX_IMPLEMENTATION_FILES:
self._raise_implementation_boundary_error(
initial,
"Adapter implementation boundary exceeds its file limit",
max_files=MAX_IMPLEMENTATION_FILES,
)
for implementation_root in implementation.roots:
if (
implementation_root.is_symlink()
or not implementation_root.is_dir()
or not implementation_root.resolve(strict=False).is_relative_to(root)
):
candidates.add(implementation_root)
continue
for path in implementation_root.rglob("*"):
if path.is_symlink() or (
(not implementation.suffixes or path.suffix in implementation.suffixes)
and path.is_file()
):
candidates.add(path)
if len(candidates) > MAX_IMPLEMENTATION_FILES:
self._raise_implementation_boundary_error(
initial,
"Adapter implementation boundary exceeds its file limit",
max_files=MAX_IMPLEMENTATION_FILES,
)
captured: list[tuple[str, str]] = []
unsafe: list[str] = []
total_bytes = 0
for path in sorted(candidates):
try:
relative = path.relative_to(root).as_posix()
if (
path.is_symlink()
or not path.is_file()
or not path.resolve(strict=True).is_relative_to(root)
):
unsafe.append(relative)
continue
size = path.stat().st_size
if size > MAX_IMPLEMENTATION_BYTES - total_bytes:
self._raise_implementation_boundary_error(
initial,
"Adapter implementation boundary exceeds its byte limit",
max_bytes=MAX_IMPLEMENTATION_BYTES,
)
raw = path.read_bytes()
total_bytes += len(raw)
if total_bytes > MAX_IMPLEMENTATION_BYTES:
self._raise_implementation_boundary_error(
initial,
"Adapter implementation boundary exceeds its byte limit",
max_bytes=MAX_IMPLEMENTATION_BYTES,
)
captured.append((relative, hashlib.sha256(raw).hexdigest()))
except (OSError, ValueError):
try:
unsafe.append(path.relative_to(root).as_posix())
except ValueError:
unsafe.append(str(path))
if unsafe:
self._raise_implementation_boundary_error(
initial,
"Adapter implementation boundary became missing or unsafe",
unsafe=sorted(unsafe),
)
digest = hashlib.sha256()
for relative, content_hash in captured:
encoded = relative.encode()
digest.update(len(encoded).to_bytes(8, "big"))
digest.update(encoded)
digest.update(bytes.fromhex(content_hash))
return _ImplementationSnapshot(digest.hexdigest(), tuple(captured))
@staticmethod
def _raise_implementation_boundary_error(
initial: bool,
message: str,
**details: object,
) -> None:
raise DocForgeError(
"invalid_adapter" if initial else "adapter_restart_required",
message,
**details,
)
@staticmethod
def _resolved_directories(