diff --git a/README.md b/README.md index 0d2f237e..d72937aa 100644 --- a/README.md +++ b/README.md @@ -215,6 +215,97 @@ or the current authority rejects that exact failure transition. Activation boundaries for File dispatch, reclaim, and delete execution are recorded in [STATUS.md](./STATUS.md). +### Scan a local File source + +The local operator can run one bounded File acquisition cycle and hand its +scheduled upserts to the existing worker. This remains an explicitly configured +local process; it adds no HTTP operation, polling daemon, publication path, or +delete authority. + +Load the generated harness database environment first, then configure the +local operator composition described by +[ADR-0069](./docs/decisions/0069-admit-an-explicit-local-operator-composition.md). +The Control operation allowlist for this workflow is: + +```text +register_source,read_source,read_source_progress,activate_file_change_feed,activate_file_delete_observations,accept_file_change_page,schedule_file_change_page +``` + +The scan and worker share the same server-owned root registry and byte ceiling. +They additionally require one durable File-import receiver, the current private +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_SERVICE_PRINCIPAL_ID +CONTEXT_ENGINE_DOGFOOD_PRINCIPAL_REF +CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_ID +CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_VERSION +CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX +CONTEXT_ENGINE_FILE_CHANGE_CHECKPOINT_SIGNING_KEY_HEX +CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX +``` + +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 +from each other and from the Control, release, dogfood, and worker secrets. The +worker signing key is already required by the explicit local operator +composition and is checked here only to preserve that cross-plane separation. +Seed the receiver together with the dogfood identity (the command is +idempotent for the exact same bindings): + +```bash +uv run context-engine-dogfood-seed \ + --organization-id "$CONTEXT_ENGINE_DOGFOOD_ORGANIZATION_ID" \ + --user-id "$CONTEXT_ENGINE_DOGFOOD_USER_ID" \ + --membership-id "$CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_ID" \ + --file-import-service-principal-id \ + "$CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID" +``` + +Register the logical root, copy the returned `sourceRef` into +`CONTEXT_ENGINE_FILE_SOURCE_REF`, activate its two existing immutable +capability transitions, and run the cycle: + +```bash +uv run context-engine-control register-file-source \ + --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \ + --display-name "Maintainer notes" \ + --root-ref "maintainer-notes" \ + --idempotency-key "maintainer-notes-v1" + +uv run context-engine-control activate-change-feed \ + --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \ + --source-ref "$CONTEXT_ENGINE_FILE_SOURCE_REF" + +uv run context-engine-control activate-delete-observations \ + --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \ + --source-ref "$CONTEXT_ENGINE_FILE_SOURCE_REF" + +uv run context-engine-control scan \ + --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \ + --source-ref "$CONTEXT_ENGINE_FILE_SOURCE_REF" + +uv run context-engine-worker --dispatch-file-once +``` + +`scan` requires that exact delete-observation activation because its complete +durable baseline is also what makes unchanged-path scheduling decisions +idempotent. A v1, v2, or v3 source is refused generically. + +Repeat the final worker command until it reports `no_work`, or run the existing +long-lived dispatcher. The scan prints deterministic, content-free JSON counts. +`advancedCursor` is the accepted durable checkpoint reference; an exact +unchanged replay reports zero accepted changes and scheduled imports while +retaining that already-advanced checkpoint when no accepted page is missing its +schedule. Before returning, scan idempotently schedules any accepted current- +scan upsert page that has no durable acquisition. Those counts are baseline deltas. +Compilation refusals are counted before handoff using the worker's exact active +Markdown configuration, but the worker remains the only publication path and +makes the authoritative terminal transition for each scheduled import. + ### Development commands ```bash diff --git a/STATUS.md b/STATUS.md index 18d104ba..e79303e6 100644 --- a/STATUS.md +++ b/STATUS.md @@ -116,6 +116,7 @@ Follow the ADR for its exact evidence boundary. | [0065](./docs/decisions/0065-recurse-file-discovery-with-anchored-descriptors.md) | Recurse File discovery through anchored descriptors under one bounded byte ceiling | | [0066](./docs/decisions/0066-embed-fragments-before-publication.md) | Embed newly published Fragments before activation through an explicit provider | | [0070](./docs/decisions/0070-activate-file-change-feed-from-registration.md) | Advance an exact registered v1 or import-enabled v2 File source to the existing immutable v3 change-feed manifest | +| [0071](./docs/decisions/0071-compose-bounded-file-scan-cycles.md) | Compose a bounded local File scan from operation-exact accept and schedule calls with checkpoint idempotence | ADR-0065 extends the active File Provider boundary from a flat root to deterministic recursive discovery of canonical nested Markdown paths. Each @@ -138,6 +139,17 @@ future candidate-discovery implementation detail and has no authorization role. This does **not** activate vector retrieval, query embedding, historical backfill, or any Runtime/AuthorizationKernel change. +ADR-0071 composes the opt-in ADR-0069 local operator process to drive one +bounded File scan over an explicitly configured anchored root, accept every new +provider page, schedule only changed upserts, reconcile accepted current-scan +upsert pages missing durable jobs, and hand those jobs to the existing +autonomous worker. Real-PostgreSQL fixture evidence covers exact unchanged +replay, interrupted scheduling recovery, one-note addition, aggregate +compilation refusal, delete observation without delete execution, and +384-dimensional Fragment publication. This does not claim that the maintainer's +private corpus has run; it activates no watcher, alternate publisher, new +tombstone authority, or network operation. + ### Wire contract, SDK, and trusted delivery | ADR | Activates | diff --git a/applications/control.py b/applications/control.py index ec231531..d88ed2a1 100644 --- a/applications/control.py +++ b/applications/control.py @@ -9,6 +9,8 @@ from datetime import UTC, datetime, timedelta from uuid import UUID, uuid4 +from applications.file_root_configuration import file_roots +from applications.file_scan import FileScanReport, scan_file_source from applications.operator_authentication import ( CONTROL_OPERATOR_SECRET_ENV, LocalOperatorAuthorities, @@ -39,6 +41,7 @@ "read-source", "activate-change-feed", "activate-delete-observations", + "scan", } ) @@ -65,6 +68,7 @@ def _parser() -> argparse.ArgumentParser: "activate-delete-observations", "activate one File source delete-observation capability", ), + ("scan", "scan one registered File source and schedule changed upserts"), ): source_command = subcommands.add_parser(name, help=help_text) _organization_argument(source_command) @@ -89,8 +93,13 @@ def main(argv: Sequence[str] | None = None) -> None: if arguments.subcommand not in _OPERATOR_SUBCOMMANDS: parser.error("unknown operation") try: - manifest = _run_operator_subcommand(arguments) - rendered = _manifest_json(manifest) + outcome = _run_operator_subcommand(arguments) + if type(outcome) is FileScanReport: + rendered = _scan_report_json(outcome) + elif type(outcome) is SourceManifest: + rendered = _manifest_json(outcome) + else: # pragma: no cover - closed application union + raise SourceNotAvailable except Exception: # Operator refusals disclose no supplied or trusted facts. parser.exit(1, "context-engine-control: operation refused\n") print(rendered, flush=True) @@ -105,13 +114,14 @@ def local_operator_authorities() -> LocalOperatorAuthorities | None: return configuration.authorities() -def _run_operator_subcommand(arguments: argparse.Namespace) -> SourceManifest: +def _run_operator_subcommand( + arguments: argparse.Namespace, +) -> SourceManifest | FileScanReport: authorities = local_operator_authorities() if authorities is None: raise SourceNotAvailable organization_id = UUID(arguments.organization_id) opaque_credential = os.environ[CONTROL_OPERATOR_SECRET_ENV] - operation = _operation(arguments.subcommand) configuration = load_database_configuration(DatabasePurpose.CONTROL_PLANE) engine = create_database_engine(configuration) @@ -119,6 +129,18 @@ def clock() -> datetime: return datetime.now(UTC) try: + if arguments.subcommand == "scan": + with file_roots() as roots: + return scan_file_source( + organization_id=organization_id, + source_ref=SourceRef(UUID(arguments.source_ref)), + authority=authorities.control, + opaque_credential=opaque_credential, + engine=engine, + clock=clock, + roots=roots, + ) + operation = _operation(arguments.subcommand) control = ContextControl( store=PostgreSQLControlStore(engine, clock=clock), authority=authorities.control, @@ -192,6 +214,24 @@ def _manifest_json(manifest: SourceManifest) -> str: return json.dumps(document, separators=(",", ":"), sort_keys=True) +def _scan_report_json(report: FileScanReport) -> str: + if type(report) is not FileScanReport: + raise SourceNotAvailable + return json.dumps( + { + "advancedCursor": report.advanced_cursor, + "changesAccepted": report.changes_accepted, + "compilationRefusals": report.compilation_refusals, + "deletesObserved": report.deletes_observed, + "importsScheduled": report.imports_scheduled, + "pathsObserved": report.paths_observed, + "sourceRef": str(report.source_ref.value), + }, + separators=(",", ":"), + sort_keys=True, + ) + + def _timestamp(value: datetime) -> str: if type(value) is not datetime or value.utcoffset() != timedelta(0): raise SourceNotAvailable diff --git a/applications/dogfood.py b/applications/dogfood.py index ef85e986..230aad59 100644 --- a/applications/dogfood.py +++ b/applications/dogfood.py @@ -1,4 +1,4 @@ -"""Explicit local identity seeding for the dogfood composition.""" +"""Explicit local identity and optional File receiver dogfood seeding.""" from __future__ import annotations @@ -26,12 +26,16 @@ def _uuid(value: str) -> UUID: def main(argv: Sequence[str] | None = None) -> None: parser = argparse.ArgumentParser( - description="Seed one local Organization/User/current Membership" + description=( + "Seed one local Organization/User/current Membership and an " + "optional File-import receiver" + ) ) parser.add_argument("--organization-id", required=True, type=_uuid) parser.add_argument("--user-id", required=True, type=_uuid) parser.add_argument("--membership-id", required=True, type=_uuid) parser.add_argument("--membership-version", default=1, type=int) + parser.add_argument("--file-import-service-principal-id", type=_uuid) args = parser.parse_args(argv) if not 1 <= args.membership_version < (1 << 63): parser.error("--membership-version must be a positive signed bigint") @@ -84,6 +88,28 @@ def main(argv: Sequence[str] | None = None) -> None: "valid_from": seeded_at, }, ) + if args.file_import_service_principal_id is not None: + connection.execute( + text( + """ + INSERT INTO service_principal ( + organization_id, service_principal_id, workload, + worker_audience, operation, enabled + ) VALUES ( + :organization_id, :service_principal_id, + 'supply.file-import', 'context-engine-worker', + 'file.import', true + ) + ON CONFLICT ( + organization_id, service_principal_id + ) DO NOTHING + """ + ), + { + "organization_id": args.organization_id, + "service_principal_id": (args.file_import_service_principal_id), + }, + ) exact = connection.execute( text( """ @@ -112,13 +138,46 @@ def main(argv: Sequence[str] | None = None) -> None: ).scalar_one() if exact is not True: raise RuntimeError("dogfood identity conflicts with durable ownership") + if args.file_import_service_principal_id is not None: + exact_receiver = connection.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 + FROM service_principal + WHERE organization_id = :organization_id + AND service_principal_id = :service_principal_id + AND workload = 'supply.file-import' + AND worker_audience = 'context-engine-worker' + AND operation = 'file.import' + AND enabled IS TRUE + ) + """ + ), + { + "organization_id": args.organization_id, + "service_principal_id": (args.file_import_service_principal_id), + }, + ).scalar_one() + if exact_receiver is not True: + raise RuntimeError( + "dogfood receiver conflicts with durable ownership" + ) finally: engine.dispose() print( "dogfood identity ready: " f"organization={args.organization_id} " f"user={args.user_id} membership={args.membership_id} " - f"version={args.membership_version}", + f"version={args.membership_version}" + + ( + "" + if args.file_import_service_principal_id is None + else ( + " file_import_service_principal=" + f"{args.file_import_service_principal_id}" + ) + ), flush=True, ) diff --git a/applications/file_root_configuration.py b/applications/file_root_configuration.py new file mode 100644 index 00000000..a09051ef --- /dev/null +++ b/applications/file_root_configuration.py @@ -0,0 +1,84 @@ +"""Shared server-owned File root registry configuration.""" + +from __future__ import annotations + +import json +import os +from collections.abc import Mapping +from pathlib import Path + +from adapters.file_source import FileReadLimits, FileRootRegistry +from engine.control import 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" + + +def required_environment( + name: str, + environment: Mapping[str, str] | None = None, +) -> str: + """Read one explicit nonblank process configuration value.""" + + source = os.environ if environment is None else environment + value = source.get(name) + if value is None or not value or value != value.strip(): + raise ValueError("Supply worker configuration is not available") + return value + + +def file_read_limits( + environment: Mapping[str, str] | None = None, +) -> FileReadLimits: + """Load the one bounded File byte ceiling shared by scan and worker.""" + + source = os.environ if environment is None else environment + raw_limit = source.get(WORKER_MAX_FILE_BYTES_ENV) + if raw_limit is None: + return FileReadLimits(max_file_bytes=DEFAULT_WORKER_MAX_FILE_BYTES) + 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)) + except ValueError: + raise ValueError("Supply worker configuration is not available") from None + + +def file_root_bindings( + environment: Mapping[str, str] | None = None, +) -> dict[FileRootRef, Path]: + """Load every explicitly configured logical root and host path.""" + + raw_registry = required_environment(WORKER_FILE_ROOTS_ENV, environment) + try: + document = json.loads(raw_registry) + except json.JSONDecodeError: + raise ValueError("Supply worker configuration is not available") from None + if type(document) is not dict or not document: + raise ValueError("Supply worker configuration is not available") + bindings: dict[FileRootRef, Path] = {} + for raw_ref, raw_path in document.items(): + path = Path(raw_path) if type(raw_path) is str else None + if ( + type(raw_ref) is not str + or type(raw_path) is not str + or not raw_path + or raw_path != raw_path.strip() + or path is None + or not path.is_absolute() + ): + raise ValueError("Supply worker configuration is not available") + bindings[FileRootRef(raw_ref)] = path + return bindings + + +def file_roots( + environment: Mapping[str, str] | None = None, +) -> FileRootRegistry: + """Open the configured roots as anchored directory capabilities.""" + + return FileRootRegistry( + file_root_bindings(environment), + limits=file_read_limits(environment), + ) diff --git a/applications/file_scan.py b/applications/file_scan.py new file mode 100644 index 00000000..8ce909e4 --- /dev/null +++ b/applications/file_scan.py @@ -0,0 +1,463 @@ +"""Compose one bounded File source acquisition cycle.""" + +from __future__ import annotations + +import hashlib +import hmac +from collections.abc import Callable +from dataclasses import dataclass, replace +from datetime import datetime +from uuid import UUID, uuid4 + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from sqlalchemy import Engine + +from adapters.file_source import FileChangeProvider, FileRootRegistry +from adapters.parsers.markdown import compile_markdown +from applications.file_root_configuration import required_environment +from applications.operator_authentication import ( + CONTROL_OPERATOR_SECRET_ENV, + DOGFOOD_SECRET_ENV, + RELEASE_OPERATOR_SECRET_ENV, + WORKER_SECRET_ENV, +) +from engine.control import ( + FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST, + ChangeCursor, + ChangeLimit, + ContextControl, + ControlOperation, + ControlOperatorAuthority, + FileChangeControlProofs, + FileChangeKind, + FileChangeProviderProofs, + FileChangeSource, + FileImportAudience, + FileImportPath, + FileImportReceiver, + FileSourceProgress, + InitialScan, + ProviderOk, + ScheduledFileChangePage, + ScheduleFileChangePage, + SourceManifest, + SourceNotAvailable, + SourceRef, +) +from engine.persistence import PostgreSQLControlStore +from engine.supply import ( + ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION, + CompilationFailure, + MarkdownCompilerConfig, +) + +PROVIDER_SIGNING_KEY_ENV = "CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX" +CHECKPOINT_SIGNING_KEY_ENV = "CONTEXT_ENGINE_FILE_CHANGE_CHECKPOINT_SIGNING_KEY_HEX" +WORKER_SERVICE_PRINCIPAL_ENV = "CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID" +DOGFOOD_MEMBERSHIP_ENV = "CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_ID" +DOGFOOD_MEMBERSHIP_VERSION_ENV = "CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_VERSION" +DOGFOOD_PRINCIPAL_ENV = "CONTEXT_ENGINE_DOGFOOD_PRINCIPAL_REF" +# One accepted path per page lets the existing all-or-none page scheduler remain +# unchanged while the application skips unchanged observations exactly. +FILE_SCAN_PAGE_LIMIT = 1 + + +@dataclass(frozen=True, slots=True) +class FileScanReport: + """Deterministic content-free summary of one completed scan cycle.""" + + source_ref: SourceRef + paths_observed: int + changes_accepted: int + imports_scheduled: int + deletes_observed: int + compilation_refusals: int + advanced_cursor: str | None + + +def scan_file_source( + *, + organization_id: UUID, + source_ref: SourceRef, + authority: ControlOperatorAuthority, + opaque_credential: str, + engine: Engine, + clock: Callable[[], datetime], + roots: FileRootRegistry, +) -> FileScanReport: + """Observe, accept, and explicitly schedule one complete File scan.""" + + provider_key, checkpoint_key = _proof_keys() + receiver = FileImportReceiver( + UUID(required_environment(WORKER_SERVICE_PRINCIPAL_ENV)) + ) + audience = FileImportAudience( + required_environment(DOGFOOD_PRINCIPAL_ENV), + UUID(required_environment(DOGFOOD_MEMBERSHIP_ENV)), + _positive_bigint(required_environment(DOGFOOD_MEMBERSHIP_VERSION_ENV)), + ) + store = PostgreSQLControlStore( + engine, + clock=clock, + file_import_receiver=receiver, + file_change_checkpoint_signing_key=checkpoint_key, + ) + control = ContextControl( + store=store, + authority=authority, + clock=clock, + file_change_proofs=FileChangeControlProofs( + provider_verification_key=provider_key.public_key() + ), + ) + manifest = _read_manifest( + control=control, + authority=authority, + opaque_credential=opaque_credential, + organization_id=organization_id, + source_ref=source_ref, + ) + if ( + manifest.active_version.capabilities + is not FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST + ): + raise SourceNotAvailable + progress = _read_progress( + control=control, + authority=authority, + opaque_credential=opaque_credential, + organization_id=organization_id, + source_ref=source_ref, + ) + imports_scheduled = 0 + compilation_refusals = 0 + reconciled_page_refs: set[str] = set() + for pending in progress.pending_change_schedules: + scheduled = _schedule_page( + control=control, + authority=authority, + opaque_credential=opaque_credential, + organization_id=organization_id, + source_ref=source_ref, + source_version_ref=pending.source_version_ref, + page_ref=pending.page_ref, + audience=audience, + ) + imports_scheduled += len(scheduled.changes) + compilation_refusals += sum( + _compilation_refused( + roots, + manifest, + change.path.value, + change.content_sha256, + change.content_length, + ) + for change in scheduled.changes + ) + reconciled_page_refs.add(pending.page_ref) + source = FileChangeSource( + organization_id, + manifest.active_version, + scan_head=progress.change_scan_head, + complete_baseline=progress.complete_change_baseline, + ) + provider = FileChangeProvider( + roots, + proofs=FileChangeProviderProofs( + provider_signing_key=provider_key, + checkpoint_verification_key=checkpoint_key.public_key(), + ), + ) + prior = _prior_identities(source) + cursor: InitialScan | ChangeCursor = InitialScan() + paths_observed: set[str] = set() + changes_accepted = 0 + deletes_observed = 0 + advanced_cursor: str | None = None + while True: + proposed = provider.read_changes( + source, + cursor, + ChangeLimit(FILE_SCAN_PAGE_LIMIT), + ) + if type(proposed) is not ProviderOk: + raise SourceNotAvailable + page = proposed.value + if page.page_limit != FILE_SCAN_PAGE_LIMIT: + raise SourceNotAvailable + if type(cursor) is InitialScan and _replays_complete_baseline( + source, + page.scan_ref, + page.scan_epoch, + ): + baseline = source.complete_baseline + if baseline is None: # pragma: no cover - proven by the predicate + raise SourceNotAvailable + return FileScanReport( + source_ref=source_ref, + paths_observed=sum( + entry.kind is FileChangeKind.UPSERT for entry in baseline.entries + ), + changes_accepted=0, + imports_scheduled=imports_scheduled, + deletes_observed=0, + compilation_refusals=compilation_refusals, + advanced_cursor=baseline.reference.checkpoint_ref, + ) + observed = tuple(page.changes) + for change in observed: + if change.kind is FileChangeKind.UPSERT: + paths_observed.add(change.path.value) + novel_upserts = tuple( + change + for change in observed + if change.kind is FileChangeKind.UPSERT + and prior.get(change.path.value) + != (change.content_sha256, change.content_length) + ) + deletes = tuple( + change for change in observed if change.kind is FileChangeKind.DELETE + ) + with authority.authorize( + opaque_credential=opaque_credential, + operation=ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + request_id=f"local-scan-accept-{uuid4().hex}", + ) as call: + if call.organization_id != organization_id: + raise SourceNotAvailable + accepted = control.accept_file_change_page(call, page) + changes_accepted += len(novel_upserts) + len(deletes) + deletes_observed += len(deletes) + if novel_upserts and accepted.page_ref not in reconciled_page_refs: + scheduled = _schedule_page( + control=control, + authority=authority, + opaque_credential=opaque_credential, + organization_id=organization_id, + source_ref=accepted.source_ref, + source_version_ref=accepted.source_version_ref, + page_ref=accepted.page_ref, + audience=audience, + ) + scheduled_changes = { + change.path.value: change + for change in scheduled.changes + if change.path.value + in {candidate.path.value for candidate in novel_upserts} + } + imports_scheduled += len(scheduled_changes) + compilation_refusals += sum( + _compilation_refused( + roots, + manifest, + path, + change.content_sha256, + change.content_length, + ) + for path, change in scheduled_changes.items() + ) + advanced_cursor = accepted.checkpoint_ref + if accepted.next_cursor is None: + break + cursor = accepted.next_cursor + source = replace(source, scan_head=accepted.scan_head) + return FileScanReport( + source_ref=source_ref, + paths_observed=len(paths_observed), + changes_accepted=changes_accepted, + imports_scheduled=imports_scheduled, + deletes_observed=deletes_observed, + compilation_refusals=compilation_refusals, + advanced_cursor=advanced_cursor, + ) + + +def _schedule_page( + *, + control: ContextControl, + authority: ControlOperatorAuthority, + opaque_credential: str, + organization_id: UUID, + source_ref: SourceRef, + source_version_ref: UUID, + page_ref: str, + audience: FileImportAudience, +) -> ScheduledFileChangePage: + with authority.authorize( + opaque_credential=opaque_credential, + operation=ControlOperation.SCHEDULE_FILE_CHANGE_PAGE, + request_id=f"local-scan-schedule-{uuid4().hex}", + ) as call: + if call.organization_id != organization_id: + raise SourceNotAvailable + return control.schedule_file_change_page( + call, + ScheduleFileChangePage( + source_ref, + source_version_ref, + page_ref, + audience, + ), + ) + + +def _read_manifest( + *, + control: ContextControl, + authority: ControlOperatorAuthority, + opaque_credential: str, + organization_id: UUID, + source_ref: SourceRef, +) -> SourceManifest: + with authority.authorize( + opaque_credential=opaque_credential, + operation=ControlOperation.READ_SOURCE, + request_id=f"local-scan-read-source-{uuid4().hex}", + ) as call: + if call.organization_id != organization_id: + raise SourceNotAvailable + return control.read_source(call, source_ref) + + +def _read_progress( + *, + control: ContextControl, + authority: ControlOperatorAuthority, + opaque_credential: str, + organization_id: UUID, + source_ref: SourceRef, +) -> FileSourceProgress: + with authority.authorize( + opaque_credential=opaque_credential, + operation=ControlOperation.READ_SOURCE_PROGRESS, + request_id=f"local-scan-read-progress-{uuid4().hex}", + ) as call: + if call.organization_id != organization_id: + raise SourceNotAvailable + return control.read_file_source_progress(call, source_ref) + + +def _private_key_material(name: str) -> bytes: + raw = required_environment(name) + if len(raw) != 64: + raise SourceNotAvailable + try: + value = bytes.fromhex(raw) + except ValueError: + raise SourceNotAvailable from None + if len(value) != 32: + raise SourceNotAvailable + return value + + +def _proof_keys() -> tuple[Ed25519PrivateKey, Ed25519PrivateKey]: + provider_material = _private_key_material(PROVIDER_SIGNING_KEY_ENV) + checkpoint_material = _private_key_material(CHECKPOINT_SIGNING_KEY_ENV) + operator_secret_values = ( + required_environment(CONTROL_OPERATOR_SECRET_ENV), + required_environment(RELEASE_OPERATOR_SECRET_ENV), + required_environment(DOGFOOD_SECRET_ENV), + ) + encoded_proof_values = ( + provider_material.hex(), + checkpoint_material.hex(), + ) + if any( + hmac.compare_digest(proof_value, operator_secret.lower()) + for proof_value in encoded_proof_values + for operator_secret in operator_secret_values + ): + raise SourceNotAvailable + configured_secrets = ( + provider_material, + checkpoint_material, + _private_key_material(WORKER_SECRET_ENV), + *(value.encode("utf-8") for value in operator_secret_values), + ) + for index, secret in enumerate(configured_secrets): + if any( + hmac.compare_digest(secret, other) + for other in configured_secrets[index + 1 :] + ): + raise SourceNotAvailable + return ( + Ed25519PrivateKey.from_private_bytes(provider_material), + Ed25519PrivateKey.from_private_bytes(checkpoint_material), + ) + + +def _positive_bigint(value: str) -> int: + if not value.isascii() or not value.isdecimal(): + raise SourceNotAvailable + parsed = int(value) + if not 1 <= parsed <= 2**63 - 1: + raise SourceNotAvailable + return parsed + + +def _prior_identities(source: FileChangeSource) -> dict[str, tuple[str, int]]: + baseline = source.complete_baseline + if baseline is None: + return {} + return { + entry.path.value: (entry.content_sha256, entry.content_length) + for entry in baseline.entries + if entry.kind is FileChangeKind.UPSERT + } + + +def _replays_complete_baseline( + source: FileChangeSource, + scan_ref: str, + scan_epoch: UUID, +) -> bool: + """Recognize the provider's exact replay of the current complete scan.""" + + baseline = source.complete_baseline + head = source.scan_head + if baseline is None or head is None or not head.complete: + return False + reference = baseline.reference + return ( + ( + head.source_version_ref, + head.scan_ref, + head.scan_epoch, + head.page_ref, + head.checkpoint_ref, + head.sequence, + ) + == ( + reference.source_version_ref, + reference.scan_ref, + reference.scan_epoch, + reference.page_ref, + reference.checkpoint_ref, + reference.sequence, + ) + and scan_ref == reference.scan_ref + and scan_epoch == reference.scan_epoch + ) + + +def _compilation_refused( + roots: FileRootRegistry, + manifest: SourceManifest, + path: str, + expected_sha256: str, + expected_length: int, +) -> int: + payload = roots.read( + manifest.active_version.root_ref, + FileImportPath(path), + ) + if ( + len(payload) != expected_length + or hashlib.sha256(payload).hexdigest() != expected_sha256 + ): + raise SourceNotAvailable + outcome = compile_markdown( + payload, + MarkdownCompilerConfig(ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION), + ) + return int(type(outcome) is CompilationFailure) diff --git a/applications/worker.py b/applications/worker.py index c6da7af0..6a230155 100644 --- a/applications/worker.py +++ b/applications/worker.py @@ -21,6 +21,21 @@ ExternalEmbeddingProvider, ) from adapters.file_source import FileReadLimits, FileRootRegistry +from applications.file_root_configuration import ( + DEFAULT_WORKER_MAX_FILE_BYTES as _DEFAULT_WORKER_MAX_FILE_BYTES, +) +from applications.file_root_configuration import ( + file_read_limits as _configured_file_read_limits, +) +from applications.file_root_configuration import ( + file_root_bindings as _file_dispatch_root_bindings, +) +from applications.file_root_configuration import ( + file_roots as _configured_file_roots, +) +from applications.file_root_configuration import ( + required_environment as _required_environment, +) from engine import BUILD_IDENTIFIER from engine.control import FileImportReceiver, FileRootRef, SourceRef from engine.persistence import ( @@ -43,6 +58,7 @@ from engine.runtime import Runtime from engine.runtime.construction import required_kernel_dependencies from engine.supply import ( + ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION, CONTEXT_FRAGMENT_EMBEDDING_DIMENSION, EmbeddingProvider, MarkdownCompilerConfig, @@ -53,12 +69,16 @@ ) _FILE_DISPATCH_POLL_SECONDS = 1.0 -DEFAULT_WORKER_MAX_FILE_BYTES = 1_048_576 -_WORKER_MAX_FILE_BYTES_ENV = "CONTEXT_ENGINE_WORKER_MAX_FILE_BYTES" +DEFAULT_WORKER_MAX_FILE_BYTES = _DEFAULT_WORKER_MAX_FILE_BYTES +_file_dispatch_roots = _configured_file_roots _WORKER_EMBEDDING_PROVIDER_ENV = "CONTEXT_ENGINE_WORKER_EMBEDDING_PROVIDER" _WORKER_EMBEDDING_DIMENSION_ENV = "CONTEXT_ENGINE_WORKER_EMBEDDING_DIMENSION" +def _file_read_limits() -> FileReadLimits: + return _configured_file_read_limits() + + class WorkerNoOpCompletionAuthority(Protocol): """Application port for one verified persistent no-op completion.""" @@ -142,13 +162,6 @@ def complete_persistent_noop_job( return authority.complete_noop(redemption) -def _required_environment(name: str) -> str: - value = os.environ.get(name) - if value is None or not value or value != value.strip(): - raise ValueError("Supply worker configuration is not available") - return value - - def _required_bounded_integer_environment( name: str, *, @@ -167,18 +180,6 @@ def _required_bounded_integer_environment( return value -def _file_read_limits() -> FileReadLimits: - raw_limit = os.environ.get(_WORKER_MAX_FILE_BYTES_ENV) - if raw_limit is None: - return FileReadLimits(max_file_bytes=DEFAULT_WORKER_MAX_FILE_BYTES) - 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)) - except ValueError: - raise ValueError("Supply worker configuration is not available") from None - - def _embedding_provider() -> EmbeddingProvider: """Compose the explicit CI twin or one environment-only external provider.""" @@ -228,8 +229,9 @@ def _run_one_file_import() -> int: engine = create_database_engine(configuration) roots = FileRootRegistry( { - FileRootRef(_required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOT_REF")): - Path(_required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOT_PATH")) + FileRootRef( + _required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOT_REF") + ): Path(_required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOT_PATH")) }, limits=_file_read_limits(), ) @@ -241,13 +243,11 @@ def _run_one_file_import() -> int: ), FileImportReceiver( UUID( - _required_environment( - "CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID" - ) + _required_environment("CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID") ) ), roots, - MarkdownCompilerConfig("markdown-config-v1"), + MarkdownCompilerConfig(ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION), embedding_provider=_embedding_provider(), clock=lambda: datetime.now(UTC).replace(microsecond=0), ).run( @@ -309,36 +309,6 @@ def _worker_signing_key() -> bytes: return signing_key -def _file_dispatch_root_bindings() -> dict[FileRootRef, Path]: - """Load the server-owned registry for every root this dispatcher serves.""" - - raw_registry = _required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON") - try: - document = json.loads(raw_registry) - except json.JSONDecodeError: - raise ValueError("Supply worker configuration is not available") from None - if type(document) is not dict or not document: - raise ValueError("Supply worker configuration is not available") - bindings: dict[FileRootRef, Path] = {} - for raw_ref, raw_path in document.items(): - if ( - type(raw_ref) is not str - or type(raw_path) is not str - or not raw_path - or raw_path != raw_path.strip() - ): - raise ValueError("Supply worker configuration is not available") - bindings[FileRootRef(raw_ref)] = Path(raw_path) - return bindings - - -def _file_dispatch_roots() -> FileRootRegistry: - return FileRootRegistry( - _file_dispatch_root_bindings(), - limits=_file_read_limits(), - ) - - def _worker_database_time(engine: Engine) -> datetime: """Read the worker authority's clock for immediate lease verification.""" @@ -377,9 +347,7 @@ def _run_file_dispatch(*, single_cycle: bool) -> int: authority = PostgreSQLFileDispatchAuthority( scheduler_engine, codec, - configured_root_refs=tuple( - root_ref.value for root_ref in root_bindings - ), + configured_root_refs=tuple(root_ref.value for root_ref in root_bindings), ) def worker_factory(receiver: FileImportReceiver) -> PostgreSQLFileImportWorker: @@ -388,7 +356,7 @@ def worker_factory(receiver: FileImportReceiver) -> PostgreSQLFileImportWorker: codec, receiver, roots, - MarkdownCompilerConfig("markdown-config-v1"), + MarkdownCompilerConfig(ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION), embedding_provider=embedding_provider, clock=lambda: _worker_database_time(worker_engine), ) diff --git a/docs/decisions/0071-compose-bounded-file-scan-cycles.md b/docs/decisions/0071-compose-bounded-file-scan-cycles.md new file mode 100644 index 00000000..91297818 --- /dev/null +++ b/docs/decisions/0071-compose-bounded-file-scan-cycles.md @@ -0,0 +1,127 @@ +--- +name: adr-0071-compose-bounded-file-scan-cycles +version: "1.0.0" +description: > + Compose one operator-invoked File acquisition cycle from separate exact + read, accept, and schedule Control calls while preserving checkpoint + idempotence. +--- + +# 0071. Compose bounded File scan cycles + +- Status: accepted +- Date: 2026-07-27 +- Refines: ADR-0054, ADR-0055, ADR-0058, ADR-0059, ADR-0068, ADR-0069 + +## Context + +The shipped local operator can register and activate a File source, while the +provider, durable page acceptance, explicit audience-bound scheduling, and +autonomous worker dispatch already exist as separate proven modules. A scan +caller must compose them without collapsing their authority boundaries. + +One scan may require several provider pages and cannot hold one +`TrustedControlCall` across the cycle. Acceptance and scheduling are distinct +operations by design: accepting content-free provider observations does not +infer a delivery audience. Process interruption can occur after either durable +step. Their transactions remain independently idempotent. This first command +needs narrow reconciliation for accepted current-scan pages missing jobs, while +broader workflow recovery remains outside its activation. + +The existing page scheduler is all-or-none. The File provider emits a complete +snapshot of upserts plus baseline-derived deletes, so scheduling a larger page +would re-import unchanged paths whenever one path changed. + +## Decision + +`context-engine-control scan` is one bounded, operator-invoked acquisition +cycle. It obtains short-lived, operation-exact trusted calls separately for +`READ_SOURCE`, `READ_SOURCE_PROGRESS`, every +`ACCEPT_FILE_CHANGE_PAGE`, and every required +`SCHEDULE_FILE_CHANGE_PAGE`. The configured Control identity must enumerate +those operations, but no call carries more than one. The process adds no HTTP +surface and does not hold an ambient Control call. + +This explicitly refines ADR-0069's “one operation per invocation” rule for a +bounded workflow command: leaf Control commands still map one invocation to +one call, while a workflow invocation may obtain a sequence of independently +consumed calls. The invariant is one operation per `TrustedControlCall`; the +workflow cannot request the full operation set, reuse a call, or invoke an +operation absent from its configured allowlist. + +The application fixes provider pages to one observation. This preserves the +existing all-or-none scheduler while allowing it to schedule only new or +content-changed upserts. Deletes are accepted and counted but never scheduled +or executed. Scan requires the exact v4 delete-observation manifest because +that carrier provides the complete durable comparison baseline; v1-v3 sources +fail closed. A complete changed scan advances that baseline. + +An exact unchanged scan is recognized only when the complete baseline is also +the durable head and the provider reproduces that scan identity. The report +retains the already accepted durable checkpoint and counts zero accepted +changes and deletes. Before returning, the same `READ_SOURCE_PROGRESS` call +also projects accepted upsert pages in the current scan epoch with no durable +acquisition. Scan schedules those missing jobs idempotently and includes them +in the scheduled-import and compilation-refusal counts. It does not create +another scan epoch or duplicate an existing job. + +Scan and worker share one server-owned anchored root registry, byte ceiling, +and exact active Markdown configuration pin. Scan preflights newly scheduled +paths only to produce the aggregate compilation-refusal count. It rechecks the +accepted byte length and SHA-256 before compiling; filesystem drift fails the +cycle generically instead of reporting on different bytes. Scan cannot mark a +job terminal or publish. The autonomous worker remains the sole compiler and +publisher and independently rechecks the accepted raw identity before content +work. + +The provider-page and checkpoint proof keys are explicit persistent Ed25519 +secrets, distinct from each other and from the Control, release, dogfood, and +worker secrets. The worker secret is already part of ADR-0069's complete local +operator configuration; scan reads it only for local cross-plane collision +checking, never for lease issuance or redemption. Scan output contains only +deterministic content-free counts, the `SourceRef`, and the accepted opaque +checkpoint reference. + +This decision explicitly refines ADR-0068 decision 7's local migrator seed +boundary. In addition to the existing Organization/User/current-Membership +identity tuple, that command may optionally create the one exact enabled +File-import ServicePrincipal required by this composition. The optional +identifier is explicit and idempotent only for the fixed `supply.file-import`, +`context-engine-worker`, `file.import` binding; a conflicting or disabled row +refuses the whole seed transaction. The seed remains a pre-process bootstrap +operation. Neither scan nor worker receives migration authority, and the +Runtime process remains unable to create identities or receivers. + +## Consequences + +- One-note additions create exactly one durable import without widening the + page scheduler or Markdown grammar. +- Exact unchanged scans create no scan epoch or Revision. They create no job + when the accepted scan is already fully scheduled; recovery may create only + the jobs missing from an accepted current-scan page. +- A source with more observations performs more short-lived Control calls, but + each authorization and durable transaction remains independently bounded. +- Because the provider revalidates the full root for each continuation and the + existing scheduler cannot select a subset of one accepted page, singleton + pages make this first local composition quadratic in observed path count. + It is suitable for the initial measured maintainer corpus, not a general + large-root synchronization loop. Larger rollout requires an exact durable + selected-upsert scheduling contract or a restart-safe provider snapshot; it + must not silently batch unchanged upserts. +- The cycle is not atomic as a whole. Existing accepted-page and scheduled-job + transactions remain its durable boundaries. A later scan reconciles an + accepted current-scan upsert page that has no acquisition only when its + durable `page_limit` is the composition's exact singleton limit. Foreign or + future larger pages are not adopted because their all-or-none scheduling + could re-import baseline-identical upserts. Broader workflow recovery remains + inactive. +- Polling, watching, full resync, delete execution, alternate publication, + non-File providers, and network operator access remain inactive. + +## Revisit trigger + +Revisit before batching more than one observation into a scheduling decision, +adding a daemon, changing the provider from complete snapshots to deltas, or +making compilation refusal a durable pre-worker state transition. Measurement +showing that singleton-page root revalidation is operationally material is also +an immediate revisit trigger. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 19dc28a4..36eb3f34 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -172,3 +172,4 @@ touched: - [0068 — Activate the loopback dogfood Runtime](0068-activate-loopback-dogfood-runtime.md) - [0069 — Admit an explicit local operator composition](0069-admit-an-explicit-local-operator-composition.md) - [0070 — Activate a File change feed from registration](0070-activate-file-change-feed-from-registration.md) +- [0071 — Compose bounded File scan cycles](0071-compose-bounded-file-scan-cycles.md) diff --git a/engine/control/__init__.py b/engine/control/__init__.py index 3122a70a..a4585c1f 100644 --- a/engine/control/__init__.py +++ b/engine/control/__init__.py @@ -84,6 +84,7 @@ FileSourceProgress, FileSourcePublishOutcome, FileSourcePublishWatermark, + PendingFileChangeSchedule, ) from engine.control.module import ContextControl, ControlStorePort @@ -130,6 +131,7 @@ "FileSourceOffboarding", "FileSourcePublishOutcome", "FileSourcePublishWatermark", + "PendingFileChangeSchedule", "FileResourceTombstone", "ExecuteFileDeleteObservation", "ExecutedFileDeleteObservation", diff --git a/engine/control/file_source_progress.py b/engine/control/file_source_progress.py index 81e49e95..b1a74688 100644 --- a/engine/control/file_source_progress.py +++ b/engine/control/file_source_progress.py @@ -38,6 +38,19 @@ class FileSourcePublishOutcome(StrEnum): TOMBSTONED = "tombstoned" +@dataclass(frozen=True, slots=True) +class PendingFileChangeSchedule: + """One accepted current-scan page whose upserts have no durable jobs.""" + + source_version_ref: UUID = field(repr=False) + page_ref: str = field(repr=False) + + def __post_init__(self) -> None: + if type(self.source_version_ref) is not UUID: + raise TypeError("pending File schedule SourceVersion must be UUID") + _require_progress_ref("pending File schedule page_ref", self.page_ref, "") + + def _require_sequence(name: str, value: object) -> int: if type(value) is not int or not 1 <= value <= _MAX_BIGINT: raise ValueError(f"{name} must fit a positive signed bigint") @@ -51,8 +64,10 @@ def _require_resource_ref(value: object) -> str: def _require_progress_ref(name: str, value: object, prefix: str) -> str: token = _require_token(name, value) digest = token.removeprefix(prefix) - if not token.startswith(prefix) or len(digest) != 64 or any( - character not in "0123456789abcdef" for character in digest + if ( + not token.startswith(prefix) + or len(digest) != 64 + or any(character not in "0123456789abcdef" for character in digest) ): raise ValueError(f"{name} is not a recognized opaque reference") return token @@ -212,9 +227,9 @@ def __post_init__(self) -> None: event_sequence=self.event_sequence, allow_unresolved_import_resource=False, ) - if ( - self.change_kind is FileSourceChangeKind.FILE_TOMBSTONE - ) is not (self.outcome is FileSourcePublishOutcome.TOMBSTONED): + if (self.change_kind is FileSourceChangeKind.FILE_TOMBSTONE) is not ( + self.outcome is FileSourcePublishOutcome.TOMBSTONED + ): raise ValueError("File Source publish outcome does not match its change") _require_utc("File Source publish published_at", self.published_at) @@ -232,19 +247,25 @@ class FileSourceProgress: default=None, repr=False, ) + pending_change_schedules: tuple[PendingFileChangeSchedule, ...] = field( + default=(), + repr=False, + ) def __post_init__(self) -> None: if type(self.organization_id) is not UUID: raise TypeError("File Source progress organization_id must be UUID") if type(self.source_ref) is not SourceRef: raise TypeError("File Source progress source_ref must be SourceRef") - if self.acquisition_checkpoint is not None and type( - self.acquisition_checkpoint - ) is not FileSourceAcquisitionCheckpoint: + if ( + self.acquisition_checkpoint is not None + and type(self.acquisition_checkpoint) is not FileSourceAcquisitionCheckpoint + ): raise TypeError("File Source acquisition checkpoint is invalid") - if self.publish_watermark is not None and type( - self.publish_watermark - ) is not FileSourcePublishWatermark: + if ( + self.publish_watermark is not None + and type(self.publish_watermark) is not FileSourcePublishWatermark + ): raise TypeError("File Source publish watermark is invalid") if self.change_scan_head is not None: from engine.control.file_change_pages import FileChangeScanHead @@ -253,14 +274,10 @@ def __post_init__(self) -> None: raise TypeError("File Source change scan head is invalid") if self.acquisition_checkpoint is None: raise ValueError("File Source change head requires a checkpoint") - if ( - self.change_scan_head.sequence - > self.acquisition_checkpoint.sequence - ): + if self.change_scan_head.sequence > self.acquisition_checkpoint.sequence: raise ValueError("File Source change head exceeds its checkpoint") if ( - self.change_scan_head.sequence - == self.acquisition_checkpoint.sequence + self.change_scan_head.sequence == self.acquisition_checkpoint.sequence and ( self.change_scan_head.checkpoint_ref != self.acquisition_checkpoint.checkpoint_ref @@ -291,10 +308,27 @@ def __post_init__(self) -> None: raise ValueError( "File Source complete baseline belongs to another SourceVersion" ) + if type(self.pending_change_schedules) is not tuple or any( + type(pending) is not PendingFileChangeSchedule + for pending in self.pending_change_schedules + ): + raise TypeError("File Source pending schedules must be a tuple") + pending_refs = tuple( + pending.page_ref for pending in self.pending_change_schedules + ) + if len(pending_refs) != len(set(pending_refs)): + raise ValueError("File Source pending schedules must be unique") + if self.pending_change_schedules and ( + self.change_scan_head is None + or any( + pending.source_version_ref != self.change_scan_head.source_version_ref + for pending in self.pending_change_schedules + ) + ): + raise ValueError("File Source pending schedules must belong to the head") if self.publish_watermark is not None and ( self.acquisition_checkpoint is None - or self.publish_watermark.sequence - > self.acquisition_checkpoint.sequence + or self.publish_watermark.sequence > self.acquisition_checkpoint.sequence ): raise ValueError("File Source publish watermark cannot exceed checkpoint") if ( diff --git a/engine/persistence/control_sources.py b/engine/persistence/control_sources.py index 47075cd6..54f63ec3 100644 --- a/engine/persistence/control_sources.py +++ b/engine/persistence/control_sources.py @@ -41,6 +41,7 @@ FileSourcePublishOutcome, FileSourcePublishWatermark, OffboardFileSource, + PendingFileChangeSchedule, RegisterFileSource, ScheduledFileChange, ScheduledFileChangePage, @@ -62,6 +63,9 @@ ) _REGISTRATION_OPERATION = "register_source" +_PENDING_FILE_CHANGE_SCHEDULES_FUNCTION = ( + "public.context_control_read_pending_file_change_schedules" +) _ACTIVE_SOURCE_SELECT = """ SELECT source.source_id, @@ -111,8 +115,7 @@ def _registration_digest(command: RegisterFileSource) -> str: "source_kind": "file", } return hashlib.sha256( - b"context-engine.register-file-source.v1\x00" - + rfc8785.dumps(document) + b"context-engine.register-file-source.v1\x00" + rfc8785.dumps(document) ).hexdigest() @@ -157,9 +160,7 @@ def __init__( file_change_checkpoint_signing_key, Ed25519PrivateKey ): raise TypeError("File change checkpoint signing key is invalid") - self._file_change_checkpoint_signing_key = ( - file_change_checkpoint_signing_key - ) + self._file_change_checkpoint_signing_key = file_change_checkpoint_signing_key def register_file_source( self, @@ -262,19 +263,23 @@ def read_source( with self._engine.begin() as connection: assert_control_role(connection) _set_organization_context(connection, call.organization_id) - row = connection.execute( - text( - _ACTIVE_SOURCE_SELECT - + """ + row = ( + connection.execute( + text( + _ACTIVE_SOURCE_SELECT + + """ WHERE source.organization_id = :organization_id AND source.source_id = :source_id """ - ), - { - "organization_id": call.organization_id, - "source_id": source_ref.value, - }, - ).mappings().one_or_none() + ), + { + "organization_id": call.organization_id, + "source_id": source_ref.value, + }, + ) + .mappings() + .one_or_none() + ) if row is None: raise SourceNotAvailable return self._manifest(cast(Mapping[str, object], row)) @@ -318,24 +323,26 @@ def activate_file_change_feed( ).one_or_none() if row is None: raise SourceNotAvailable - source_row = connection.execute( - text( - _ACTIVE_SOURCE_SELECT - + """ + source_row = ( + connection.execute( + text( + _ACTIVE_SOURCE_SELECT + + """ WHERE source.organization_id = :organization_id AND source.source_id = :source_id """ - ), - { - "organization_id": call.organization_id, - "source_id": command.source_ref.value, - }, - ).mappings().one_or_none() + ), + { + "organization_id": call.organization_id, + "source_id": command.source_ref.value, + }, + ) + .mappings() + .one_or_none() + ) if source_row is None or source_row["version_id"] != row[0]: raise SourceNotAvailable - manifest = self._manifest( - cast(Mapping[str, object], source_row) - ) + manifest = self._manifest(cast(Mapping[str, object], source_row)) if ( manifest.active_version.capabilities is not FILE_CHANGE_CAPABILITY_MANIFEST @@ -382,24 +389,26 @@ def activate_file_delete_observations( ).one_or_none() if row is None: raise SourceNotAvailable - source_row = connection.execute( - text( - _ACTIVE_SOURCE_SELECT - + """ + source_row = ( + connection.execute( + text( + _ACTIVE_SOURCE_SELECT + + """ WHERE source.organization_id = :organization_id AND source.source_id = :source_id """ - ), - { - "organization_id": call.organization_id, - "source_id": command.source_ref.value, - }, - ).mappings().one_or_none() + ), + { + "organization_id": call.organization_id, + "source_id": command.source_ref.value, + }, + ) + .mappings() + .one_or_none() + ) if source_row is None or source_row["version_id"] != row[0]: raise SourceNotAvailable - manifest = self._manifest( - cast(Mapping[str, object], source_row) - ) + manifest = self._manifest(cast(Mapping[str, object], source_row)) if ( manifest.active_version.capabilities is not FILE_DELETE_OBSERVATION_CAPABILITY_MANIFEST @@ -449,9 +458,7 @@ def accept_file_change_page( "scanEpoch": str(value.baseline_ref.scan_epoch), "scanRef": value.baseline_ref.scan_ref, "sequence": value.baseline_ref.sequence, - "sourceVersionId": str( - value.baseline_ref.source_version_ref - ), + "sourceVersionId": str(value.baseline_ref.source_version_ref), } ) delete_observations = ( @@ -495,16 +502,16 @@ def accept_file_change_page( ), "predecessor_sequence": value.predecessor_sequence, "superseded_scan_epoch": value.superseded_scan_epoch, - "changes": rfc8785.dumps( - cast(Any, changes_document) - ).decode("utf-8"), + "changes": rfc8785.dumps(cast(Any, changes_document)).decode( + "utf-8" + ), "complete": value.complete, "baseline": ( None if baseline_document is None - else rfc8785.dumps( - cast(Any, baseline_document) - ).decode("utf-8") + else rfc8785.dumps(cast(Any, baseline_document)).decode( + "utf-8" + ) ), }, ).one_or_none() @@ -809,6 +816,22 @@ def read_file_source_progress( cast(Mapping[str, object], snapshot_row) for snapshot_row in snapshot_rows ) + pending_schedule_rows = tuple( + connection.execute( + text( + f""" + SELECT * + FROM {_PENDING_FILE_CHANGE_SCHEDULES_FUNCTION}( + :organization_id, :source_id + ) + """ + ), + { + "organization_id": call.organization_id, + "source_id": source_ref.value, + }, + ).mappings() + ) checkpoint = ( None if row["acquisition_sequence"] is None @@ -820,17 +843,13 @@ def read_file_source_progress( ), acquisition_ref=row["acquisition_acquisition_id"], job_ref=row["acquisition_job_id"], - cleanup_intent_ref=row[ - "acquisition_cleanup_intent_id" - ], + cleanup_intent_ref=row["acquisition_cleanup_intent_id"], resource_ref=row["acquisition_resource_ref"], revision_ref=row["acquisition_revision_id"], event_ref=row["acquisition_event_ref"], event_sequence=row["acquisition_event_sequence"], accepted_at=row["acquisition_accepted_at"], - source_version_ref=row[ - "acquisition_source_version_id" - ], + source_version_ref=row["acquisition_source_version_id"], change_page_ref=row["acquisition_change_page_ref"], ) ) @@ -841,9 +860,7 @@ def read_file_source_progress( sequence=row["publish_sequence"], watermark_ref=row["publish_watermark_ref"], checkpoint_ref=row["publish_checkpoint_ref"], - change_kind=FileSourceChangeKind( - row["publish_change_kind"] - ), + change_kind=FileSourceChangeKind(row["publish_change_kind"]), outcome=FileSourcePublishOutcome(row["publish_outcome"]), acquisition_ref=row["publish_acquisition_id"], job_ref=row["publish_job_id"], @@ -864,15 +881,11 @@ def read_file_source_progress( None if row["change_scan_epoch"] is None else FileChangeScanHead( - source_version_ref=row[ - "change_source_version_id" - ], + source_version_ref=row["change_source_version_id"], scan_ref=row["change_scan_ref"], scan_epoch=row["change_scan_epoch"], page_limit=row["change_page_limit"], - superseded_scan_epoch=row[ - "change_superseded_scan_epoch" - ], + superseded_scan_epoch=row["change_superseded_scan_epoch"], page_ref=row["change_page_ref"], checkpoint_ref=row["change_checkpoint_ref"], sequence=row["change_sequence"], @@ -882,6 +895,13 @@ def read_file_source_progress( complete_change_baseline=self._complete_change_baseline( baseline_rows, ), + pending_change_schedules=tuple( + PendingFileChangeSchedule( + source_version_ref=row["pending_source_version_id"], + page_ref=row["pending_page_ref"], + ) + for row in pending_schedule_rows + ), ) except SourceNotAvailable: raise @@ -933,12 +953,8 @@ def _complete_change_baseline( continue entries.append( FileChangeBaselineEntry( - kind=FileChangeKind( - cast(str, value["baseline_entry_kind"]) - ), - path=FileImportPath( - cast(str, value["baseline_entry_path"]) - ), + kind=FileChangeKind(cast(str, value["baseline_entry_kind"])), + path=FileImportPath(cast(str, value["baseline_entry_path"])), content_sha256=cast( str, value["baseline_entry_content_sha256"], @@ -1079,21 +1095,25 @@ def _select_registration( organization_id: UUID, idempotency_key: str, ) -> Mapping[str, object] | None: - row = connection.execute( - text( - _ACTIVE_SOURCE_SELECT - + """ + row = ( + connection.execute( + text( + _ACTIVE_SOURCE_SELECT + + """ WHERE source.organization_id = :organization_id AND source.registration_operation = :registration_operation AND source.idempotency_key = :idempotency_key """ - ), - { - "organization_id": organization_id, - "registration_operation": _REGISTRATION_OPERATION, - "idempotency_key": idempotency_key, - }, - ).mappings().one_or_none() + ), + { + "organization_id": organization_id, + "registration_operation": _REGISTRATION_OPERATION, + "idempotency_key": idempotency_key, + }, + ) + .mappings() + .one_or_none() + ) if row is None: return None return cast(Mapping[str, object], row) @@ -1107,9 +1127,7 @@ def _manifest(row: Mapping[str, object]) -> SourceManifest: ) declaration_version_value = capabilities.get("declarationVersion") declaration_version = ( - declaration_version_value - if type(declaration_version_value) is str - else "" + declaration_version_value if type(declaration_version_value) is str else "" ) capability_manifest = _KNOWN_CAPABILITY_DOCUMENTS.get(declaration_version) if ( diff --git a/engine/persistence/schema_security_manifest.yaml b/engine/persistence/schema_security_manifest.yaml index ca36fe74..962caea0 100644 --- a/engine/persistence/schema_security_manifest.yaml +++ b/engine/persistence/schema_security_manifest.yaml @@ -1,5 +1,5 @@ { - "manifestVersion": "33.0.0", + "manifestVersion": "34.0.0", "controlOperations": [ { "name": "register_file_source", @@ -213,7 +213,10 @@ }, { "name": "read_file_source_progress", - "databaseFunction": "context_control_read_file_source_progress", + "databaseFunctions": [ + "context_control_read_file_source_progress", + "context_control_read_pending_file_change_schedules" + ], "role": "context_engine_control", "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false, @@ -230,7 +233,8 @@ "file_source_change_page", "file_source_change", "file_source_acquisition_checkpoint", - "file_source_publish_watermark" + "file_source_publish_watermark", + "file_acquisition" ] }, { @@ -1549,6 +1553,7 @@ "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_offboard_file_source" ], "context_engine_learning": [], @@ -1722,7 +1727,8 @@ "SELECT", "INSERT", "EXECUTE context_control_activate_file_change_feed", - "EXECUTE context_control_activate_file_delete_observations" + "EXECUTE context_control_activate_file_delete_observations", + "EXECUTE context_control_read_pending_file_change_schedules" ], "context_engine_learning": [], "context_engine_runtime": [], @@ -5299,7 +5305,8 @@ "permittedOperations": { "context_engine_control": [ "EXECUTE context_control_prepare_file_import", - "EXECUTE context_control_schedule_file_change_page" + "EXECUTE context_control_schedule_file_change_page", + "EXECUTE context_control_read_pending_file_change_schedules" ], "context_engine_runtime": [], "context_engine_worker": [], @@ -7455,7 +7462,7 @@ {"name": "file_source_delete_observation_page_definer_delete", "command": "DELETE", "roles": ["context_engine_worker_lease_definer"], "using": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"} ]}, "functionOnlyMutation": {"databaseFunctions": ["context_control_accept_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_read_complete_file_change_baseline"], "context_engine_runtime": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "INSERT", "DELETE"]}, + "permittedOperations": {"context_engine_control": ["EXECUTE context_control_accept_file_delete_observation_page", "EXECUTE context_control_read_complete_file_change_baseline", "EXECUTE context_control_read_pending_file_change_schedules"], "context_engine_runtime": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "INSERT", "DELETE"]}, "partitions": [], "securityInvariantIds": ["TENANT-OWNERSHIP-001", "TENANT-FK-002", "RLS-FAIL-CLOSED-003", "WORKER-LEASE-007"], "negativeTestIds": ["DB-001", "DB-004", "DB-008", "PG-FILE-DELETE-PAGE-085", "PG-FILE-DELETE-NO-EFFECT-085"] @@ -7526,7 +7533,7 @@ ]}, "immutableRows": {"trigger": "file_source_change_page_immutable", "function": "context_content_reject_mutation", "events": ["UPDATE", "DELETE"], "sqlstate": "55000"}, "functionOnlyMutation": {"databaseFunctions": ["context_control_accept_file_change_page", "context_control_accept_file_delete_observation_page"], "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false}, - "permittedOperations": {"context_engine_control": ["EXECUTE context_control_accept_file_change_page", "EXECUTE context_control_accept_file_delete_observation_page", "EXECUTE context_control_read_complete_file_change_baseline", "EXECUTE context_control_schedule_file_change_page"], "context_engine_runtime": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "INSERT"]}, + "permittedOperations": {"context_engine_control": ["EXECUTE context_control_accept_file_change_page", "EXECUTE context_control_accept_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"], "context_engine_runtime": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "INSERT"]}, "partitions": [], "securityInvariantIds": ["TENANT-OWNERSHIP-001", "TENANT-FK-002", "RLS-FAIL-CLOSED-003", "WORKER-LEASE-007"], "negativeTestIds": ["DB-001", "DB-004", "DB-008", "PG-FILE-CHANGE-PAGE-081", "PG-FILE-CHANGE-DENY-081"] @@ -7560,7 +7567,7 @@ ]}, "immutableRows": {"trigger": "file_source_change_immutable", "function": "context_content_reject_mutation", "events": ["UPDATE", "DELETE"], "sqlstate": "55000"}, "functionOnlyMutation": {"databaseFunctions": ["context_control_accept_file_change_page", "context_control_accept_file_delete_observation_page"], "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false}, - "permittedOperations": {"context_engine_control": ["EXECUTE context_control_accept_file_change_page", "EXECUTE context_control_accept_file_delete_observation_page", "EXECUTE context_control_schedule_file_change_page"], "context_engine_runtime": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "INSERT"]}, + "permittedOperations": {"context_engine_control": ["EXECUTE context_control_accept_file_change_page", "EXECUTE context_control_accept_file_delete_observation_page", "EXECUTE context_control_read_pending_file_change_schedules", "EXECUTE context_control_schedule_file_change_page"], "context_engine_runtime": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "INSERT"]}, "partitions": [], "securityInvariantIds": ["TENANT-OWNERSHIP-001", "TENANT-FK-002", "RLS-FAIL-CLOSED-003", "WORKER-LEASE-007"], "negativeTestIds": ["DB-001", "DB-004", "DB-008", "PG-FILE-CHANGE-PAGE-081", "PG-FILE-CHANGE-DENY-081", "PG-FILE-CHANGE-SCHEDULE-083"] @@ -7786,6 +7793,7 @@ "EXECUTE context_control_accept_file_change_page", "EXECUTE context_control_accept_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_progress", "EXECUTE context_control_tombstone_file_resource" diff --git a/engine/supply/__init__.py b/engine/supply/__init__.py index 9a3772fc..4469daa1 100644 --- a/engine/supply/__init__.py +++ b/engine/supply/__init__.py @@ -31,6 +31,7 @@ worker_lease_nonce_digest, ) from engine.supply.markdown import ( + ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION, MARKDOWN_CANONICALIZATION_PROFILE, MARKDOWN_CANONICALIZATION_V1_PROFILE, MARKDOWN_CODE_LANGUAGE_MAX_LENGTH, @@ -58,6 +59,7 @@ ) __all__ = [ + "ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION", "CONTEXT_FRAGMENT_EMBEDDING_DIMENSION", "MARKDOWN_CANONICALIZATION_PROFILE", "MARKDOWN_CODE_LANGUAGE_MAX_LENGTH", diff --git a/engine/supply/markdown.py b/engine/supply/markdown.py index 8de3b4bc..588b3a62 100644 --- a/engine/supply/markdown.py +++ b/engine/supply/markdown.py @@ -12,6 +12,7 @@ MARKDOWN_COMPILER_V1_VERSION: Final = "context-engine-markdown-v1" MARKDOWN_COMPILER_VERSION: Final = "context-engine-markdown-v2" +ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION: Final = "markdown-config-v1" MARKDOWN_CANONICALIZATION_V1_PROFILE: Final = "markdown-heading-paragraph-v1" MARKDOWN_CANONICALIZATION_PROFILE: Final = "markdown-structural-units-v2" MARKDOWN_CONTENT_HASH_PROFILE: Final = "sha256-canonical-utf8-v1" diff --git a/migrations/versions/20260727_0038_pending_file_change_schedules.py b/migrations/versions/20260727_0038_pending_file_change_schedules.py new file mode 100644 index 00000000..02d3e6d9 --- /dev/null +++ b/migrations/versions/20260727_0038_pending_file_change_schedules.py @@ -0,0 +1,158 @@ +"""Expose accepted current-scan pages that still require scheduling. + +Revision ID: 20260727_0038 +Revises: 20260727_0037 +Create Date: 2026-07-27 +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "20260727_0038" +down_revision: str | None = "20260727_0037" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_CONTROL = "context_engine_control" +_DEFINER = "context_engine_worker_lease_definer" +_FUNCTION = "context_control_read_pending_file_change_schedules" +_SIGNATURE = "(uuid, uuid)" +_V4 = "file-capabilities-v4" + + +def upgrade() -> None: + """Add one tenant-scoped read model for restart-safe scheduling handoff.""" + + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute( + f""" + CREATE FUNCTION public.{_FUNCTION}( + requested_organization_id uuid, + requested_source_id uuid + ) RETURNS TABLE ( + pending_source_version_id uuid, + pending_page_ref text + ) + 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 active AS ( + SELECT source.active_version_id + FROM public.context_source AS source + JOIN public.source_version AS version + ON version.organization_id = source.organization_id + AND version.source_id = source.source_id + AND version.version_id = source.active_version_id + WHERE source.organization_id = requested_organization_id + AND source.source_id = requested_source_id + AND source.lifecycle_state = 'active' + AND version.capability_manifest->>'declarationVersion' + = '{_V4}' + ), head AS ( + SELECT page.source_version_id, page.scan_epoch + FROM active + 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 = active.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 + ) + SELECT page.source_version_id, page.page_ref + FROM head + JOIN public.file_source_change_page AS page + ON page.organization_id = requested_organization_id + AND page.source_id = requested_source_id + AND page.source_version_id = head.source_version_id + AND page.scan_epoch = head.scan_epoch + 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 + LEFT JOIN public.file_source_change_page AS baseline_terminal + ON baseline_terminal.organization_id = binding.organization_id + AND baseline_terminal.source_id = binding.source_id + AND baseline_terminal.source_version_id = binding.source_version_id + AND baseline_terminal.page_ref = binding.baseline_page_ref + WHERE EXISTS ( + SELECT 1 + FROM public.file_source_change AS change + WHERE change.organization_id = page.organization_id + AND change.source_id = page.source_id + AND change.source_version_id = page.source_version_id + AND change.page_ref = page.page_ref + AND change.change_kind = 'upsert' + AND ( + baseline_terminal.page_ref IS NULL + OR NOT EXISTS ( + SELECT 1 + FROM public.file_source_change_page AS baseline_page + JOIN public.file_source_change AS baseline_change + ON baseline_change.organization_id = + baseline_page.organization_id + AND baseline_change.source_id = baseline_page.source_id + AND baseline_change.source_version_id = + baseline_page.source_version_id + AND baseline_change.page_ref = baseline_page.page_ref + WHERE baseline_page.organization_id = + baseline_terminal.organization_id + AND baseline_page.source_id = + baseline_terminal.source_id + AND baseline_page.source_version_id = + baseline_terminal.source_version_id + AND baseline_page.scan_epoch = + baseline_terminal.scan_epoch + AND baseline_change.change_kind = 'upsert' + AND baseline_change.relative_path = + change.relative_path + AND baseline_change.content_sha256 = + change.content_sha256 + AND baseline_change.content_length = + change.content_length + ) + ) + ) + AND page.page_limit = 1 + AND NOT EXISTS ( + SELECT 1 + FROM public.file_acquisition AS acquisition + WHERE acquisition.organization_id = page.organization_id + AND acquisition.source_id = page.source_id + AND acquisition.source_version_id = page.source_version_id + AND acquisition.change_page_ref = page.page_ref + ) + ORDER BY page.page_ordinal; + END; + $function$ + """ + ) + op.execute(f"REVOKE ALL ON FUNCTION public.{_FUNCTION}{_SIGNATURE} FROM PUBLIC") + op.execute(f"ALTER FUNCTION public.{_FUNCTION}{_SIGNATURE} OWNER TO {_DEFINER}") + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_FUNCTION}{_SIGNATURE} TO {_CONTROL}" + ) + op.execute("RESET ROLE") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + + +def downgrade() -> None: + """Remove the read model without changing accepted pages or jobs.""" + + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute(f"DROP FUNCTION public.{_FUNCTION}{_SIGNATURE}") + op.execute("RESET ROLE") diff --git a/tests/integration/test_dogfood_runtime_activation.py b/tests/integration/test_dogfood_runtime_activation.py index e7a42bd1..04a904de 100644 --- a/tests/integration/test_dogfood_runtime_activation.py +++ b/tests/integration/test_dogfood_runtime_activation.py @@ -353,11 +353,14 @@ def test_dogfood_served_composition_delivers_release_scoped_file_evidence_before guarded_worker_engine, ) _add_policy_out_of_scope_distractors(migration_configuration, scenario) - assert _strictly_closer_distractor_count( - migration_configuration, - scenario, - target, - ) > DEFAULT_VECTOR_CANDIDATE_LIMIT + assert ( + _strictly_closer_distractor_count( + migration_configuration, + scenario, + target, + ) + > DEFAULT_VECTOR_CANDIDATE_LIMIT + ) configuration = _configuration(scenario, user_id) served: dict[str, object] = {} for name, value in _environment(configuration, runtime_configuration).items(): @@ -683,6 +686,7 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( organization_id = uuid4() user_id = uuid4() membership_id = uuid4() + receiver_id = uuid4() command = ( "context-engine-dogfood-seed", "--organization-id", @@ -691,6 +695,8 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( str(user_id), "--membership-id", str(membership_id), + "--file-import-service-principal-id", + str(receiver_id), ) engine = create_database_engine(migration_configuration) try: @@ -700,7 +706,17 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( text( """ SELECT user_id, status, membership_version, valid_from, - valid_until, xmin::text + valid_until, xmin::text, + ( + SELECT count(*) + FROM service_principal + WHERE organization_id = :organization_id + AND service_principal_id = :receiver_id + AND workload = 'supply.file-import' + AND worker_audience = 'context-engine-worker' + AND operation = 'file.import' + AND enabled IS TRUE + ) AS receivers FROM membership WHERE organization_id = :organization_id AND membership_id = :membership_id @@ -709,6 +725,7 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( { "organization_id": organization_id, "membership_id": membership_id, + "receiver_id": receiver_id, }, ).one() second = subprocess.run(command, check=True, capture_output=True, text=True) @@ -718,7 +735,17 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( text( """ SELECT user_id, status, membership_version, valid_from, - valid_until, xmin::text + valid_until, xmin::text, + ( + SELECT count(*) + FROM service_principal + WHERE organization_id = :organization_id + AND service_principal_id = :receiver_id + AND workload = 'supply.file-import' + AND worker_audience = 'context-engine-worker' + AND operation = 'file.import' + AND enabled IS TRUE + ) AS receivers FROM membership WHERE organization_id = :organization_id AND membership_id = :membership_id @@ -727,11 +754,13 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( { "organization_id": organization_id, "membership_id": membership_id, + "receiver_id": receiver_id, }, ).one() assert row == first_row assert tuple(row)[:3] == (user_id, "active", 1) assert row.valid_until is None + assert row.receivers == 1 with engine.begin() as connection: connection.execute( text( @@ -752,6 +781,19 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( assert "dogfood identity ready" not in future.stdout finally: with engine.begin() as connection: + connection.execute( + text( + """ + DELETE FROM service_principal + WHERE organization_id = :organization_id + AND service_principal_id = :receiver_id + """ + ), + { + "organization_id": organization_id, + "receiver_id": receiver_id, + }, + ) connection.execute( text( """ @@ -776,3 +818,179 @@ def test_dogfood_seed_cli_creates_one_idempotent_current_membership( {"user_id": user_id}, ) engine.dispose() + + +@pytest.mark.parametrize( + ("workload", "operation", "enabled"), + ( + ("supply.file-import", "file.import", False), + ("supply.noop", "noop.complete", True), + ), + ids=("disabled-exact-receiver", "conflicting-receiver-binding"), +) +def test_dogfood_seed_cli_rolls_back_when_file_import_receiver_conflicts( + migration_configuration: DatabaseConfiguration, + workload: str, + operation: str, + enabled: bool, +) -> None: + organization_id = uuid4() + user_id = uuid4() + membership_id = uuid4() + receiver_id = uuid4() + command = ( + "context-engine-dogfood-seed", + "--organization-id", + str(organization_id), + "--user-id", + str(user_id), + "--membership-id", + str(membership_id), + "--file-import-service-principal-id", + str(receiver_id), + ) + engine = create_database_engine(migration_configuration) + try: + with engine.begin() as connection: + connection.execute( + text( + """ + INSERT INTO organization (organization_id) + VALUES (:organization_id) + """ + ), + {"organization_id": organization_id}, + ) + connection.execute( + text( + """ + INSERT INTO service_principal ( + organization_id, service_principal_id, workload, + worker_audience, operation, enabled + ) VALUES ( + :organization_id, :receiver_id, :workload, + 'context-engine-worker', :operation, :enabled + ) + """ + ), + { + "organization_id": organization_id, + "receiver_id": receiver_id, + "workload": workload, + "operation": operation, + "enabled": enabled, + }, + ) + with engine.connect() as connection: + before = connection.execute( + text( + """ + SELECT organization.xmin::text AS organization_xmin, + principal.xmin::text AS receiver_xmin, + principal.workload, + principal.worker_audience, + principal.operation, + principal.enabled + FROM organization + JOIN service_principal AS principal + USING (organization_id) + WHERE organization_id = :organization_id + AND principal.service_principal_id = :receiver_id + """ + ), + { + "organization_id": organization_id, + "receiver_id": receiver_id, + }, + ).one() + + refused = subprocess.run(command, check=False, capture_output=True, text=True) + + assert refused.returncode != 0 + assert "dogfood identity ready" not in refused.stdout + with engine.connect() as connection: + after = connection.execute( + text( + """ + SELECT organization.xmin::text AS organization_xmin, + principal.xmin::text AS receiver_xmin, + principal.workload, + principal.worker_audience, + principal.operation, + principal.enabled + FROM organization + JOIN service_principal AS principal + USING (organization_id) + WHERE organization_id = :organization_id + AND principal.service_principal_id = :receiver_id + """ + ), + { + "organization_id": organization_id, + "receiver_id": receiver_id, + }, + ).one() + attempted_identity_rows = connection.execute( + text( + """ + SELECT + ( + SELECT count(*) + FROM user_account + WHERE user_id = :user_id + ) AS users, + ( + SELECT count(*) + FROM membership + WHERE organization_id = :organization_id + AND membership_id = :membership_id + ) AS memberships + """ + ), + { + "organization_id": organization_id, + "user_id": user_id, + "membership_id": membership_id, + }, + ).one() + assert after == before + assert attempted_identity_rows == (0, 0) + finally: + with engine.begin() as connection: + connection.execute( + text( + """ + DELETE FROM membership + WHERE organization_id = :organization_id + AND membership_id = :membership_id + """ + ), + { + "organization_id": organization_id, + "membership_id": membership_id, + }, + ) + connection.execute( + text( + """ + DELETE FROM service_principal + WHERE organization_id = :organization_id + AND service_principal_id = :receiver_id + """ + ), + { + "organization_id": organization_id, + "receiver_id": receiver_id, + }, + ) + connection.execute( + text( + "DELETE FROM organization WHERE organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ) + connection.execute( + text("DELETE FROM user_account WHERE user_id = :user_id"), + {"user_id": user_id}, + ) + engine.dispose() diff --git a/tests/integration/test_file_scan_operator_process.py b/tests/integration/test_file_scan_operator_process.py new file mode 100644 index 00000000..6b7e4c3f --- /dev/null +++ b/tests/integration/test_file_scan_operator_process.py @@ -0,0 +1,860 @@ +from __future__ import annotations + +import json +import os +import subprocess +from collections.abc import Iterator +from datetime import UTC, datetime +from pathlib import Path +from typing import cast +from uuid import UUID, uuid4 + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from sqlalchemy import Engine, text + +from adapters.file_source import FileChangeProvider, FileReadLimits, FileRootRegistry +from applications.operator_authentication import ( + CONTROL_OPERATOR_OPERATIONS_ENV, + CONTROL_OPERATOR_SECRET_ENV, + DOGFOOD_SECRET_ENV, + OPERATOR_ORGANIZATION_ENV, + RELEASE_OPERATOR_SECRET_ENV, + WORKER_SECRET_ENV, + LocalOperatorConfiguration, +) +from engine.control import ( + ChangeLimit, + ContextControl, + ControlOperation, + FileChangeControlProofs, + FileChangeProviderProofs, + FileChangeSource, + FileImportReceiver, + InitialScan, + ProviderOk, + SourceRef, +) +from engine.persistence import ( + DatabaseConfiguration, + PostgreSQLControlStore, + create_database_engine, +) +from engine.supply import CONTEXT_FRAGMENT_EMBEDDING_DIMENSION +from tests.integration.test_file_change_pages import ( + _SCENARIOS, + _delete_scenarios, +) + +pytestmark = pytest.mark.integration +ROOT = Path(__file__).parents[2] +CONTROL_SECRET = "issue-112-control-operator-secret-0001" +RELEASE_SECRET = "issue-112-release-operator-secret-0001" +DOGFOOD_SECRET = "issue-112-dogfood-runtime-secret-0001" +WORKER_KEY = bytes.fromhex("ab" * 32) +PROVIDER_KEY = bytes.fromhex("cd" * 32) +CHECKPOINT_KEY = bytes.fromhex("ef" * 32) + + +@pytest.fixture +def file_scan_scenario( + migration_configuration: DatabaseConfiguration, + tmp_path: Path, +) -> Iterator[tuple[UUID, UUID, UUID, Path, dict[str, str]]]: + organization_id = uuid4() + user_id = uuid4() + membership_id = uuid4() + receiver_id = uuid4() + root = tmp_path / "operator-scan-root" + root.mkdir() + scenarios: list[tuple[UUID, UUID]] = [] + _SCENARIOS.append(scenarios) + scenarios.append((organization_id, user_id)) + engine = create_database_engine(migration_configuration) + try: + with 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, + statement_timestamp() - interval '1 day' + ) + """ + ), + { + "org": organization_id, + "membership": membership_id, + "user": user_id, + }, + ) + 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}, + ) + environment = { + **os.environ, + OPERATOR_ORGANIZATION_ENV: str(organization_id), + CONTROL_OPERATOR_SECRET_ENV: CONTROL_SECRET, + RELEASE_OPERATOR_SECRET_ENV: RELEASE_SECRET, + DOGFOOD_SECRET_ENV: DOGFOOD_SECRET, + WORKER_SECRET_ENV: WORKER_KEY.hex(), + CONTROL_OPERATOR_OPERATIONS_ENV: ( + "register_source,read_source,read_source_progress," + "activate_file_change_feed," + "activate_file_delete_observations," + "accept_file_change_page,schedule_file_change_page" + ), + "CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_ID": str(membership_id), + "CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_VERSION": "1", + "CONTEXT_ENGINE_DOGFOOD_PRINCIPAL_REF": "principal:file-reader", + "CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID": str(receiver_id), + "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON": json.dumps( + {"operator-scan-root": str(root)} + ), + "CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX": (PROVIDER_KEY.hex()), + "CONTEXT_ENGINE_FILE_CHANGE_CHECKPOINT_SIGNING_KEY_HEX": ( + CHECKPOINT_KEY.hex() + ), + "CONTEXT_ENGINE_WORKER_EMBEDDING_PROVIDER": "twin", + "CONTEXT_ENGINE_WORKER_EMBEDDING_DIMENSION": str( + CONTEXT_FRAGMENT_EMBEDDING_DIMENSION + ), + } + yield organization_id, membership_id, receiver_id, root, environment + finally: + engine.dispose() + _SCENARIOS.remove(scenarios) + _delete_scenarios(migration_configuration, scenarios) + + +def _control( + arguments: list[str], + *, + environment: dict[str, str], + check: bool = True, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["context-engine-control", *arguments], + cwd=ROOT, + env=environment, + check=check, + capture_output=True, + text=True, + ) + + +def _worker(environment: dict[str, str]) -> dict[str, object]: + completed = subprocess.run( + ["context-engine-worker", "--dispatch-file-once"], + cwd=ROOT, + env=environment, + check=True, + capture_output=True, + text=True, + ) + assert completed.stderr == "" + return cast(dict[str, object], json.loads(completed.stdout)) + + +def _register_activated_source( + organization_id: UUID, + environment: dict[str, str], +) -> UUID: + registered = _control( + [ + "register-file-source", + "--organization-id", + str(organization_id), + "--display-name", + "Operator scan fixture", + "--root-ref", + "operator-scan-root", + "--idempotency-key", + "operator-scan-fixture-v1", + ], + environment=environment, + ) + source_ref = UUID(json.loads(registered.stdout)["sourceRef"]) + for subcommand in ( + "activate-change-feed", + "activate-delete-observations", + ): + _control( + [ + subcommand, + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=environment, + ) + return source_ref + + +def _register_change_feed_source( + organization_id: UUID, + environment: dict[str, str], +) -> UUID: + registered = _control( + [ + "register-file-source", + "--organization-id", + str(organization_id), + "--display-name", + "Operator v3 scan fixture", + "--root-ref", + "operator-scan-root", + "--idempotency-key", + "operator-v3-scan-fixture-v1", + ], + environment=environment, + ) + source_ref = UUID(json.loads(registered.stdout)["sourceRef"]) + _control( + [ + "activate-change-feed", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=environment, + ) + return source_ref + + +def _scan( + organization_id: UUID, + source_ref: UUID, + environment: dict[str, str], +) -> dict[str, object]: + completed = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=environment, + ) + assert completed.stderr == "" + return cast(dict[str, object], json.loads(completed.stdout)) + + +def test_scan_process_schedules_only_changed_upserts_and_existing_worker_consumes( + 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 / "a.md").write_text("# A\n\nFirst note.\n", encoding="utf-8") + (root / "refused.md").write_text( + "# Refused\n\n> blockquotes remain unsupported\n", + encoding="utf-8", + ) + source_ref = _register_activated_source(organization_id, environment) + + first = _scan(organization_id, source_ref, environment) + + assert first == { + "advancedCursor": first["advancedCursor"], + "changesAccepted": 2, + "compilationRefusals": 1, + "deletesObserved": 0, + "importsScheduled": 2, + "pathsObserved": 2, + "sourceRef": str(source_ref), + } + assert type(first["advancedCursor"]) is str + assert str(first["advancedCursor"]).startswith("facp_") + assert [_worker(environment)["outcome"] for _ in range(3)] == [ + "dispatched", + "refused", + "no_work", + ] + + engine = create_database_engine(migration_configuration) + try: + with engine.connect() as connection: + first_snapshot = tuple( + connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM file_import_job + WHERE organization_id = :org), + (SELECT count(*) FROM context_fragment + WHERE organization_id = :org), + (SELECT count(*) FROM context_fragment + WHERE organization_id = :org + AND vector_dims(embedding) = 384) + """ + ), + {"org": organization_id}, + ).one() + ) + assert first_snapshot == (2, 1, 1) + + unchanged = _scan(organization_id, source_ref, environment) + assert unchanged == { + "advancedCursor": first["advancedCursor"], + "changesAccepted": 0, + "compilationRefusals": 0, + "deletesObserved": 0, + "importsScheduled": 0, + "pathsObserved": 2, + "sourceRef": str(source_ref), + } + with engine.connect() as connection: + assert ( + connection.execute( + text( + "SELECT count(*) FROM file_import_job " + "WHERE organization_id = :org" + ), + {"org": organization_id}, + ).scalar_one() + == 2 + ) + + (root / "new.md").write_text("# New\n\nSecond note.\n", encoding="utf-8") + changed = _scan(organization_id, source_ref, environment) + assert changed == { + "advancedCursor": changed["advancedCursor"], + "changesAccepted": 1, + "compilationRefusals": 0, + "deletesObserved": 0, + "importsScheduled": 1, + "pathsObserved": 3, + "sourceRef": str(source_ref), + } + assert changed["advancedCursor"] != first["advancedCursor"] + with engine.connect() as connection: + assert ( + connection.execute( + text( + "SELECT count(*) FROM file_import_job " + "WHERE organization_id = :org" + ), + {"org": organization_id}, + ).scalar_one() + == 3 + ) + + assert _worker(environment)["outcome"] == "dispatched" + assert _worker(environment)["outcome"] == "no_work" + with engine.connect() as connection: + final_snapshot = tuple( + connection.execute( + text( + """ + SELECT count(*), + count(*) FILTER ( + WHERE vector_dims(embedding) = 384 + ) + FROM context_fragment + WHERE organization_id = :org + """ + ), + {"org": organization_id}, + ).one() + ) + assert final_snapshot == (2, 2) + + with engine.connect() as connection: + delete_effects_before = tuple( + connection.execute( + text( + """ + SELECT + (SELECT policy_epoch + FROM organization_policy_epoch + WHERE organization_id = :org), + (SELECT count(*) + FROM file_resource_cleanup_intent + WHERE organization_id = :org), + (SELECT count(*) + FROM file_delete_observation_execution + WHERE organization_id = :org), + (SELECT count(*) + FROM context_resource + WHERE organization_id = :org + AND tombstoned IS TRUE), + (SELECT count(*) + FROM context_resource + WHERE organization_id = :org + AND tombstoned IS FALSE), + (SELECT count(*) + FROM context_revision + WHERE organization_id = :org) + """ + ), + {"org": organization_id}, + ).one() + ) + + (root / "a.md").unlink() + deleted = _scan(organization_id, source_ref, environment) + assert deleted == { + "advancedCursor": deleted["advancedCursor"], + "changesAccepted": 1, + "compilationRefusals": 0, + "deletesObserved": 1, + "importsScheduled": 0, + "pathsObserved": 2, + "sourceRef": str(source_ref), + } + assert deleted["advancedCursor"] != changed["advancedCursor"] + with engine.connect() as connection: + after_delete = tuple( + connection.execute( + text( + """ + SELECT + (SELECT policy_epoch + FROM organization_policy_epoch + WHERE organization_id = :org), + (SELECT count(*) + FROM file_resource_cleanup_intent + WHERE organization_id = :org), + (SELECT count(*) + FROM file_delete_observation_execution + WHERE organization_id = :org), + (SELECT count(*) + FROM context_resource + WHERE organization_id = :org + AND tombstoned IS TRUE), + (SELECT count(*) + FROM context_resource + WHERE organization_id = :org + AND tombstoned IS FALSE), + (SELECT count(*) + FROM context_revision + WHERE organization_id = :org), + (SELECT count(*) + FROM file_import_job + WHERE organization_id = :org) + """ + ), + {"org": organization_id}, + ).one() + ) + assert after_delete[:-1] == delete_effects_before + assert after_delete[-1] == 3 + + unchanged_after_delete = _scan( + organization_id, + source_ref, + environment, + ) + assert unchanged_after_delete == { + "advancedCursor": deleted["advancedCursor"], + "changesAccepted": 0, + "compilationRefusals": 0, + "deletesObserved": 0, + "importsScheduled": 0, + "pathsObserved": 2, + "sourceRef": str(source_ref), + } + with engine.connect() as connection: + unchanged_counts = tuple( + connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM file_import_job + WHERE organization_id = :org), + (SELECT count(*) FROM context_revision + WHERE organization_id = :org) + """ + ), + {"org": organization_id}, + ).one() + ) + assert unchanged_counts == (3, 2) + finally: + engine.dispose() + + +def test_scan_process_recovers_a_complete_accepted_page_missing_its_schedule( + 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 / "recover.md").write_text( + "# Recover\n\nSchedule this accepted note.\n", + encoding="utf-8", + ) + source_ref = _register_activated_source(organization_id, environment) + engine = create_database_engine(migration_configuration) + try: + with engine.begin() as connection: + connection.execute( + text( + """ + UPDATE service_principal + SET enabled = false + WHERE organization_id = :org + AND service_principal_id = :receiver + """ + ), + {"org": organization_id, "receiver": receiver_id}, + ) + + interrupted = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=environment, + check=False, + ) + + assert interrupted.returncode != 0 + assert interrupted.stdout == "" + assert interrupted.stderr == "context-engine-control: operation refused\n" + with engine.connect() as connection: + stranded = tuple( + connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM file_source_change_page + WHERE organization_id = :org + AND source_id = :source + AND complete IS TRUE), + (SELECT count(*) FROM file_import_job + WHERE organization_id = :org + AND source_id = :source) + """ + ), + {"org": organization_id, "source": source_ref}, + ).one() + ) + assert stranded == (1, 0) + + with engine.begin() as connection: + connection.execute( + text( + """ + UPDATE service_principal + SET enabled = true + WHERE organization_id = :org + AND service_principal_id = :receiver + """ + ), + {"org": organization_id, "receiver": receiver_id}, + ) + + recovered = _scan(organization_id, source_ref, environment) + + assert recovered == { + "advancedCursor": recovered["advancedCursor"], + "changesAccepted": 0, + "compilationRefusals": 0, + "deletesObserved": 0, + "importsScheduled": 1, + "pathsObserved": 1, + "sourceRef": str(source_ref), + } + assert _worker(environment)["outcome"] == "dispatched" + assert _worker(environment)["outcome"] == "no_work" + with engine.connect() as connection: + assert tuple( + connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM file_import_job + WHERE organization_id = :org + AND source_id = :source), + (SELECT count(*) FROM context_fragment + WHERE organization_id = :org) + """ + ), + {"org": organization_id, "source": source_ref}, + ).one() + ) == (1, 1) + finally: + engine.dispose() + + +def test_scan_process_does_not_reconcile_a_foreign_larger_mixed_page( + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + file_scan_scenario: tuple[UUID, UUID, UUID, Path, dict[str, str]], +) -> None: + organization_id, _membership_id, receiver_id, root, environment = file_scan_scenario + (root / "a.md").write_text("# A\n\nOriginal.\n", encoding="utf-8") + (root / "b.md").write_text("# B\n\nUnchanged.\n", encoding="utf-8") + source_ref = _register_activated_source(organization_id, environment) + baseline = _scan(organization_id, source_ref, environment) + assert baseline["importsScheduled"] == 2 + + (root / "a.md").write_text("# A\n\nChanged.\n", encoding="utf-8") + configuration = LocalOperatorConfiguration.load(environment) + assert configuration is not None + + def clock() -> datetime: + return datetime.now(UTC) + + authority = configuration.authorities(clock=clock).control + provider_key = Ed25519PrivateKey.from_private_bytes(PROVIDER_KEY) + checkpoint_key = Ed25519PrivateKey.from_private_bytes(CHECKPOINT_KEY) + control = ContextControl( + store=PostgreSQLControlStore( + guarded_control_engine, + clock=clock, + file_import_receiver=FileImportReceiver(receiver_id), + file_change_checkpoint_signing_key=checkpoint_key, + ), + authority=authority, + clock=clock, + file_change_proofs=FileChangeControlProofs( + provider_verification_key=provider_key.public_key() + ), + ) + source = SourceRef(source_ref) + with authority.authorize( + opaque_credential=CONTROL_SECRET, + operation=ControlOperation.READ_SOURCE, + request_id="foreign-larger-read-source", + ) as call: + manifest = control.read_source(call, source) + with authority.authorize( + opaque_credential=CONTROL_SECRET, + operation=ControlOperation.READ_SOURCE_PROGRESS, + request_id="foreign-larger-read-baseline", + ) as call: + progress = control.read_file_source_progress(call, source) + provider_source = FileChangeSource( + organization_id, + manifest.active_version, + scan_head=progress.change_scan_head, + complete_baseline=progress.complete_change_baseline, + ) + with FileRootRegistry( + {manifest.active_version.root_ref: root}, + limits=FileReadLimits(max_file_bytes=1_048_576), + ) as roots: + page = FileChangeProvider( + roots, + proofs=FileChangeProviderProofs( + provider_signing_key=provider_key, + checkpoint_verification_key=checkpoint_key.public_key(), + ), + ).read_changes(provider_source, InitialScan(), ChangeLimit(2)) + assert type(page) is ProviderOk + assert page.value.page_limit == 2 + assert page.value.complete is True + assert tuple(change.path.value for change in page.value.changes) == ( + "a.md", + "b.md", + ) + with authority.authorize( + opaque_credential=CONTROL_SECRET, + operation=ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + request_id="foreign-larger-accept-page", + ) as call: + control.accept_file_change_page(call, page.value) + with authority.authorize( + opaque_credential=CONTROL_SECRET, + operation=ControlOperation.READ_SOURCE_PROGRESS, + request_id="foreign-larger-read-pending", + ) as call: + accepted_progress = control.read_file_source_progress(call, source) + + assert accepted_progress.pending_change_schedules == () + refused = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=environment, + check=False, + ) + assert refused.returncode != 0 + assert refused.stdout == "" + assert refused.stderr == "context-engine-control: operation refused\n" + engine = create_database_engine(migration_configuration) + try: + with engine.connect() as connection: + assert connection.execute( + text( + "SELECT count(*) FROM file_import_job " + "WHERE organization_id = :org AND source_id = :source" + ), + {"org": organization_id, "source": source_ref}, + ).scalar_one() == 2 + finally: + engine.dispose() + + +def test_scan_process_refuses_absent_configuration_generically( + file_scan_scenario: tuple[UUID, UUID, UUID, Path, dict[str, str]], +) -> None: + organization_id, _membership_id, _receiver_id, _root, environment = ( + file_scan_scenario + ) + source_ref = uuid4() + absent = environment.copy() + del absent["CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX"] + + refused = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=absent, + check=False, + ) + + assert refused.returncode != 0 + assert refused.stdout == "" + assert refused.stderr == "context-engine-control: operation refused\n" + rendered = refused.stdout + refused.stderr + assert str(organization_id) not in rendered + assert str(source_ref) not in rendered + + no_operator = environment.copy() + for name in ( + OPERATOR_ORGANIZATION_ENV, + CONTROL_OPERATOR_SECRET_ENV, + RELEASE_OPERATOR_SECRET_ENV, + DOGFOOD_SECRET_ENV, + WORKER_SECRET_ENV, + CONTROL_OPERATOR_OPERATIONS_ENV, + ): + del no_operator[name] + refused_without_operator = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=no_operator, + check=False, + ) + assert refused_without_operator.returncode != 0 + assert refused_without_operator.stdout == "" + assert ( + refused_without_operator.stderr == "context-engine-control: operation refused\n" + ) + + wrong_organization = uuid4() + wrong_org = _control( + [ + "scan", + "--organization-id", + str(wrong_organization), + "--source-ref", + str(source_ref), + ], + environment=environment, + check=False, + ) + assert wrong_org.returncode != 0 + assert wrong_org.stdout == "" + assert wrong_org.stderr == "context-engine-control: operation refused\n" + assert str(wrong_organization) not in wrong_org.stderr + assert str(source_ref) not in wrong_org.stderr + + reused_proof_key = environment.copy() + reused_proof_key["CONTEXT_ENGINE_FILE_CHANGE_CHECKPOINT_SIGNING_KEY_HEX"] = ( + reused_proof_key["CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX"] + ) + refused_reuse = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=reused_proof_key, + check=False, + ) + assert refused_reuse.returncode != 0 + assert refused_reuse.stdout == "" + assert refused_reuse.stderr == "context-engine-control: operation refused\n" + + reused_across_planes = environment.copy() + reused_across_planes[CONTROL_OPERATOR_SECRET_ENV] = reused_across_planes[ + "CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX" + ] + refused_cross_plane = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=reused_across_planes, + check=False, + ) + assert refused_cross_plane.returncode != 0 + assert refused_cross_plane.stdout == "" + assert refused_cross_plane.stderr == "context-engine-control: operation refused\n" + + +def test_scan_process_refuses_a_v3_source_without_complete_baseline_carrier( + file_scan_scenario: tuple[UUID, UUID, UUID, Path, dict[str, str]], +) -> None: + organization_id, _membership_id, _receiver_id, root, environment = ( + file_scan_scenario + ) + (root / "v3.md").write_text("# V3\n\nNot scan-active.\n", encoding="utf-8") + source_ref = _register_change_feed_source(organization_id, environment) + + refused = _control( + [ + "scan", + "--organization-id", + str(organization_id), + "--source-ref", + str(source_ref), + ], + environment=environment, + check=False, + ) + + assert refused.returncode != 0 + assert refused.stdout == "" + assert refused.stderr == "context-engine-control: operation refused\n" diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index 16600eee..80bf5c99 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -2369,6 +2369,63 @@ def definition() -> str: assert "selected_capabilities NOT IN" in definition() +def test_pending_file_schedule_projection_revision_downgrades_and_reapplies( + migration_configuration: DatabaseConfiguration, +) -> None: + """Issue #112 adds only one reversible read-only reconciliation function.""" + + alembic_configuration = Config(ROOT / "alembic.ini") + try: + command.downgrade(alembic_configuration, "20260727_0037") + assert _revision_rows(migration_configuration) == ["20260727_0037"] + engine = create_database_engine(migration_configuration) + try: + with engine.connect() as connection: + assert connection.execute( + text( + """ + SELECT to_regprocedure( + 'public.context_control_read_pending_file_change_schedules(uuid,uuid)' + ) IS NULL + """ + ) + ).scalar_one() is True + finally: + engine.dispose() + finally: + command.upgrade(alembic_configuration, "head") + + assert _revision_rows(migration_configuration) == [HEAD_REVISION] + engine = create_database_engine(migration_configuration) + try: + with engine.connect() as connection: + assert connection.execute( + text( + """ + SELECT ARRAY[ + has_function_privilege( + 'context_engine_control', + 'public.context_control_read_pending_file_change_schedules(uuid,uuid)', + 'EXECUTE' + ), + has_function_privilege( + 'context_engine_runtime', + 'public.context_control_read_pending_file_change_schedules(uuid,uuid)', + 'EXECUTE' + ), + has_function_privilege( + 'context_engine_worker', + 'public.context_control_read_pending_file_change_schedules(uuid,uuid)', + 'EXECUTE' + ) + ] + """ + ) + ).scalar_one() == [True, False, False] + finally: + engine.dispose() + + def test_fragment_embedding_revision_preserves_retained_fragments( tmp_path: Path, migration_configuration: DatabaseConfiguration, diff --git a/tests/process/test_processes.py b/tests/process/test_processes.py index c463e116..0d41b0d5 100644 --- a/tests/process/test_processes.py +++ b/tests/process/test_processes.py @@ -53,6 +53,7 @@ def test_control_process_help_and_unknown_subcommand() -> None: "read-source", "activate-change-feed", "activate-delete-observations", + "scan", ): assert subcommand in help_result.stdout diff --git a/tests/unit/test_file_dispatch.py b/tests/unit/test_file_dispatch.py index d95581e7..e111ca20 100644 --- a/tests/unit/test_file_dispatch.py +++ b/tests/unit/test_file_dispatch.py @@ -384,9 +384,7 @@ def test_worker_default_file_limit_accepts_above_legacy_ceiling_and_refuses_over root = tmp_path / "configured-root" root.mkdir() (root / "above-legacy.md").write_bytes(b"a" * 4_097) - (root / "oversize.md").write_bytes( - b"b" * (DEFAULT_WORKER_MAX_FILE_BYTES + 1) - ) + (root / "oversize.md").write_bytes(b"b" * (DEFAULT_WORKER_MAX_FILE_BYTES + 1)) monkeypatch.setenv( "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON", json.dumps({"configured-root": str(root)}), @@ -394,12 +392,15 @@ def test_worker_default_file_limit_accepts_above_legacy_ceiling_and_refuses_over monkeypatch.delenv("CONTEXT_ENGINE_WORKER_MAX_FILE_BYTES", raising=False) with _file_dispatch_roots() as roots: - assert len( - roots.read( - FileRootRef("configured-root"), - FileImportPath("above-legacy.md"), + assert ( + len( + roots.read( + FileRootRef("configured-root"), + FileImportPath("above-legacy.md"), + ) ) - ) == 4_097 + == 4_097 + ) with pytest.raises(LookupError, match="regular configured-root file"): roots.read( FileRootRef("configured-root"), @@ -418,7 +419,10 @@ def test_worker_rejects_invalid_file_byte_limits( _file_read_limits() -@pytest.mark.parametrize("document", ["[]", "{}", '{"root": 1}', "not-json"]) +@pytest.mark.parametrize( + "document", + ["[]", "{}", '{"root": 1}', '{"root": "relative"}', "not-json"], +) def test_dispatch_rejects_invalid_server_root_registry( monkeypatch: pytest.MonkeyPatch, document: str, diff --git a/tests/unit/test_file_scan.py b/tests/unit/test_file_scan.py new file mode 100644 index 00000000..e62ee6a1 --- /dev/null +++ b/tests/unit/test_file_scan.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import hashlib +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +import pytest + +from adapters.file_source import FileReadLimits, FileRootRegistry +from applications.file_scan import _compilation_refused +from engine.control import FileRootRef, SourceManifest, SourceNotAvailable, SourceRef + + +def _manifest(root_ref: FileRootRef) -> SourceManifest: + now = datetime(2026, 7, 27, tzinfo=UTC) + return SourceManifest.registered_file( + source_ref=SourceRef(uuid4()), + version_ref=uuid4(), + display_name="scan preflight", + root_ref=root_ref, + created_at=now, + ) + + +def test_scan_preflight_compiles_only_the_exact_accepted_file_identity( + tmp_path: Path, +) -> None: + root_ref = FileRootRef("scan-preflight-root") + root = tmp_path / "root" + root.mkdir() + target = root / "note.md" + accepted = b"# Accepted\n\nStable note.\n" + target.write_bytes(accepted) + + with FileRootRegistry( + {root_ref: root}, + limits=FileReadLimits(max_file_bytes=1_048_576), + ) as roots: + assert ( + _compilation_refused( + roots, + _manifest(root_ref), + "note.md", + hashlib.sha256(accepted).hexdigest(), + len(accepted), + ) + == 0 + ) + + target.write_bytes(b"# Changed\n\nDifferent bytes.\n") + with pytest.raises(SourceNotAvailable): + _compilation_refused( + roots, + _manifest(root_ref), + "note.md", + hashlib.sha256(accepted).hexdigest(), + len(accepted), + ) diff --git a/tests/unit/test_file_source_progress.py b/tests/unit/test_file_source_progress.py index 6fa7c89b..e34be883 100644 --- a/tests/unit/test_file_source_progress.py +++ b/tests/unit/test_file_source_progress.py @@ -188,6 +188,7 @@ def test_progress_contracts_keep_checkpoint_and_watermark_semantics_separate() - "publish_watermark", "change_scan_head", "complete_change_baseline", + "pending_change_schedules", ] assert FileSourceChangeKind.FILE_IMPORT.value == "file_import" assert FileSourceChangeKind.FILE_TOMBSTONE.value == "file_tombstone" @@ -427,9 +428,7 @@ def test_complete_change_baseline_is_distinct_from_an_incomplete_head() -> None: complete_change_baseline=FileChangeBaseline( reference=replace( complete_reference, - source_version_ref=UUID( - "3ea05cf1-29d9-46c8-a082-0798dc46cdfd" - ), + source_version_ref=UUID("3ea05cf1-29d9-46c8-a082-0798dc46cdfd"), ), entries=baseline.entries, ), diff --git a/tests/unit/test_schema_security_manifest.py b/tests/unit/test_schema_security_manifest.py index 4ec34a87..bd6f3fa3 100644 --- a/tests/unit/test_schema_security_manifest.py +++ b/tests/unit/test_schema_security_manifest.py @@ -32,7 +32,7 @@ def test_manifest_classifies_the_exact_current_release_schema() -> None: document = manifest() tables = table_entries(document) - assert document["manifestVersion"] == "33.0.0" + assert document["manifestVersion"] == "34.0.0" assert set(tables) == { "active_release_manifest", "action_delivery_attempt", @@ -831,8 +831,13 @@ def test_issue_21_file_source_manifest_is_closed_and_role_separated() -> None: for operation in manifest()["controlOperations"] if operation["name"] == "read_file_source_progress" ) + assert progress_read["databaseFunctions"] == [ + "context_control_read_file_source_progress", + "context_control_read_pending_file_change_schedules", + ] assert "file_source_change_page" in progress_read["reads"] assert "file_source_delete_observation_page" in progress_read["reads"] + assert "file_acquisition" in progress_read["reads"] schedule = next( operation for operation in manifest()["controlOperations"] @@ -846,9 +851,7 @@ def test_issue_21_file_source_manifest_is_closed_and_role_separated() -> None: assert schedule["filesystemAccessAllowed"] is False assert schedule["completePageValidation"] is True assert schedule["migrationFence"] == { - "sharedAdvisoryLock": ( - "context-engine.file-change-scheduling-migration-fence" - ), + "sharedAdvisoryLock": ("context-engine.file-change-scheduling-migration-fence"), "definerOnlyGenerationRead": ( "alembic_version SELECT revoked by 0032 downgrade" ), @@ -879,9 +882,7 @@ def test_issue_21_file_source_manifest_is_closed_and_role_separated() -> None: assert watermark_fence["function"] == ( "context_file_source_fence_scheduled_publication_epoch" ) - assert watermark_fence["negativeTestIds"] == [ - "PG-FILE-CHANGE-SUPERSESSION-083" - ] + assert watermark_fence["negativeTestIds"] == ["PG-FILE-CHANGE-SUPERSESSION-083"] assert { key["name"]: key["columns"] for key in entries["file_acquisition"]["organizationInclusiveKeys"] @@ -908,6 +909,7 @@ 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_offboard_file_source", ], "context_engine_learning": [], @@ -915,12 +917,12 @@ def test_issue_21_file_source_manifest_is_closed_and_role_separated() -> None: "context_engine_security_operator": [], "context_engine_worker": [], "context_engine_worker_lease_definer": ["SELECT", "UPDATE"], - "context_engine_action_prepare_definer": ["SELECT"], - "context_engine_action_execute_definer": ["SELECT"], - "context_engine_file_dispatch_definer": [ - "SELECT", - "UPDATE lifecycle_state, active_version_id", - ], + "context_engine_action_prepare_definer": ["SELECT"], + "context_engine_action_execute_definer": ["SELECT"], + "context_engine_file_dispatch_definer": [ + "SELECT", + "UPDATE lifecycle_state, active_version_id", + ], } assert version["permittedOperations"] == { "context_engine_control": [ @@ -928,6 +930,7 @@ 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", ], "context_engine_learning": [], "context_engine_runtime": [],