diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index 0bdf6ca1..c819d197 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -4,6 +4,19 @@ This file is generated by `scripts/third_party_governance.py`. Do not edit it by hand. Each registered source is pinned and distributed with its license text. +## onyx + +- Upstream: https://github.com/onyx-dot-app/onyx.git +- Commit: `2fb3dd10493b3883870fa8adced5b1a0e114feff` +- License: MIT (`third_party/onyx/LICENSE.upstream`) +- Reuse mode: `copy-patch` +- Approval: https://github.com/stone16/context-engine/issues/126 +- Copied paths: + - `backend/onyx/connectors/interfaces.py` + - `backend/onyx/connectors/connector_runner.py` + - `backend/onyx/connectors/models.py` + - `backend/onyx/connectors/registry.py` + ## ragflow - Upstream: https://github.com/infiniflow/ragflow.git diff --git a/THIRD_PARTY_SBOM.cyclonedx.json b/THIRD_PARTY_SBOM.cyclonedx.json index e72924d7..e8b6f527 100644 --- a/THIRD_PARTY_SBOM.cyclonedx.json +++ b/THIRD_PARTY_SBOM.cyclonedx.json @@ -1,6 +1,47 @@ { "bomFormat": "CycloneDX", "components": [ + { + "bom-ref": "context-engine:third-party:onyx", + "externalReferences": [ + { + "type": "vcs", + "url": "https://github.com/onyx-dot-app/onyx.git#2fb3dd10493b3883870fa8adced5b1a0e114feff" + } + ], + "licenses": [ + { + "license": { + "id": "MIT" + } + } + ], + "name": "onyx", + "properties": [ + { + "name": "context-engine:scope", + "value": "third_party/onyx" + }, + { + "name": "context-engine:file-sha256", + "value": "third_party/onyx/connectors/interfaces.py=6a9f3f753972d2036c9fe90d4a394b4cd22b325e92e00a01a550d7f8cc94f87f" + }, + { + "name": "context-engine:file-sha256", + "value": "third_party/onyx/connectors/connector_runner.py=b650097cb44b1f56a07f80befa7ca6123771ab7244db77bb5637305c92b834de" + }, + { + "name": "context-engine:file-sha256", + "value": "third_party/onyx/connectors/models.py=e3b95634e689457be8dbb99a253c26924143a0d59ac5bd5fb7629b57c3f06454" + }, + { + "name": "context-engine:file-sha256", + "value": "third_party/onyx/connectors/registry.py=b134c07f20959ed44802c1fc753f9c0a8e5a9388134f4bc57b4bcc5ba4166618" + } + ], + "type": "library", + "version": "2fb3dd10493b3883870fa8adced5b1a0e114feff" + }, { "bom-ref": "context-engine:third-party:ragflow", "externalReferences": [ diff --git a/adapters/connectors/__init__.py b/adapters/connectors/__init__.py new file mode 100644 index 00000000..2aece734 --- /dev/null +++ b/adapters/connectors/__init__.py @@ -0,0 +1,15 @@ +"""ContextEngine-owned connector adapters.""" + +from adapters.connectors.file import ( + FileConnectorAdapter, + FileConnectorProcessAdapter, + FileRootVaultSource, + PermissionObservationFailed, +) + +__all__ = [ + "FileConnectorAdapter", + "FileConnectorProcessAdapter", + "FileRootVaultSource", + "PermissionObservationFailed", +] diff --git a/adapters/connectors/file.py b/adapters/connectors/file.py new file mode 100644 index 00000000..1013a9e8 --- /dev/null +++ b/adapters/connectors/file.py @@ -0,0 +1,537 @@ +"""File/Obsidian connector translated onto the closed Supply execution seam.""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import subprocess +import sys +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Protocol, cast +from uuid import UUID + +from adapters.file_source import FileRootRegistry +from engine.control import FileRootRef +from engine.supply import ( + ConnectorCheckpointBinding, + SourceAclEvidenceClass, + SourceAclObservation, + SupplyChangePage, + SupplyDocumentDeleteObservation, + SupplyDocumentEnvelope, + WorkerLeaseToken, + deserialize_supply_change_page, +) +from third_party.onyx.connectors.connector_runner import ConnectorRunner +from third_party.onyx.connectors.interfaces import ( + CheckpointedConnectorWithPermSync, + CheckpointOutput, +) +from third_party.onyx.connectors.models import ( + ConnectorCheckpoint, + ConnectorFailure, + DeletedDocument, + Document, +) + +_CHECKPOINT_VERSION = 1 +_BATCH_SIZE = 100 +_WEAK_ACL_JUSTIFICATION = "local File/Obsidian has no corpus ACL API" + + +class PermissionObservationFailed(RuntimeError): + """The source ACL observation failed, so no Article may be emitted.""" + + +@dataclass(frozen=True, slots=True) +class VaultSnapshotEntry: + """One stable source snapshot entry presented to the connector.""" + + path: str + content: bytes + + def __post_init__(self) -> None: + if ( + type(self.path) is not str + or not self.path + or self.path != self.path.strip() + ): + raise ValueError("vault snapshot path must be canonical") + if type(self.content) is not bytes or not self.content: + raise ValueError("vault snapshot content must be nonempty bytes") + + +class VaultSource(Protocol): + def snapshot(self) -> tuple[VaultSnapshotEntry, ...]: ... + + +class VaultPermissionObserver(Protocol): + def observe_acl(self, path: str) -> None: ... + + +class FileConnectorTwin(VaultSource, VaultPermissionObserver, Protocol): + policy_epoch: int + + def now(self) -> datetime: ... + + +@dataclass(frozen=True, slots=True) +class FileCheckpoint: + """Decoded connector state; the engine stores only its opaque bytes.""" + + entries: tuple[tuple[str, str], ...] + + def __post_init__(self) -> None: + if type(self.entries) is not tuple: + raise TypeError("File checkpoint entries must be a tuple") + paths: list[str] = [] + for entry in self.entries: + if type(entry) is not tuple or len(entry) != 2: + raise TypeError("File checkpoint entries must be exact pairs") + path, digest = entry + if type(path) is not str or not path or path != path.strip(): + raise ValueError("File checkpoint path must be canonical") + if ( + type(digest) is not str + or len(digest) != 64 + or any(character not in "0123456789abcdef" for character in digest) + ): + raise ValueError("File checkpoint digest must be lowercase SHA-256") + paths.append(path) + if paths != sorted(set(paths), key=lambda value: value.encode("utf-8")): + raise ValueError("File checkpoint paths must be sorted and unique") + + @property + def paths(self) -> tuple[str, ...]: + return tuple(path for path, _digest in self.entries) + + +def encode_file_checkpoint(checkpoint: FileCheckpoint) -> bytes: + if type(checkpoint) is not FileCheckpoint: + raise TypeError("File checkpoint encoding requires an exact value") + return json.dumps( + {"entries": [list(entry) for entry in checkpoint.entries], "version": 1}, + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ).encode("ascii") + + +def decode_file_checkpoint(payload: bytes) -> FileCheckpoint: + if type(payload) is not bytes or not payload: + raise ValueError("File checkpoint must be nonempty bytes") + try: + decoded = json.loads(payload) + if type(decoded) is not dict or set(decoded) != {"entries", "version"}: + raise ValueError + if decoded["version"] != _CHECKPOINT_VERSION: + raise ValueError + entries = decoded["entries"] + if type(entries) is not list: + raise ValueError + return FileCheckpoint( + tuple((cast(str, item[0]), cast(str, item[1])) for item in entries) + ) + except (IndexError, KeyError, TypeError, ValueError, json.JSONDecodeError): + raise ValueError("File checkpoint is unavailable") from None + + +class FileRootVaultSource: + """Use the existing anchored File registry as the sole filesystem reader.""" + + __slots__ = ("_registry", "_root_ref") + + def __init__(self, registry: FileRootRegistry, root_ref: FileRootRef) -> None: + if type(registry) is not FileRootRegistry or type(root_ref) is not FileRootRef: + raise TypeError("File vault source requires registered root contracts") + self._registry = registry + self._root_ref = root_ref + + def snapshot(self) -> tuple[VaultSnapshotEntry, ...]: + return tuple( + VaultSnapshotEntry(path.value, content) + for path, content in self._registry.observe_markdown_files(self._root_ref) + ) + + def observe_acl(self, path: str) -> None: + if type(path) is not str or not path: + raise PermissionObservationFailed("File ACL observation is unavailable") + # The local source has no ACL endpoint. Successfully classifying that absence + # is honest Weak evidence; failures must raise instead of falling back here. + + +class FileConnectorProcessAdapter: + """Invoke one isolated File scan per engine checkpoint proposal.""" + + __slots__ = ( + "_checkpoint", + "_policy_epoch", + "_root_path", + "_root_ref", + "_service_actor_expires_at", + "_service_principal_id", + "_idempotency_key", + "_worker_lease", + ) + + def __init__( + self, + root_ref: FileRootRef, + root_path: Path, + *, + policy_epoch: int, + worker_lease: WorkerLeaseToken, + service_principal_id: UUID, + idempotency_key: str, + service_actor_expires_at: datetime, + ) -> None: + if type(root_ref) is not FileRootRef: + raise TypeError("File connector process requires FileRootRef") + if not isinstance(root_path, Path) or not root_path.is_absolute(): + raise ValueError("File connector process requires an absolute root") + if type(policy_epoch) is not int or policy_epoch < 1: + raise ValueError("File connector process Policy Epoch must be positive") + if type(worker_lease) is not WorkerLeaseToken: + raise TypeError("File connector process requires WorkerLeaseToken") + if type(service_principal_id) is not UUID: + raise TypeError("File connector process requires ServiceActor UUID") + if ( + type(idempotency_key) is not str + or len(idempotency_key) != 64 + or any(character not in "0123456789abcdef" for character in idempotency_key) + ): + raise ValueError("File connector process requires an idempotency digest") + if ( + type(service_actor_expires_at) is not datetime + or service_actor_expires_at.tzinfo is None + or service_actor_expires_at.utcoffset() != timedelta(0) + ): + raise ValueError("File connector process requires a UTC actor expiry") + self._root_ref = root_ref + self._root_path = root_path + self._policy_epoch = policy_epoch + self._worker_lease = worker_lease + self._service_principal_id = service_principal_id + self._idempotency_key = idempotency_key + self._service_actor_expires_at = service_actor_expires_at + self._checkpoint: bytes | None = None + + def load_checkpoint(self, opaque_checkpoint: bytes | None) -> None: + if opaque_checkpoint is not None: + decode_file_checkpoint(opaque_checkpoint) + self._checkpoint = opaque_checkpoint + + def load(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + return self._run(binding) + + def poll(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + return self._run(binding) + + def _run(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + if type(binding) is not ConnectorCheckpointBinding: + raise TypeError("File connector process requires exact binding") + request = json.dumps( + { + "opaque_checkpoint": ( + None + if self._checkpoint is None + else base64.b64encode(self._checkpoint).decode("ascii") + ), + "idempotency_key": self._idempotency_key, + "organization_id": str(binding.organization_id), + "policy_epoch": self._policy_epoch, + "root_path": str(self._root_path), + "root_ref": self._root_ref.value, + "service_actor_expires_at": self._service_actor_expires_at.isoformat(), + "service_principal_id": str(self._service_principal_id), + "source_version_id": str(binding.source_version_id), + "worker_job_id": str(binding.worker_job_id), + "worker_lease": self._worker_lease.serialize(), + }, + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ).encode("ascii") + try: + completed = subprocess.run( + [sys.executable, "-m", "applications.connector_runner", "--scan-file"], + input=request, + capture_output=True, + check=False, + env={ + "PATH": os.defpath, + "PYTHONPATH": os.pathsep.join(sys.path), + "PYTHONUTF8": "1", + }, + timeout=30.0, + ) + except Exception: + raise RuntimeError("File connector process is unavailable") from None + if completed.returncode != 0 or completed.stderr: + raise RuntimeError("File connector process is unavailable") + try: + page = deserialize_supply_change_page(completed.stdout) + except ValueError: + raise RuntimeError("File connector process output is unavailable") from None + if page.binding != binding: + raise RuntimeError("File connector process binding is unavailable") + return page + + +class _VaultConnector(CheckpointedConnectorWithPermSync): + """Native File source logic executed through the registered Onyx runner shape.""" + + __slots__ = ("_permission_observer", "_source") + + def __init__( + self, + source: VaultSource, + permission_observer: VaultPermissionObserver, + ) -> None: + self._source = source + self._permission_observer = permission_observer + + def build_dummy_checkpoint(self) -> ConnectorCheckpoint: + return ConnectorCheckpoint(encode_file_checkpoint(FileCheckpoint(()))) + + def validate_checkpoint(self, payload: bytes) -> ConnectorCheckpoint: + decode_file_checkpoint(payload) + return ConnectorCheckpoint(payload) + + def load_from_checkpoint( + self, + checkpoint: ConnectorCheckpoint, + ) -> CheckpointOutput: + return self._generate(checkpoint, include_permissions=False) + + def load_from_checkpoint_with_perm_sync( + self, + checkpoint: ConnectorCheckpoint, + ) -> CheckpointOutput: + return self._generate(checkpoint, include_permissions=True) + + def _generate( + self, + checkpoint: ConnectorCheckpoint, + *, + include_permissions: bool, + ) -> CheckpointOutput: + prior = decode_file_checkpoint(checkpoint.payload) + snapshot = self._source.snapshot() + paths = tuple(item.path for item in snapshot) + if paths != tuple(sorted(set(paths), key=lambda value: value.encode("utf-8"))): + raise RuntimeError("vault snapshot must be sorted and unique") + current_entries = tuple( + (item.path, hashlib.sha256(item.content).hexdigest()) for item in snapshot + ) + current_by_path = dict(current_entries) + prior_by_path = dict(prior.entries) + changed_paths = tuple( + path + for path in sorted( + set(prior_by_path) | set(current_by_path), + key=lambda value: value.encode("utf-8"), + ) + if prior_by_path.get(path) != current_by_path.get(path) + ) + selected_paths = changed_paths[:_BATCH_SIZE] + snapshot_by_path = {item.path: item for item in snapshot} + next_entries = dict(prior_by_path) + for path in selected_paths: + item = snapshot_by_path.get(path) + if item is None: + if include_permissions: + self._permission_observer.observe_acl(path) + next_entries.pop(path, None) + yield DeletedDocument(_document_ref(path)) + continue + if include_permissions: + self._permission_observer.observe_acl(item.path) + next_entries[item.path] = current_by_path[item.path] + yield Document( + document_id=_document_ref(item.path), + content=item.content, + content_type="text/markdown", + metadata=( + ("content_sha256", current_by_path[item.path]), + ("path", item.path), + ), + ) + return ConnectorCheckpoint( + encode_file_checkpoint( + FileCheckpoint( + tuple( + sorted( + next_entries.items(), + key=lambda item: item[0].encode("utf-8"), + ) + ) + ) + ) + ) + + +class FileConnectorAdapter: + """Translate admitted runner outputs into CE Supply envelopes and deletes.""" + + __slots__ = ( + "_checkpoint", + "_clock", + "_connector", + "_policy_epoch", + "emitted_pages", + "last_emitted_page", + ) + + def __init__( + self, + source: VaultSource, + permission_observer: VaultPermissionObserver, + *, + policy_epoch: int, + clock: Callable[[], datetime] | None = None, + ) -> None: + if type(policy_epoch) is not int or policy_epoch < 1: + raise ValueError("File connector Policy Epoch must be positive") + if not callable(clock or _utc_now): + raise TypeError("File connector clock must be callable") + self._connector = _VaultConnector(source, permission_observer) + self._policy_epoch = policy_epoch + self._clock = clock or _utc_now + self._checkpoint: ConnectorCheckpoint | None = None + self.emitted_pages: list[SupplyChangePage] = [] + self.last_emitted_page: SupplyChangePage | None = None + + @classmethod + def from_twin(cls, twin: FileConnectorTwin) -> FileConnectorAdapter: + return cls(twin, twin, policy_epoch=twin.policy_epoch, clock=twin.now) + + def load_checkpoint(self, opaque_checkpoint: bytes | None) -> None: + self.last_emitted_page = None + self._checkpoint = ( + self._connector.build_dummy_checkpoint() + if opaque_checkpoint is None + else self._connector.validate_checkpoint(opaque_checkpoint) + ) + + def load(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + if self._checkpoint is None: + raise RuntimeError("File connector checkpoint was not loaded") + return self._run(binding) + + def poll(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + if self._checkpoint is None: + raise RuntimeError("File connector checkpoint was not loaded") + return self._run(binding) + + def _run(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + if type(binding) is not ConnectorCheckpointBinding: + raise TypeError("File connector requires an exact checkpoint binding") + assert self._checkpoint is not None + documents: list[Document] = [] + deleted: list[DeletedDocument] = [] + failures: list[ConnectorFailure] = [] + proposed: ConnectorCheckpoint | None = None + for batch in ConnectorRunner( + self._connector, + self._checkpoint, + batch_size=_BATCH_SIZE, + include_permissions=True, + ).run(): + documents.extend(batch.documents) + deleted.extend(batch.deleted_documents) + failures.extend(batch.failures) + if batch.checkpoint is not None: + if proposed is not None: + raise RuntimeError("File connector returned multiple checkpoints") + proposed = batch.checkpoint + if failures or proposed is None: + raise RuntimeError("File connector output is unavailable") + observed_at = self._clock() + acl = _weak_acl(binding, observed_at, self._policy_epoch) + page = SupplyChangePage( + binding=binding, + page_ref=_page_ref( + binding, + self._checkpoint.payload, + proposed.payload, + ), + documents=tuple( + SupplyDocumentEnvelope( + organization_id=binding.organization_id, + source_version_id=binding.source_version_id, + worker_job_id=binding.worker_job_id, + document_ref=document.document_id, + content=document.content, + content_type=document.content_type, + acl_observation=acl, + metadata=document.metadata, + ) + for document in documents + ), + deleted_document_refs=tuple( + SupplyDocumentDeleteObservation( + document_ref=document.document_id, + acl_observation=acl, + ) + for document in deleted + ), + checkpoint_proposal=proposed.payload, + terminal=not documents and not deleted, + ) + self.emitted_pages.append(page) + self.last_emitted_page = page + return page + + +def _utc_now() -> datetime: + return datetime.now(UTC) + + +def _document_ref(path: str) -> str: + return f"file:{hashlib.sha256(path.encode('utf-8')).hexdigest()}" + + +def _page_ref( + binding: ConnectorCheckpointBinding, + prior: bytes, + proposed: bytes, +) -> str: + digest = hashlib.sha256() + digest.update(binding.organization_id.bytes) + digest.update(binding.source_version_id.bytes) + digest.update(binding.worker_job_id.bytes) + digest.update(hashlib.sha256(prior).digest()) + digest.update(hashlib.sha256(proposed).digest()) + return f"file-page:{digest.hexdigest()}" + + +def _weak_acl( + binding: ConnectorCheckpointBinding, + observed_at: datetime, + policy_epoch: int, +) -> SourceAclObservation: + return SourceAclObservation( + organization_id=binding.organization_id, + observed_at=observed_at, + policy_epoch=policy_epoch, + evidence_class=SourceAclEvidenceClass.WEAK, + source_lacks_stronger_acl=_WEAK_ACL_JUSTIFICATION, + ) + + +__all__ = [ + "FileCheckpoint", + "FileConnectorAdapter", + "FileConnectorProcessAdapter", + "FileRootVaultSource", + "PermissionObservationFailed", + "VaultSnapshotEntry", + "decode_file_checkpoint", + "encode_file_checkpoint", +] diff --git a/adapters/file_source.py b/adapters/file_source.py index 7b53312b..1976d87f 100644 --- a/adapters/file_source.py +++ b/adapters/file_source.py @@ -178,6 +178,14 @@ def read(self, root_ref: FileRootRef, path: FileImportPath) -> bytes: payload, _metadata = self._read_regular(root_ref, path) return payload + def observe_markdown_files( + self, + root_ref: FileRootRef, + ) -> tuple[tuple[FileImportPath, bytes], ...]: + """Expose the one anchored acquisition truth to admitted File consumers.""" + + return self._observe_markdown_files(root_ref) + def _read_regular( self, root_ref: FileRootRef, path: FileImportPath ) -> tuple[bytes, os.stat_result]: diff --git a/applications/connector_runner.py b/applications/connector_runner.py new file mode 100644 index 00000000..3864df2f --- /dev/null +++ b/applications/connector_runner.py @@ -0,0 +1,169 @@ +"""ContextEngine-owned process boundary for one leased connector job.""" + +from __future__ import annotations + +import argparse +import base64 +import binascii +import json +import sys +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import cast +from uuid import UUID + +from adapters.connectors.file import FileConnectorAdapter, FileRootVaultSource +from adapters.file_source import FileReadLimits, FileRootRegistry +from engine.control import FileRootRef +from engine.supply import ( + SupplyBridgeExecution, + WorkerLeaseToken, + serialize_supply_change_page, +) + +_MAX_JOB_BYTES = 64 * 1024 +_MAX_FILE_BYTES = 64 * 1024 * 1024 + + +@dataclass(frozen=True, slots=True) +class ConnectorRunnerRequest: + """Closed serialized input for exactly one engine-minted connector job.""" + + organization_id: UUID + source_version_id: UUID + worker_job_id: UUID + service_principal_id: UUID + worker_lease: WorkerLeaseToken + policy_epoch: int + idempotency_key: str + service_actor_expires_at: datetime + root_ref: FileRootRef + root_path: Path + opaque_checkpoint: bytes | None + + @classmethod + def from_json(cls, payload: bytes) -> ConnectorRunnerRequest: + if type(payload) is not bytes or not 1 <= len(payload) <= _MAX_JOB_BYTES: + raise ValueError("connector runner job is unavailable") + try: + decoded = json.loads(payload) + if type(decoded) is not dict or set(decoded) != { + "idempotency_key", + "opaque_checkpoint", + "organization_id", + "policy_epoch", + "root_path", + "root_ref", + "service_actor_expires_at", + "service_principal_id", + "source_version_id", + "worker_job_id", + "worker_lease", + }: + raise ValueError + checkpoint = decoded["opaque_checkpoint"] + parsed = cls( + organization_id=UUID(cast(str, decoded["organization_id"])), + source_version_id=UUID(cast(str, decoded["source_version_id"])), + worker_job_id=UUID(cast(str, decoded["worker_job_id"])), + service_principal_id=UUID( + cast(str, decoded["service_principal_id"]) + ), + worker_lease=WorkerLeaseToken(cast(str, decoded["worker_lease"])), + policy_epoch=cast(int, decoded["policy_epoch"]), + idempotency_key=cast(str, decoded["idempotency_key"]), + service_actor_expires_at=datetime.fromisoformat( + cast(str, decoded["service_actor_expires_at"]) + ), + root_ref=FileRootRef(cast(str, decoded["root_ref"])), + root_path=Path(cast(str, decoded["root_path"])), + opaque_checkpoint=( + None + if checkpoint is None + else base64.b64decode(cast(str, checkpoint), validate=True) + ), + ) + except (KeyError, TypeError, ValueError, binascii.Error, json.JSONDecodeError): + raise ValueError("connector runner job is unavailable") from None + parsed._validate() + return parsed + + def _validate(self) -> None: + if type(self.policy_epoch) is not int or self.policy_epoch < 1: + raise ValueError("connector runner job is unavailable") + if ( + type(self.idempotency_key) is not str + or len(self.idempotency_key) != 64 + or any( + character not in "0123456789abcdef" + for character in self.idempotency_key + ) + ): + raise ValueError("connector runner job is unavailable") + if ( + type(self.service_actor_expires_at) is not datetime + or self.service_actor_expires_at.tzinfo is None + ): + raise ValueError("connector runner job is unavailable") + if not isinstance(self.root_path, Path) or not self.root_path.is_absolute(): + raise ValueError("connector runner job is unavailable") + + @property + def execution(self) -> SupplyBridgeExecution: + return SupplyBridgeExecution( + organization_id=self.organization_id, + source_version_id=self.source_version_id, + worker_job_id=self.worker_job_id, + worker_lease=self.worker_lease, + ) + + def create_adapter(self) -> tuple[FileConnectorAdapter, FileRootRegistry]: + """Compose the sole admitted connector from explicitly passed root facts.""" + + roots = FileRootRegistry( + {self.root_ref: self.root_path}, + limits=FileReadLimits(_MAX_FILE_BYTES), + ) + adapter = FileConnectorAdapter( + FileRootVaultSource(roots, self.root_ref), + FileRootVaultSource(roots, self.root_ref), + policy_epoch=self.policy_epoch, + ) + return adapter, roots + + def execute(self) -> bytes: + adapter, roots = self.create_adapter() + try: + adapter.load_checkpoint(self.opaque_checkpoint) + page = ( + adapter.load(self.execution.binding) + if self.opaque_checkpoint is None + else adapter.poll(self.execution.binding) + ) + return serialize_supply_change_page(page) + finally: + roots.close() + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--scan-file", action="store_true") + args = parser.parse_args(argv) + if not args.scan_file: + parser.error("one runner operation is required") + try: + sys.stdout.buffer.write( + ConnectorRunnerRequest.from_json(sys.stdin.buffer.read()).execute() + ) + except Exception: + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +__all__ = ["ConnectorRunnerRequest", "main"] diff --git a/applications/worker.py b/applications/worker.py index 690db355..fc71fd33 100644 --- a/applications/worker.py +++ b/applications/worker.py @@ -15,6 +15,7 @@ from sqlalchemy import Engine, text from sqlalchemy.exc import SQLAlchemyError +from adapters.connectors.file import FileConnectorProcessAdapter from adapters.embeddings import ( DeterministicEmbeddingTwin, ExternalEmbeddingConfiguration, @@ -51,8 +52,12 @@ FileImportLeaseRedemption, FileImportRefused, FileImportUnavailable, + PostgreSQLConnectorCheckpointStore, PostgreSQLFileDispatchAuthority, PostgreSQLFileImportWorker, + PostgreSQLStagedArtifactSink, + PostgreSQLSupplyExecutionBridge, + SupplyBridgeExecutionIdentity, create_database_engine, load_database_configuration, ) @@ -66,8 +71,11 @@ from engine.supply import ( ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION, CONTEXT_FRAGMENT_EMBEDDING_DIMENSION, + SUPPLY_CONNECTOR_WORKER_LEASE_OPERATION, EmbeddingProvider, MarkdownCompilerConfig, + SupplyBridgeExecution, + SupplyExecutionConfiguration, WorkerLeaseCodec, WorkerLeaseKeyring, WorkerLeaseToken, @@ -80,6 +88,7 @@ _file_dispatch_roots = _configured_file_roots _WORKER_EMBEDDING_PROVIDER_ENV = "CONTEXT_ENGINE_WORKER_EMBEDDING_PROVIDER" _WORKER_EMBEDDING_DIMENSION_ENV = "CONTEXT_ENGINE_WORKER_EMBEDDING_DIMENSION" +_SUPPLY_CONNECTOR_EXECUTION_CONFIGURATION = SupplyExecutionConfiguration() def _file_read_limits() -> FileReadLimits: @@ -375,6 +384,106 @@ def _run_one_file_import() -> int: engine.dispose() +def _run_one_file_connector_job() -> int: + """Execute one exact leased File connector job through the Supply bridge.""" + + codec = WorkerLeaseCodec( + WorkerLeaseKeyring(active_version=1, keys={1: _worker_signing_key()}) + ) + engine = create_database_engine( + load_database_configuration(DatabasePurpose.SUPPLY_WORKER) + ) + organization_id = UUID( + _required_environment("CONTEXT_ENGINE_WORKER_ORGANIZATION_ID") + ) + source_version_id = UUID( + _required_environment("CONTEXT_ENGINE_WORKER_SOURCE_VERSION_ID") + ) + worker_job_id = UUID(_required_environment("CONTEXT_ENGINE_WORKER_JOB_ID")) + service_principal_id = UUID( + _required_environment("CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID") + ) + worker_lease = WorkerLeaseToken( + _required_environment("CONTEXT_ENGINE_WORKER_LEASE_TOKEN") + ) + root_ref = FileRootRef( + _required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOT_REF") + ) + root_path = Path(_required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOT_PATH")) + try: + checked_at = _worker_database_time(engine) + claims = codec.verify( + worker_lease, + expected_organization_id=organization_id, + expected_job_id=worker_job_id, + expected_service_principal_id=service_principal_id, + expected_workload="supply.connector", + expected_operation=SUPPLY_CONNECTOR_WORKER_LEASE_OPERATION, + expected_worker_audience="context-engine-connector-runner", + expected_source_version_ref=str(source_version_id), + now=checked_at, + ) + if ( + claims.policy_epoch is None + or claims.idempotency_key is None + or claims.allowed_source_version_refs is None + or claims.allowed_operations is None + or claims.service_actor_expires_at is None + ): + raise FileImportUnavailable("Connector execution identity is unavailable") + execution = SupplyBridgeExecution( + organization_id=organization_id, + source_version_id=source_version_id, + worker_job_id=worker_job_id, + worker_lease=worker_lease, + ) + result = PostgreSQLSupplyExecutionBridge( + engine, + codec, + SupplyBridgeExecutionIdentity( + organization_id=organization_id, + service_principal_id=service_principal_id, + allowed_source_version_ids=tuple( + UUID(value) for value in claims.allowed_source_version_refs + ), + allowed_operations=claims.allowed_operations, + policy_epoch=claims.policy_epoch, + idempotency_key=claims.idempotency_key, + expires_at=claims.service_actor_expires_at, + ), + PostgreSQLConnectorCheckpointStore(engine), + PostgreSQLStagedArtifactSink(engine), + configuration=_SUPPLY_CONNECTOR_EXECUTION_CONFIGURATION, + clock=lambda: _worker_database_time(engine), + ).execute( + execution, + FileConnectorProcessAdapter( + root_ref, + root_path, + policy_epoch=claims.policy_epoch, + worker_lease=worker_lease, + service_principal_id=service_principal_id, + idempotency_key=claims.idempotency_key, + service_actor_expires_at=claims.service_actor_expires_at, + ), + ) + print( + json.dumps( + { + "acceptedPageCount": len(result.accepted_page_refs), + "jobBehavior": "connector.execute", + "service": "context-engine-worker", + "status": "complete", + }, + sort_keys=True, + ), + flush=True, + ) + return 0 + finally: + engine.dispose() + + def _worker_signing_key() -> bytes: signing_key_hex = _required_environment( "CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX" @@ -502,7 +611,10 @@ def run( run_file_job: bool = False, dispatch_file_once: bool = False, dispatch_files: bool = False, + run_file_connector_job: bool = False, ) -> int: + if run_file_connector_job: + return _run_one_file_connector_job() if dispatch_file_once: return _run_file_dispatch(single_cycle=True) if dispatch_files: @@ -535,6 +647,11 @@ def main(argv: Sequence[str] | None = None) -> int: action="store_true", help="complete the deterministic no-op lifecycle and exit", ) + parser.add_argument( + "--run-file-connector-job", + action="store_true", + help="consume one exact leased File connector job and exit", + ) parser.add_argument( "--run-file-job", action="store_true", @@ -557,6 +674,7 @@ def main(argv: Sequence[str] | None = None) -> int: args.run_file_job, args.dispatch_file_once, args.dispatch_files, + args.run_file_connector_job, ) ) if selected_modes > 1: @@ -566,6 +684,7 @@ def main(argv: Sequence[str] | None = None) -> int: run_file_job=args.run_file_job, dispatch_file_once=args.dispatch_file_once, dispatch_files=args.dispatch_files, + run_file_connector_job=args.run_file_connector_job, ) diff --git a/engine/supply/__init__.py b/engine/supply/__init__.py index 50927709..85c52fd4 100644 --- a/engine/supply/__init__.py +++ b/engine/supply/__init__.py @@ -42,6 +42,7 @@ SupplyExecutionBoundReason, SupplyExecutionConfiguration, SupplyStagedPageByteLimitExceeded, + deserialize_supply_change_page, serialize_supply_change_page, ) from engine.supply.jobs import ( @@ -161,6 +162,7 @@ "SupplyExecutionBoundReason", "SupplyExecutionConfiguration", "SupplyStagedPageByteLimitExceeded", + "deserialize_supply_change_page", "serialize_supply_change_page", "UnsupportedConstruct", "WorkNotAvailable", diff --git a/engine/supply/execution.py b/engine/supply/execution.py index e8629a45..cef22f4d 100644 --- a/engine/supply/execution.py +++ b/engine/supply/execution.py @@ -3,11 +3,12 @@ from __future__ import annotations import base64 +import binascii import json from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import StrEnum -from typing import Final, Protocol +from typing import Final, Protocol, cast from uuid import UUID from sqlalchemy import Connection @@ -480,6 +481,160 @@ def serialize_supply_change_page(page: SupplyChangePage) -> bytes: return payload +def deserialize_supply_change_page(payload: bytes) -> SupplyChangePage: + """Parse one runner response back into exact engine-owned contracts.""" + + _require_bytes( + "serialized staged page", + payload, + maximum_length=_MAX_STAGED_PAGE_BYTES, + ) + try: + decoded = json.loads(payload) + if type(decoded) is not dict: + raise ValueError + document = cast(dict[str, object], decoded) + if set(document) != { + "binding", + "checkpoint_proposal", + "deleted_document_refs", + "documents", + "page_ref", + "terminal", + }: + raise ValueError + binding_document = _exact_dict( + document["binding"], + {"organization_id", "source_version_id", "worker_job_id"}, + ) + binding = ConnectorCheckpointBinding( + organization_id=UUID(cast(str, binding_document["organization_id"])), + source_version_id=UUID( + cast(str, binding_document["source_version_id"]) + ), + worker_job_id=UUID(cast(str, binding_document["worker_job_id"])), + ) + raw_documents = document["documents"] + raw_deletes = document["deleted_document_refs"] + if type(raw_documents) is not list or type(raw_deletes) is not list: + raise ValueError + page = SupplyChangePage( + binding=binding, + page_ref=cast(str, document["page_ref"]), + documents=tuple( + _deserialize_supply_document_envelope(value) + for value in raw_documents + ), + deleted_document_refs=tuple( + _deserialize_supply_delete_observation(value) + for value in raw_deletes + ), + checkpoint_proposal=base64.b64decode( + cast(str, document["checkpoint_proposal"]), + validate=True, + ), + terminal=cast(bool, document["terminal"]), + ) + _validate_supply_change_page(page) + return page + except ( + KeyError, + binascii.Error, + TypeError, + ValueError, + UnicodeDecodeError, + json.JSONDecodeError, + ): + raise ValueError("serialized Supply change page is unavailable") from None + + +def _exact_dict(value: object, keys: set[str]) -> dict[str, object]: + if type(value) is not dict or set(value) != keys: + raise ValueError + return cast(dict[str, object], value) + + +def _deserialize_supply_document_envelope( + value: object, +) -> SupplyDocumentEnvelope: + document = _exact_dict( + value, + { + "acl_observation", + "content", + "content_type", + "document_ref", + "metadata", + "organization_id", + "source_version_id", + "worker_job_id", + }, + ) + metadata = document["metadata"] + if type(metadata) is not list: + raise ValueError + if any(type(item) is not list or len(item) != 2 for item in metadata): + raise ValueError + return SupplyDocumentEnvelope( + organization_id=UUID(cast(str, document["organization_id"])), + source_version_id=UUID(cast(str, document["source_version_id"])), + worker_job_id=UUID(cast(str, document["worker_job_id"])), + document_ref=cast(str, document["document_ref"]), + content=base64.b64decode(cast(str, document["content"]), validate=True), + content_type=cast(str, document["content_type"]), + acl_observation=_deserialize_source_acl_observation( + document["acl_observation"] + ), + metadata=tuple( + (cast(str, item[0]), cast(str, item[1])) + for item in metadata + ), + ) + + +def _deserialize_supply_delete_observation( + value: object, +) -> SupplyDocumentDeleteObservation: + document = _exact_dict(value, {"acl_observation", "document_ref"}) + return SupplyDocumentDeleteObservation( + document_ref=cast(str, document["document_ref"]), + acl_observation=_deserialize_source_acl_observation( + document["acl_observation"] + ), + ) + + +def _deserialize_source_acl_observation(value: object) -> SourceAclObservation: + document = _exact_dict( + value, + { + "evidence_class", + "evidence_payload", + "observed_at", + "organization_id", + "policy_epoch", + "source_lacks_stronger_acl", + }, + ) + evidence_payload = document["evidence_payload"] + if evidence_payload is not None: + evidence_payload = base64.b64decode( + cast(str, evidence_payload), + validate=True, + ) + return SourceAclObservation( + organization_id=UUID(cast(str, document["organization_id"])), + observed_at=datetime.fromisoformat(cast(str, document["observed_at"])), + policy_epoch=cast(int, document["policy_epoch"]), + evidence_class=SourceAclEvidenceClass(cast(str, document["evidence_class"])), + evidence_payload=evidence_payload, + source_lacks_stronger_acl=cast( + str | None, + document["source_lacks_stronger_acl"], + ), + ) + + def _serialize_source_acl_observation( observation: SourceAclObservation, ) -> dict[str, object]: @@ -618,5 +773,6 @@ def failure(self, observation: ConnectorFailure) -> None: ... "SupplyExecutionBoundReason", "SupplyExecutionConfiguration", "SupplyStagedPageByteLimitExceeded", + "deserialize_supply_change_page", "serialize_supply_change_page", ] diff --git a/pyproject.toml b/pyproject.toml index 4fab231a..88d32a0b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ context-engine-embedding-benchmark = "applications.embedding_benchmark:main" context-engine-eval = "applications.eval_v1:main" context-engine-golden-backup = "applications.golden_backup:main" context-engine-worker = "applications.worker:main" +context-engine-connector-runner = "applications.connector_runner:main" [project.optional-dependencies] benchmark = [ diff --git a/tests/integration/test_connector_checkpoint_resume.py b/tests/integration/test_connector_checkpoint_resume.py new file mode 100644 index 00000000..28ae7409 --- /dev/null +++ b/tests/integration/test_connector_checkpoint_resume.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from dataclasses import replace + +import pytest +from sqlalchemy import Connection, Engine + +from adapters.connectors.file import FileConnectorAdapter +from engine.persistence import ( + DatabaseConfiguration, + PostgreSQLConnectorCheckpointStore, + PostgreSQLStagedArtifactSink, + PostgreSQLSupplyBridgeLeaseIssuer, + SupplyBridgeLeasePreemptionRequest, +) +from engine.supply import ( + ConnectorCheckpointBinding, + StagedArtifact, + StagedArtifactSink, + SupplyChangePage, + WorkerLeaseClaims, +) +from tests.integration.test_connector_checkpoint_store import ( + _bridge, + _claims, + _Scenario, + _seed_scenario, +) +from tests.integration.test_connector_checkpoint_store import ( + scenarios as _checkpoint_scenarios, +) +from tests.support.file_connector_twin import SyntheticVaultTwin + +pytestmark = pytest.mark.integration +scenarios = _checkpoint_scenarios + + +class _FailSecondAcceptance(StagedArtifactSink): + def __init__(self, inner: StagedArtifactSink) -> None: + self._inner = inner + self._calls = 0 + + def accept_change_page( + self, + connection: Connection, + page: SupplyChangePage, + *, + lease_claims: WorkerLeaseClaims, + ) -> None: + self._calls += 1 + if self._calls == 2: + raise RuntimeError("injected failure before atomic acceptance") + self._inner.accept_change_page(connection, page, lease_claims=lease_claims) + + def load( + self, + binding: ConnectorCheckpointBinding, + artifact_ref: str, + *, + lease_claims: WorkerLeaseClaims, + ) -> StagedArtifact | None: + return self._inner.load(binding, artifact_ref, lease_claims=lease_claims) + + +def test_unaccepted_page_keeps_prior_checkpoint_and_resume_reemits_once( + scenarios: list[_Scenario], + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + scenario = _seed_scenario(migration_configuration, guarded_control_engine) + scenarios.append(scenario) + claims = _claims(scenario) + assert claims.policy_epoch is not None + twin = SyntheticVaultTwin( + {"alpha.md": b"# Alpha v1\n"}, + snapshots=( + {"alpha.md": b"# Alpha v1\n"}, + {"alpha.md": b"# Alpha v2\n"}, + {"alpha.md": b"# Alpha v2\n"}, + {"alpha.md": b"# Alpha v2\n"}, + ), + policy_epoch=claims.policy_epoch, + ) + adapter = FileConnectorAdapter.from_twin(twin) + store = PostgreSQLConnectorCheckpointStore(guarded_worker_engine) + sink = PostgreSQLStagedArtifactSink(guarded_worker_engine) + + with pytest.raises(RuntimeError, match="injected failure"): + _bridge( + scenario, + guarded_worker_engine, + store, + _FailSecondAcceptance(sink), + ).execute(scenario.execution, adapter) + + prior = adapter.emitted_pages[0].checkpoint_proposal + failed_page_ref = adapter.emitted_pages[1].page_ref + assert store.load(scenario.execution.binding, lease_claims=claims) == prior + + resumed_token = PostgreSQLSupplyBridgeLeaseIssuer( + guarded_control_engine, + scenario.codec, + ).preempt( + SupplyBridgeLeasePreemptionRequest( + organization_id=scenario.organization_id, + source_id=scenario.source_id, + source_version_id=scenario.source_version_id, + worker_job_id=scenario.job_id, + service_principal_id=scenario.service_principal_id, + reason_digest="c" * 64, + ) + ) + resumed_scenario = replace( + scenario, + execution=replace(scenario.execution, worker_lease=resumed_token), + ) + resumed = FileConnectorAdapter.from_twin(twin) + result = _bridge(resumed_scenario, guarded_worker_engine, store, sink).execute( + resumed_scenario.execution, + resumed, + ) + + assert result.accepted_page_refs[0] == failed_page_ref + assert sum(page.page_ref == failed_page_ref for page in resumed.emitted_pages) == 1 + assert len(result.accepted_page_refs) == 2 diff --git a/tests/integration/test_connector_checkpoint_store.py b/tests/integration/test_connector_checkpoint_store.py index 3fe03248..6498d124 100644 --- a/tests/integration/test_connector_checkpoint_store.py +++ b/tests/integration/test_connector_checkpoint_store.py @@ -124,6 +124,28 @@ def _emit(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: return page +class _ScriptedAdapter(ConnectorAdapter): + def __init__(self, pages: tuple[SupplyChangePage, ...]) -> None: + self._pages = iter(pages) + self.loaded_checkpoints: list[bytes | None] = [] + self.emitted_page_refs: list[str] = [] + + def load_checkpoint(self, opaque_checkpoint: bytes | None) -> None: + self.loaded_checkpoints.append(opaque_checkpoint) + + def load(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + return self._emit(binding) + + def poll(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + return self._emit(binding) + + def _emit(self, binding: ConnectorCheckpointBinding) -> SupplyChangePage: + page = next(self._pages) + assert page.binding == binding + self.emitted_page_refs.append(page.page_ref) + return page + + class _FailBeforeAtomicAcceptance(StagedArtifactSink): def __init__(self, inner: StagedArtifactSink, page_ref: str) -> None: self._inner = inner @@ -731,6 +753,76 @@ def test_repeating_empty_pages_without_cursor_progress_terminate_content_free( ) +def test_no_progress_count_resets_after_content_and_checkpoint_progress( + scenarios: list[_Scenario], + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + scenario = _seed_scenario(migration_configuration, guarded_control_engine) + scenarios.append(scenario) + binding = scenario.execution.binding + bootstrap = _page(scenario, 1, terminal=False) + first_empty = SupplyChangePage( + binding=binding, + page_ref="empty-before-progress", + documents=(), + deleted_document_refs=(), + checkpoint_proposal=bootstrap.checkpoint_proposal, + terminal=False, + ) + progress = _page(scenario, 2, terminal=False) + second_empty = SupplyChangePage( + binding=binding, + page_ref="empty-after-progress", + documents=(), + deleted_document_refs=(), + checkpoint_proposal=progress.checkpoint_proposal, + terminal=False, + ) + terminal = SupplyChangePage( + binding=binding, + page_ref="terminal-after-reset", + documents=(), + deleted_document_refs=(), + checkpoint_proposal=b"terminal-checkpoint", + terminal=True, + ) + adapter = _ScriptedAdapter( + (bootstrap, first_empty, progress, second_empty, terminal) + ) + store = PostgreSQLConnectorCheckpointStore(guarded_worker_engine) + + result = _bridge( + scenario, + guarded_worker_engine, + store, + configuration=SupplyExecutionConfiguration( + page_limit=5, + no_progress_page_limit=1, + ), + ).execute(scenario.execution, adapter) + + assert result.accepted_page_refs == ( + "page:1", + "empty-before-progress", + "page:2", + "empty-after-progress", + "terminal-after-reset", + ) + assert adapter.loaded_checkpoints == [ + None, + bootstrap.checkpoint_proposal, + bootstrap.checkpoint_proposal, + progress.checkpoint_proposal, + progress.checkpoint_proposal, + ] + assert ( + store.load(binding, lease_claims=_claims(scenario)) + == b"terminal-checkpoint" + ) + + def test_empty_terminal_page_is_success_not_a_bound_failure( scenarios: list[_Scenario], migration_configuration: DatabaseConfiguration, diff --git a/tests/integration/test_connector_runner_lease.py b/tests/integration/test_connector_runner_lease.py new file mode 100644 index 00000000..77c24aa0 --- /dev/null +++ b/tests/integration/test_connector_runner_lease.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +import json +import os +import subprocess +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from pathlib import Path +from uuid import uuid4 + +import pytest +from sqlalchemy import Engine + +from adapters.connectors.file import FileConnectorAdapter, FileConnectorProcessAdapter +from engine.control import FileRootRef +from engine.persistence import ( + DatabaseConfiguration, + PostgreSQLConnectorCheckpointStore, + PostgreSQLStagedArtifactSink, + PostgreSQLSupplyExecutionBridge, + SupplyBridgeExecutionIdentity, +) +from engine.supply import SupplyBridgeExecution, WorkNotAvailable +from tests.integration.test_connector_checkpoint_store import ( + _claims, + _Scenario, + _seed_scenario, +) +from tests.integration.test_connector_checkpoint_store import ( + scenarios as _checkpoint_scenarios, +) +from tests.support.file_connector_twin import SyntheticVaultTwin + +pytestmark = pytest.mark.integration +scenarios = _checkpoint_scenarios + + +def _bridge( + scenario: _Scenario, + engine: Engine, + *, + checked_at: datetime, +) -> PostgreSQLSupplyExecutionBridge: + claims = _claims(scenario) + assert claims.policy_epoch is not None + assert claims.idempotency_key is not None + assert claims.service_actor_expires_at is not None + return PostgreSQLSupplyExecutionBridge( + engine, + scenario.codec, + SupplyBridgeExecutionIdentity( + organization_id=scenario.organization_id, + service_principal_id=scenario.service_principal_id, + allowed_source_version_ids=(scenario.source_version_id,), + allowed_operations=("connector.execute",), + policy_epoch=claims.policy_epoch, + idempotency_key=claims.idempotency_key, + expires_at=claims.service_actor_expires_at, + ), + PostgreSQLConnectorCheckpointStore(engine), + PostgreSQLStagedArtifactSink(engine), + clock=lambda: checked_at, + ) + + +def test_file_runner_refuses_absent_expired_or_wrong_job_lease_before_scan( + scenarios: list[_Scenario], + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + scenario = _seed_scenario(migration_configuration, guarded_control_engine) + scenarios.append(scenario) + claims = _claims(scenario) + twin = SyntheticVaultTwin({"alpha.md": b"# Alpha\n"}) + adapter = FileConnectorAdapter.from_twin(twin) + now = datetime.now(UTC).replace(microsecond=0) + + with pytest.raises(TypeError): + _bridge(scenario, guarded_worker_engine, checked_at=now).execute( + None, # type: ignore[arg-type] + adapter, + ) + with pytest.raises(WorkNotAvailable, match="^work not available$"): + _bridge(scenario, guarded_worker_engine, checked_at=now).execute( + replace(scenario.execution, worker_job_id=uuid4()), + adapter, + ) + with pytest.raises(WorkNotAvailable, match="^work not available$"): + _bridge( + scenario, + guarded_worker_engine, + checked_at=claims.expires_at + timedelta(seconds=1), + ).execute(scenario.execution, adapter) + + assert twin.snapshot_calls == 0 + assert type(scenario.execution) is SupplyBridgeExecution + + +def test_valid_exact_lease_executes_file_scan_in_independent_process( + tmp_path: Path, + scenarios: list[_Scenario], + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + scenario = _seed_scenario(migration_configuration, guarded_control_engine) + scenarios.append(scenario) + claims = _claims(scenario) + assert claims.policy_epoch is not None + vault = tmp_path / "connector-vault" + vault.mkdir() + (vault / "alpha.md").write_bytes(b"# Alpha\n") + adapter = FileConnectorProcessAdapter( + FileRootRef("synthetic-root"), + vault, + policy_epoch=claims.policy_epoch, + worker_lease=scenario.execution.worker_lease, + service_principal_id=scenario.service_principal_id, + idempotency_key=claims.idempotency_key or "", + service_actor_expires_at=(claims.service_actor_expires_at or claims.expires_at), + ) + + result = _bridge( + scenario, + guarded_worker_engine, + checked_at=datetime.now(UTC).replace(microsecond=0), + ).execute(scenario.execution, adapter) + + assert len(result.accepted_page_refs) == 2 + + +def test_worker_process_executes_exact_leased_file_connector_job( + tmp_path: Path, + scenarios: list[_Scenario], + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, +) -> None: + scenario = _seed_scenario(migration_configuration, guarded_control_engine) + scenarios.append(scenario) + vault = tmp_path / "worker-connector-vault" + vault.mkdir() + (vault / "alpha.md").write_bytes(b"# Alpha\n") + environment = os.environ.copy() + environment.update( + { + "CONTEXT_ENGINE_WORKER_FILE_ROOT_PATH": str(vault), + "CONTEXT_ENGINE_WORKER_FILE_ROOT_REF": "synthetic-root", + "CONTEXT_ENGINE_WORKER_JOB_ID": str(scenario.job_id), + "CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX": bytes(range(32)).hex(), + "CONTEXT_ENGINE_WORKER_LEASE_TOKEN": ( + scenario.execution.worker_lease.serialize() + ), + "CONTEXT_ENGINE_WORKER_ORGANIZATION_ID": str( + scenario.organization_id + ), + "CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID": str( + scenario.service_principal_id + ), + "CONTEXT_ENGINE_WORKER_SOURCE_VERSION_ID": str( + scenario.source_version_id + ), + "CONTEXT_ENGINE_WORKER_SUPPLY_CUMULATIVE_BYTE_LIMIT": "1", + "CONTEXT_ENGINE_WORKER_SUPPLY_NO_PROGRESS_PAGE_LIMIT": "1", + "CONTEXT_ENGINE_WORKER_SUPPLY_PAGE_LIMIT": "1", + } + ) + + completed = subprocess.run( + ["context-engine-worker", "--run-file-connector-job"], + check=False, + capture_output=True, + env=environment, + text=True, + timeout=60, + ) + + assert completed.returncode == 0, completed.stderr + assert completed.stderr == "" + assert json.loads(completed.stdout) == { + "acceptedPageCount": 2, + "jobBehavior": "connector.execute", + "service": "context-engine-worker", + "status": "complete", + } diff --git a/tests/integration/test_file_connector_delete.py b/tests/integration/test_file_connector_delete.py new file mode 100644 index 00000000..b7a9312c --- /dev/null +++ b/tests/integration/test_file_connector_delete.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import json + +import pytest +from sqlalchemy import Engine + +from adapters.connectors.file import FileConnectorAdapter +from engine.persistence import ( + DatabaseConfiguration, + PostgreSQLConnectorCheckpointStore, + PostgreSQLStagedArtifactSink, +) +from tests.integration.test_connector_checkpoint_store import ( + _bridge, + _claims, + _Scenario, + _seed_scenario, +) +from tests.integration.test_connector_checkpoint_store import ( + scenarios as _checkpoint_scenarios, +) +from tests.support.file_connector_twin import SyntheticVaultTwin + +pytestmark = pytest.mark.integration +scenarios = _checkpoint_scenarios + + +def test_deleted_file_emits_durable_delete_observation( + scenarios: list[_Scenario], + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + scenario = _seed_scenario(migration_configuration, guarded_control_engine) + scenarios.append(scenario) + claims = _claims(scenario) + assert claims.policy_epoch is not None + twin = SyntheticVaultTwin( + {"alpha.md": b"# Alpha\n"}, + snapshots=( + {"alpha.md": b"# Alpha\n"}, + {}, + {}, + ), + policy_epoch=claims.policy_epoch, + ) + adapter = FileConnectorAdapter.from_twin(twin) + sink = PostgreSQLStagedArtifactSink(guarded_worker_engine) + + result = _bridge( + scenario, + guarded_worker_engine, + PostgreSQLConnectorCheckpointStore(guarded_worker_engine), + ).execute(scenario.execution, adapter) + + assert len(result.accepted_page_refs) == 3 + deleted = sink.load( + scenario.execution.binding, + result.accepted_page_refs[1], + lease_claims=claims, + ) + assert deleted is not None + payload = json.loads(deleted.payload) + assert payload["documents"] == [] + assert len(payload["deleted_document_refs"]) == 1 + assert payload["deleted_document_refs"][0]["acl_observation"][ + "evidence_class" + ] == "weak" + assert payload["terminal"] is False diff --git a/tests/integration/test_file_connector_incremental.py b/tests/integration/test_file_connector_incremental.py new file mode 100644 index 00000000..a07ea996 --- /dev/null +++ b/tests/integration/test_file_connector_incremental.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import json + +import pytest +from sqlalchemy import Engine + +from adapters.connectors.file import FileConnectorAdapter, decode_file_checkpoint +from engine.persistence import ( + DatabaseConfiguration, + PostgreSQLConnectorCheckpointStore, + PostgreSQLStagedArtifactSink, +) +from engine.supply import ConnectorCheckpointBinding +from tests.integration.test_connector_checkpoint_store import ( + _bridge, + _claims, + _Scenario, + _seed_scenario, +) +from tests.integration.test_connector_checkpoint_store import ( + scenarios as _checkpoint_scenarios, +) +from tests.support.file_connector_twin import SyntheticVaultTwin + +pytestmark = pytest.mark.integration +scenarios = _checkpoint_scenarios + + +def test_first_scan_ingests_and_unchanged_second_scan_is_empty( + scenarios: list[_Scenario], + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, +) -> None: + scenario = _seed_scenario(migration_configuration, guarded_control_engine) + scenarios.append(scenario) + claims = _claims(scenario) + assert claims.policy_epoch is not None + twin = SyntheticVaultTwin( + {"alpha.md": b"# Alpha\n"}, + policy_epoch=claims.policy_epoch, + ) + adapter = FileConnectorAdapter.from_twin(twin) + store = PostgreSQLConnectorCheckpointStore(guarded_worker_engine) + sink = PostgreSQLStagedArtifactSink(guarded_worker_engine) + + result = _bridge(scenario, guarded_worker_engine, store).execute( + scenario.execution, + adapter, + ) + + assert len(result.accepted_page_refs) == 2 + first = sink.load( + scenario.execution.binding, + result.accepted_page_refs[0], + lease_claims=claims, + ) + second = sink.load( + scenario.execution.binding, + result.accepted_page_refs[1], + lease_claims=claims, + ) + assert first is not None and second is not None + assert len(json.loads(first.payload)["documents"]) == 1 + assert json.loads(first.payload)["terminal"] is False + assert json.loads(second.payload)["documents"] == [] + assert json.loads(second.payload)["deleted_document_refs"] == [] + assert json.loads(second.payload)["terminal"] is True + checkpoint = store.load(scenario.execution.binding, lease_claims=claims) + assert checkpoint is not None + assert decode_file_checkpoint(checkpoint).paths == ("alpha.md",) + + +def test_changed_file_emits_exactly_that_change() -> None: + twin = SyntheticVaultTwin( + { + "alpha.md": b"# Alpha v1\n", + "bravo.md": b"# Bravo\n", + } + ) + adapter = FileConnectorAdapter.from_twin(twin) + binding = scenario_binding(twin) + adapter.load_checkpoint(None) + initial = adapter.load(binding) + twin.replace("alpha.md", b"# Alpha v2\n") + adapter.load_checkpoint(initial.checkpoint_proposal) + + changed = adapter.poll(binding) + + assert len(changed.documents) == 1 + assert changed.documents[0].content == b"# Alpha v2\n" + assert dict(changed.documents[0].metadata)["path"] == "alpha.md" + + +def scenario_binding(twin: SyntheticVaultTwin) -> ConnectorCheckpointBinding: + return ConnectorCheckpointBinding( + twin.organization_id, + twin.source_version_id, + twin.worker_job_id, + ) diff --git a/tests/support/file_connector_twin.py b/tests/support/file_connector_twin.py new file mode 100644 index 00000000..84b9b0ad --- /dev/null +++ b/tests/support/file_connector_twin.py @@ -0,0 +1,65 @@ +"""Deterministic offline twin for the admitted File/Obsidian connector.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +from adapters.connectors.file import PermissionObservationFailed, VaultSnapshotEntry + + +class SyntheticVaultTwin: + """In-memory vault surface with explicit permission-observation failures.""" + + def __init__( + self, + files: dict[str, bytes], + *, + fail_acl_for: set[str] | None = None, + organization_id: UUID | None = None, + source_version_id: UUID | None = None, + worker_job_id: UUID | None = None, + policy_epoch: int = 1, + snapshots: tuple[dict[str, bytes], ...] | None = None, + ) -> None: + self.organization_id = organization_id or uuid4() + self.source_version_id = source_version_id or uuid4() + self.worker_job_id = worker_job_id or uuid4() + self.policy_epoch = policy_epoch + self._files = dict(files) + self._snapshots = tuple(dict(snapshot) for snapshot in snapshots or ()) + self._snapshot_index = 0 + self._fail_acl_for = set(fail_acl_for or ()) + self.filesystem_accesses = 0 + self.credential_accesses = 0 + self.observed_acl_paths: list[str] = [] + self.snapshot_calls = 0 + self._now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC) + + def snapshot(self) -> tuple[VaultSnapshotEntry, ...]: + self.snapshot_calls += 1 + if self._snapshots: + index = min(self._snapshot_index, len(self._snapshots) - 1) + self._files = dict(self._snapshots[index]) + self._snapshot_index += 1 + return tuple( + VaultSnapshotEntry(path, self._files[path]) + for path in sorted(self._files, key=lambda value: value.encode("utf-8")) + ) + + def observe_acl(self, path: str) -> None: + self.observed_acl_paths.append(path) + if path in self._fail_acl_for: + raise PermissionObservationFailed("File ACL observation is unavailable") + + def now(self) -> datetime: + return self._now + + def replace(self, path: str, content: bytes) -> None: + self._files[path] = content + + def delete(self, path: str) -> None: + del self._files[path] + + +__all__ = ["SyntheticVaultTwin"] diff --git a/tests/unit/test_connector_runner_isolation.py b/tests/unit/test_connector_runner_isolation.py new file mode 100644 index 00000000..289aac0f --- /dev/null +++ b/tests/unit/test_connector_runner_isolation.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import ast +import os +import sys +from dataclasses import fields +from datetime import UTC, datetime +from pathlib import Path +from uuid import UUID, uuid4 + +import pytest + +from adapters.connectors.file import FileConnectorProcessAdapter +from applications.connector_runner import ConnectorRunnerRequest +from engine.control import FileRootRef +from engine.supply import ConnectorCheckpointBinding, WorkerLeaseToken + +REPOSITORY_ROOT = Path(__file__).parents[2] +RUNNER_PATHS = ( + REPOSITORY_ROOT / "applications/connector_runner.py", + REPOSITORY_ROOT / "adapters/connectors/file.py", +) + + +def test_runner_request_is_one_explicit_job_without_ambient_connector_credentials() -> ( + None +): + assert [item.name for item in fields(ConnectorRunnerRequest)] == [ + "organization_id", + "source_version_id", + "worker_job_id", + "service_principal_id", + "worker_lease", + "policy_epoch", + "idempotency_key", + "service_actor_expires_at", + "root_ref", + "root_path", + "opaque_checkpoint", + ] + + rendered = "\n".join(path.read_text(encoding="utf-8") for path in RUNNER_PATHS) + assert "CONTEXT_ENGINE_CONNECTOR_CREDENTIAL" not in rendered + assert "CONTEXT_ENGINE_WORKER_FILE_ROOT_PATH" not in rendered + assert "credential_cache" not in rendered + + +def test_runner_has_no_independent_database_index_or_write_surface() -> None: + imports: set[str] = set() + forbidden_calls: set[str] = set() + for path in RUNNER_PATHS: + tree = ast.parse(path.read_bytes(), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imports.update(alias.name.partition(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.level == 0: + assert node.module is not None + imports.add(node.module.partition(".")[0]) + elif ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr + in { + "mkdir", + "open", + "write_bytes", + "write_text", + "unlink", + } + ): + forbidden_calls.add(node.func.attr) + + assert imports.isdisjoint({"alembic", "celery", "psycopg", "redis", "sqlalchemy"}) + assert not forbidden_calls + assert not (REPOSITORY_ROOT / "contract_kit").exists() + + +def test_malformed_or_missing_runner_job_refuses_before_root_access( + monkeypatch: pytest.MonkeyPatch, +) -> None: + accessed = False + + def forbidden_path(*_args: object, **_kwargs: object) -> None: + nonlocal accessed + accessed = True + raise AssertionError("invalid runner request reached its root") + + monkeypatch.setattr(Path, "is_dir", forbidden_path) + with pytest.raises(ValueError): + ConnectorRunnerRequest.from_json(b"{}") + assert not accessed + + +def test_process_adapter_rejects_malformed_child_output( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class MalformedResult: + returncode = 0 + stdout = b"not-a-page" + stderr = b"" + + monkeypatch.setattr( + "adapters.connectors.file.subprocess.run", + lambda *_args, **_kwargs: MalformedResult(), + ) + adapter = FileConnectorProcessAdapter( + FileRootRef("synthetic-root"), + Path("/synthetic/root"), + policy_epoch=1, + worker_lease=WorkerLeaseToken("synthetic.opaque.lease"), + service_principal_id=UUID("00000000-0000-4000-8000-000000000001"), + idempotency_key="0" * 64, + service_actor_expires_at=datetime(2026, 7, 30, 9, tzinfo=UTC), + ) + adapter.load_checkpoint(None) + + with pytest.raises(RuntimeError, match="output is unavailable"): + adapter.load(ConnectorCheckpointBinding(uuid4(), uuid4(), uuid4())) + + +def test_process_adapter_scrubs_parent_environment_from_runner( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class EmptyPageResult: + returncode = 2 + stdout = b"" + stderr = b"" + + captured: dict[str, object] = {} + + def recording_run(*args: object, **kwargs: object) -> EmptyPageResult: + captured.update(kwargs) + return EmptyPageResult() + + monkeypatch.setenv( + "CONTEXT_ENGINE_WORKER_DATABASE_URL", + "postgresql+psycopg://secret.invalid/context-engine", + ) + monkeypatch.setattr("adapters.connectors.file.subprocess.run", recording_run) + adapter = FileConnectorProcessAdapter( + FileRootRef("synthetic-root"), + Path("/synthetic/root"), + policy_epoch=1, + worker_lease=WorkerLeaseToken("synthetic.opaque.lease"), + service_principal_id=UUID("00000000-0000-4000-8000-000000000001"), + idempotency_key="0" * 64, + service_actor_expires_at=datetime(2026, 7, 30, 9, tzinfo=UTC), + ) + adapter.load_checkpoint(None) + + with pytest.raises(RuntimeError, match="process is unavailable"): + adapter.load(ConnectorCheckpointBinding(uuid4(), uuid4(), uuid4())) + + assert captured["env"] == { + "PATH": os.defpath, + "PYTHONPATH": os.pathsep.join(sys.path), + "PYTHONUTF8": "1", + } + environment = captured["env"] + assert isinstance(environment, dict) + assert "CONTEXT_ENGINE_WORKER_DATABASE_URL" not in environment diff --git a/tests/unit/test_connector_runner_process.py b/tests/unit/test_connector_runner_process.py new file mode 100644 index 00000000..6af1ef4d --- /dev/null +++ b/tests/unit/test_connector_runner_process.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from uuid import UUID, uuid4 + +from adapters.connectors.file import FileConnectorProcessAdapter +from engine.control import FileRootRef +from engine.supply import ConnectorCheckpointBinding, WorkerLeaseToken + + +def test_process_adapter_scans_in_independent_runner(tmp_path: Path) -> None: + root = tmp_path / "vault" + root.mkdir() + (root / "alpha.md").write_bytes(b"# Alpha\n") + adapter = FileConnectorProcessAdapter( + FileRootRef("synthetic-root"), + root, + policy_epoch=7, + worker_lease=WorkerLeaseToken("synthetic.opaque.lease"), + service_principal_id=UUID("00000000-0000-4000-8000-000000000001"), + idempotency_key="0" * 64, + service_actor_expires_at=datetime(2026, 7, 30, 9, tzinfo=UTC), + ) + binding = ConnectorCheckpointBinding(uuid4(), uuid4(), uuid4()) + adapter.load_checkpoint(None) + + first = adapter.load(binding) + adapter.load_checkpoint(first.checkpoint_proposal) + second = adapter.poll(binding) + + assert len(first.documents) == 1 + assert first.terminal is False + assert second.documents == () + assert second.deleted_document_refs == () + assert second.terminal is True diff --git a/tests/unit/test_connector_twin_offline.py b/tests/unit/test_connector_twin_offline.py new file mode 100644 index 00000000..6a09af2f --- /dev/null +++ b/tests/unit/test_connector_twin_offline.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import socket + +import pytest + +from adapters.connectors.file import FileConnectorAdapter, decode_file_checkpoint +from engine.supply import ConnectorCheckpointBinding +from tests.support.file_connector_twin import SyntheticVaultTwin + + +def test_synthetic_vault_twin_runs_offline_without_files_or_credentials( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def forbidden_network(*_args: object, **_kwargs: object) -> None: + raise AssertionError("offline connector twin attempted network access") + + monkeypatch.setattr(socket, "create_connection", forbidden_network) + twin = SyntheticVaultTwin( + { + "alpha.md": b"# Alpha\n", + "nested/bravo.md": b"# Bravo\n", + } + ) + adapter = FileConnectorAdapter.from_twin(twin) + binding = ConnectorCheckpointBinding( + organization_id=twin.organization_id, + source_version_id=twin.source_version_id, + worker_job_id=twin.worker_job_id, + ) + adapter.load_checkpoint(None) + + page = adapter.load(binding) + + assert len(page.documents) == 2 + assert decode_file_checkpoint(page.checkpoint_proposal).paths == ( + "alpha.md", + "nested/bravo.md", + ) + assert twin.filesystem_accesses == 0 + assert twin.credential_accesses == 0 + + +def test_connector_batches_at_most_one_hundred_changes_per_page() -> None: + twin = SyntheticVaultTwin( + {f"note-{index:03}.md": f"# Note {index}\n".encode() for index in range(101)} + ) + adapter = FileConnectorAdapter.from_twin(twin) + binding = ConnectorCheckpointBinding( + organization_id=twin.organization_id, + source_version_id=twin.source_version_id, + worker_job_id=twin.worker_job_id, + ) + adapter.load_checkpoint(None) + + first = adapter.load(binding) + adapter.load_checkpoint(first.checkpoint_proposal) + second = adapter.poll(binding) + adapter.load_checkpoint(second.checkpoint_proposal) + terminal = adapter.poll(binding) + + assert len(first.documents) == 100 + assert len(second.documents) == 1 + assert terminal.documents == () + assert terminal.terminal is True diff --git a/tests/unit/test_file_connector_acl_observation.py b/tests/unit/test_file_connector_acl_observation.py new file mode 100644 index 00000000..1f6c71d5 --- /dev/null +++ b/tests/unit/test_file_connector_acl_observation.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import pytest + +from adapters.connectors.file import FileConnectorAdapter, PermissionObservationFailed +from engine.supply import ConnectorCheckpointBinding, SourceAclEvidenceClass +from tests.support.file_connector_twin import SyntheticVaultTwin + + +def test_local_vault_records_explicit_honest_weak_evidence() -> None: + twin = SyntheticVaultTwin({"alpha.md": b"# Alpha\n"}) + adapter = FileConnectorAdapter.from_twin(twin) + binding = ConnectorCheckpointBinding( + organization_id=twin.organization_id, + source_version_id=twin.source_version_id, + worker_job_id=twin.worker_job_id, + ) + adapter.load_checkpoint(None) + + page = adapter.load(binding) + + assert len(page.documents) == 1 + observation = page.documents[0].acl_observation + assert observation.evidence_class is SourceAclEvidenceClass.WEAK + assert observation.evidence_payload is None + assert observation.source_lacks_stronger_acl == ( + "local File/Obsidian has no corpus ACL API" + ) + + +def test_failed_permission_observation_emits_no_article_or_checkpoint() -> None: + twin = SyntheticVaultTwin( + {"alpha.md": b"# Alpha\n"}, + fail_acl_for={"alpha.md"}, + ) + adapter = FileConnectorAdapter.from_twin(twin) + binding = ConnectorCheckpointBinding( + organization_id=twin.organization_id, + source_version_id=twin.source_version_id, + worker_job_id=twin.worker_job_id, + ) + adapter.load_checkpoint(None) + + with pytest.raises(PermissionObservationFailed, match="unavailable"): + adapter.load(binding) + + assert adapter.last_emitted_page is None diff --git a/tests/unit/test_supply_bridge_no_authorization.py b/tests/unit/test_supply_bridge_no_authorization.py index bb6de584..cf93b97c 100644 --- a/tests/unit/test_supply_bridge_no_authorization.py +++ b/tests/unit/test_supply_bridge_no_authorization.py @@ -119,9 +119,10 @@ def test_acl_observation_is_not_consumed_as_authorization_outside_kernel() -> No forbidden_consumers: list[str] = [] definition_path = ROOT / "engine" / "supply" / "execution.py" public_reexport_path = ROOT / "engine" / "supply" / "__init__.py" + evidence_producer_paths = {ROOT / "adapters" / "connectors" / "file.py"} for path in _production_python_files(): - if path in {definition_path, public_reexport_path}: + if path in {definition_path, public_reexport_path} | evidence_producer_paths: continue tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) if _consumes_acl_observation(tree): @@ -130,6 +131,22 @@ def test_acl_observation_is_not_consumed_as_authorization_outside_kernel() -> No assert forbidden_consumers == [] +def test_registered_supply_adapter_only_constructs_acl_evidence() -> None: + path = ROOT / "adapters" / "connectors" / "file.py" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + calls = { + node.func.id + for node in ast.walk(tree) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) + } + names = {node.id for node in ast.walk(tree) if isinstance(node, ast.Name)} + + assert "SourceAclObservation" in calls + assert "AuthorizationKernel" not in names + assert "AuthorizedProjection" not in names + assert not {"authorize", "grant", "resolve"}.intersection(calls) + + def test_acl_observation_module_cannot_construct_runtime_authority() -> None: execution_module = ast.parse( (ROOT / "engine" / "supply" / "execution.py").read_text(encoding="utf-8") diff --git a/tests/unit/test_third_party_onyx_registration.py b/tests/unit/test_third_party_onyx_registration.py new file mode 100644 index 00000000..a944e5eb --- /dev/null +++ b/tests/unit/test_third_party_onyx_registration.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import ast +import hashlib +import json +import re +import tomllib +from collections import Counter +from pathlib import Path + +import pytest + +from adapters.connectors.file import FileConnectorAdapter +from engine.supply import ConnectorCheckpointBinding +from tests.support.file_connector_twin import SyntheticVaultTwin +from third_party.onyx.connectors.connector_runner import ConnectorRunner + +REPOSITORY_ROOT = Path(__file__).parents[2] +REGISTRATION_ROOT = REPOSITORY_ROOT / "third_party/onyx" +REGISTRATION_PATH = REGISTRATION_ROOT / "UPSTREAM.toml" +PINNED_COMMIT = "2fb3dd10493b3883870fa8adced5b1a0e114feff" +REQUIRED_SOURCE_PATHS = { + "backend/onyx/connectors/interfaces.py", + "backend/onyx/connectors/connector_runner.py", + "backend/onyx/connectors/models.py", + "backend/onyx/connectors/registry.py", +} +REQUIRED_EXCLUSIONS = {"backend/ee", "web/src/app/ee", "web/src/ee"} +UPSTREAM_SHA256 = { + "backend/onyx/connectors/interfaces.py": ( + "293c0dcca9230b75ea3eef1475262e0b4010ca4df9321880f41a9dad05561756" + ), + "backend/onyx/connectors/connector_runner.py": ( + "dc41c82425287c039b0897c135bc45f520eeb88be9b2ef16df159f835a63f311" + ), + "backend/onyx/connectors/models.py": ( + "8edcf633de61d2c769c0959ce744e9efae436083917b7ddb1d692f89eaa4f44b" + ), + "backend/onyx/connectors/registry.py": ( + "439c49bcb7dcc522545176d015dde73ce93d991e4b3e875db4b4d23d94cad9c4" + ), +} +ALLOWED_IMPORT_ROOTS = { + "__future__", + "abc", + "collections", + "dataclasses", + "enum", + "types", + "typing", + "third_party", +} + + +def _registration() -> dict[str, object]: + return tomllib.loads(REGISTRATION_PATH.read_text(encoding="utf-8")) + + +def test_vendored_bytes_match_complete_pinned_registration() -> None: + registration = _registration() + + assert registration["repository"] == "https://github.com/onyx-dot-app/onyx.git" + commit = registration["commit"] + assert isinstance(commit, str) + assert re.fullmatch(r"[0-9a-f]{40}", commit) + assert commit == PINNED_COMMIT + assert registration["reuse_mode"] == "copy-patch" + assert registration["license"] == "MIT" + assert registration["approval"] == ( + "https://github.com/stone16/context-engine/issues/126" + ) + source_paths = registration["source_paths"] + excluded_paths = registration["excluded_paths"] + assert isinstance(source_paths, list) + assert isinstance(excluded_paths, list) + assert set(source_paths) == REQUIRED_SOURCE_PATHS + assert set(excluded_paths) >= REQUIRED_EXCLUSIONS + + files = registration["files"] + assert isinstance(files, list) and files + registered_paths: set[Path] = set() + for entry in files: + assert isinstance(entry, dict) + assert set(entry) == { + "upstream_path", + "vendored_path", + "sha256", + } + upstream_path = entry["upstream_path"] + vendored_path = entry["vendored_path"] + assert isinstance(upstream_path, str) + assert isinstance(vendored_path, str) + assert "ee" not in Path(upstream_path).parts + assert "ee" not in Path(vendored_path).parts + expected_hash = entry["sha256"] + assert isinstance(expected_hash, str) + assert re.fullmatch(r"[0-9a-f]{64}", expected_hash) + path = REPOSITORY_ROOT / vendored_path + path.relative_to(REGISTRATION_ROOT) + assert path.is_file() + assert hashlib.sha256(path.read_bytes()).hexdigest() == expected_hash + registered_paths.add(path) + + vendored_files = { + path + for path in REGISTRATION_ROOT.rglob("*.py") + if "__pycache__" not in path.relative_to(REGISTRATION_ROOT).parts + } + assert registered_paths == vendored_files + assert (REGISTRATION_ROOT / "LICENSE.upstream").is_file() + assert hashlib.sha256( + (REGISTRATION_ROOT / "LICENSE.upstream").read_bytes() + ).hexdigest() == "d4847240794058c7ac3cfdf8e5d528fe8b0edf15b32a96612ecb9b3e182092b7" + assert (REGISTRATION_ROOT / "MODIFICATIONS.md").is_file() + modifications = (REGISTRATION_ROOT / "MODIFICATIONS.md").read_text( + encoding="utf-8" + ) + for upstream_path, upstream_sha256 in UPSTREAM_SHA256.items(): + assert f"`{upstream_path}`" in modifications + assert f"`{upstream_sha256}`" in modifications + assert (REGISTRATION_ROOT / "patches").is_dir() + notices = (REPOSITORY_ROOT / "THIRD_PARTY_NOTICES.md").read_text( + encoding="utf-8" + ) + assert "## onyx" in notices + assert f"- Commit: `{PINNED_COMMIT}`" in notices + assert "- License: MIT (`third_party/onyx/LICENSE.upstream`)" in notices + sbom = json.loads( + (REGISTRATION_ROOT / "sbom.cyclonedx.json").read_text(encoding="utf-8") + ) + assert sbom["bomFormat"] == "CycloneDX" + assert sbom["metadata"]["component"]["bom-ref"] == ( + "context-engine:third-party:onyx" + ) + assert {component["name"] for component in sbom["components"]} == { + "Onyx connector framework" + } + + +def test_vendored_subtree_imports_only_approved_dependencies() -> None: + registration = _registration() + files = registration["files"] + assert isinstance(files, list) + + imports: set[str] = set() + for entry in files: + assert isinstance(entry, dict) + path = REPOSITORY_ROOT / str(entry["vendored_path"]) + tree = ast.parse(path.read_bytes(), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imports.update(alias.name.partition(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.level == 0: + assert node.module is not None + imports.add(node.module.partition(".")[0]) + + assert imports <= ALLOWED_IMPORT_ROOTS + assert imports.isdisjoint( + {"alembic", "celery", "onyx", "psycopg", "redis", "sqlalchemy"} + ) + + +def test_registered_runner_region_is_executed_by_file_adapter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + twin = SyntheticVaultTwin({"notes/alpha.md": b"# Alpha\n"}) + binding = ConnectorCheckpointBinding( + organization_id=twin.organization_id, + source_version_id=twin.source_version_id, + worker_job_id=twin.worker_job_id, + ) + calls: Counter[str] = Counter() + original = ConnectorRunner.run + + def recording_run(self: ConnectorRunner) -> object: + calls["run"] += 1 + return original(self) + + monkeypatch.setattr(ConnectorRunner, "run", recording_run) + adapter = FileConnectorAdapter.from_twin(twin) + adapter.load_checkpoint(None) + + page = adapter.load(binding) + + assert page.documents + assert calls == {"run": 1} diff --git a/third_party/onyx/LICENSE.upstream b/third_party/onyx/LICENSE.upstream new file mode 100644 index 00000000..adedc9c9 --- /dev/null +++ b/third_party/onyx/LICENSE.upstream @@ -0,0 +1,28 @@ +Copyright (c) 2023-present DanswerAI, Inc. + +Portions of this software are licensed as follows: + +- All content that resides under "ee" directories of this repository is licensed under the Onyx Enterprise License. Each ee directory contains an identical copy of this license at its root: + - backend/ee/LICENSE + - web/src/app/ee/LICENSE + - web/src/ee/LICENSE +- All third party components incorporated into the Onyx Software are licensed under the original license provided by the owner of the applicable component. +- Content outside of the above mentioned directories or restrictions above is available under the "MIT Expat" license as defined below. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/third_party/onyx/MODIFICATIONS.md b/third_party/onyx/MODIFICATIONS.md new file mode 100644 index 00000000..052d3e50 --- /dev/null +++ b/third_party/onyx/MODIFICATIONS.md @@ -0,0 +1,27 @@ +# Onyx connector framework modifications + +The registered source is copied and aggressively patched from the four MIT-region +files named in `UPSTREAM.toml`. The patch retains the checkpoint-return generator, +batch runner, connector interface, wire-model, and lazy-registry shapes while +removing all Onyx control-plane, database, index, Redis, Celery, tenant, file-store, +Pydantic, logging, hierarchy, heartbeat, and enterprise permission dependencies. + +The registry is closed to the File/Obsidian connector. The wire types are reduced +to dependency-free transient values and are translated immediately into +ContextEngine's Supply contracts. Exception handling avoids logging local variables +or credentials, and no code or behavior was copied from an `ee/` path. + +The governance registration records each post-patch vendored hash. These independently +verified hashes identify the corresponding original files at pinned commit +`2fb3dd10493b3883870fa8adced5b1a0e114feff`: + +| Upstream path | Original SHA-256 | +|---|---| +| `backend/onyx/connectors/interfaces.py` | `293c0dcca9230b75ea3eef1475262e0b4010ca4df9321880f41a9dad05561756` | +| `backend/onyx/connectors/connector_runner.py` | `dc41c82425287c039b0897c135bc45f520eeb88be9b2ef16df159f835a63f311` | +| `backend/onyx/connectors/models.py` | `8edcf633de61d2c769c0959ce744e9efae436083917b7ddb1d692f89eaa4f44b` | +| `backend/onyx/connectors/registry.py` | `439c49bcb7dcc522545176d015dde73ce93d991e4b3e875db4b4d23d94cad9c4` | + +The patch directory is reserved for future machine-readable refresh patches; this +initial cut is an intentionally narrow manual cut, not whole-file vendoring of +dependency-entangled upstream modules. diff --git a/third_party/onyx/UPSTREAM.toml b/third_party/onyx/UPSTREAM.toml new file mode 100644 index 00000000..6ec024a4 --- /dev/null +++ b/third_party/onyx/UPSTREAM.toml @@ -0,0 +1,37 @@ +repository = "https://github.com/onyx-dot-app/onyx.git" +commit = "2fb3dd10493b3883870fa8adced5b1a0e114feff" +source_paths = [ + "backend/onyx/connectors/interfaces.py", + "backend/onyx/connectors/connector_runner.py", + "backend/onyx/connectors/models.py", + "backend/onyx/connectors/registry.py", +] +excluded_paths = [ + "backend/ee", + "web/src/app/ee", + "web/src/ee", + "backend/onyx/connectors/file/connector.py", +] +reuse_mode = "copy-patch" +approval = "https://github.com/stone16/context-engine/issues/126" +license = "MIT" + +[[files]] +upstream_path = "backend/onyx/connectors/interfaces.py" +vendored_path = "third_party/onyx/connectors/interfaces.py" +sha256 = "6a9f3f753972d2036c9fe90d4a394b4cd22b325e92e00a01a550d7f8cc94f87f" + +[[files]] +upstream_path = "backend/onyx/connectors/connector_runner.py" +vendored_path = "third_party/onyx/connectors/connector_runner.py" +sha256 = "b650097cb44b1f56a07f80befa7ca6123771ab7244db77bb5637305c92b834de" + +[[files]] +upstream_path = "backend/onyx/connectors/models.py" +vendored_path = "third_party/onyx/connectors/models.py" +sha256 = "e3b95634e689457be8dbb99a253c26924143a0d59ac5bd5fb7629b57c3f06454" + +[[files]] +upstream_path = "backend/onyx/connectors/registry.py" +vendored_path = "third_party/onyx/connectors/registry.py" +sha256 = "b134c07f20959ed44802c1fc753f9c0a8e5a9388134f4bc57b4bcc5ba4166618" diff --git a/third_party/onyx/connectors/connector_runner.py b/third_party/onyx/connectors/connector_runner.py new file mode 100644 index 00000000..637882bc --- /dev/null +++ b/third_party/onyx/connectors/connector_runner.py @@ -0,0 +1,119 @@ +"""Patched checkpoint extraction and batching from the pinned Onyx MIT runner.""" + +from __future__ import annotations + +from collections.abc import Generator +from dataclasses import dataclass + +from third_party.onyx.connectors.interfaces import ( + CheckpointedConnector, + CheckpointedConnectorWithPermSync, + CheckpointOutput, +) +from third_party.onyx.connectors.models import ( + ConnectorCheckpoint, + ConnectorFailure, + DeletedDocument, + Document, +) + + +class CheckpointOutputWrapper: + """Expose a generator's final checkpoint only after all items.""" + + def __init__(self) -> None: + self.next_checkpoint: ConnectorCheckpoint | None = None + + def __call__( + self, + checkpoint_connector_generator: CheckpointOutput, + ) -> Generator[Document | DeletedDocument | ConnectorFailure | ConnectorCheckpoint]: + def _inner_wrapper( + output: CheckpointOutput, + ) -> CheckpointOutput: + self.next_checkpoint = yield from output + return self.next_checkpoint + + yield from _inner_wrapper(checkpoint_connector_generator) + if self.next_checkpoint is None: + raise RuntimeError("connector did not return a checkpoint") + yield self.next_checkpoint + + +@dataclass(frozen=True, slots=True) +class ConnectorBatch: + """One ordered item batch or the final connector checkpoint.""" + + documents: tuple[Document, ...] = () + deleted_documents: tuple[DeletedDocument, ...] = () + failures: tuple[ConnectorFailure, ...] = () + checkpoint: ConnectorCheckpoint | None = None + + +class ConnectorRunner: + """Batch one checkpointed connector without persistence or logging content.""" + + def __init__( + self, + connector: CheckpointedConnector, + checkpoint: ConnectorCheckpoint, + *, + batch_size: int, + include_permissions: bool, + ) -> None: + if not isinstance(connector, CheckpointedConnector): + raise TypeError("runner requires a checkpointed connector") + if type(checkpoint) is not ConnectorCheckpoint: + raise TypeError("runner requires an exact checkpoint") + if type(batch_size) is not int or batch_size < 1: + raise ValueError("runner batch size must be positive") + if include_permissions and not isinstance( + connector, + CheckpointedConnectorWithPermSync, + ): + raise ValueError("connector does not support permission observation") + self._connector = connector + self._checkpoint = checkpoint + self._batch_size = batch_size + self._include_permissions = include_permissions + + def run(self) -> Generator[ConnectorBatch]: + load = ( + self._connector.load_from_checkpoint_with_perm_sync + if self._include_permissions + and isinstance(self._connector, CheckpointedConnectorWithPermSync) + else self._connector.load_from_checkpoint + ) + documents: list[Document] = [] + deleted: list[DeletedDocument] = [] + failures: list[ConnectorFailure] = [] + for item in CheckpointOutputWrapper()(load(self._checkpoint)): + if type(item) is ConnectorCheckpoint: + if documents or deleted or failures: + yield ConnectorBatch( + tuple(documents), + tuple(deleted), + tuple(failures), + ) + yield ConnectorBatch(checkpoint=item) + continue + if type(item) is Document: + documents.append(item) + elif type(item) is DeletedDocument: + deleted.append(item) + elif type(item) is ConnectorFailure: + failures.append(item) + else: + raise ValueError("connector returned an invalid item") + if len(documents) + len(deleted) + len(failures) >= self._batch_size: + yield ConnectorBatch( + tuple(documents), + tuple(deleted), + tuple(failures), + ) + documents = [] + deleted = [] + failures = [] + + +__all__ = ["CheckpointOutputWrapper", "ConnectorBatch", "ConnectorRunner"] diff --git a/third_party/onyx/connectors/interfaces.py b/third_party/onyx/connectors/interfaces.py new file mode 100644 index 00000000..f8ba9194 --- /dev/null +++ b/third_party/onyx/connectors/interfaces.py @@ -0,0 +1,54 @@ +"""Patched checkpoint connector interfaces from the pinned Onyx MIT region.""" + +from __future__ import annotations + +import abc +from collections.abc import Generator + +from third_party.onyx.connectors.models import ( + ConnectorCheckpoint, + ConnectorItem, +) + +type CheckpointOutput = Generator[ + ConnectorItem, + None, + ConnectorCheckpoint, +] + + +class CheckpointedConnector(abc.ABC): + """A source connector whose generator returns exactly one checkpoint.""" + + @abc.abstractmethod + def load_from_checkpoint( + self, + checkpoint: ConnectorCheckpoint, + ) -> CheckpointOutput: + raise NotImplementedError + + @abc.abstractmethod + def build_dummy_checkpoint(self) -> ConnectorCheckpoint: + raise NotImplementedError + + @abc.abstractmethod + def validate_checkpoint(self, payload: bytes) -> ConnectorCheckpoint: + raise NotImplementedError + + +class CheckpointedConnectorWithPermSync(CheckpointedConnector): + """Checkpoint connector that observes permissions with every item.""" + + @abc.abstractmethod + def load_from_checkpoint_with_perm_sync( + self, + checkpoint: ConnectorCheckpoint, + ) -> CheckpointOutput: + raise NotImplementedError + + +__all__ = [ + "CheckpointOutput", + "CheckpointedConnector", + "CheckpointedConnectorWithPermSync", +] diff --git a/third_party/onyx/connectors/models.py b/third_party/onyx/connectors/models.py new file mode 100644 index 00000000..c2574cfa --- /dev/null +++ b/third_party/onyx/connectors/models.py @@ -0,0 +1,74 @@ +"""Patched connector wire models from the pinned Onyx connector framework. + +This registered MIT region is deliberately not ContextEngine canonical state. +The CE adapter translates every value into the Supply execution contracts. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True, slots=True) +class ConnectorCheckpoint: + """One opaque connector-owned progress value.""" + + payload: bytes + + def __post_init__(self) -> None: + if type(self.payload) is not bytes or not self.payload: + raise ValueError("connector checkpoint must be nonempty bytes") + + +@dataclass(frozen=True, slots=True) +class Document: + """Small runner-side document shape translated at the CE boundary.""" + + document_id: str + content: bytes + content_type: str + metadata: tuple[tuple[str, str], ...] = () + + def __post_init__(self) -> None: + if type(self.document_id) is not str or not self.document_id: + raise ValueError("connector document requires an identity") + if type(self.content) is not bytes or not self.content: + raise ValueError("connector document requires content bytes") + if type(self.content_type) is not str or not self.content_type: + raise ValueError("connector document requires a content type") + if type(self.metadata) is not tuple: + raise TypeError("connector document metadata must be a tuple") + + +@dataclass(frozen=True, slots=True) +class DeletedDocument: + """One source identity observed absent from the current snapshot.""" + + document_id: str + + def __post_init__(self) -> None: + if type(self.document_id) is not str or not self.document_id: + raise ValueError("connector delete requires an identity") + + +@dataclass(frozen=True, slots=True) +class ConnectorFailure: + """Content-free connector failure returned through the batch runner.""" + + category: str + + def __post_init__(self) -> None: + if self.category not in {"retryable", "terminal"}: + raise ValueError("connector failure category must be closed") + + +ConnectorItem = Document | DeletedDocument | ConnectorFailure + + +__all__ = [ + "ConnectorCheckpoint", + "ConnectorFailure", + "ConnectorItem", + "DeletedDocument", + "Document", +] diff --git a/third_party/onyx/connectors/registry.py b/third_party/onyx/connectors/registry.py new file mode 100644 index 00000000..6b4f7dda --- /dev/null +++ b/third_party/onyx/connectors/registry.py @@ -0,0 +1,30 @@ +"""Closed, patched registry derived from the pinned Onyx MIT registry shape.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from types import MappingProxyType + + +class ConnectorKind(StrEnum): + FILE_OBSIDIAN = "file-obsidian" + + +@dataclass(frozen=True, slots=True) +class ConnectorMapping: + module_path: str + class_name: str + + +CONNECTOR_CLASS_MAP = MappingProxyType( + { + ConnectorKind.FILE_OBSIDIAN: ConnectorMapping( + module_path="adapters.connectors.file", + class_name="FileConnectorAdapter", + ) + } +) + + +__all__ = ["CONNECTOR_CLASS_MAP", "ConnectorKind", "ConnectorMapping"] diff --git a/third_party/onyx/patches/.gitkeep b/third_party/onyx/patches/.gitkeep new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/third_party/onyx/patches/.gitkeep @@ -0,0 +1 @@ + diff --git a/third_party/onyx/sbom.cyclonedx.json b/third_party/onyx/sbom.cyclonedx.json new file mode 100644 index 00000000..f45e426b --- /dev/null +++ b/third_party/onyx/sbom.cyclonedx.json @@ -0,0 +1,31 @@ +{ + "bomFormat": "CycloneDX", + "metadata": { + "component": { + "bom-ref": "context-engine:third-party:onyx", + "name": "ContextEngine registered Onyx connector subtree", + "type": "library" + }, + "properties": [ + { + "name": "context-engine:sbom:scope", + "value": "third_party/onyx" + }, + { + "name": "context-engine:sbom:artifact-wide", + "value": "false" + } + ] + }, + "components": [ + { + "licenses": [{"license": {"id": "MIT"}}], + "name": "Onyx connector framework", + "purl": "pkg:github/onyx-dot-app/onyx@2fb3dd10493b3883870fa8adced5b1a0e114feff#backend/onyx/connectors", + "type": "library", + "version": "2fb3dd10493b3883870fa8adced5b1a0e114feff" + } + ], + "specVersion": "1.6", + "version": 1 +}