diff --git a/README.md b/README.md index c74aefe8..04bcd4c5 100644 --- a/README.md +++ b/README.md @@ -245,6 +245,8 @@ dogfood audience, and two distinct persistent Ed25519 proof keys: ```text CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON CONTEXT_ENGINE_WORKER_MAX_FILE_BYTES # optional +CONTEXT_ENGINE_WORKER_MAX_FILE_CHANGE_BASELINE_SIZE # optional +CONTEXT_ENGINE_WORKER_FILE_CURATED_SUBTREES_JSON # optional CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID CONTEXT_ENGINE_DOGFOOD_PRINCIPAL_REF CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_ID @@ -254,6 +256,32 @@ CONTEXT_ENGINE_FILE_CHANGE_CHECKPOINT_SIGNING_KEY_HEX CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX ``` +File scans keep ADR-0065's default limit of 10,000 Markdown paths. An operator +may explicitly set `CONTEXT_ENGINE_WORKER_MAX_FILE_CHANGE_BASELINE_SIZE` to a +positive integer no greater than 15,000; invalid values fail process +configuration. The effective value is signed with provider pages and retained +on durable scan provenance. Crossing it fails closed before a partial baseline +is accepted, and `status` reports the content-free closed condition +`scan_bound_exceeded` plus the effective bound. + +The alternative configuration path keeps the default bound and selects a +curated subtree per logical root. The value is JSON from each configured root +reference to one nonempty canonical relative directory, for example: + +```text +CONTEXT_ENGINE_WORKER_FILE_CURATED_SUBTREES_JSON={"maintainer-notes":"curated/notes"} +``` + +The registered root remains the descriptor-anchored read capability. Scan +traversal starts at the selected subtree while keeping full registered-root +relative path identities; switching a whole-root baseline to a selection that +would reinterpret active paths refuses before accepting a page. Absolute +paths, empty components, `.` / `..`, backslashes, and unknown root references +are refused at configuration time. Whole-vault versus curated subtree remains +a maintainer decision. Their synthetic measured costs and the single-command +reproduction are recorded in +[`2026-07-30-file-scan-baseline-measurement.md`](./docs/design/2026-07-30-file-scan-baseline-measurement.md). + Each proof-key value is exactly 32 random bytes encoded as 64 lowercase or uppercase hexadecimal characters. Keep both in the same local secret source across process restarts and never print or commit them. They must be distinct diff --git a/adapters/file_source.py b/adapters/file_source.py index 1976d87f..ea259fa0 100644 --- a/adapters/file_source.py +++ b/adapters/file_source.py @@ -19,9 +19,10 @@ from engine._opaque import decode_base64url, encode_base64url from engine.control import ( + DEFAULT_FILE_CHANGE_BASELINE_SIZE, FILE_CHANGE_CAPABILITY_MANIFEST, FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, - MAX_FILE_CHANGE_BASELINE_SIZE, + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, CapabilityStatus, ChangeCursor, ChangeLimit, @@ -39,6 +40,7 @@ ProviderInvalidCheckpoint, ProviderOk, ProviderRetryableUnavailable, + ProviderScanBoundExceeded, ProviderUnsupported, SourceChange, ) @@ -65,6 +67,7 @@ class FileReadLimits: """Server-owned hard ceiling for one acquired File payload.""" max_file_bytes: int + max_baseline_entries: int = DEFAULT_FILE_CHANGE_BASELINE_SIZE def __post_init__(self) -> None: if ( @@ -72,6 +75,13 @@ def __post_init__(self) -> None: or not 1 <= self.max_file_bytes <= MAX_CONFIGURED_FILE_BYTES ): raise ValueError("File byte ceiling must be a bounded positive integer") + if ( + type(self.max_baseline_entries) is not int + or not 1 + <= self.max_baseline_entries + <= MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE + ): + raise ValueError("File scan bound must be a bounded positive integer") @dataclass(frozen=True, slots=True) @@ -80,6 +90,11 @@ class _AnchoredRoot: display_path: Path descriptor: int + curated_subtree: tuple[str, ...] | None + + +class _FileScanBoundExceeded(LookupError): + """Internal traversal signal for the configured all-or-none scan fence.""" @dataclass(frozen=True, slots=True) @@ -136,11 +151,20 @@ def __init__( roots: Mapping[FileRootRef, Path], *, limits: FileReadLimits, + curated_subtrees: Mapping[FileRootRef, str] | None = None, ) -> None: if not isinstance(roots, Mapping) or not roots: raise ValueError("File root registry requires explicit bindings") if type(limits) is not FileReadLimits: raise TypeError("File root registry requires FileReadLimits") + selections = {} if curated_subtrees is None else curated_subtrees + if not isinstance(selections, Mapping) or any( + type(root_ref) is not FileRootRef + or root_ref not in roots + or not _canonical_relative_directory(value) + for root_ref, value in selections.items() + ): + raise ValueError("File curated subtrees require canonical root bindings") copied: dict[FileRootRef, _AnchoredRoot] = {} try: for root_ref, root_path in roots.items(): @@ -156,7 +180,12 @@ def __init__( raise ValueError( "File root must be an existing non-symlink directory" ) from None - copied[root_ref] = _AnchoredRoot(display_path, descriptor) + selection = selections.get(root_ref) + copied[root_ref] = _AnchoredRoot( + display_path, + descriptor, + None if selection is None else tuple(selection.split("/")), + ) except Exception: for root in copied.values(): os.close(root.descriptor) @@ -266,19 +295,50 @@ def _observe_markdown_files( raise LookupError("File root is not configured") observed: list[tuple[FileImportPath, bytes]] = [] snapshots: dict[str, tuple[_DirectoryEntry, ...]] = {} - self._observe_directory( - anchored.descriptor, - relative_prefix="", - observed=observed, - snapshots=snapshots, - ) - self._revalidate_directory_tree( - anchored.descriptor, - relative_prefix="", - snapshots=snapshots, - ) + descriptor = os.dup(anchored.descriptor) + relative_prefix = "" + try: + for component in anchored.curated_subtree or (): + try: + child = os.open( + component, + _DIRECTORY_OPEN_FLAGS, + dir_fd=descriptor, + ) + except OSError: + raise LookupError( + "File curated subtree is not available" + ) from None + os.close(descriptor) + descriptor = child + relative_prefix = ( + f"{relative_prefix}/{component}" + if relative_prefix + else component + ) + self._observe_directory( + descriptor, + relative_prefix=relative_prefix, + observed=observed, + snapshots=snapshots, + ) + self._revalidate_directory_tree( + descriptor, + relative_prefix=relative_prefix, + snapshots=snapshots, + ) + finally: + os.close(descriptor) return tuple(sorted(observed, key=lambda item: item[0].value.encode("utf-8"))) + def _curated_subtree_prefix(self, root_ref: FileRootRef) -> str | None: + anchored = self._roots.get(root_ref) + if anchored is None: + raise LookupError("File root is not configured") + if anchored.curated_subtree is None: + return None + return "/".join(anchored.curated_subtree) + "/" + def _observe_directory( self, descriptor: int, @@ -342,8 +402,8 @@ def _observe_directory( ): raise RuntimeError("File root observation is unstable") observed.append((path, payload)) - if len(observed) > MAX_FILE_CHANGE_BASELINE_SIZE: - raise LookupError("File root exceeds the configured scan bound") + if len(observed) > self._limits.max_baseline_entries: + raise _FileScanBoundExceeded if _directory_snapshot(descriptor, relative_prefix) != initial: raise RuntimeError("File root observation is unstable") @@ -470,6 +530,20 @@ def _safe_directory_component(name: object) -> bool: ) +def _canonical_relative_directory(value: object) -> bool: + """Return whether configuration names one nonempty root-relative directory.""" + + return ( + type(value) is str + and bool(value) + and value == value.strip() + and not value.startswith("/") + and "\\" not in value + and all(component not in {"", ".", ".."} for component in value.split("/")) + and not any(ord(character) < 0x20 for character in value) + ) + + _CURSOR_DOMAIN = "context-engine.file-change-cursor.v1" _SCAN_DOMAIN = b"context-engine.file-change-scan.v1\x00" @@ -540,6 +614,19 @@ def read_changes( ): return ProviderUnsupported("readChanges") try: + selection_prefix = self._registry._curated_subtree_prefix( + source.source_version.root_ref + ) + if ( + selection_prefix is not None + and source.complete_baseline is not None + and any( + entry.kind is FileChangeKind.UPSERT + and not entry.path.value.startswith(selection_prefix) + for entry in source.complete_baseline.entries + ) + ): + return ProviderGenericDenied() observed = tuple( _ObservedFile( path=path, @@ -550,14 +637,25 @@ def read_changes( source.source_version.root_ref ) ) + except _FileScanBoundExceeded: + return ProviderScanBoundExceeded( + scan_bound=self._registry._limits.max_baseline_entries + ) except LookupError: return ProviderGenericDenied() except RuntimeError: return ProviderRetryableUnavailable(timedelta(seconds=1)) changes, baseline_ref = self._changes(source, observed) - if len(changes) > MAX_FILE_CHANGE_BASELINE_SIZE: - return ProviderGenericDenied() - scan_ref = self._scan_ref(source, changes, baseline_ref) + if len(changes) > self._registry._limits.max_baseline_entries: + return ProviderScanBoundExceeded( + scan_bound=self._registry._limits.max_baseline_entries + ) + scan_ref = self._scan_ref( + source, + changes, + baseline_ref, + scan_bound=self._registry._limits.max_baseline_entries, + ) requested_limit = limit if ( type(cursor) is InitialScan @@ -594,6 +692,7 @@ def read_changes( scan_epoch=scan_epoch, limit=requested_limit, observed_count=len(changes), + scan_bound=self._registry._limits.max_baseline_entries, ): return ProviderInvalidCheckpoint() offset = cast(int, claims["offset"]) @@ -644,6 +743,7 @@ def read_changes( complete=complete, provider_proof="A" * 86, capability_version=capabilities.declaration_version, + scan_bound=self._registry._limits.max_baseline_entries, ) return ProviderOk( replace(unsigned, provider_proof=self._proofs._seal_page(unsigned)) @@ -750,11 +850,14 @@ def _scan_ref( source: FileChangeSource, observed: tuple[_ObservedChange, ...], baseline_ref: FileChangeBaselineRef | None, + *, + scan_bound: int = DEFAULT_FILE_CHANGE_BASELINE_SIZE, ) -> str: document: dict[str, object] = { "organizationId": str(source.organization_id), "sourceId": str(source.source_version.source_ref.value), "sourceVersionId": str(source.source_version.version_ref), + "scanBound": scan_bound, "entries": [ { "contentLength": item.content_length, @@ -783,6 +886,7 @@ def _scan_ref( "scanEpoch": str(baseline_ref.scan_epoch), "scanRef": baseline_ref.scan_ref, "sequence": baseline_ref.sequence, + "scanBound": baseline_ref.scan_bound, "sourceVersionId": str(baseline_ref.source_version_ref), } ) @@ -839,6 +943,7 @@ def _encode_cursor( "organizationId": str(source.organization_id), "scanEpoch": str(scan_epoch), "scanRef": scan_ref, + "scanBound": self._registry._limits.max_baseline_entries, "sourceId": str(source.source_version.source_ref.value), "sourceVersionId": str(source.source_version.version_ref), "version": 1, @@ -870,6 +975,7 @@ def _decode_cursor( "organizationId", "scanEpoch", "scanRef", + "scanBound", "sourceId", "sourceVersionId", "version", @@ -889,6 +995,7 @@ def _cursor_matches( scan_epoch: UUID, limit: ChangeLimit, observed_count: int, + scan_bound: int, ) -> bool: offset = claims.get("offset") return ( @@ -898,6 +1005,7 @@ def _cursor_matches( and claims.get("sourceId") == str(source.source_version.source_ref.value) and claims.get("sourceVersionId") == str(source.source_version.version_ref) and claims.get("scanRef") == scan_ref + and claims.get("scanBound") == scan_bound and claims.get("scanEpoch") == str(scan_epoch) and claims.get("limit") == limit.value and type(offset) is int diff --git a/applications/control.py b/applications/control.py index a2eb9cb0..7abd7a8d 100644 --- a/applications/control.py +++ b/applications/control.py @@ -462,6 +462,7 @@ def _scan_report_document(report: FileScanReport) -> dict[str, object]: "deletesObserved": report.deletes_observed, "importsScheduled": report.imports_scheduled, "pathsObserved": report.paths_observed, + "scanBound": report.scan_bound, "sourceRef": str(report.source_ref.value), } @@ -502,6 +503,9 @@ def _multi_scan_report_json(report: MultiSourceScanReport) -> str: source.paths_observed for source in sources ), "refusalCount": len(refusals), + "scanBounds": sorted( + {source.scan_bound for source in sources} + ), "sourceCount": len(report.outcomes), }, }, @@ -577,12 +581,16 @@ def _status_document_with_refusals( "scanEpoch": str(head.scan_epoch), "scanRef": head.scan_ref, "sequence": head.sequence, + "scanBound": head.scan_bound, "sourceVersionRef": str(head.source_version_ref), } ), "completeChangeBaselineSize": ( 0 if baseline is None else len(baseline.entries) ), + "completeChangeBaselineScanBound": ( + None if baseline is None else baseline.reference.scan_bound + ), "lastSuccessfulAcquisition": last_successful_acquisition, "publishWatermark": ( None @@ -596,6 +604,14 @@ def _status_document_with_refusals( } ), "refusals": refusals, + "scanRefusal": ( + None + if status.scan_refusal_category is None + else { + "category": status.scan_refusal_category.value, + "scanBound": status.scan_refusal_bound, + } + ), "sourceRef": str(progress.source_ref.value), } diff --git a/applications/file_root_configuration.py b/applications/file_root_configuration.py index a09051ef..09fb5d48 100644 --- a/applications/file_root_configuration.py +++ b/applications/file_root_configuration.py @@ -8,11 +8,15 @@ from pathlib import Path from adapters.file_source import FileReadLimits, FileRootRegistry -from engine.control import FileRootRef +from engine.control import DEFAULT_FILE_CHANGE_BASELINE_SIZE, FileRootRef DEFAULT_WORKER_MAX_FILE_BYTES = 1_048_576 WORKER_MAX_FILE_BYTES_ENV = "CONTEXT_ENGINE_WORKER_MAX_FILE_BYTES" WORKER_FILE_ROOTS_ENV = "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON" +WORKER_FILE_CURATED_SUBTREES_ENV = "CONTEXT_ENGINE_WORKER_FILE_CURATED_SUBTREES_JSON" +WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV = ( + "CONTEXT_ENGINE_WORKER_MAX_FILE_CHANGE_BASELINE_SIZE" +) def required_environment( @@ -35,12 +39,32 @@ def file_read_limits( source = os.environ if environment is None else environment raw_limit = source.get(WORKER_MAX_FILE_BYTES_ENV) + raw_scan_bound = source.get(WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV) + if raw_scan_bound is None: + scan_bound = DEFAULT_FILE_CHANGE_BASELINE_SIZE + elif ( + not raw_scan_bound + or raw_scan_bound != raw_scan_bound.strip() + or not raw_scan_bound.isdecimal() + ): + raise ValueError("Supply worker configuration is not available") + else: + scan_bound = int(raw_scan_bound) if raw_limit is None: - return FileReadLimits(max_file_bytes=DEFAULT_WORKER_MAX_FILE_BYTES) + try: + return FileReadLimits( + max_file_bytes=DEFAULT_WORKER_MAX_FILE_BYTES, + max_baseline_entries=scan_bound, + ) + except ValueError: + raise ValueError("Supply worker configuration is not available") from None if not raw_limit or raw_limit != raw_limit.strip() or not raw_limit.isdecimal(): raise ValueError("Supply worker configuration is not available") try: - return FileReadLimits(max_file_bytes=int(raw_limit)) + return FileReadLimits( + max_file_bytes=int(raw_limit), + max_baseline_entries=scan_bound, + ) except ValueError: raise ValueError("Supply worker configuration is not available") from None @@ -73,6 +97,48 @@ def file_root_bindings( return bindings +def file_curated_subtrees( + environment: Mapping[str, str] | None = None, +) -> dict[FileRootRef, str]: + """Load explicit root-relative traversal selections without remapping roots.""" + + source = os.environ if environment is None else environment + bindings = file_root_bindings(environment) + raw_subtrees = source.get(WORKER_FILE_CURATED_SUBTREES_ENV) + if raw_subtrees is None: + return {} + try: + document = json.loads(raw_subtrees) + except json.JSONDecodeError: + raise ValueError("Supply worker configuration is not available") from None + raw_refs = {root_ref.value for root_ref in bindings} + if ( + type(document) is not dict + or any( + type(raw_ref) is not str + or raw_ref not in raw_refs + or not _canonical_subtree(selection) + for raw_ref, selection in document.items() + ) + ): + raise ValueError("Supply worker configuration is not available") + return { + FileRootRef(raw_ref): selection for raw_ref, selection in document.items() + } + + +def _canonical_subtree(value: object) -> bool: + return ( + type(value) is str + and bool(value) + and value == value.strip() + and not value.startswith("/") + and "\\" not in value + and all(component not in {"", ".", ".."} for component in value.split("/")) + and not any(ord(character) < 0x20 for character in value) + ) + + def file_roots( environment: Mapping[str, str] | None = None, ) -> FileRootRegistry: @@ -81,4 +147,5 @@ def file_roots( return FileRootRegistry( file_root_bindings(environment), limits=file_read_limits(environment), + curated_subtrees=file_curated_subtrees(environment), ) diff --git a/applications/file_scan.py b/applications/file_scan.py index b80c2844..4884382a 100644 --- a/applications/file_scan.py +++ b/applications/file_scan.py @@ -42,6 +42,7 @@ FileSourceProgress, InitialScan, ProviderOk, + ProviderScanBoundExceeded, ScheduledFileChangePage, ScheduleFileChangePage, SourceManifest, @@ -77,6 +78,7 @@ class FileScanReport: deletes_observed: int compilation_refusals: int advanced_cursor: str | None + scan_bound: int class SourceScanRefused(SourceNotAvailable): @@ -188,6 +190,20 @@ def scan_file_source( cursor, ChangeLimit(FILE_SCAN_PAGE_LIMIT), ) + if type(proposed) is ProviderScanBoundExceeded: + with authority.authorize( + opaque_credential=opaque_credential, + operation=ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + request_id=f"local-scan-bound-refusal-{uuid4().hex}", + ) as call: + if call.organization_id != organization_id: + raise SourceNotAvailable + control.report_file_scan_bound_refusal( + call, + source_ref, + proposed.scan_bound, + ) + raise SourceNotAvailable if type(proposed) is not ProviderOk: raise SourceScanRefused page = proposed.value @@ -201,6 +217,14 @@ def scan_file_source( baseline = source.complete_baseline if baseline is None: # pragma: no cover - proven by the predicate raise SourceNotAvailable + with authority.authorize( + opaque_credential=opaque_credential, + operation=ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + request_id=f"local-scan-bound-clear-{uuid4().hex}", + ) as call: + if call.organization_id != organization_id: + raise SourceNotAvailable + control.clear_file_scan_bound_refusal(call, source_ref) return FileScanReport( source_ref=source_ref, paths_observed=sum( @@ -211,6 +235,7 @@ def scan_file_source( deletes_observed=0, compilation_refusals=compilation_refusals, advanced_cursor=baseline.reference.checkpoint_ref, + scan_bound=baseline.reference.scan_bound, ) observed = tuple(page.changes) for change in observed: @@ -277,6 +302,7 @@ def scan_file_source( deletes_observed=deletes_observed, compilation_refusals=compilation_refusals, advanced_cursor=advanced_cursor, + scan_bound=roots._limits.max_baseline_entries, ) @@ -463,6 +489,7 @@ def _replays_complete_baseline( head.page_ref, head.checkpoint_ref, head.sequence, + head.scan_bound, ) == ( reference.source_version_ref, @@ -471,6 +498,7 @@ def _replays_complete_baseline( reference.page_ref, reference.checkpoint_ref, reference.sequence, + reference.scan_bound, ) and scan_ref == reference.scan_ref and scan_epoch == reference.scan_epoch diff --git a/applications/file_scan_measurement.py b/applications/file_scan_measurement.py new file mode 100644 index 00000000..c5b94355 --- /dev/null +++ b/applications/file_scan_measurement.py @@ -0,0 +1,297 @@ +"""Measure the bounded File provider against generated synthetic trees.""" + +from __future__ import annotations + +import argparse +import json +import math +import platform +import sys +import tracemalloc +from collections.abc import Callable, Sequence +from dataclasses import replace +from datetime import UTC, datetime +from pathlib import Path +from tempfile import TemporaryDirectory +from time import perf_counter +from typing import cast +from uuid import UUID + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from adapters.file_source import FileChangeProvider, FileReadLimits, FileRootRegistry +from applications.file_scan import FILE_SCAN_PAGE_LIMIT +from engine._opaque import encode_base64url +from engine.control import ( + FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, + ChangeCursor, + ChangeLimit, + FileChangeControlProofs, + FileChangeProviderProofs, + FileChangeScanHead, + FileChangeSource, + FileRootRef, + InitialScan, + PendingChangeCursor, + ProviderOk, + SourceManifest, + SourceRef, +) +from engine.control.file_change_pages import _accepted_cursor_payload + +MEASUREMENT_SIZES = (1_000, 5_000, 10_000, 15_000) +SCHEMA_VERSION = "context-engine-file-scan-measurement-v1" +_ROOT_REF = FileRootRef("synthetic-scan-root") +_ORGANIZATION_ID = UUID("0c499906-b9b0-4865-a6f5-45bc35178a90") +_SOURCE_ID = UUID("b1712c37-d2d1-4e23-834a-0f49e137268c") +_SOURCE_VERSION_ID = UUID("4452cf6a-88c3-470b-b3e4-5be2b4897b0a") +_SCAN_EPOCH_CHECKPOINT = "facp_" + "b" * 64 +_PROVIDER_KEY = Ed25519PrivateKey.from_private_bytes(bytes.fromhex("11" * 32)) +_CHECKPOINT_KEY = Ed25519PrivateKey.from_private_bytes(bytes.fromhex("22" * 32)) + + +class MeasurementUnavailable(RuntimeError): + """The synthetic provider measurement could not produce its exact contract.""" + + +def _source() -> FileChangeSource: + manifest = SourceManifest.registered_file( + source_ref=SourceRef(_SOURCE_ID), + version_ref=_SOURCE_VERSION_ID, + display_name="Synthetic scan measurement", + root_ref=_ROOT_REF, + created_at=datetime(2026, 7, 30, tzinfo=UTC), + capabilities=FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, + ) + return FileChangeSource(_ORGANIZATION_ID, manifest.active_version) + + +def _timed(call: Callable[[], object]) -> tuple[object, float, int]: + tracemalloc.start() + started = perf_counter() + try: + value = call() + elapsed = perf_counter() - started + _current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + return value, elapsed, peak + + +def _generate_tree(root: Path, path_count: int) -> None: + directories = tuple(root / f"group-{ordinal:03d}" for ordinal in range(100)) + for directory in directories: + directory.mkdir() + for ordinal in range(path_count): + directory = directories[ordinal % len(directories)] + (directory / f"entry-{ordinal:05d}.md").write_bytes(b"# Synthetic\n") + + +def measure_size( + path_count: int, + *, + curated_subtree: bool = False, +) -> dict[str, int | float | str]: + """Measure one generated tree through the production provider seam.""" + + if type(path_count) is not int or path_count < 2 or path_count > 15_000: + raise MeasurementUnavailable("synthetic path count is unavailable") + with TemporaryDirectory(prefix="context-engine-synthetic-scan-") as directory: + root = Path(directory).resolve(strict=True) + observed_root = root / "curated" if curated_subtree else root + if curated_subtree: + observed_root.mkdir() + _generate_tree(observed_root, path_count) + registry = FileRootRegistry( + {_ROOT_REF: root}, + limits=FileReadLimits( + max_file_bytes=1_024, + max_baseline_entries=path_count, + ), + curated_subtrees=({_ROOT_REF: "curated"} if curated_subtree else None), + ) + try: + provider_proofs = FileChangeProviderProofs( + provider_signing_key=_PROVIDER_KEY, + checkpoint_verification_key=_CHECKPOINT_KEY.public_key(), + ) + provider = FileChangeProvider(registry, proofs=provider_proofs) + source = _source() + initial_raw, initial_seconds, initial_peak = _timed( + lambda: provider.read_changes( + source, + InitialScan(), + ChangeLimit(FILE_SCAN_PAGE_LIMIT), + ) + ) + if type(initial_raw) is not ProviderOk: + raise MeasurementUnavailable("initial synthetic scan was refused") + initial = initial_raw.value + pending = initial.next_cursor + if type(pending) is not PendingChangeCursor or initial.complete: + raise MeasurementUnavailable("synthetic continuation was unavailable") + verified = FileChangeControlProofs( + provider_verification_key=_PROVIDER_KEY.public_key() + ).verify_page(initial) + if verified is None: + raise MeasurementUnavailable("synthetic page proof was unavailable") + payload = _accepted_cursor_payload( + organization_id=_ORGANIZATION_ID, + source_ref=SourceRef(_SOURCE_ID), + source_version_ref=_SOURCE_VERSION_ID, + scan_ref=initial.scan_ref, + scan_epoch=initial.scan_epoch, + page_ref=verified.page_ref, + checkpoint_ref=_SCAN_EPOCH_CHECKPOINT, + sequence=1, + pending_cursor=pending, + ) + cursor = ChangeCursor( + f"{encode_base64url(payload)}." + f"{encode_base64url(_CHECKPOINT_KEY.sign(payload))}" + ) + continued_source = replace( + source, + scan_head=FileChangeScanHead( + source_version_ref=_SOURCE_VERSION_ID, + scan_ref=initial.scan_ref, + scan_epoch=initial.scan_epoch, + page_limit=FILE_SCAN_PAGE_LIMIT, + page_ref=verified.page_ref, + checkpoint_ref=_SCAN_EPOCH_CHECKPOINT, + sequence=1, + complete=False, + scan_bound=path_count, + ), + ) + continuation_raw, continuation_seconds, continuation_peak = _timed( + lambda: provider.read_changes( + continued_source, + cursor, + ChangeLimit(FILE_SCAN_PAGE_LIMIT), + ) + ) + if type(continuation_raw) is not ProviderOk: + raise MeasurementUnavailable("synthetic continuation was refused") + page_count = math.ceil(path_count / FILE_SCAN_PAGE_LIMIT) + estimated_cycle_seconds = initial_seconds + ( + (page_count - 1) * continuation_seconds + ) + return { + "continuationPeakMemoryBytes": continuation_peak, + "continuationWallClockSeconds": round(continuation_seconds, 6), + "estimatedSingletonCycleSeconds": round( + estimated_cycle_seconds, 3 + ), + "initialPeakMemoryBytes": initial_peak, + "initialWallClockSeconds": round(initial_seconds, 6), + "measurementRef": ( + f"synthetic-curated-{path_count}" + if curated_subtree + else f"synthetic-{path_count}" + ), + "pageCount": page_count, + "pathCount": path_count, + "peakMemoryBytes": max(initial_peak, continuation_peak), + } + finally: + registry.close() + + +def run_measurement() -> dict[str, object]: + """Return aggregate-only results for the fixed representative sizes.""" + + measurements = [measure_size(size) for size in MEASUREMENT_SIZES] + by_size = {cast(int, value["pathCount"]): value for value in measurements} + curated_measurement = measure_size(5_000, curated_subtree=True) + + def option( + name: str, + measured: dict[str, int | float | str], + ) -> dict[str, object]: + return { + "label": name, + "representativePathCount": measured["pathCount"], + "measurementRef": measured["measurementRef"], + "initialWallClockSeconds": measured["initialWallClockSeconds"], + "continuationWallClockSeconds": measured[ + "continuationWallClockSeconds" + ], + "peakMemoryBytes": measured["peakMemoryBytes"], + "pageCount": measured["pageCount"], + "estimatedSingletonCycleSeconds": measured[ + "estimatedSingletonCycleSeconds" + ], + } + + return { + "schemaVersion": SCHEMA_VERSION, + "measuredAt": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "environment": { + "machine": platform.machine(), + "python": platform.python_version(), + "system": platform.system(), + }, + "method": { + "generatedTree": True, + "pageLimit": FILE_SCAN_PAGE_LIMIT, + "productionProviderSeam": True, + "curatedOptionUsesConfiguredTraversal": True, + "singletonCycleEstimate": ( + "initial call plus pageCount minus one times the measured signed " + "continuation call" + ), + }, + "measurements": measurements, + "options": { + "configurableWholeVault": option( + "configurable whole-vault bound", by_size[15_000] + ), + "curatedSubtree": option( + "illustrative curated subtree", + curated_measurement, + ), + }, + } + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="context-engine-file-scan-measurement", + description="Measure recursive File scans over generated synthetic trees.", + ) + parser.add_argument("--output", type=Path, required=True) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + arguments = _parser().parse_args(argv) + try: + report = run_measurement() + output = cast(Path, arguments.output) + repository_root = Path.cwd().resolve() + state_root = repository_root / ".context-engine" + tracked_report = ( + repository_root + / "docs/evaluation/2026-07-30-file-scan-measurement.json" + ) + if ( + not output.resolve().is_relative_to(state_root) + and output.resolve() != tracked_report + ): + raise MeasurementUnavailable( + "measurement output target is unavailable" + ) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text( + json.dumps(report, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return 0 + except (MeasurementUnavailable, OSError) as failure: + print(str(failure), file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/applications/worker.py b/applications/worker.py index fc71fd33..4d86865a 100644 --- a/applications/worker.py +++ b/applications/worker.py @@ -25,6 +25,9 @@ from applications.file_root_configuration import ( DEFAULT_WORKER_MAX_FILE_BYTES as _DEFAULT_WORKER_MAX_FILE_BYTES, ) +from applications.file_root_configuration import ( + file_curated_subtrees as _file_dispatch_curated_subtrees, +) from applications.file_root_configuration import ( file_read_limits as _configured_file_read_limits, ) @@ -532,6 +535,7 @@ def _run_file_dispatch(*, single_cycle: bool) -> int: roots = FileRootRegistry( root_bindings, limits=_file_read_limits(), + curated_subtrees=_file_dispatch_curated_subtrees(), ) try: authority = PostgreSQLFileDispatchAuthority( diff --git a/docs/design/2026-07-30-file-scan-baseline-measurement.md b/docs/design/2026-07-30-file-scan-baseline-measurement.md new file mode 100644 index 00000000..3e6a258a --- /dev/null +++ b/docs/design/2026-07-30-file-scan-baseline-measurement.md @@ -0,0 +1,63 @@ +# File scan bound measurement + +**Measured:** 2026-07-30 on Python 3.13.5, Darwin arm64. The tracked aggregate +result is +[`2026-07-30-file-scan-measurement.json`](../evaluation/2026-07-30-file-scan-measurement.json). + +## Reproduction + +Run the measurement from the repository root. It generates all inputs beneath +a temporary directory and overwrites the aggregate-only tracked report: + +```bash +uv run context-engine-file-scan-measurement \ + --output docs/evaluation/2026-07-30-file-scan-measurement.json +``` + +The command creates 100 directories and distributes fixed, constant-content +Markdown files across them. It calls the production `FileChangeProvider` at +1,000, 5,000, 10,000, and 15,000 paths with the production singleton page +limit. For each size it measures one initial call and one signed continuation +call with `perf_counter`, and records peak Python allocations with +`tracemalloc`. The full-cycle value is an estimate, not an executed full +cycle: initial-call time plus page count minus one multiplied by the measured +continuation-call time. That calculation exposes ADR-0071's current full-root +revalidation cost without spending many hours executing every singleton page. + +## Results + +| Paths | Initial wall time | Continuation wall time | Peak Python memory | Pages | Estimated singleton cycle | +|---:|---:|---:|---:|---:|---:| +| 1,000 | 0.261 s | 0.236 s | 0.89 MiB | 1,000 | 236 s | +| 5,000 | 1.188 s | 1.149 s | 4.14 MiB | 5,000 | 5,744 s | +| 10,000 | 2.359 s | 2.433 s | 8.20 MiB | 10,000 | 24,333 s | +| 15,000 | 3.647 s | 3.612 s | 12.23 MiB | 15,000 | 54,181 s | + +Peak memory grows approximately linearly across the measured sizes and stays +bounded by the configured path ceiling. Wall time for a single snapshot also +grows approximately linearly. The composed singleton cycle is operationally +material because every continuation revalidates the whole tree; the estimated +cost grows quadratically, which trips ADR-0071's revisit trigger. + +## Maintainer-gated options + +Both supported configuration paths are measured and neither is selected here: + +- **Configurable whole-vault bound:** the 15,000-path measurement uses 12.23 + MiB peak Python memory, takes 3.647 seconds for the initial snapshot and + 3.612 seconds for one continuation, emits 15,000 singleton pages, and yields + a 54,181-second full-cycle estimate. +- **Curated subtree:** the configured 5,000-path selection uses 4.26 MiB peak + Python memory, takes 1.165 seconds for the initial snapshot and 1.165 seconds + for one continuation, emits 5,000 singleton pages, and yields a 5,826-second + full-cycle estimate. Actual cost follows the selected subtree's path count. + +The configurable path does not change ADR-0065's 10,000 default. Changing the +default later requires an ADR that refines ADR-0065. Issue #126 may re-house +File traversal in a connector runner; if it replaces this provider path, rerun +this measurement against that implementation. Making the composed scan fast +remains the selected-upsert or restart-safe-snapshot work owned by ADR-0071, +not this measurement. +The curated option is measured through the explicit curated-subtree registry +configuration while retaining the registered root as the read capability and +preserving root-relative path identities. diff --git a/docs/evaluation/2026-07-30-file-scan-measurement.json b/docs/evaluation/2026-07-30-file-scan-measurement.json new file mode 100644 index 00000000..220d62da --- /dev/null +++ b/docs/evaluation/2026-07-30-file-scan-measurement.json @@ -0,0 +1,84 @@ +{ + "environment": { + "machine": "arm64", + "python": "3.13.5", + "system": "Darwin" + }, + "measuredAt": "2026-07-30T16:29:30.286524Z", + "measurements": [ + { + "continuationPeakMemoryBytes": 849094, + "continuationWallClockSeconds": 0.236323, + "estimatedSingletonCycleSeconds": 236.348, + "initialPeakMemoryBytes": 934926, + "initialWallClockSeconds": 0.261369, + "measurementRef": "synthetic-1000", + "pageCount": 1000, + "pathCount": 1000, + "peakMemoryBytes": 934926 + }, + { + "continuationPeakMemoryBytes": 4345406, + "continuationWallClockSeconds": 1.148789, + "estimatedSingletonCycleSeconds": 5743.982, + "initialPeakMemoryBytes": 4345710, + "initialWallClockSeconds": 1.187794, + "measurementRef": "synthetic-5000", + "pageCount": 5000, + "pathCount": 5000, + "peakMemoryBytes": 4345710 + }, + { + "continuationPeakMemoryBytes": 8593552, + "continuationWallClockSeconds": 2.433297, + "estimatedSingletonCycleSeconds": 24332.896, + "initialPeakMemoryBytes": 8579008, + "initialWallClockSeconds": 2.359169, + "measurementRef": "synthetic-10000", + "pageCount": 10000, + "pathCount": 10000, + "peakMemoryBytes": 8593552 + }, + { + "continuationPeakMemoryBytes": 12819760, + "continuationWallClockSeconds": 3.612044, + "estimatedSingletonCycleSeconds": 54180.696, + "initialPeakMemoryBytes": 12805464, + "initialWallClockSeconds": 3.647206, + "measurementRef": "synthetic-15000", + "pageCount": 15000, + "pathCount": 15000, + "peakMemoryBytes": 12819760 + } + ], + "method": { + "curatedOptionUsesConfiguredTraversal": true, + "generatedTree": true, + "pageLimit": 1, + "productionProviderSeam": true, + "singletonCycleEstimate": "initial call plus pageCount minus one times the measured signed continuation call" + }, + "options": { + "configurableWholeVault": { + "continuationWallClockSeconds": 3.612044, + "estimatedSingletonCycleSeconds": 54180.696, + "initialWallClockSeconds": 3.647206, + "label": "configurable whole-vault bound", + "measurementRef": "synthetic-15000", + "pageCount": 15000, + "peakMemoryBytes": 12819760, + "representativePathCount": 15000 + }, + "curatedSubtree": { + "continuationWallClockSeconds": 1.165121, + "estimatedSingletonCycleSeconds": 5825.604, + "initialWallClockSeconds": 1.165012, + "label": "illustrative curated subtree", + "measurementRef": "synthetic-curated-5000", + "pageCount": 5000, + "peakMemoryBytes": 4466215, + "representativePathCount": 5000 + } + }, + "schemaVersion": "context-engine-file-scan-measurement-v1" +} diff --git a/engine/control/__init__.py b/engine/control/__init__.py index e76cb8d0..a0f1a764 100644 --- a/engine/control/__init__.py +++ b/engine/control/__init__.py @@ -35,6 +35,8 @@ SourceVersion, ) from engine.control.file_change_pages import ( + DEFAULT_FILE_CHANGE_BASELINE_SIZE, + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, MAX_FILE_CHANGE_BASELINE_SIZE, MAX_FILE_CHANGE_PAGE_SIZE, AcceptedChangePage, @@ -56,6 +58,7 @@ ProviderInvalidCheckpoint, ProviderOk, ProviderRetryableUnavailable, + ProviderScanBoundExceeded, ProviderUnsupported, SourceChange, VerifiedChangePage, @@ -84,6 +87,7 @@ from engine.control.file_source_progress import ( FileCompilationRefusal, FileCompilationRefusalCategory, + FileScanRefusalCategory, FileSourceAcquisitionCheckpoint, FileSourceChangeKind, FileSourceProgress, @@ -101,6 +105,8 @@ from engine.source_acl import SourceAclEvidenceMode __all__ = [ + "DEFAULT_FILE_CHANGE_BASELINE_SIZE", + "MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE", "MAX_FILE_CHANGE_BASELINE_SIZE", "MAX_FILE_CHANGE_PAGE_SIZE", "AcceptedChangePage", @@ -142,6 +148,7 @@ "FileSourceAcquisitionCheckpoint", "FileSourceChangeKind", "FileSourceProgress", + "FileScanRefusalCategory", "FileSourceCleanupState", "FileSourceOffboarding", "FileSourcePublishOutcome", @@ -163,6 +170,7 @@ "ProviderInvalidCheckpoint", "ProviderOk", "ProviderRetryableUnavailable", + "ProviderScanBoundExceeded", "ProviderUnsupported", "SourceAclEvidenceMode", "SourceControlUnavailable", diff --git a/engine/control/file_change_pages.py b/engine/control/file_change_pages.py index 659ec5e2..f9f1bffe 100644 --- a/engine/control/file_change_pages.py +++ b/engine/control/file_change_pages.py @@ -28,7 +28,20 @@ from engine.control.file_imports import FileImportPath MAX_FILE_CHANGE_PAGE_SIZE = 100 -MAX_FILE_CHANGE_BASELINE_SIZE = 10_000 +DEFAULT_FILE_CHANGE_BASELINE_SIZE = 10_000 +MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE = 15_000 +# ADR-0065's unchanged default. Individual scans carry an explicit configured +# bound up to MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE as provenance. +MAX_FILE_CHANGE_BASELINE_SIZE = DEFAULT_FILE_CHANGE_BASELINE_SIZE + + +def _require_scan_bound(value: object) -> int: + if ( + type(value) is not int + or not 1 <= value <= MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE + ): + raise ValueError("File scan bound must be a bounded positive integer") + return value @dataclass(frozen=True, slots=True) @@ -96,6 +109,7 @@ class FileChangeScanHead: checkpoint_ref: str = field(repr=False) sequence: int complete: bool + scan_bound: int = DEFAULT_FILE_CHANGE_BASELINE_SIZE superseded_scan_epoch: UUID | None = field(default=None, repr=False) def __post_init__(self) -> None: @@ -115,6 +129,7 @@ def __post_init__(self) -> None: raise ValueError("File change scan head sequence is invalid") if type(self.complete) is not bool: raise TypeError("File change scan head complete must be bool") + _require_scan_bound(self.scan_bound) if ( self.superseded_scan_epoch is not None and type(self.superseded_scan_epoch) is not UUID @@ -132,6 +147,7 @@ class FileChangeBaselineRef: page_ref: str = field(repr=False) checkpoint_ref: str = field(repr=False) sequence: int + scan_bound: int = DEFAULT_FILE_CHANGE_BASELINE_SIZE comparison_baseline_ref: FileChangeBaselineRef | None = field( default=None, repr=False, @@ -147,6 +163,7 @@ def __post_init__(self) -> None: _require_checkpoint_ref(self.checkpoint_ref) if type(self.sequence) is not int or not 1 <= self.sequence <= 2**63 - 1: raise ValueError("File change baseline sequence is invalid") + _require_scan_bound(self.scan_bound) if self.comparison_baseline_ref is not None: if type(self.comparison_baseline_ref) is not FileChangeBaselineRef: raise TypeError("File change comparison baseline is invalid") @@ -191,7 +208,8 @@ def __post_init__(self) -> None: raise TypeError("File change baseline reference is invalid") if ( type(self.entries) is not tuple - or len(self.entries) > MAX_FILE_CHANGE_BASELINE_SIZE + or len(self.entries) > MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE + or len(self.entries) > self.reference.scan_bound or any(type(entry) is not FileChangeBaselineEntry for entry in self.entries) ): raise TypeError("File change baseline entries must be bounded") @@ -298,6 +316,7 @@ class ChangePage: provider_proof: str = field(repr=False) baseline_ref: FileChangeBaselineRef | None = field(default=None, repr=False) capability_version: str = "file-capabilities-v3" + scan_bound: int = DEFAULT_FILE_CHANGE_BASELINE_SIZE def __post_init__(self) -> None: if any( @@ -322,6 +341,7 @@ def __post_init__(self) -> None: or not 1 <= self.page_limit <= MAX_FILE_CHANGE_PAGE_SIZE ): raise ValueError("ChangePage page_limit is invalid") + _require_scan_bound(self.scan_bound) if self.predecessor_page_ref is not None: _require_sha256( "ChangePage predecessor_page_ref", self.predecessor_page_ref @@ -424,6 +444,7 @@ def _page_document(page: ChangePage) -> dict[str, object]: "predecessorPageRef": page.predecessor_page_ref, "predecessorSequence": page.predecessor_sequence, "scanEpoch": str(page.scan_epoch), + "scanBound": page.scan_bound, "scanRef": page.scan_ref, "supersededScanEpoch": ( None @@ -450,6 +471,7 @@ def _baseline_reference_document( "scanEpoch": str(reference.scan_epoch), "scanRef": reference.scan_ref, "sequence": reference.sequence, + "scanBound": reference.scan_bound, "sourceVersionId": str(reference.source_version_ref), } @@ -543,6 +565,7 @@ class AcceptedChangePage: complete: bool next_cursor: ChangeCursor | None = field(repr=False) accepted_at: datetime + scan_bound: int = DEFAULT_FILE_CHANGE_BASELINE_SIZE def __post_init__(self) -> None: _validate_acceptance_fields( @@ -563,6 +586,7 @@ def __post_init__(self) -> None: or not 1 <= self.page_limit <= MAX_FILE_CHANGE_PAGE_SIZE ): raise ValueError("accepted page page_limit is invalid") + _require_scan_bound(self.scan_bound) if ( self.superseded_scan_epoch is not None and type(self.superseded_scan_epoch) is not UUID @@ -586,6 +610,7 @@ def scan_head(self) -> FileChangeScanHead: checkpoint_ref=self.checkpoint_ref, sequence=self.sequence, complete=self.complete, + scan_bound=self.scan_bound, superseded_scan_epoch=self.superseded_scan_epoch, ) @@ -803,10 +828,21 @@ class ProviderGenericDenied: """One non-enumerating refusal for an unavailable source binding.""" +@dataclass(frozen=True, slots=True) +class ProviderScanBoundExceeded: + """Closed content-free refusal that tells an operator which limit to act on.""" + + scan_bound: int + + def __post_init__(self) -> None: + _require_scan_bound(self.scan_bound) + + FileChangeProviderOutcome = ( ProviderOk[ChangePage] | ProviderUnsupported | ProviderRetryableUnavailable | ProviderInvalidCheckpoint | ProviderGenericDenied + | ProviderScanBoundExceeded ) diff --git a/engine/control/file_source_progress.py b/engine/control/file_source_progress.py index dbe01ed8..b270bc19 100644 --- a/engine/control/file_source_progress.py +++ b/engine/control/file_source_progress.py @@ -14,6 +14,7 @@ _require_token, _require_utc, ) +from engine.control.file_change_pages import MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE if TYPE_CHECKING: from engine.control.file_change_pages import FileChangeBaseline, FileChangeScanHead @@ -46,6 +47,12 @@ class FileCompilationRefusalCategory(StrEnum): UNSUPPORTED_DOCUMENT_SHAPE = "unsupported_document_shape" +class FileScanRefusalCategory(StrEnum): + """Closed content-free scan-level conditions retained for operations.""" + + SCAN_BOUND_EXCEEDED = "scan_bound_exceeded" + + @dataclass(frozen=True, slots=True) class PendingFileChangeSchedule: """One accepted current-scan page whose upserts have no durable jobs.""" @@ -83,6 +90,8 @@ class FileSourceStatus: last_successful_acquisition_at: datetime | None last_successful_acquisition_age_seconds: int | None refusals: tuple[FileCompilationRefusal, ...] = () + scan_refusal_category: FileScanRefusalCategory | None = None + scan_refusal_bound: int | None = None def __post_init__(self) -> None: _require_utc("File Source status observed_at", self.observed_at) @@ -113,6 +122,17 @@ def __post_init__(self) -> None: raise ValueError("File Source refusal paths require canonical order") if len(paths) != len(set(paths)): raise ValueError("File Source refusal paths must be unique") + if self.scan_refusal_category is None: + if self.scan_refusal_bound is not None: + raise ValueError("File Source absent scan refusal cannot have a bound") + elif ( + type(self.scan_refusal_category) is not FileScanRefusalCategory + or type(self.scan_refusal_bound) is not int + or not 1 + <= self.scan_refusal_bound + <= MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE + ): + raise ValueError("File Source scan refusal is invalid") def _require_sequence(name: str, value: object) -> int: diff --git a/engine/control/module.py b/engine/control/module.py index 12ed9153..6d8255c0 100644 --- a/engine/control/module.py +++ b/engine/control/module.py @@ -133,6 +133,19 @@ def accept_file_change_page( page: VerifiedChangePage, ) -> AcceptedChangePage: ... + def report_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + scan_bound: int, + ) -> None: ... + + def clear_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> None: ... + class ArticlePolicyDefaultStorePort(Protocol): """Narrow persistence capability for future-Article default writes.""" @@ -183,7 +196,12 @@ def __init__( ] if file_change_proofs is not None: required_methods.extend( - ("accept_file_change_page", "schedule_file_change_page") + ( + "accept_file_change_page", + "clear_file_scan_bound_refusal", + "report_file_scan_bound_refusal", + "schedule_file_change_page", + ) ) for method_name in required_methods: if not callable(getattr(store, method_name, None)): @@ -310,6 +328,7 @@ def accept_file_change_page( or accepted.source_version_ref != page.source_version_ref or accepted.scan_ref != page.scan_ref or accepted.scan_epoch != page.scan_epoch + or accepted.scan_bound != page.scan_bound or accepted.page_limit != page.page_limit or ( page.predecessor_page_ref is None @@ -332,6 +351,66 @@ def accept_file_change_page( "File change page acceptance is unavailable" ) from None + def report_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + scan_bound: int, + ) -> None: + """Retain one closed operator condition under exact page-accept authority.""" + + if type(source_ref) is not SourceRef or type(scan_bound) is not int: + raise TypeError("File scan bound refusal is invalid") + try: + _validate_and_consume_control_call( + call, + authority=self._authority, + expected_operation=ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + checked_at=self._clock(), + ) + cast(FileChangePageStorePort, self._store).report_file_scan_bound_refusal( + call, + source_ref, + scan_bound, + ) + except (ControlOperatorAuthenticationRejected, SourceNotAvailable): + raise SourceNotAvailable from None + except SourceControlUnavailable: + raise + except Exception: + raise SourceControlUnavailable( + "File scan bound refusal reporting is unavailable" + ) from None + + def clear_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> None: + """Clear the closed condition after a complete snapshot revalidation.""" + + if type(source_ref) is not SourceRef: + raise TypeError("File scan bound refusal clear is invalid") + try: + _validate_and_consume_control_call( + call, + authority=self._authority, + expected_operation=ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + checked_at=self._clock(), + ) + cast(FileChangePageStorePort, self._store).clear_file_scan_bound_refusal( + call, + source_ref, + ) + except (ControlOperatorAuthenticationRejected, SourceNotAvailable): + raise SourceNotAvailable from None + except SourceControlUnavailable: + raise + except Exception: + raise SourceControlUnavailable( + "File scan bound refusal clear is unavailable" + ) from None + def activate_file_change_feed( self, call: TrustedControlCall, diff --git a/engine/persistence/control_sources.py b/engine/persistence/control_sources.py index 73aa89df..d75e6971 100644 --- a/engine/persistence/control_sources.py +++ b/engine/persistence/control_sources.py @@ -5,7 +5,8 @@ import hashlib from collections.abc import Callable, Mapping, Sequence from datetime import datetime -from typing import Any, cast +from threading import Lock +from typing import Any, Literal, cast from uuid import UUID, uuid4 import rfc8785 @@ -15,6 +16,7 @@ from engine._opaque import encode_base64url from engine.control import ( + DEFAULT_FILE_CHANGE_BASELINE_SIZE, FILE_CAPABILITY_MANIFEST, FILE_CHANGE_CAPABILITY_MANIFEST, FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, @@ -35,6 +37,7 @@ FileImportPath, FileResourceTombstone, FileRootRef, + FileScanRefusalCategory, FileSourceAcquisitionCheckpoint, FileSourceChangeKind, FileSourceCleanupState, @@ -68,6 +71,18 @@ ) _REGISTRATION_OPERATION = "register_source" +_FILE_STATUS_MIGRATION_FENCE = "context-engine.file-status-migration-fence" +_BOUNDED_FILE_DELETE_ACCEPT = ( + "context_control_accept_bounded_file_delete_observation_page", + "uuid, uuid, uuid, text, uuid, smallint, text, text, text, bigint, uuid, " + "jsonb, boolean, integer, jsonb", +) +_LEGACY_FILE_DELETE_ACCEPT = ( + "context_control_accept_file_delete_observation_page", + "uuid, uuid, uuid, text, uuid, smallint, text, text, text, bigint, uuid, " + "jsonb, boolean, jsonb", +) +type _FileDeleteAcceptGeneration = Literal["bounded", "legacy"] _ACTIVE_SOURCE_SELECT = """ SELECT source.source_id, @@ -135,6 +150,30 @@ def _set_organization_context(connection: Any, organization_id: UUID) -> None: ) +def _has_file_delete_accept_authority( + connection: Any, + identity: tuple[str, str], +) -> bool: + return ( + connection.execute( + text( + "SELECT count(*) = 1 FROM pg_catalog.pg_proc p " + "JOIN pg_catalog.pg_namespace n ON n.oid = p.pronamespace " + "WHERE n.nspname = 'public' " + "AND p.proname = :function_name " + "AND pg_catalog.oidvectortypes(p.proargtypes) = :argument_types " + "AND pg_catalog.has_function_privilege(" + "SESSION_USER, p.oid, 'EXECUTE')" + ), + { + "function_name": identity[0], + "argument_types": identity[1], + }, + ).scalar_one() + is True + ) + + class PostgreSQLControlStore: """Register/read File source manifests under the exact non-owner Control role.""" @@ -163,6 +202,8 @@ def __init__( ): raise TypeError("File change checkpoint signing key is invalid") self._file_change_checkpoint_signing_key = file_change_checkpoint_signing_key + self._file_delete_accept_generation: _FileDeleteAcceptGeneration | None = None + self._file_delete_accept_generation_lock = Lock() def register_file_source( self, @@ -577,6 +618,38 @@ def accept_file_change_page( } for change in value.changes ] + delete_observations = ( + value.capability_version + == FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST.declaration_version + ) + generation: _FileDeleteAcceptGeneration | None = None + if delete_observations: + with self._file_delete_accept_generation_lock: + generation = self._file_delete_accept_generation + if generation is None: + with self._engine.begin() as connection: + assert_control_role(connection) + bounded_available = _has_file_delete_accept_authority( + connection, + _BOUNDED_FILE_DELETE_ACCEPT, + ) + legacy_available = _has_file_delete_accept_authority( + connection, + _LEGACY_FILE_DELETE_ACCEPT, + ) + if bounded_available: + generation = "bounded" + elif ( + legacy_available + and value.scan_bound == DEFAULT_FILE_CHANGE_BASELINE_SIZE + ): + # Bind this store to the historical 10,000-bound + # schema generation. It must never switch to the + # legacy authority after observing revision 0045. + generation = "legacy" + else: + raise SourceNotAvailable + self._file_delete_accept_generation = generation with self._engine.begin() as connection: assert_control_role(connection) baseline_document = ( @@ -591,18 +664,38 @@ def accept_file_change_page( "sourceVersionId": str(value.baseline_ref.source_version_ref), } ) - delete_observations = ( - value.capability_version - == FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST.declaration_version - ) - function_name = ( - "context_control_accept_file_delete_observation_page" - if delete_observations - else "context_control_accept_file_change_page" - ) - baseline_argument = ( - ", CAST(:baseline AS jsonb)" if delete_observations else "" - ) + bounded_delete_observations = False + if delete_observations: + if generation is None: + raise SourceNotAvailable + connection.execute( + text( + "SELECT pg_catalog.pg_advisory_xact_lock_shared(" + "pg_catalog.hashtextextended(:fence, 0))" + ), + {"fence": _FILE_STATUS_MIGRATION_FENCE}, + ) + selected_accept = ( + _BOUNDED_FILE_DELETE_ACCEPT + if generation == "bounded" + else _LEGACY_FILE_DELETE_ACCEPT + ) + if not _has_file_delete_accept_authority( + connection, + selected_accept, + ): + raise SourceNotAvailable + bounded_delete_observations = generation == "bounded" + if bounded_delete_observations: + baseline_argument = ", :scan_bound, CAST(:baseline AS jsonb)" + elif value.scan_bound == DEFAULT_FILE_CHANGE_BASELINE_SIZE: + baseline_argument = ", CAST(:baseline AS jsonb)" + else: + raise SourceNotAvailable + function_name = selected_accept[0] + else: + function_name = "context_control_accept_file_change_page" + baseline_argument = "" row = connection.execute( text( f""" @@ -636,6 +729,7 @@ def accept_file_change_page( "utf-8" ), "complete": value.complete, + "scan_bound": value.scan_bound, "baseline": ( None if baseline_document is None @@ -681,6 +775,11 @@ def accept_file_change_page( complete=row.complete, next_cursor=next_cursor, accepted_at=row.accepted_at, + scan_bound=( + row.scan_bound + if bounded_delete_observations + else value.scan_bound + ), ) except SourceNotAvailable: raise @@ -689,6 +788,80 @@ def accept_file_change_page( "File change page database authority is unavailable" ) from None + def report_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + scan_bound: int, + ) -> None: + """Persist one closed scan-bound condition on the active File source.""" + + if ( + type(call) is not TrustedControlCall + or type(source_ref) is not SourceRef + or type(scan_bound) is not int + ): + raise SourceNotAvailable + try: + with self._engine.begin() as connection: + assert_control_role(connection) + row = connection.execute( + text( + "SELECT * FROM public." + "context_control_report_file_scan_bound_refusal(" + ":organization_id, :source_id, :scan_bound)" + ), + { + "organization_id": call.organization_id, + "source_id": source_ref.value, + "scan_bound": scan_bound, + }, + ).one_or_none() + if ( + row is None + or row.refusal_category != "scan_bound_exceeded" + or row.scan_bound != scan_bound + ): + raise SourceNotAvailable + except SourceNotAvailable: + raise + except (DBAPIError, SQLAlchemyError, AssertionError, TypeError, ValueError): + raise SourceControlUnavailable( + "File scan bound refusal database authority is unavailable" + ) from None + + def clear_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> None: + """Clear one retained bound condition after complete revalidation.""" + + if type(call) is not TrustedControlCall or type(source_ref) is not SourceRef: + raise SourceNotAvailable + try: + with self._engine.begin() as connection: + assert_control_role(connection) + row = connection.execute( + text( + "SELECT * FROM public." + "context_control_clear_file_scan_bound_refusal(" + ":organization_id, :source_id)" + ), + { + "organization_id": call.organization_id, + "source_id": source_ref.value, + }, + ).one_or_none() + if row is None or row.cleared is not True: + raise SourceNotAvailable + except SourceNotAvailable: + raise + except (DBAPIError, SQLAlchemyError, AssertionError, TypeError, ValueError): + raise SourceControlUnavailable( + "File scan bound refusal clear database authority is unavailable" + ) from None + def offboard_file_source( self, call: TrustedControlCall, @@ -953,6 +1126,11 @@ def read_file_source_progress( context_control_read_file_source_status( :organization_id, :source_id ) + ), scan_bound_status AS MATERIALIZED ( + SELECT * FROM public. + context_control_read_file_scan_bound_status( + :organization_id, :source_id + ) ), status AS MATERIALIZED ( SELECT max(status_observed_at) AS status_observed_at, @@ -976,11 +1154,13 @@ def read_file_source_progress( ) AS refusal_documents FROM status_rows ) - SELECT progress.*, baseline.*, pending.*, status.* + SELECT progress.*, baseline.*, pending.*, status.*, + scan_bound_status.* FROM progress LEFT JOIN baseline ON true CROSS JOIN pending CROSS JOIN status + CROSS JOIN scan_bound_status """ ), { @@ -1064,6 +1244,7 @@ def read_file_source_progress( checkpoint_ref=row["change_checkpoint_ref"], sequence=row["change_sequence"], complete=row["change_complete"], + scan_bound=row["head_scan_bound"], ) ), complete_change_baseline=self._complete_change_baseline( @@ -1104,6 +1285,12 @@ def read_file_source_progress( ) for document in refusal_documents ), + scan_refusal_category=( + None + if row["refusal_category"] is None + else FileScanRefusalCategory(row["refusal_category"]) + ), + scan_refusal_bound=row["refusal_scan_bound"], ), ) except SourceNotAvailable: @@ -1139,6 +1326,7 @@ def _complete_change_baseline( row["baseline_parent_checkpoint_ref"], ), sequence=cast(int, row["baseline_parent_sequence"]), + scan_bound=cast(int, row["baseline_parent_scan_bound"]), ) ) reference = FileChangeBaselineRef( @@ -1148,6 +1336,7 @@ def _complete_change_baseline( page_ref=cast(str, row["baseline_page_ref"]), checkpoint_ref=cast(str, row["baseline_checkpoint_ref"]), sequence=cast(int, row["baseline_sequence"]), + scan_bound=cast(int, row["baseline_scan_bound"]), comparison_baseline_ref=comparison_reference, ) entries: list[FileChangeBaselineEntry] = [] diff --git a/engine/persistence/schema_security_manifest.yaml b/engine/persistence/schema_security_manifest.yaml index 30681a1d..6c8f6f36 100644 --- a/engine/persistence/schema_security_manifest.yaml +++ b/engine/persistence/schema_security_manifest.yaml @@ -68,7 +68,7 @@ }, { "name": "accept_file_delete_observation_page", - "databaseFunction": "context_control_accept_file_delete_observation_page", + "databaseFunction": "context_control_accept_bounded_file_delete_observation_page", "role": "context_engine_control", "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false, @@ -81,8 +81,10 @@ "file_source_delete_observation_page", "file_source_change_page", "file_source_change", - "file_source_acquisition_checkpoint" - ] + "file_source_acquisition_checkpoint", + "context_source" + ], + "scanBoundProvenance": "signed provider page plus durable scan_bound, default 10000, configured ceiling 15000" }, { "name": "schedule_file_change_page", @@ -216,7 +218,8 @@ "databaseFunctions": [ "context_control_read_file_source_progress", "context_control_read_pending_file_change_schedules", - "context_control_read_file_source_status" + "context_control_read_file_source_status", + "context_control_read_file_scan_bound_status" ], "role": "context_engine_control", "definerRole": "context_engine_worker_lease_definer", @@ -1607,6 +1610,10 @@ { "name": "ck_context_source_lifecycle", "expression": "active has no disabled version/time; disabled binds disabled_version_id to active_version_id and has disabled_at" + }, + { + "name": "ck_context_source_file_scan_refusal", + "expression": "both refusal fields are null or category is scan_bound_exceeded and bound is between 1 and 15000" } ], "rowLevelSecurity": { @@ -1727,7 +1734,10 @@ "EXECUTE context_control_activate_file_delete_observations", "EXECUTE context_control_read_pending_file_change_schedules", "EXECUTE context_control_read_file_source_status", - "EXECUTE context_control_offboard_file_source" + "EXECUTE context_control_offboard_file_source", + "EXECUTE context_control_report_file_scan_bound_refusal", + "EXECUTE context_control_clear_file_scan_bound_refusal", + "EXECUTE context_control_read_file_scan_bound_status" ], "context_engine_learning": [], "context_engine_release_operator": [ @@ -1738,7 +1748,7 @@ "context_engine_worker": [], "context_engine_worker_lease_definer": [ "SELECT", - "UPDATE" + "UPDATE active_version_id, file_scan_refusal_category, file_scan_refusal_bound" ], "context_engine_action_prepare_definer": [ "SELECT" @@ -9851,14 +9861,14 @@ }, "functionOnlyMutation": { "databaseFunctions": [ - "context_control_accept_file_delete_observation_page" + "context_control_accept_bounded_file_delete_observation_page" ], "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false }, "permittedOperations": { "context_engine_control": [ - "EXECUTE context_control_accept_file_delete_observation_page", + "EXECUTE context_control_accept_bounded_file_delete_observation_page", "EXECUTE context_control_read_complete_file_change_baseline", "EXECUTE context_control_read_pending_file_change_schedules" ], @@ -10139,6 +10149,10 @@ { "name": "ck_file_source_change_page_bounds", "expression": "page_ordinal is positive bigint, page_limit is between 1 and 100, and change_count is between 0 and page_limit" + }, + { + "name": "ck_file_source_change_page_scan_bound", + "expression": "scan_bound is between 1 and 15000" } ], "rowLevelSecurity": { @@ -10192,7 +10206,7 @@ "functionOnlyMutation": { "databaseFunctions": [ "context_control_accept_file_change_page", - "context_control_accept_file_delete_observation_page" + "context_control_accept_bounded_file_delete_observation_page" ], "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false @@ -10200,11 +10214,12 @@ "permittedOperations": { "context_engine_control": [ "EXECUTE context_control_accept_file_change_page", - "EXECUTE context_control_accept_file_delete_observation_page", + "EXECUTE context_control_accept_bounded_file_delete_observation_page", "EXECUTE context_control_read_complete_file_change_baseline", "EXECUTE context_control_read_pending_file_change_schedules", "EXECUTE context_control_schedule_file_change_page", - "EXECUTE context_control_read_file_source_status" + "EXECUTE context_control_read_file_source_status", + "EXECUTE context_control_read_file_scan_bound_status" ], "context_engine_runtime": [], "context_engine_worker": [], @@ -10352,7 +10367,7 @@ "functionOnlyMutation": { "databaseFunctions": [ "context_control_accept_file_change_page", - "context_control_accept_file_delete_observation_page" + "context_control_accept_bounded_file_delete_observation_page" ], "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false @@ -10360,7 +10375,7 @@ "permittedOperations": { "context_engine_control": [ "EXECUTE context_control_accept_file_change_page", - "EXECUTE context_control_accept_file_delete_observation_page", + "EXECUTE context_control_accept_bounded_file_delete_observation_page", "EXECUTE context_control_read_pending_file_change_schedules", "EXECUTE context_control_schedule_file_change_page", "EXECUTE context_control_read_file_source_status" @@ -10592,14 +10607,14 @@ "context_file_source_checkpoint_import_job", "context_file_source_checkpoint_tombstone", "context_control_accept_file_change_page", - "context_control_accept_file_delete_observation_page", + "context_control_accept_bounded_file_delete_observation_page", "context_control_schedule_file_change_page" ], "causalDatabaseFunctions": [ "context_control_prepare_file_import", "context_control_tombstone_file_resource", "context_control_accept_file_change_page", - "context_control_accept_file_delete_observation_page", + "context_control_accept_bounded_file_delete_observation_page", "context_control_schedule_file_change_page" ], "definerRole": "context_engine_worker_lease_definer", @@ -10609,7 +10624,7 @@ "context_engine_control": [ "EXECUTE context_control_prepare_file_import", "EXECUTE context_control_accept_file_change_page", - "EXECUTE context_control_accept_file_delete_observation_page", + "EXECUTE context_control_accept_bounded_file_delete_observation_page", "EXECUTE context_control_read_complete_file_change_baseline", "EXECUTE context_control_read_pending_file_change_schedules", "EXECUTE context_control_schedule_file_change_page", diff --git a/migrations/versions/20260730_0045_file_scan_bound_provenance.py b/migrations/versions/20260730_0045_file_scan_bound_provenance.py new file mode 100644 index 00000000..5d82d065 --- /dev/null +++ b/migrations/versions/20260730_0045_file_scan_bound_provenance.py @@ -0,0 +1,541 @@ +"""Record configured File scan bounds and closed bound refusals. + +Revision ID: 20260730_0045 +Revises: 20260730_0044 +Create Date: 2026-07-30 +""" + +# ruff: noqa: E501 + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260730_0045" +down_revision: str | None = "20260730_0044" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_CONTROL = "context_engine_control" +_DEFINER = "context_engine_worker_lease_definer" +_OLD_ACCEPT = "context_control_accept_file_delete_observation_page" +_OLD_ACCEPT_SIGNATURE = ( + "(uuid,uuid,uuid,text,uuid,smallint,text,text,text,bigint,uuid,jsonb,boolean,jsonb)" +) +_ACCEPT = "context_control_accept_bounded_file_delete_observation_page" +_ACCEPT_SIGNATURE = ( + "(uuid,uuid,uuid,text,uuid,smallint,text,text,text,bigint,uuid,jsonb,boolean,integer,jsonb)" +) +_REPORT = "context_control_report_file_scan_bound_refusal" +_REPORT_SIGNATURE = "(uuid,uuid,integer)" +_CLEAR = "context_control_clear_file_scan_bound_refusal" +_CLEAR_SIGNATURE = "(uuid,uuid)" +_READ = "context_control_read_file_scan_bound_status" +_READ_SIGNATURE = "(uuid,uuid)" +_BOUND_TRIGGER = "context_file_change_set_scan_bound" +_DEFAULT_BOUND = 10_000 +_MAX_BOUND = 15_000 +_STATUS_MIGRATION_FENCE = "context-engine.file-status-migration-fence" +_UPSTREAM_MIGRATION_FENCES = ( + "context-engine.file-change-scheduling-migration-fence", + "context-engine.file-dispatch-migration-fence", +) + + +def _shared_status_fence() -> str: + return f""" + PERFORM pg_catalog.pg_advisory_xact_lock_shared( + pg_catalog.hashtextextended('{_STATUS_MIGRATION_FENCE}', 0) + ); + PERFORM 1 + FROM pg_catalog.pg_attribute AS attribute + WHERE attribute.attrelid = 'public.context_source'::regclass + AND attribute.attname = 'file_scan_refusal_category' + AND attribute.attnum > 0 + AND attribute.attisdropped IS FALSE; + IF NOT FOUND THEN RETURN; END IF; +""" + + +def _acquire_exclusive_status_fences() -> None: + for migration_fence in ( + *_UPSTREAM_MIGRATION_FENCES, + _STATUS_MIGRATION_FENCE, + ): + op.execute( + "SELECT pg_catalog.pg_advisory_xact_lock(" + f"pg_catalog.hashtextextended('{migration_fence}', 0))" + ) + + +def _set_absolute_acceptance_ceiling(*, raised: bool) -> None: + """Move the existing durable function's absolute fence in both directions.""" + + old, new = ((_DEFAULT_BOUND, _MAX_BOUND) if raised else (_MAX_BOUND, _DEFAULT_BOUND)) + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute( + f""" + DO $block$ + DECLARE + definition text; + replacement_definition text; + searched text := E'> {old}\\n'; + replacement_text text := E'> {new}\\n'; + BEGIN + definition := pg_catalog.pg_get_functiondef( + 'public.{_OLD_ACCEPT}{_OLD_ACCEPT_SIGNATURE}'::regprocedure + ); + IF pg_catalog.strpos(definition, replacement_text) > 0 THEN + RETURN; + END IF; + replacement_definition := pg_catalog.replace( + definition, searched, replacement_text + ); + IF replacement_definition = definition THEN + RAISE EXCEPTION 'File baseline durable fence was not recognized'; + END IF; + EXECUTE replacement_definition; + END; + $block$ + """ + ) + op.execute("RESET ROLE") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + + +def upgrade() -> None: + """Add per-scan provenance while keeping ADR-0065's default unchanged.""" + + _acquire_exclusive_status_fences() + op.add_column( + "file_source_change_page", + sa.Column( + "scan_bound", + sa.Integer(), + nullable=False, + server_default=str(_DEFAULT_BOUND), + ), + ) + op.create_check_constraint( + "ck_file_source_change_page_scan_bound", + "file_source_change_page", + f"scan_bound BETWEEN 1 AND {_MAX_BOUND}", + ) + op.add_column( + "context_source", + sa.Column("file_scan_refusal_category", sa.Text(), nullable=True), + ) + op.add_column( + "context_source", + sa.Column("file_scan_refusal_bound", sa.Integer(), nullable=True), + ) + op.create_check_constraint( + "ck_context_source_file_scan_refusal", + "context_source", + "(file_scan_refusal_category IS NULL AND file_scan_refusal_bound IS NULL) OR " + "(file_scan_refusal_category = 'scan_bound_exceeded' AND " + f"file_scan_refusal_bound BETWEEN 1 AND {_MAX_BOUND})", + ) + op.execute( + "GRANT UPDATE (file_scan_refusal_category, file_scan_refusal_bound) " + f"ON TABLE context_source TO {_DEFINER}" + ) + _set_absolute_acceptance_ceiling(raised=True) + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute( + f""" + CREATE FUNCTION public.{_BOUND_TRIGGER}() + RETURNS trigger + LANGUAGE plpgsql SECURITY INVOKER + SET search_path = pg_catalog + AS $function$ + DECLARE configured text; + BEGIN + configured := NULLIF(current_setting('app.file_scan_bound', true), ''); + NEW.scan_bound := COALESCE(configured::integer, {_DEFAULT_BOUND}); + RETURN NEW; + END; + $function$ + """ + ) + op.execute(f"REVOKE ALL ON FUNCTION public.{_BOUND_TRIGGER}() FROM PUBLIC") + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_BOUND_TRIGGER}() " + f"TO context_engine_migrator, {_DEFINER}" + ) + op.execute("RESET ROLE") + op.execute( + "CREATE TRIGGER file_source_change_page_set_scan_bound " + "BEFORE INSERT ON file_source_change_page FOR EACH ROW " + f"EXECUTE FUNCTION public.{_BOUND_TRIGGER}()" + ) + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute( + f""" + CREATE FUNCTION public.{_REPORT}( + requested_organization_id uuid, + requested_source_id uuid, + requested_scan_bound integer + ) RETURNS TABLE (refusal_category text, scan_bound integer) + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + SET row_security = on + AS $function$ + BEGIN + IF SESSION_USER <> '{_CONTROL}' + OR requested_organization_id IS NULL + OR requested_source_id IS NULL + OR requested_scan_bound NOT BETWEEN 1 AND {_MAX_BOUND} + THEN RETURN; END IF; + {_shared_status_fence()} + PERFORM pg_catalog.set_config( + 'app.organization_id', requested_organization_id::text, true + ); + UPDATE public.context_source AS source + SET file_scan_refusal_category = 'scan_bound_exceeded', + file_scan_refusal_bound = requested_scan_bound + WHERE source.organization_id = requested_organization_id + AND source.source_id = requested_source_id + AND source.source_kind = 'file' + AND source.lifecycle_state = 'active'; + IF NOT FOUND THEN RETURN; END IF; + refusal_category := 'scan_bound_exceeded'; + scan_bound := requested_scan_bound; + RETURN NEXT; + END; + $function$ + """ + ) + op.execute( + f""" + CREATE FUNCTION public.{_CLEAR}( + requested_organization_id uuid, + requested_source_id uuid + ) RETURNS TABLE (cleared boolean) + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + SET row_security = on + AS $function$ + BEGIN + IF SESSION_USER <> '{_CONTROL}' + OR requested_organization_id IS NULL + OR requested_source_id IS NULL + THEN RETURN; END IF; + {_shared_status_fence()} + PERFORM pg_catalog.set_config( + 'app.organization_id', requested_organization_id::text, true + ); + UPDATE public.context_source AS source + SET file_scan_refusal_category = NULL, + file_scan_refusal_bound = NULL + WHERE source.organization_id = requested_organization_id + AND source.source_id = requested_source_id + AND source.source_kind = 'file' + AND source.lifecycle_state = 'active'; + IF NOT FOUND THEN RETURN; END IF; + cleared := true; + RETURN NEXT; + END; + $function$ + """ + ) + op.execute( + f""" + CREATE FUNCTION public.{_READ}( + requested_organization_id uuid, + requested_source_id uuid + ) RETURNS TABLE ( + head_scan_bound integer, + baseline_scan_bound integer, + baseline_parent_scan_bound integer, + refusal_category text, + refusal_scan_bound integer + ) + LANGUAGE plpgsql STABLE SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + SET row_security = on + AS $function$ + BEGIN + IF SESSION_USER <> '{_CONTROL}' + OR NULLIF(current_setting('app.organization_id', true), '')::uuid + IS DISTINCT FROM requested_organization_id + THEN RETURN; END IF; + RETURN QUERY + WITH selected_source AS ( + SELECT source.active_version_id, + source.file_scan_refusal_category, + source.file_scan_refusal_bound + FROM public.context_source AS source + WHERE source.organization_id = requested_organization_id + AND source.source_id = requested_source_id + AND source.source_kind = 'file' + ), head AS ( + SELECT page.scan_bound, page.complete + FROM selected_source + JOIN public.file_source_acquisition_checkpoint AS checkpoint + ON checkpoint.organization_id = requested_organization_id + AND checkpoint.source_id = requested_source_id + AND checkpoint.source_version_id = + selected_source.active_version_id + AND checkpoint.change_kind = 'file_change_page' + JOIN public.file_source_change_page AS page + ON page.organization_id = checkpoint.organization_id + AND page.source_id = checkpoint.source_id + AND page.source_version_id = checkpoint.source_version_id + AND page.page_ref = checkpoint.change_page_ref + ORDER BY checkpoint.sequence DESC LIMIT 1 + ), baseline AS ( + SELECT page.scan_bound, binding.baseline_page_ref, + page.organization_id, page.source_id, + page.source_version_id + FROM selected_source + JOIN public.file_source_acquisition_checkpoint AS checkpoint + ON checkpoint.organization_id = requested_organization_id + AND checkpoint.source_id = requested_source_id + AND checkpoint.source_version_id = + selected_source.active_version_id + AND checkpoint.change_kind = 'file_change_page' + JOIN public.file_source_change_page AS page + ON page.organization_id = checkpoint.organization_id + AND page.source_id = checkpoint.source_id + AND page.source_version_id = checkpoint.source_version_id + AND page.page_ref = checkpoint.change_page_ref + AND page.complete IS TRUE + JOIN public.file_source_delete_observation_page AS binding + ON binding.organization_id = page.organization_id + AND binding.source_id = page.source_id + AND binding.source_version_id = page.source_version_id + AND binding.page_ref = page.page_ref + ORDER BY checkpoint.sequence DESC LIMIT 1 + ), parent AS ( + SELECT page.scan_bound + FROM baseline + JOIN public.file_source_change_page AS page + ON page.organization_id = baseline.organization_id + AND page.source_id = baseline.source_id + AND page.source_version_id = baseline.source_version_id + AND page.page_ref = baseline.baseline_page_ref + ) + SELECT head.scan_bound, baseline.scan_bound, parent.scan_bound, + selected_source.file_scan_refusal_category, + selected_source.file_scan_refusal_bound + FROM selected_source + LEFT JOIN head ON true + LEFT JOIN baseline ON true + LEFT JOIN parent ON true; + END; + $function$ + """ + ) + op.execute( + f""" + CREATE FUNCTION public.{_ACCEPT}( + requested_organization_id uuid, + requested_source_id uuid, + requested_source_version_id uuid, + requested_scan_ref text, + requested_scan_epoch uuid, + requested_page_limit smallint, + requested_page_ref text, + requested_predecessor_page_ref text, + requested_predecessor_checkpoint_ref text, + requested_predecessor_sequence bigint, + requested_superseded_scan_epoch uuid, + requested_changes jsonb, + requested_complete boolean, + requested_scan_bound integer, + requested_baseline jsonb + ) RETURNS TABLE ( + source_id uuid, source_version_id uuid, page_ref text, + checkpoint_ref text, sequence bigint, change_count smallint, + complete boolean, accepted_at timestamptz, + superseded_scan_epoch uuid, page_limit smallint, + scan_bound integer + ) + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + SET row_security = on + AS $function$ + DECLARE result record; + existing_page boolean := false; + BEGIN + IF SESSION_USER <> '{_CONTROL}' + OR requested_scan_bound IS NULL + OR requested_scan_bound NOT BETWEEN 1 AND {_MAX_BOUND} + OR pg_catalog.jsonb_typeof(requested_changes) <> 'array' + THEN RETURN; END IF; + {_shared_status_fence()} + PERFORM pg_catalog.set_config( + 'app.organization_id', requested_organization_id::text, true + ); + PERFORM pg_catalog.pg_advisory_xact_lock( + pg_catalog.hashtextextended( + 'context-engine.file-source-progress:' + || requested_organization_id::text || ':' + || requested_source_id::text, + 0 + ) + ); + SELECT EXISTS ( + SELECT 1 FROM public.file_source_change_page AS page + WHERE page.organization_id = requested_organization_id + AND page.source_id = requested_source_id + AND page.source_version_id = requested_source_version_id + AND page.page_ref = requested_page_ref + ) INTO existing_page; + IF EXISTS ( + SELECT 1 FROM public.file_source_change_page AS page + WHERE page.organization_id = requested_organization_id + AND page.source_id = requested_source_id + AND page.source_version_id = requested_source_version_id + AND page.scan_ref = requested_scan_ref + AND page.scan_epoch = requested_scan_epoch + AND page.scan_bound <> requested_scan_bound + ) OR ( + SELECT COALESCE(sum(page.change_count), 0) + FROM public.file_source_change_page AS page + WHERE page.organization_id = requested_organization_id + AND page.source_id = requested_source_id + AND page.source_version_id = requested_source_version_id + AND page.scan_ref = requested_scan_ref + AND page.scan_epoch = requested_scan_epoch + AND page.page_ref <> requested_page_ref + ) + pg_catalog.jsonb_array_length(requested_changes) + > requested_scan_bound + THEN RETURN; END IF; + PERFORM pg_catalog.set_config( + 'app.file_scan_bound', requested_scan_bound::text, true + ); + SELECT * INTO result + FROM public.{_OLD_ACCEPT}( + requested_organization_id, requested_source_id, + requested_source_version_id, requested_scan_ref, + requested_scan_epoch, requested_page_limit, + requested_page_ref, requested_predecessor_page_ref, + requested_predecessor_checkpoint_ref, + requested_predecessor_sequence, + requested_superseded_scan_epoch, requested_changes, + requested_complete, requested_baseline + ); + IF NOT FOUND THEN RETURN; END IF; + IF NOT EXISTS ( + SELECT 1 FROM public.file_source_change_page AS page + WHERE page.organization_id = requested_organization_id + AND page.source_id = requested_source_id + AND page.page_ref = result.page_ref + AND page.scan_bound = requested_scan_bound + ) THEN + RAISE EXCEPTION USING ERRCODE = '55000', + MESSAGE = 'File scan bound provenance was not retained'; + END IF; + IF requested_complete AND existing_page IS FALSE THEN + UPDATE public.context_source AS source + SET file_scan_refusal_category = NULL, + file_scan_refusal_bound = NULL + WHERE source.organization_id = requested_organization_id + AND source.source_id = requested_source_id + AND source.lifecycle_state = 'active'; + END IF; + source_id := result.source_id; + source_version_id := result.source_version_id; + page_ref := result.page_ref; + checkpoint_ref := result.checkpoint_ref; + sequence := result.sequence; + change_count := result.change_count; + complete := result.complete; + accepted_at := result.accepted_at; + superseded_scan_epoch := result.superseded_scan_epoch; + page_limit := result.page_limit; + scan_bound := requested_scan_bound; + RETURN NEXT; + END; + $function$ + """ + ) + for function_name, signature in ( + (_REPORT, _REPORT_SIGNATURE), + (_CLEAR, _CLEAR_SIGNATURE), + (_READ, _READ_SIGNATURE), + (_ACCEPT, _ACCEPT_SIGNATURE), + ): + op.execute( + f"REVOKE ALL ON FUNCTION public.{function_name}{signature} FROM PUBLIC" + ) + op.execute( + f"ALTER FUNCTION public.{function_name}{signature} OWNER TO {_DEFINER}" + ) + op.execute( + f"REVOKE EXECUTE ON FUNCTION public.{_OLD_ACCEPT}{_OLD_ACCEPT_SIGNATURE} " + f"FROM {_CONTROL}" + ) + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_REPORT}{_REPORT_SIGNATURE} TO {_CONTROL}" + ) + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_CLEAR}{_CLEAR_SIGNATURE} TO {_CONTROL}" + ) + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_READ}{_READ_SIGNATURE} TO {_CONTROL}" + ) + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_ACCEPT}{_ACCEPT_SIGNATURE} TO {_CONTROL}" + ) + op.execute("RESET ROLE") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + + +def downgrade() -> None: + """Drop provenance only when every retained value is the historical default.""" + + _acquire_exclusive_status_fences() + op.execute( + "LOCK TABLE public.context_source, public.file_source_change_page " + "IN ACCESS EXCLUSIVE MODE" + ) + retained = op.get_bind().execute( + sa.text( + "SELECT EXISTS (SELECT 1 FROM file_source_change_page " + f"WHERE scan_bound <> {_DEFAULT_BOUND}) OR EXISTS (" + "SELECT 1 FROM context_source " + "WHERE file_scan_refusal_category IS NOT NULL)" + ) + ).scalar_one() + if retained: + raise RuntimeError( + "File scan bound downgrade requires default-only retained provenance" + ) + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + for function_name, signature in ( + (_ACCEPT, _ACCEPT_SIGNATURE), + (_READ, _READ_SIGNATURE), + (_CLEAR, _CLEAR_SIGNATURE), + (_REPORT, _REPORT_SIGNATURE), + ): + op.execute(f"DROP FUNCTION public.{function_name}{signature}") + op.execute(f"DROP FUNCTION public.{_BOUND_TRIGGER}() CASCADE") + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_OLD_ACCEPT}{_OLD_ACCEPT_SIGNATURE} " + f"TO {_CONTROL}" + ) + op.execute("RESET ROLE") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + _set_absolute_acceptance_ceiling(raised=False) + op.execute( + "REVOKE UPDATE (file_scan_refusal_category, file_scan_refusal_bound) " + f"ON TABLE context_source FROM {_DEFINER}" + ) + op.drop_constraint( + "ck_context_source_file_scan_refusal", "context_source", type_="check" + ) + op.drop_column("context_source", "file_scan_refusal_bound") + op.drop_column("context_source", "file_scan_refusal_category") + op.drop_constraint( + "ck_file_source_change_page_scan_bound", + "file_source_change_page", + type_="check", + ) + op.drop_column("file_source_change_page", "scan_bound") diff --git a/pyproject.toml b/pyproject.toml index c4d8552b..bc5ce6a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,6 +47,7 @@ context-engine-dogfood-seed = "applications.dogfood:main" context-engine-dogfood-eval = "applications.dogfood_evaluation:main" context-engine-embedding-benchmark = "applications.embedding_benchmark:main" context-engine-eval = "applications.eval_v1:main" +context-engine-file-scan-measurement = "applications.file_scan_measurement:main" context-engine-golden-backup = "applications.golden_backup:main" context-engine-local-context = "applications.local_context:main" context-engine-worker = "applications.worker:main" diff --git a/tests/integration/test_file_change_pages.py b/tests/integration/test_file_change_pages.py index e7f9a261..12a31a5b 100644 --- a/tests/integration/test_file_change_pages.py +++ b/tests/integration/test_file_change_pages.py @@ -24,6 +24,7 @@ from adapters.http.app import create_app from adapters.parsers.markdown import compile_markdown as compile_markdown_original from engine.control import ( + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, MAX_FILE_CHANGE_BASELINE_SIZE, ActivateFileChangeFeed, ActivateFileDeleteObservations, @@ -52,6 +53,7 @@ ProviderGenericDenied, ProviderInvalidCheckpoint, ProviderOk, + ProviderScanBoundExceeded, RegisterFileSource, ScheduleFileChangePage, SourceControlUnavailable, @@ -2566,7 +2568,9 @@ def test_oversized_delete_diff_is_denied_before_durable_progress( outcome = provider.read_changes(source, InitialScan(), ChangeLimit(1)) - assert type(outcome) is ProviderGenericDenied + assert outcome == ProviderScanBoundExceeded( + scan_bound=MAX_FILE_CHANGE_BASELINE_SIZE + ) migration_engine = create_database_engine(migration_configuration) try: with migration_engine.connect() as connection: @@ -2593,6 +2597,279 @@ def test_oversized_delete_diff_is_denied_before_durable_progress( assert after == before +def test_bounded_delete_page_replay_at_exact_bound_is_idempotent( + tmp_path: Path, + guarded_control_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + root = tmp_path / "root" + root.mkdir() + (root / "only.md").write_bytes(b"# Only\n") + provider_proofs, control_proofs = _proofs() + organization_id = uuid4() + control, authority, source = _seed_file_change_source( + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + organization_id=organization_id, + receiver=FileImportReceiver(uuid4()), + root_ref=FileRootRef("exact-bound-replay-root"), + control_proofs=control_proofs, + ) + source = _activate_delete_observations( + control, + authority, + organization_id, + source, + ) + registry = FileRootRegistry( + {source.source_version.root_ref: root}, + limits=FileReadLimits(max_file_bytes=1_024, max_baseline_entries=1), + ) + provider = FileChangeProvider( + registry, + proofs=provider_proofs, + ) + try: + page = provider.read_changes(source, InitialScan(), ChangeLimit(1)) + finally: + registry.close() + assert type(page) is ProviderOk + verified = control_proofs.verify_page(page.value) + assert verified is not None + changes_document = [ + { + "contentLength": change.content_length, + "contentSha256": change.content_sha256, + "kind": change.kind.value, + "path": change.path.value, + } + for change in page.value.changes + ] + with guarded_control_engine.begin() as connection: + invalid = connection.execute( + text( + """ + SELECT * FROM public. + context_control_accept_bounded_file_delete_observation_page( + :organization_id, :source_id, :source_version_id, + :scan_ref, :scan_epoch, :page_limit, :page_ref, + NULL, NULL, NULL, :superseded_scan_epoch, + CAST(:changes AS jsonb), :complete, NULL, NULL + ) + """ + ), + { + "organization_id": organization_id, + "source_id": page.value.source_ref, + "source_version_id": page.value.source_version_ref, + "scan_ref": page.value.scan_ref, + "scan_epoch": page.value.scan_epoch, + "page_limit": page.value.page_limit, + "page_ref": verified.page_ref, + "superseded_scan_epoch": page.value.superseded_scan_epoch, + "changes": json.dumps(changes_document, separators=(",", ":")), + "complete": page.value.complete, + }, + ).one_or_none() + assert invalid is None + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.connect() as connection: + assert connection.execute( + text( + "SELECT count(*) FROM file_source_change_page " + "WHERE organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ).scalar_one() == 0 + finally: + migration_engine.dispose() + + with _authorize( + authority, + organization_id, + ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + "accept-exact-bound-page", + ) as call: + accepted = control.accept_file_change_page(call, page.value) + with _authorize( + authority, + organization_id, + ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + "record-newer-bound-refusal", + ) as call: + control.report_file_scan_bound_refusal( + call, + accepted.source_ref, + 1, + ) + with _authorize( + authority, + organization_id, + ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + "replay-exact-bound-page", + ) as call: + replayed = control.accept_file_change_page(call, page.value) + with _authorize( + authority, + organization_id, + ControlOperation.READ_SOURCE_PROGRESS, + "read-refusal-after-stale-terminal-replay", + ) as call: + progress = control.read_file_source_progress(call, accepted.source_ref) + + assert replayed == accepted + status = progress.status + assert status is not None + assert status.scan_refusal_category is not None + assert status.scan_refusal_category.value == "scan_bound_exceeded" + assert status.scan_refusal_bound == 1 + + +def test_durable_configured_ceiling_accepts_exactly_then_refuses_oversize_pages( + guarded_control_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + provider_proofs, control_proofs = _proofs() + del provider_proofs + organization_id = uuid4() + _control, _authority, source = _seed_file_change_source( + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + organization_id=organization_id, + receiver=FileImportReceiver(uuid4()), + root_ref=FileRootRef("durable-configured-ceiling-root"), + control_proofs=control_proofs, + ) + source = _activate_delete_observations( + _control, + _authority, + organization_id, + source, + ) + scan_ref = "e" * 64 + scan_epoch = uuid4() + predecessor_page_ref: str | None = None + predecessor_checkpoint_ref: str | None = None + predecessor_sequence: int | None = None + page_size = 100 + page_count = MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE // page_size + + with guarded_control_engine.begin() as connection: + for page_ordinal in range(page_count): + first_index = page_ordinal * page_size + changes = [ + { + "contentLength": 1, + "contentSha256": f"{index % 16:x}" * 64, + "kind": "upsert", + "path": f"{index:05d}.md", + } + for index in range(first_index, first_index + page_size) + ] + page_ref = f"{page_ordinal + 1:064x}" + accepted = connection.execute( + text( + """ + SELECT * FROM public. + context_control_accept_bounded_file_delete_observation_page( + :organization_id, :source_id, :source_version_id, + :scan_ref, :scan_epoch, :page_limit, :page_ref, + :predecessor_page_ref, :predecessor_checkpoint_ref, + :predecessor_sequence, NULL, + CAST(:changes AS jsonb), false, :scan_bound, NULL + ) + """ + ), + { + "organization_id": organization_id, + "source_id": source.source_version.source_ref.value, + "source_version_id": source.source_version.version_ref, + "scan_ref": scan_ref, + "scan_epoch": scan_epoch, + "page_limit": page_size, + "page_ref": page_ref, + "predecessor_page_ref": predecessor_page_ref, + "predecessor_checkpoint_ref": predecessor_checkpoint_ref, + "predecessor_sequence": predecessor_sequence, + "changes": json.dumps(changes, separators=(",", ":")), + "scan_bound": MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, + }, + ).one() + predecessor_page_ref = accepted.page_ref + predecessor_checkpoint_ref = accepted.checkpoint_ref + predecessor_sequence = accepted.sequence + + for page_ref, change_count in (("f" * 64, 1), ("d" * 64, page_size)): + changes = [ + { + "contentLength": 1, + "contentSha256": "a" * 64, + "kind": "upsert", + "path": f"overflow-{index:03d}.md", + } + for index in range(change_count) + ] + refused = connection.execute( + text( + """ + SELECT * FROM public. + context_control_accept_bounded_file_delete_observation_page( + :organization_id, :source_id, :source_version_id, + :scan_ref, :scan_epoch, :page_limit, :page_ref, + :predecessor_page_ref, :predecessor_checkpoint_ref, + :predecessor_sequence, NULL, + CAST(:changes AS jsonb), false, :scan_bound, NULL + ) + """ + ), + { + "organization_id": organization_id, + "source_id": source.source_version.source_ref.value, + "source_version_id": source.source_version.version_ref, + "scan_ref": scan_ref, + "scan_epoch": scan_epoch, + "page_limit": page_size, + "page_ref": page_ref, + "predecessor_page_ref": predecessor_page_ref, + "predecessor_checkpoint_ref": predecessor_checkpoint_ref, + "predecessor_sequence": predecessor_sequence, + "changes": json.dumps(changes, separators=(",", ":")), + "scan_bound": MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, + }, + ).one_or_none() + assert refused is None + + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.connect() as connection: + retained = connection.execute( + text( + """ + SELECT count(*), COALESCE(sum(change_count), 0), + min(scan_bound), max(scan_bound) + FROM file_source_change_page + WHERE organization_id = :organization_id + AND source_id = :source_id + AND scan_epoch = :scan_epoch + """ + ), + { + "organization_id": organization_id, + "source_id": source.source_version.source_ref.value, + "scan_epoch": scan_epoch, + }, + ).one() + finally: + migration_engine.dispose() + assert tuple(retained) == ( + page_count, + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, + ) + + @pytest.mark.security_evidence(id="PG-FILE-DELETE-DETECT-085", layer="postgres") def test_delete_observation_refuses_forged_incomplete_and_stale_baselines( tmp_path: Path, diff --git a/tests/integration/test_file_scan_operator_process.py b/tests/integration/test_file_scan_operator_process.py index d7ec1ab2..52fd8536 100644 --- a/tests/integration/test_file_scan_operator_process.py +++ b/tests/integration/test_file_scan_operator_process.py @@ -302,10 +302,12 @@ def test_status_reports_progress_freshness_and_current_compilation_refusals( "acquisitionCheckpoint": None, "activeResourceCount": 0, "changeScanHead": None, + "completeChangeBaselineScanBound": None, "completeChangeBaselineSize": 0, "lastSuccessfulAcquisition": {"state": "never"}, "publishWatermark": None, "refusals": [], + "scanRefusal": None, "sourceRef": str(source_ref), } scanned = _scan(organization_id, source_ref, environment) @@ -315,10 +317,12 @@ def test_status_reports_progress_freshness_and_current_compilation_refusals( "acquisitionCheckpoint": before_publication["acquisitionCheckpoint"], "activeResourceCount": 0, "changeScanHead": before_publication["changeScanHead"], + "completeChangeBaselineScanBound": 10_000, "completeChangeBaselineSize": 2, "lastSuccessfulAcquisition": {"state": "never"}, "publishWatermark": None, "refusals": [], + "scanRefusal": None, "sourceRef": str(source_ref), } assert before_publication["acquisitionCheckpoint"] is not None @@ -330,6 +334,7 @@ def test_status_reports_progress_freshness_and_current_compilation_refusals( "pageRef": scan_head["pageRef"], "scanEpoch": scan_head["scanEpoch"], "scanRef": scan_head["scanRef"], + "scanBound": 10_000, "sequence": scan_head["sequence"], "sourceVersionRef": scan_head["sourceVersionRef"], } @@ -422,6 +427,7 @@ def test_scan_process_schedules_only_changed_upserts_and_existing_worker_consume "deletesObserved": 0, "importsScheduled": 2, "pathsObserved": 2, + "scanBound": 10_000, "sourceRef": str(source_ref), } assert type(first["advancedCursor"]) is str @@ -462,6 +468,7 @@ def test_scan_process_schedules_only_changed_upserts_and_existing_worker_consume "deletesObserved": 0, "importsScheduled": 0, "pathsObserved": 2, + "scanBound": 10_000, "sourceRef": str(source_ref), } with engine.connect() as connection: @@ -485,6 +492,7 @@ def test_scan_process_schedules_only_changed_upserts_and_existing_worker_consume "deletesObserved": 0, "importsScheduled": 1, "pathsObserved": 3, + "scanBound": 10_000, "sourceRef": str(source_ref), } assert changed["advancedCursor"] != first["advancedCursor"] @@ -561,6 +569,7 @@ def test_scan_process_schedules_only_changed_upserts_and_existing_worker_consume "deletesObserved": 1, "importsScheduled": 0, "pathsObserved": 2, + "scanBound": 10_000, "sourceRef": str(source_ref), } assert deleted["advancedCursor"] != changed["advancedCursor"] @@ -613,6 +622,7 @@ def test_scan_process_schedules_only_changed_upserts_and_existing_worker_consume "deletesObserved": 0, "importsScheduled": 0, "pathsObserved": 2, + "scanBound": 10_000, "sourceRef": str(source_ref), } with engine.connect() as connection: @@ -717,6 +727,7 @@ def test_scan_process_recovers_a_complete_accepted_page_missing_its_schedule( "deletesObserved": 0, "importsScheduled": 1, "pathsObserved": 1, + "scanBound": 10_000, "sourceRef": str(source_ref), } assert _worker(environment)["outcome"] == "dispatched" diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index 8ae7654c..2c6ebd22 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -34,8 +34,11 @@ FileChangeSource, FileImportAudience, FileImportPath, + FileImportReceiver, + FileRootRef, InitialScan, ProviderOk, + RegisterFileSource, ScheduledFileChangePage, ScheduleFileChangePage, SourceNotAvailable, @@ -296,6 +299,9 @@ MIGRATION_TEST_HEAD_PRECONDITIONS = { "test_article_policy_downgrade_rejects_every_deferred_admin_state", "test_article_policy_downgrade_refuses_state_that_would_reauthorize_content", + "test_file_scan_bound_acceptance_fails_closed_after_downgrade", + "test_file_scan_bound_revision_downgrades_and_reapplies_when_default_only", + "test_file_scan_bound_downgrade_observes_in_flight_refusal_report", "test_file_reclaim_revision_refuses_retained_higher_generation", "test_mixed_file_upsert_downgrade_waits_for_in_flight_scheduler", "test_file_delete_observation_revision_refuses_accepted_baseline_downgrade", @@ -3472,6 +3478,383 @@ def test_file_source_status_revision_downgrades_and_reapplies_when_empty( engine.dispose() +def test_file_scan_bound_revision_downgrades_and_reapplies_when_default_only( + migration_configuration: DatabaseConfiguration, +) -> None: + """Issue #135 removes provenance only when no non-default fact is retained.""" + + alembic_configuration = Config(ROOT / "alembic.ini") + try: + command.downgrade(alembic_configuration, "20260730_0044") + assert _revision_rows(migration_configuration) == ["20260730_0044"] + engine = create_database_engine(migration_configuration) + try: + with engine.connect() as connection: + assert connection.execute( + text( + """ + SELECT ARRAY[ + to_regprocedure( + 'public.context_control_accept_bounded_file_delete_observation_page(uuid,uuid,uuid,text,uuid,smallint,text,text,text,bigint,uuid,jsonb,boolean,integer,jsonb)' + ) IS NULL, + to_regprocedure( + 'public.context_control_report_file_scan_bound_refusal(uuid,uuid,integer)' + ) IS NULL, + NOT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'file_source_change_page' + AND column_name = 'scan_bound' + ) + ] + """ + ) + ).scalar_one() == [True, True, True] + finally: + engine.dispose() + finally: + command.upgrade(alembic_configuration, "head") + + assert _revision_rows(migration_configuration) == [HEAD_REVISION] + + +def test_file_scan_bound_acceptance_fails_closed_after_downgrade( + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, +) -> None: + """A store bound to provenance authority never falls back after rollback.""" + + scenario = _prepare_file_import_scenario( + tmp_path, + migration_configuration, + guarded_control_engine, + issue_lease=False, + ) + provider_key = Ed25519PrivateKey.from_private_bytes(bytes(range(32))) + checkpoint_key = Ed25519PrivateKey.from_private_bytes(bytes(range(32, 64))) + authority = ControlOperatorAuthority( + _ControlAuthenticator(scenario.organization_id), + call_ttl=timedelta(minutes=5), + clock=lambda: NOW, + ) + control = ContextControl( + store=PostgreSQLControlStore( + guarded_control_engine, + clock=lambda: NOW, + file_import_receiver=scenario.receiver, + file_change_checkpoint_signing_key=checkpoint_key, + ), + authority=authority, + clock=lambda: NOW, + file_change_proofs=FileChangeControlProofs( + provider_verification_key=provider_key.public_key() + ), + ) + with authority.authorize( + opaque_credential="control-secret", + operation=ControlOperation.ACTIVATE_FILE_CHANGE_FEED, + request_id="bound-rollback-activate-v3", + ) as call: + control.activate_file_change_feed( + call, + ActivateFileChangeFeed(scenario.source_ref), + ) + with authority.authorize( + opaque_credential="control-secret", + operation=ControlOperation.ACTIVATE_FILE_DELETE_OBSERVATIONS, + request_id="bound-rollback-activate-v4", + ) as call: + v4 = control.activate_file_delete_observations( + call, + ActivateFileDeleteObservations(scenario.source_ref), + ) + provider = FileChangeProvider( + FileRootRegistry( + {scenario.root_ref: scenario.root}, + limits=FileReadLimits(max_file_bytes=1_024 * 1_024), + ), + proofs=FileChangeProviderProofs( + provider_signing_key=provider_key, + checkpoint_verification_key=checkpoint_key.public_key(), + ), + ) + page = provider.read_changes( + FileChangeSource(scenario.organization_id, v4.active_version), + InitialScan(), + ChangeLimit(2), + ) + assert type(page) is ProviderOk + def accept_page() -> object: + with authority.authorize( + opaque_credential="control-secret", + operation=ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + request_id="bound-rollback-accept-loses-fence", + ) as call: + return control.accept_file_change_page(call, page.value) + + migration_engine = create_database_engine(migration_configuration) + alembic_configuration = Config(ROOT / "alembic.ini") + fence = "context-engine.file-status-migration-fence" + try: + with migration_engine.connect() as blocker: + transaction = blocker.begin() + try: + blocker.execute(text("LOCK TABLE context_source IN ACCESS SHARE MODE")) + with ThreadPoolExecutor(max_workers=2) as executor: + pending_downgrade = executor.submit( + command.downgrade, + alembic_configuration, + "20260730_0044", + ) + with migration_engine.connect() as observer: + deadline = monotonic() + 10 + while monotonic() < deadline: + downgrade_holds_fence = observer.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 FROM pg_locks + WHERE locktype = 'advisory' + AND mode = 'ExclusiveLock' + AND granted IS TRUE + AND classid = (( + hashtextextended(:fence, 0) >> 32 + ) & 4294967295)::oid + AND objid = (hashtextextended(:fence, 0) + & 4294967295)::oid + AND objsubid = 1 + ) + """ + ), + {"fence": fence}, + ).scalar_one() + if downgrade_holds_fence: + break + sleep(0.01) + assert downgrade_holds_fence + pending_accept = executor.submit(accept_page) + with migration_engine.connect() as observer: + deadline = monotonic() + 10 + while monotonic() < deadline: + accept_waits_for_fence = observer.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 FROM pg_locks + WHERE locktype = 'advisory' + AND mode = 'ShareLock' + AND granted IS FALSE + AND classid = (( + hashtextextended(:fence, 0) >> 32 + ) & 4294967295)::oid + AND objid = (hashtextextended(:fence, 0) + & 4294967295)::oid + AND objsubid = 1 + ) + """ + ), + {"fence": fence}, + ).scalar_one() + if accept_waits_for_fence: + break + sleep(0.01) + assert accept_waits_for_fence + transaction.commit() + pending_downgrade.result(timeout=10) + with pytest.raises(SourceNotAvailable): + pending_accept.result(timeout=10) + finally: + if transaction.is_active: + transaction.rollback() + finally: + command.upgrade(alembic_configuration, "head") + _delete_file_import_scenario( + migration_configuration, + scenario.organization_id, + ) + migration_engine.dispose() + + +def test_file_scan_bound_downgrade_observes_in_flight_refusal_report( + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, +) -> None: + """The exclusive status fence cannot discard a concurrent bound refusal.""" + + organization_id, user_id, membership_id, receiver_id = ( + uuid4(), + uuid4(), + uuid4(), + uuid4(), + ) + migration_engine = create_database_engine(migration_configuration) + with migration_engine.begin() as connection: + connection.execute( + text("INSERT INTO organization (organization_id) VALUES (:org)"), + {"org": organization_id}, + ) + connection.execute( + text("INSERT INTO user_account (user_id) VALUES (:user)"), + {"user": user_id}, + ) + connection.execute( + text( + """ + INSERT INTO membership ( + organization_id, membership_id, user_id, status, + membership_version, valid_from + ) VALUES (:org, :membership, :user, 'active', 1, :now) + """ + ), + { + "org": organization_id, + "membership": membership_id, + "user": user_id, + "now": NOW - timedelta(days=1), + }, + ) + connection.execute( + text( + """ + INSERT INTO service_principal ( + organization_id, service_principal_id, workload, + worker_audience, operation, enabled + ) VALUES ( + :org, :receiver, 'supply.file-import', + 'context-engine-worker', 'file.import', true + ) + """ + ), + {"org": organization_id, "receiver": receiver_id}, + ) + authority = ControlOperatorAuthority( + _ControlAuthenticator(organization_id), + call_ttl=timedelta(minutes=5), + clock=lambda: NOW, + ) + control = ContextControl( + store=PostgreSQLControlStore( + guarded_control_engine, + clock=lambda: NOW, + file_import_receiver=FileImportReceiver(receiver_id), + ), + authority=authority, + clock=lambda: NOW, + ) + with authority.authorize( + opaque_credential="control-secret", + operation=ControlOperation.REGISTER_SOURCE, + request_id="register-bound-race-source", + ) as call: + source = control.register_source( + call, + RegisterFileSource( + "Bound race source", + FileRootRef("bound-race-root"), + "bound-race-source", + ), + ) + fence = "context-engine.file-status-migration-fence" + + def report_refusal() -> str: + with guarded_control_engine.begin() as connection: + row = connection.execute( + text( + "SELECT * FROM public." + "context_control_report_file_scan_bound_refusal(" + ":organization_id, :source_id, 10000)" + ), + { + "organization_id": organization_id, + "source_id": source.source_ref.value, + }, + ).one() + category = row.refusal_category + assert type(category) is str + return category + + try: + with migration_engine.connect() as blocker: + transaction = blocker.begin() + try: + blocker.execute( + text( + "SELECT 1 FROM context_source " + "WHERE organization_id = :organization_id " + "AND source_id = :source_id FOR UPDATE" + ), + { + "organization_id": organization_id, + "source_id": source.source_ref.value, + }, + ) + with ThreadPoolExecutor(max_workers=2) as executor: + pending_report = executor.submit(report_refusal) + worker_holds_fence = False + with migration_engine.connect() as observer: + deadline = monotonic() + 10 + while monotonic() < deadline: + worker_holds_fence = observer.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 FROM pg_locks + WHERE locktype = 'advisory' + AND mode = 'ShareLock' + AND granted IS TRUE + AND classid = (( + hashtextextended(:fence, 0) >> 32 + ) & 4294967295)::oid + AND objid = (hashtextextended(:fence, 0) + & 4294967295)::oid + AND objsubid = 1 + ) + """ + ), + {"fence": fence}, + ).scalar_one() + if worker_holds_fence: + break + sleep(0.01) + assert worker_holds_fence + pending_downgrade = executor.submit( + command.downgrade, + Config(ROOT / "alembic.ini"), + "20260730_0044", + ) + transaction.commit() + assert pending_report.result(timeout=10) == "scan_bound_exceeded" + with pytest.raises( + RuntimeError, + match="default-only retained provenance", + ): + pending_downgrade.result(timeout=10) + finally: + if transaction.is_active: + transaction.rollback() + assert _revision_rows(migration_configuration) == [HEAD_REVISION] + finally: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE context_source SET file_scan_refusal_category = NULL, " + "file_scan_refusal_bound = NULL " + "WHERE organization_id = :organization_id " + "AND source_id = :source_id" + ), + { + "organization_id": organization_id, + "source_id": source.source_ref.value, + }, + ) + _delete_file_import_scenario( + migration_configuration, + organization_id, + ) + migration_engine.dispose() + + def test_file_source_status_downgrade_observes_in_flight_compilation_refusal( tmp_path: Path, migration_configuration: DatabaseConfiguration, diff --git a/tests/integration/test_multi_source_scan_status.py b/tests/integration/test_multi_source_scan_status.py index 748f01f6..34a164d1 100644 --- a/tests/integration/test_multi_source_scan_status.py +++ b/tests/integration/test_multi_source_scan_status.py @@ -13,10 +13,16 @@ OPERATOR_ORGANIZATION_ENV, ) from engine.persistence import DatabaseConfiguration, create_database_engine -from tests.integration.test_file_scan_operator_process import _control, _worker +from tests.integration.test_file_scan_operator_process import ( + _control, + _worker, +) +from tests.integration.test_file_scan_operator_process import ( + file_scan_scenario as _file_scan_scenario, +) +file_scan_scenario = _file_scan_scenario pytestmark = pytest.mark.integration -pytest_plugins = ("tests.integration.test_file_scan_operator_process",) def _register_activated( @@ -199,6 +205,7 @@ def test_scan_all_and_status_discover_every_active_source_without_source_args( "importsScheduled": 3, "pathsObserved": 3, "refusalCount": 0, + "scanBounds": [10_000], "sourceCount": 2, } assert scan["refusals"] == [] @@ -320,6 +327,7 @@ def test_scan_all_reports_one_refusal_and_continues_with_later_sources( "importsScheduled": 2, "pathsObserved": 2, "refusalCount": 1, + "scanBounds": [10_000], "sourceCount": 3, } diff --git a/tests/integration/test_no_false_deletes_at_bound.py b/tests/integration/test_no_false_deletes_at_bound.py new file mode 100644 index 00000000..5f24c943 --- /dev/null +++ b/tests/integration/test_no_false_deletes_at_bound.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import json +from pathlib import Path +from uuid import UUID + +import pytest +from sqlalchemy import text + +from applications.file_root_configuration import ( + WORKER_FILE_CURATED_SUBTREES_ENV, + WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV, +) +from engine.persistence import DatabaseConfiguration, create_database_engine +from tests.integration.test_file_scan_operator_process import ( + _control, + _register_activated_source, + _scan, + _status, + _worker, +) +from tests.integration.test_file_scan_operator_process import ( + file_scan_scenario as _file_scan_scenario, +) + +file_scan_scenario = _file_scan_scenario + +pytestmark = pytest.mark.integration + + +def _delete_effects( + configuration: DatabaseConfiguration, + organization_id: UUID, + source_ref: UUID, +) -> tuple[int, ...]: + engine = create_database_engine(configuration) + try: + with engine.connect() as connection: + row = connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM file_source_change_page + WHERE organization_id = :organization_id + AND source_id = :source_id), + (SELECT count(*) FROM file_source_change + WHERE organization_id = :organization_id + AND source_id = :source_id + AND change_kind = 'delete'), + (SELECT count(*) FROM file_delete_observation_execution + WHERE organization_id = :organization_id + AND source_id = :source_id), + (SELECT count(*) FROM file_resource_cleanup_intent + WHERE organization_id = :organization_id + AND source_id = :source_id), + (SELECT count(*) FROM context_resource + WHERE organization_id = :organization_id + AND source_ref = CAST(:source_id AS text) + AND tombstoned IS TRUE) + """ + ), + { + "organization_id": organization_id, + "source_id": source_ref, + }, + ).one() + finally: + engine.dispose() + return tuple(row) + + +def test_bound_refusal_never_turns_an_unobserved_path_into_a_delete( + migration_configuration: DatabaseConfiguration, + file_scan_scenario: tuple[UUID, UUID, UUID, Path, dict[str, str]], +) -> None: + organization_id, _membership_id, _receiver_id, root, environment = ( + file_scan_scenario + ) + (root / "alpha.md").write_text("# Alpha\n", encoding="utf-8") + (root / "beta.md").write_text("# Beta\n", encoding="utf-8") + source_ref = _register_activated_source(organization_id, environment) + _scan(organization_id, source_ref, environment) + assert [_worker(environment)["outcome"] for _ in range(3)] == [ + "refused", + "refused", + "no_work", + ] + + before = _delete_effects( + migration_configuration, + organization_id, + source_ref, + ) + (root / "beta.md").unlink() + (root / "gamma.md").write_text("# Gamma\n", encoding="utf-8") + bounded = environment | {WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV: "1"} + + refused = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=bounded, + check=False, + ) + + assert refused.returncode != 0 + assert refused.stdout == "" + assert refused.stderr == "context-engine-control: operation refused\n" + assert _delete_effects( + migration_configuration, + organization_id, + source_ref, + ) == before + assert _status(organization_id, source_ref, bounded)["scanRefusal"] == { + "category": "scan_bound_exceeded", + "scanBound": 1, + } + + completed = _scan(organization_id, source_ref, environment) + assert completed["deletesObserved"] == 1 + assert _status(organization_id, source_ref, environment)["scanRefusal"] is None + + +def test_curated_selection_change_refuses_without_reinterpreting_baseline_paths( + migration_configuration: DatabaseConfiguration, + file_scan_scenario: tuple[UUID, UUID, UUID, Path, dict[str, str]], +) -> None: + organization_id, _membership_id, _receiver_id, root, environment = ( + file_scan_scenario + ) + curated = root / "curated" + curated.mkdir() + (curated / "inside.md").write_text("# Inside\n", encoding="utf-8") + (root / "outside.md").write_text("# Outside\n", encoding="utf-8") + source_ref = _register_activated_source(organization_id, environment) + _scan(organization_id, source_ref, environment) + before = _delete_effects(migration_configuration, organization_id, source_ref) + selected = environment | { + WORKER_FILE_CURATED_SUBTREES_ENV: json.dumps( + {"operator-scan-root": "curated"} + ) + } + + refused = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=selected, + check=False, + ) + + assert refused.returncode != 0 + assert refused.stdout == "" + assert refused.stderr == "context-engine-control: operation refused\n" + assert ( + _delete_effects(migration_configuration, organization_id, source_ref) == before + ) diff --git a/tests/integration/test_operator_lease_binding_unchanged.py b/tests/integration/test_operator_lease_binding_unchanged.py index b083689f..f5d05d9c 100644 --- a/tests/integration/test_operator_lease_binding_unchanged.py +++ b/tests/integration/test_operator_lease_binding_unchanged.py @@ -6,8 +6,12 @@ import pytest from sqlalchemy import Engine, text +from sqlalchemy.exc import DBAPIError from adapters.embeddings import DeterministicEmbeddingTwin +from applications.file_root_configuration import ( + WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV, +) from applications.worker import dispatch_one_file_import from applications.worker_progress import FileDispatchFailureCategory from engine.control import FileImportReceiver, FileRootRef @@ -28,10 +32,13 @@ _register_activated_source, _scan, ) +from tests.integration.test_file_scan_operator_process import ( + file_scan_scenario as _file_scan_scenario, +) from tests.support.worker_batch_progress import file_root_registry +file_scan_scenario = _file_scan_scenario pytestmark = pytest.mark.integration -pytest_plugins = ("tests.integration.test_file_scan_operator_process",) WORKER_KEY = bytes.fromhex("ab" * 32) @@ -49,7 +56,8 @@ def test_batch_convenience_preserves_exact_wrong_job_and_expiry_rejection( encoding="utf-8", ) source_ref = _register_activated_source(organization_id, environment) - _scan(organization_id, source_ref, environment) + raised = environment | {WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV: "15000"} + assert _scan(organization_id, source_ref, raised)["scanBound"] == 15_000 codec = WorkerLeaseCodec(WorkerLeaseKeyring(active_version=1, keys={1: WORKER_KEY})) authority = PostgreSQLFileDispatchAuthority( guarded_scheduler_engine, @@ -61,6 +69,18 @@ def test_batch_convenience_preserves_exact_wrong_job_and_expiry_rejection( exact_claim = claim assert claim.organization_id == organization_id assert claim.source_ref.value == source_ref + with ( + guarded_worker_engine.connect() as connection, + pytest.raises(DBAPIError), + ): + connection.execute( + text( + "SELECT count(*) FROM file_import_job " + "WHERE organization_id = :organization_id " + "AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": claim.job_id}, + ).scalar_one() roots = file_root_registry(FileRootRef("operator-scan-root"), root) try: wrong_job_claim = FileDispatchLease( diff --git a/tests/integration/test_scan_provenance_records_bound.py b/tests/integration/test_scan_provenance_records_bound.py new file mode 100644 index 00000000..d9ed838f --- /dev/null +++ b/tests/integration/test_scan_provenance_records_bound.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from pathlib import Path +from typing import cast +from uuid import UUID + +import pytest +from sqlalchemy import text + +from applications.file_root_configuration import ( + WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV, +) +from engine.persistence import DatabaseConfiguration, create_database_engine +from tests.integration.test_file_scan_operator_process import ( + _register_activated_source, + _scan, + _status, +) +from tests.integration.test_file_scan_operator_process import ( + file_scan_scenario as _file_scan_scenario, +) + +file_scan_scenario = _file_scan_scenario + +pytestmark = pytest.mark.integration + + +def test_restart_read_retains_each_scan_bound_as_durable_provenance( + migration_configuration: DatabaseConfiguration, + file_scan_scenario: tuple[UUID, UUID, UUID, Path, dict[str, str]], +) -> None: + organization_id, _membership_id, _receiver_id, root, environment = ( + file_scan_scenario + ) + (root / "alpha.md").write_text("# Alpha\n", encoding="utf-8") + source_ref = _register_activated_source(organization_id, environment) + + first = _scan(organization_id, source_ref, environment) + assert first["scanBound"] == 10_000 + + raised = environment | {WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV: "15000"} + (root / "beta.md").write_text("# Beta\n", encoding="utf-8") + second = _scan(organization_id, source_ref, raised) + assert second["scanBound"] == 15_000 + restarted_status = _status(organization_id, source_ref, raised) + assert restarted_status["completeChangeBaselineScanBound"] == 15_000 + restarted_head = cast(dict[str, object], restarted_status["changeScanHead"]) + assert restarted_head["scanBound"] == 15_000 + + engine = create_database_engine(migration_configuration) + try: + with engine.connect() as connection: + retained = tuple( + connection.execute( + text( + """ + SELECT scan_bound, count(*) + FROM file_source_change_page + WHERE organization_id = :organization_id + AND source_id = :source_id + GROUP BY scan_bound + ORDER BY scan_bound + """ + ), + { + "organization_id": organization_id, + "source_id": source_ref, + }, + ) + ) + finally: + engine.dispose() + + assert [tuple(row) for row in retained] == [ + (10_000, 1), + (15_000, 2), + ] diff --git a/tests/unit/test_bound_configuration_validation.py b/tests/unit/test_bound_configuration_validation.py new file mode 100644 index 00000000..0668e1cb --- /dev/null +++ b/tests/unit/test_bound_configuration_validation.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from uuid import UUID + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from adapters.file_source import FileChangeProvider +from applications.file_root_configuration import ( + WORKER_FILE_CURATED_SUBTREES_ENV, + WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV, + file_curated_subtrees, + file_read_limits, + file_root_bindings, + file_roots, +) +from engine.control import ( + DEFAULT_FILE_CHANGE_BASELINE_SIZE, + FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE, + ChangeLimit, + FileChangeProviderProofs, + FileChangeSource, + FileRootRef, + InitialScan, + ProviderOk, + SourceManifest, + SourceRef, +) + + +@pytest.mark.parametrize("value", ["0", "-1", "1.5", "ten-thousand", "", " 10000"]) +def test_scan_bound_rejects_invalid_values_at_configuration_time(value: str) -> None: + with pytest.raises(ValueError, match="configuration is not available"): + file_read_limits({WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV: value}) + + +def test_scan_bound_defaults_to_adr_0065_and_accepts_explicit_ceiling() -> None: + assert file_read_limits({}).max_baseline_entries == ( + DEFAULT_FILE_CHANGE_BASELINE_SIZE + ) + assert file_read_limits( + { + WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV: str( + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE + ) + } + ).max_baseline_entries == MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE + with pytest.raises(ValueError, match="configuration is not available"): + file_read_limits( + { + WORKER_MAX_FILE_CHANGE_BASELINE_SIZE_ENV: str( + MAX_CONFIGURED_FILE_CHANGE_BASELINE_SIZE + 1 + ) + } + ) + + +def test_curated_subtree_is_an_explicit_anchored_root_selection(tmp_path: Path) -> None: + configured = tmp_path / "configured" + curated = configured / "curated" / "notes" + curated.mkdir(parents=True) + environment = { + "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON": json.dumps( + {"maintainer-notes": str(configured)} + ), + WORKER_FILE_CURATED_SUBTREES_ENV: json.dumps( + {"maintainer-notes": "curated/notes"} + ), + } + + assert file_root_bindings(environment) == { + FileRootRef("maintainer-notes"): configured + } + assert file_curated_subtrees(environment) == { + FileRootRef("maintainer-notes"): "curated/notes" + } + + +def test_curated_subtree_scan_preserves_registered_root_relative_paths( + tmp_path: Path, +) -> None: + configured = tmp_path / "configured" + curated = configured / "curated" / "notes" + curated.mkdir(parents=True) + (curated / "inside.md").write_bytes(b"# Inside\n") + (configured / "outside.md").write_bytes(b"# Outside\n") + environment = { + "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON": json.dumps( + {"maintainer-notes": str(configured)} + ), + WORKER_FILE_CURATED_SUBTREES_ENV: json.dumps( + {"maintainer-notes": "curated/notes"} + ), + } + roots = file_roots(environment) + try: + key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex("11" * 32)) + checkpoint_key = Ed25519PrivateKey.from_private_bytes( + bytes.fromhex("22" * 32) + ) + manifest = SourceManifest.registered_file( + source_ref=SourceRef(UUID("efea559b-2aa7-4714-ac07-41e91b2a6f67")), + version_ref=UUID("86a1088b-70f8-4423-8e43-f4baa4277948"), + display_name="Synthetic curated source", + root_ref=FileRootRef("maintainer-notes"), + created_at=datetime(2026, 7, 30, tzinfo=UTC), + capabilities=FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, + ) + provider = FileChangeProvider( + roots, + proofs=FileChangeProviderProofs( + provider_signing_key=key, + checkpoint_verification_key=checkpoint_key.public_key(), + ), + ) + outcome = provider.read_changes( + FileChangeSource( + UUID("ac6bbbf4-dc09-4197-862a-3c84d711a3cc"), + manifest.active_version, + ), + InitialScan(), + ChangeLimit(10), + ) + finally: + roots.close() + + assert type(outcome) is ProviderOk + assert [change.path.value for change in outcome.value.changes] == [ + "curated/notes/inside.md" + ] + + +@pytest.mark.parametrize( + "selection", + ["/absolute", "../escape", "curated/../escape", "curated\\notes", "", None], +) +def test_curated_subtree_rejects_noncanonical_selection( + tmp_path: Path, + selection: object, +) -> None: + configured = tmp_path / "configured" + configured.mkdir() + environment = { + "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON": json.dumps( + {"maintainer-notes": str(configured)} + ), + WORKER_FILE_CURATED_SUBTREES_ENV: json.dumps( + {"maintainer-notes": selection} + ), + } + + with pytest.raises(ValueError, match="configuration is not available"): + file_curated_subtrees(environment) diff --git a/tests/unit/test_file_change_control.py b/tests/unit/test_file_change_control.py index 92399bbc..1f89de13 100644 --- a/tests/unit/test_file_change_control.py +++ b/tests/unit/test_file_change_control.py @@ -91,6 +91,21 @@ def accept_file_change_page( accepted_at=NOW, ) + def report_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + scan_bound: int, + ) -> None: + raise AssertionError("unexpected Control operation") + + def clear_file_scan_bound_refusal( + self, + call: TrustedControlCall, + source_ref: SourceRef, + ) -> None: + raise AssertionError("unexpected Control operation") + def activate_file_change_feed( self, call: TrustedControlCall, command: ActivateFileChangeFeed ) -> SourceManifest: diff --git a/tests/unit/test_file_delete_observation_contracts.py b/tests/unit/test_file_delete_observation_contracts.py index 4d640db4..7b31d3d1 100644 --- a/tests/unit/test_file_delete_observation_contracts.py +++ b/tests/unit/test_file_delete_observation_contracts.py @@ -27,8 +27,8 @@ FileImportPath, FileRootRef, InitialScan, - ProviderGenericDenied, ProviderOk, + ProviderScanBoundExceeded, SourceChange, SourceManifest, SourceRef, @@ -350,7 +350,9 @@ def test_oversized_mixed_diff_is_denied_before_first_page( outcome = provider.read_changes(source, InitialScan(), ChangeLimit(1)) - assert type(outcome) is ProviderGenericDenied + assert outcome == ProviderScanBoundExceeded( + scan_bound=MAX_FILE_CHANGE_BASELINE_SIZE + ) def test_completed_delete_scan_replays_its_original_parent_baseline( diff --git a/tests/unit/test_file_scan_bound_edges.py b/tests/unit/test_file_scan_bound_edges.py new file mode 100644 index 00000000..c4051929 --- /dev/null +++ b/tests/unit/test_file_scan_bound_edges.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from uuid import UUID + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from adapters.file_source import FileChangeProvider, FileReadLimits, FileRootRegistry +from engine.control import ( + FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, + ChangeLimit, + FileChangeBaseline, + FileChangeBaselineEntry, + FileChangeBaselineRef, + FileChangeKind, + FileChangeProviderProofs, + FileChangeSource, + FileImportPath, + FileRootRef, + InitialScan, + ProviderGenericDenied, + ProviderOk, + ProviderScanBoundExceeded, + SourceManifest, + SourceRef, +) + +ORGANIZATION_ID = UUID("7fbe09ee-ef27-4d8d-af10-d676fe72c740") +SOURCE_ID = UUID("d5f943d7-cab2-4f30-a12c-75c402fb5683") +SOURCE_VERSION_ID = UUID("37b78436-46d5-4a61-a2df-130c90868c32") + + +def _provider( + tmp_path: Path, + *, + bound: int, +) -> tuple[FileChangeProvider, FileRootRegistry]: + root = tmp_path / "root" + root.mkdir() + registry = FileRootRegistry( + {FileRootRef("synthetic-root"): root}, + limits=FileReadLimits(max_file_bytes=1_024, max_baseline_entries=bound), + ) + provider_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex("11" * 32)) + checkpoint_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex("22" * 32)) + return ( + FileChangeProvider( + registry, + proofs=FileChangeProviderProofs( + provider_signing_key=provider_key, + checkpoint_verification_key=checkpoint_key.public_key(), + ), + ), + registry, + ) + + +def _source() -> FileChangeSource: + manifest = SourceManifest.registered_file( + source_ref=SourceRef(SOURCE_ID), + version_ref=SOURCE_VERSION_ID, + display_name="Synthetic bound fixture", + root_ref=FileRootRef("synthetic-root"), + created_at=datetime(2026, 7, 30, tzinfo=UTC), + capabilities=FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, + ) + return FileChangeSource(ORGANIZATION_ID, manifest.active_version) + + +@pytest.mark.parametrize( + ("observed_count", "expected_type"), + [ + (10, ProviderOk), + (11, ProviderScanBoundExceeded), + (101, ProviderScanBoundExceeded), + ], +) +def test_scan_bound_has_defined_exact_and_oversized_outcomes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + observed_count: int, + expected_type: type[object], +) -> None: + provider, registry = _provider(tmp_path, bound=10) + observed = tuple( + (FileImportPath(f"synthetic-{index:05d}.md"), b"synthetic") + for index in range(observed_count) + ) + monkeypatch.setattr( + FileRootRegistry, + "_observe_markdown_files", + lambda _registry, _root_ref: observed, + ) + + outcome = provider.read_changes(_source(), InitialScan(), ChangeLimit(10)) + + assert type(outcome) is expected_type + if type(outcome) is ProviderOk: + assert outcome.value.complete is True + assert len(outcome.value.changes) == 10 + assert outcome.value.scan_bound == 10 + else: + assert outcome == ProviderScanBoundExceeded(scan_bound=10) + registry.close() + + +@pytest.mark.parametrize( + ("observed_count", "expected_type"), + [ + (10, ProviderOk), + (11, ProviderScanBoundExceeded), + (101, ProviderScanBoundExceeded), + ], +) +def test_actual_traversal_has_defined_exact_and_oversized_outcomes( + tmp_path: Path, + observed_count: int, + expected_type: type[object], +) -> None: + provider, registry = _provider(tmp_path, bound=10) + root = registry.resolve( + FileRootRef("synthetic-root"), FileImportPath("seed.md") + ).parent + for index in range(observed_count): + (root / f"actual-{index:05d}.md").write_bytes(b"synthetic") + + outcome = provider.read_changes(_source(), InitialScan(), ChangeLimit(10)) + + assert type(outcome) is expected_type + if type(outcome) is ProviderOk: + assert outcome.value.complete is True + assert len(outcome.value.changes) == 10 + else: + assert outcome == ProviderScanBoundExceeded(scan_bound=10) + registry.close() + + +def test_curated_selection_refuses_incompatible_whole_root_baseline_before_scan( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = tmp_path / "root" + (root / "curated").mkdir(parents=True) + registry = FileRootRegistry( + {FileRootRef("synthetic-root"): root}, + limits=FileReadLimits(max_file_bytes=1_024, max_baseline_entries=10), + curated_subtrees={FileRootRef("synthetic-root"): "curated"}, + ) + key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex("11" * 32)) + checkpoint_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex("22" * 32)) + provider = FileChangeProvider( + registry, + proofs=FileChangeProviderProofs( + provider_signing_key=key, + checkpoint_verification_key=checkpoint_key.public_key(), + ), + ) + baseline = FileChangeBaseline( + FileChangeBaselineRef( + source_version_ref=SOURCE_VERSION_ID, + scan_ref="a" * 64, + scan_epoch=UUID("390adef6-f348-47c9-9807-078460819635"), + page_ref="b" * 64, + checkpoint_ref="facp_" + "c" * 64, + sequence=1, + scan_bound=10, + ), + ( + FileChangeBaselineEntry( + FileChangeKind.UPSERT, + FileImportPath("outside.md"), + "d" * 64, + 1, + ), + ), + ) + source = FileChangeSource( + ORGANIZATION_ID, + _source().source_version, + complete_baseline=baseline, + ) + traversed = False + + def observe(_registry: FileRootRegistry, _root_ref: FileRootRef) -> object: + nonlocal traversed + traversed = True + return () + + monkeypatch.setattr(FileRootRegistry, "_observe_markdown_files", observe) + + outcome = provider.read_changes(source, InitialScan(), ChangeLimit(10)) + + assert type(outcome) is ProviderGenericDenied + assert traversed is False + registry.close() diff --git a/tests/unit/test_file_source_progress.py b/tests/unit/test_file_source_progress.py index 219e3634..a953ecf8 100644 --- a/tests/unit/test_file_source_progress.py +++ b/tests/unit/test_file_source_progress.py @@ -480,6 +480,7 @@ def test_database_baseline_projection_restores_global_canonical_path_order() -> "baseline_page_ref": "4" * 64, "baseline_checkpoint_ref": "facp_" + "5" * 64, "baseline_sequence": 2, + "baseline_scan_bound": 10_000, "baseline_parent_scan_epoch": None, } rows = tuple( diff --git a/tests/unit/test_scan_measurement_report.py b/tests/unit/test_scan_measurement_report.py new file mode 100644 index 00000000..99d97994 --- /dev/null +++ b/tests/unit/test_scan_measurement_report.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from applications import file_scan_measurement as measurement + +REPORT_PATH = Path("docs/evaluation/2026-07-30-file-scan-measurement.json") + + +def test_tracked_measurement_is_aggregate_synthetic_and_covers_both_options() -> None: + report = json.loads(REPORT_PATH.read_text(encoding="utf-8")) + + assert report["schemaVersion"] == measurement.SCHEMA_VERSION + assert report["method"] == { + "curatedOptionUsesConfiguredTraversal": True, + "generatedTree": True, + "pageLimit": 1, + "productionProviderSeam": True, + "singletonCycleEstimate": ( + "initial call plus pageCount minus one times the measured signed " + "continuation call" + ), + } + rows = report["measurements"] + assert [row["pathCount"] for row in rows] == list(measurement.MEASUREMENT_SIZES) + for row in rows: + assert set(row) == { + "continuationPeakMemoryBytes", + "continuationWallClockSeconds", + "estimatedSingletonCycleSeconds", + "initialPeakMemoryBytes", + "initialWallClockSeconds", + "measurementRef", + "pageCount", + "pathCount", + "peakMemoryBytes", + } + assert row["measurementRef"] == f"synthetic-{row['pathCount']}" + assert row["pageCount"] == row["pathCount"] + assert row["peakMemoryBytes"] > 0 + assert row["initialWallClockSeconds"] > 0 + assert row["continuationWallClockSeconds"] > 0 + options = report["options"] + assert options["configurableWholeVault"]["representativePathCount"] == 15_000 + assert options["curatedSubtree"]["representativePathCount"] == 5_000 + assert options["curatedSubtree"]["measurementRef"] == "synthetic-curated-5000" + + serialized = REPORT_PATH.read_text(encoding="utf-8") + assert "/Users/" not in serialized + assert "Obsidian" not in serialized + assert ".md" not in serialized + + +@pytest.mark.parametrize("path_count", [1, 15_001, True]) +def test_measurement_refuses_sizes_outside_its_synthetic_contract( + path_count: int, +) -> None: + with pytest.raises( + measurement.MeasurementUnavailable, + match="synthetic path count is unavailable", + ): + measurement.measure_size(path_count) + + +@pytest.mark.parametrize("curated_subtree", [False, True]) +def test_measurement_reexecutes_production_traversal_without_personal_data( + curated_subtree: bool, +) -> None: + row = measurement.measure_size(10, curated_subtree=curated_subtree) + + assert row["pathCount"] == 10 + assert row["pageCount"] == 10 + assert type(row["peakMemoryBytes"]) is int + assert row["peakMemoryBytes"] > 0 + assert row["measurementRef"] == ( + "synthetic-curated-10" if curated_subtree else "synthetic-10" + ) + serialized = json.dumps(row) + assert "/Users/" not in serialized + assert "Obsidian" not in serialized + assert ".md" not in serialized diff --git a/tests/unit/test_schema_security_manifest.py b/tests/unit/test_schema_security_manifest.py index cd2b65a4..5eabcb46 100644 --- a/tests/unit/test_schema_security_manifest.py +++ b/tests/unit/test_schema_security_manifest.py @@ -870,6 +870,7 @@ def test_issue_21_file_source_manifest_is_closed_and_role_separated() -> None: "context_control_read_file_source_progress", "context_control_read_pending_file_change_schedules", "context_control_read_file_source_status", + "context_control_read_file_scan_bound_status", ] assert "file_source_change_page" in progress_read["reads"] assert "file_source_delete_observation_page" in progress_read["reads"] @@ -984,10 +985,13 @@ def test_issue_21_file_source_manifest_is_closed_and_role_separated() -> None: "INSERT", "EXECUTE context_control_activate_file_change_feed", "EXECUTE context_control_activate_file_delete_observations", - "EXECUTE context_control_read_pending_file_change_schedules", - "EXECUTE context_control_read_file_source_status", - "EXECUTE context_control_offboard_file_source", - ], + "EXECUTE context_control_read_pending_file_change_schedules", + "EXECUTE context_control_read_file_source_status", + "EXECUTE context_control_offboard_file_source", + "EXECUTE context_control_report_file_scan_bound_refusal", + "EXECUTE context_control_clear_file_scan_bound_refusal", + "EXECUTE context_control_read_file_scan_bound_status", + ], "context_engine_learning": [], "context_engine_release_operator": [ "EXECUTE context_release_observe_candidate_snapshot" @@ -995,7 +999,13 @@ def test_issue_21_file_source_manifest_is_closed_and_role_separated() -> None: "context_engine_runtime": [], "context_engine_security_operator": [], "context_engine_worker": [], - "context_engine_worker_lease_definer": ["SELECT", "UPDATE"], + "context_engine_worker_lease_definer": [ + "SELECT", + ( + "UPDATE active_version_id, file_scan_refusal_category, " + "file_scan_refusal_bound" + ), + ], "context_engine_action_prepare_definer": ["SELECT"], "context_engine_action_execute_definer": ["SELECT"], "context_engine_file_dispatch_definer": [