diff --git a/adapters/http/app.py b/adapters/http/app.py index 72130240..f52b8d70 100644 --- a/adapters/http/app.py +++ b/adapters/http/app.py @@ -93,6 +93,7 @@ CitationAuthorityUnavailable, ) from engine.runtime.construction import required_kernel_dependencies +from engine.runtime.content_io import CandidateIndexUnavailable from engine.runtime.context_run import ContextRunPersistenceUnavailable from engine.runtime.contracts import ( Acquire, @@ -722,6 +723,8 @@ def resolve_context( raise TrustedAuthorityUnavailable from None except ActiveReleaseUnavailable: raise TrustedAuthorityUnavailable from None + except CandidateIndexUnavailable: + raise TrustedAuthorityUnavailable from None except InvalidTrustedScopeSnapshot: raise TrustedAuthorityUnavailable from None diff --git a/adapters/pgvector.py b/adapters/pgvector.py new file mode 100644 index 00000000..38f000c0 --- /dev/null +++ b/adapters/pgvector.py @@ -0,0 +1,86 @@ +"""PostgreSQL ANN candidate discovery using one retained Runtime transaction.""" + +from __future__ import annotations + +from engine.runtime.content_io import CandidateIndexUnavailable +from engine.runtime.contracts import Acquire +from engine.runtime.evidence import CandidateRef +from engine.runtime.materialized import ( + MaterializedProjectionSession, + _discover_materialized_vector, +) +from engine.supply import ( + CONTEXT_FRAGMENT_EMBEDDING_DIMENSION, + EmbeddingProfile, + EmbeddingProvider, + EmbeddingProviderUnavailable, + validate_embedding_batch, +) + +DEFAULT_VECTOR_CANDIDATE_LIMIT = 16 +MAX_VECTOR_CANDIDATE_LIMIT = 64 + + +class VectorCandidateIndexUnavailable(CandidateIndexUnavailable): + """Content-free failure of query embedding or ANN candidate discovery.""" + + +class PostgreSQLVectorCandidateIndex: + """Return bounded content-free pgvector candidates for one Acquire.""" + + __slots__ = ("_embedding_profile", "_embedding_provider", "_limit") + + def __init__( + self, + embedding_provider: EmbeddingProvider, + *, + limit: int = DEFAULT_VECTOR_CANDIDATE_LIMIT, + ) -> None: + try: + embedding_profile = embedding_provider.profile + except (AttributeError, TypeError, ValueError): + raise TypeError( + "Vector candidate index requires an embedding provider" + ) from None + if type(embedding_profile) is not EmbeddingProfile: + raise TypeError("Vector candidate index requires an embedding provider") + if embedding_profile.dimension != CONTEXT_FRAGMENT_EMBEDDING_DIMENSION: + raise ValueError("Vector candidate dimension does not match storage") + if type(limit) is not int or not 1 <= limit <= MAX_VECTOR_CANDIDATE_LIMIT: + raise ValueError("Vector candidate limit is not available") + self._embedding_provider = embedding_provider + self._embedding_profile = embedding_profile + self._limit = limit + + def discover( + self, + request: Acquire, + projection_session: MaterializedProjectionSession, + ) -> tuple[CandidateRef, ...]: + if type(request) is not Acquire: + raise TypeError("Vector candidate discovery requires Acquire") + try: + query_embedding = validate_embedding_batch( + (request.need.query,), + self._embedding_provider.embed((request.need.query,)), + self._embedding_profile, + )[0] + return _discover_materialized_vector( + projection_session, + query_embedding, + self._limit, + source_refs=( + request.narrowing.source_refs + if request.narrowing is not None + else None + ), + resource_refs=( + request.narrowing.resource_refs + if request.narrowing is not None + else None + ), + ) + except EmbeddingProviderUnavailable: + raise VectorCandidateIndexUnavailable( + "Vector candidate discovery is unavailable" + ) from None diff --git a/docs/decisions/0067-discover-vector-candidates-in-the-current-runtime-transaction.md b/docs/decisions/0067-discover-vector-candidates-in-the-current-runtime-transaction.md new file mode 100644 index 00000000..9c115964 --- /dev/null +++ b/docs/decisions/0067-discover-vector-candidates-in-the-current-runtime-transaction.md @@ -0,0 +1,140 @@ +--- +name: adr-0067-discover-vector-candidates-in-the-current-runtime-transaction +version: "1.0.0" +description: > + Embed Acquire queries with the publication-compatible profile and discover + bounded content-free pgvector candidates in the retained UserActor transaction. +--- + +# 0067. Discover vector candidates in the current Runtime transaction + +- Status: accepted +- Date: 2026-07-27 +- Refines: ADR-0009, ADR-0025, ADR-0066 + +## Context + +ADR-0066 published fixed-dimension Fragment vectors and reserved query-time +embedding, compatibility, and filtered-ANN evidence for the first vector +CandidateIndex decision. Runtime already owns one current UserActor transaction +whose transaction-local settings drive FORCE RLS and whose nominal projection +session remains live through the sealed AuthorizationKernel. Opening a second +database session for retrieval would detach discovery from that trusted context. + +HNSW applies post-scan filters, including RLS and active-resource predicates. +With selective Organization, Membership, source, or policy filters, a bounded +one-pass approximate scan can underfill even when visible rows exist. Candidate +ordering cannot be authorization input, and ADR-0025 deliberately makes current +deterministic assembly independent of untrusted candidate order. + +## Decision + +`PostgreSQLVectorCandidateIndex` uses the explicit `EmbeddingProvider` seam and +requires exact dimensional compatibility with the stored 384-dimensional +vectors. Each +`Acquire` embeds only its exact query, validates the provider response through +the same float32 contract used by publication, and collapses provider failure +to one content-free availability category. HTTP maps that category to the +existing generic service-unavailable response. + +The index calls a narrow vector-discovery operation on the current +`MaterializedProjectionSession`. Its PostgreSQL port executes on the retained +UserActor connection; it opens no connection and creates no transaction. The +query selects only Organization/source/resource/revision/fragment lineage, +orders by cosine distance with `<=>`, and applies an exact 1–64 limit. Body, +path, title, provider metadata, and distance score never leave persistence. +Every result remains an untrusted `CandidateRef` and still passes the unchanged +AuthorizationKernel before projection. + +Caller-supplied source and resource narrowing is passed through only as an ANN +pre-filter before the candidate limit. It can remove candidates but cannot +grant access; the Kernel independently recomputes the authoritative +`EffectiveScope` intersection and exactly reauthorizes every surviving +`CandidateRef`. + +Before the ANN query, the port sets pgvector `hnsw.iterative_scan` to strict +order and a bounded maximum scan-tuple ceiling with transaction-local settings. +This allows pgvector to continue scanning after RLS and live-resource filters +remove approximate neighbors while retaining bounded work and exact distance +order at the index seam. The pinned harness uses the pgvector version recorded +by the served database topology. + +Candidate rank is intentionally not preserved through the current Kernel. +ADR-0025 makes assembly independent of candidate order so index rank cannot +affect access or the deterministic Package. This slice therefore provides +bounded vector recall, not relevance-ranked budget selection. Changing that +behavior requires a separately designed authorized ranking stage and is not +smuggled into this product-lane index change. + +The served composition does not activate the index here. Its default remains +reject-all and reports Runtime delivery as not active. + +The current `EmbeddingProfile` records dimension only. It does not prove that +publication and query vectors share one immutable model and input profile, so +this slice does not claim semantic embedding-profile compatibility. A served +external provider is prohibited until storage or immutable Release lineage +binds that identity and Runtime rejects mismatches. The bounded dogfood +activation may use only the deterministic network-free twin over a freshly +reimported corpus produced by that same twin. + +An external query-embedding provider is also not eligible for served +composition while Runtime cannot meter and enforce that call against the +effective PackageBudget. The follow-up activation must either add that usage +propagation or explicitly compose only the deterministic network-free twin, +classified as zero external provider calls and zero provider cost. External +provider configuration must fail closed until the metered path exists. + +The follow-up served activation must also pass the Kernel-computed +`EffectiveScope` to discovery as a removal-only pre-filter before the ANN +limit. Caller narrowing alone is insufficient when trusted policy operands are +stricter than the RLS-visible set. Every returned candidate must still pass the +Kernel's exact authorization and projection; the pre-filter never grants +authority. Until that propagation exists, the served vector carrier remains +`NOT_ACTIVE`. + +## Rationale + +Keeping discovery on the retained UserActor transaction preserves the exact +FORCE-RLS and Membership lifetime already established for resolve; a second +connection would require reconstructing trusted context and create a drift-prone +authorization boundary. Strict-order iterative HNSW scanning bounds work while +continuing past rows removed by tenant, lifecycle, ACL, and narrowing filters. +Keeping rank outside the current Kernel preserves ADR-0025's deterministic, +rank-independent assembly until a separately measured authorized-ranking stage +can justify changing the sealed path. + +## Consequences + +- Query embedding and publication share one dimension contract; semantic + model/input-profile identity is not yet persisted or release-bound. +- FORCE RLS narrows discovery as defense in depth; it never grants authority. +- Filtered HNSW scans can continue within a fixed work ceiling instead of + silently stopping after the first filtered approximate batch. +- Provider diagnostics, distances, bodies, paths, titles, and denied lineage do + not enter the candidate output or HTTP error response. +- Current PackageBudget selection remains deterministic and rank-independent; + semantic ranking quality is not claimed by this slice. +- Served external query embedding remains `NOT_ACTIVE` until its provider call, + cost, and elapsed usage can be enforced and recorded by Runtime. +- Served vector discovery remains `NOT_ACTIVE` until the Kernel-computed + EffectiveScope can restrict candidates before the ANN limit without becoming + an authorization decision at the index. + +## Revisit trigger + +Revisit after the real golden set and dogfood corpus exist, or before activating +rank-sensitive budget selection. The exit evidence for tuning or replacing the +current strategy includes exact-search recall delta, underfilled-result rate, +`EXPLAIN ANALYZE`, corpus size, filter selectivity, hardware, iterative-scan or +oversampling settings, latency, and cost. Hybrid fusion and content-bearing +reranking remain separate evidence-triggered decisions. + +Revisit before any served composition admits an external query-embedding +provider without exact PackageBudget usage propagation. + +Revisit before any served external provider or mixed-origin corpus can be used +without immutable publication/query embedding-profile identity and mismatch +rejection. + +Revisit before any served composition can limit ANN candidates without first +applying the current Kernel-computed EffectiveScope as a removal-only filter. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 19a0e62c..373287dc 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -52,6 +52,7 @@ kernel, capability separation, and publication visibility model. | Bounded File reclaim | [0060 — Reclaim expired File imports with bounded retries](0060-reclaim-expired-file-imports-with-bounded-retries.md) | The same function-only scheduler prefers database-timed expired work, revalidates exact current authority, and advances at most three higher WorkerLease generations through durable-boundary recovery | Caller-selected retry routing/timing/generation, terminal-failure retry, dead-letter or requeue authority, delete execution, Runtime authority, or a new process/queue | | Recursive File discovery | [0065 — Recurse File discovery with anchored descriptors](0065-recurse-file-discovery-with-anchored-descriptors.md) | Canonical nested Markdown paths are discovered through stable descriptor-relative no-follow traversal under one server-owned 1–64 MiB byte ceiling | Path-string traversal, symlink following, truncated baselines, unbounded or caller-owned read limits, non-Markdown discovery, or new polling/resync authority | | Fragment embedding publication | [0066 — Embed Fragments before publication](0066-embed-fragments-before-publication.md) | New immutable Fragments receive one validated 384-dimensional vector before activation through an explicit external provider or network-free CI twin | Implicit twin fallback, query-time retrieval, backfill authority, missing-vector activation, or vector-based authorization | +| Vector candidate discovery | [0067 — Discover vector candidates in the current Runtime transaction](0067-discover-vector-candidates-in-the-current-runtime-transaction.md) | Query embedding uses the stored profile and bounded filtered HNSW discovery stays inside the retained UserActor transaction, returning content-free CandidateRefs only | Second retrieval connections, vector rank as authority, content/score-bearing candidates, implicit provider fallback, or served carrier activation | | Private delivery ingress | [0045 — Redeem private delivery evidence at ingress](0045-redeem-private-delivery-evidence-at-ingress.md) | One digest-only service/request/asker/audience/epoch-bound DeliveryEvidenceRef constructs private TrustedDeliveryContext inside the current UserActor transaction before content work | Raw trusted delivery facts on the wire, bearer persistence, application-role minting/table reads, alternate Runtime paths, or claiming later M2 carriers | | Exact Package egress | [0046 — Bind egress to one exact Package hop](0046-bind-egress-to-one-exact-package-hop.md) | One digest-only grant binds one exact audience-bound Package to one model or channel preflight hop and redeems atomically | Treating Package construction as disclosure authority, arbitrary content at egress, cross-hop reuse, or bypassing final policy | | Public OpenAPI v0 | [0047 — Freeze OpenAPI v0 through one Runtime path](0047-freeze-openapi-v0-through-one-runtime-path.md) | One public `/v0/resolve` schema and a hidden provisional v1 bridge share the same sealed Runtime; Package release lineage is read-only from the Learning-published active manifest | Two authorization compositions, caller-authored release facts, Runtime publication/fallback, or in-place mutation of historical snapshots | @@ -166,3 +167,4 @@ touched: - [0064 — Split process ceremony along the kernel-seam boundary](0064-split-process-ceremony-along-the-kernel-seam-boundary.md) - [0065 — Recursive File discovery with anchored descriptors](0065-recurse-file-discovery-with-anchored-descriptors.md) - [0066 — Embed Fragments before publication](0066-embed-fragments-before-publication.md) +- [0067 — Discover vector candidates in the current Runtime transaction](0067-discover-vector-candidates-in-the-current-runtime-transaction.md) diff --git a/engine/persistence/membership_context.py b/engine/persistence/membership_context.py index 23c6b072..380efd26 100644 --- a/engine/persistence/membership_context.py +++ b/engine/persistence/membership_context.py @@ -164,6 +164,35 @@ def _canonical_candidate_revision(value: str) -> UUID | None: return parsed +_VECTOR_ITERATIVE_SCAN = "strict_order" +_VECTOR_MAX_SCAN_TUPLES = 20_000 +_VECTOR_CANDIDATE_SQL = """ + SELECT + resource.organization_id, + resource.source_ref, + fragment.resource_ref, + fragment.revision_id, + fragment.fragment_ref + FROM context_fragment AS fragment + JOIN context_resource AS resource + ON resource.organization_id = fragment.organization_id + AND resource.resource_ref = fragment.resource_ref + AND resource.active_revision_id = fragment.revision_id + AND resource.tombstoned IS FALSE + WHERE fragment.embedding IS NOT NULL + AND ( + CAST(:source_refs AS text[]) IS NULL + OR resource.source_ref = ANY(CAST(:source_refs AS text[])) + ) + AND ( + CAST(:resource_refs AS text[]) IS NULL + OR fragment.resource_ref = ANY(CAST(:resource_refs AS text[])) + ) + ORDER BY fragment.embedding <=> CAST(:query_embedding AS vector) + LIMIT :limit +""" + + class _PostgreSQLMaterializedProjectionPort: """Two-stage Fragment reads on the owning current-UserActor transaction.""" @@ -184,6 +213,51 @@ def source_is_active(self, source_ref: UUID) -> bool: ).scalar_one() return observed is True + def discover_vector( + self, + query_embedding: tuple[float, ...], + limit: int, + source_refs: tuple[str, ...] | None, + resource_refs: tuple[str, ...] | None, + ) -> tuple[CandidateRef, ...]: + self._connection.execute( + text( + "SELECT set_config('hnsw.iterative_scan', :iterative_scan, true), " + "set_config('hnsw.max_scan_tuples', :max_scan_tuples, true)" + ), + { + "iterative_scan": _VECTOR_ITERATIVE_SCAN, + "max_scan_tuples": str(_VECTOR_MAX_SCAN_TUPLES), + }, + ) + rows = self._connection.execute( + text(_VECTOR_CANDIDATE_SQL), + { + "query_embedding": "[" + + ",".join(repr(value) for value in query_embedding) + + "]", + "limit": limit, + "source_refs": list(source_refs) if source_refs is not None else None, + "resource_refs": ( + list(resource_refs) if resource_refs is not None else None + ), + }, + ) + candidates: list[CandidateRef] = [] + for row in rows: + if type(row.revision_id) is not UUID: + raise TypeError("vector discovery returned invalid revision lineage") + candidates.append( + CandidateRef( + organization_id=row.organization_id, + source_ref=row.source_ref, + resource_ref=row.resource_ref, + revision_ref=str(row.revision_id), + fragment_ref=row.fragment_ref, + ) + ) + return tuple(candidates) + def discover_exact_phrase( self, phrase_digest: str, diff --git a/engine/runtime/content_io.py b/engine/runtime/content_io.py index aac58ee3..0137539c 100644 --- a/engine/runtime/content_io.py +++ b/engine/runtime/content_io.py @@ -8,6 +8,16 @@ from engine.runtime.evidence import CandidateRef from engine.runtime.materialized import MaterializedProjectionSession +__all__ = [ + "CandidateIndex", + "CandidateIndexUnavailable", + "ContextProvider", + "RuntimeContentIo", + "SourceContentReader", + "exact_phrase_digest", + "prohibited_empty_path_content_io", +] + _EXACT_PHRASE_DIGEST_DOMAIN = b"context-engine.exact-phrase.v1\x00" @@ -29,6 +39,10 @@ def discover( ) -> tuple[CandidateRef, ...]: ... +class CandidateIndexUnavailable(RuntimeError): + """Content-free transient failure of one configured candidate index.""" + + class ContextProvider(Protocol): """Future provider projection seam.""" diff --git a/engine/runtime/materialized.py b/engine/runtime/materialized.py index 144b4ceb..d3f85be7 100644 --- a/engine/runtime/materialized.py +++ b/engine/runtime/materialized.py @@ -196,6 +196,14 @@ def __post_init__(self) -> None: class MaterializedProjectionPort(Protocol): """Narrow operations executed by the owning current database transaction.""" + def discover_vector( + self, + query_embedding: tuple[float, ...], + limit: int, + source_refs: tuple[str, ...] | None, + resource_refs: tuple[str, ...] | None, + ) -> tuple[CandidateRef, ...]: ... + def discover_exact_phrase( self, phrase_digest: str, @@ -306,6 +314,7 @@ def _construct_materialized_projection_session( not callable(getattr(port, "source_is_active", None)) or not callable(getattr(port, "locate", None)) or not callable(getattr(port, "project", None)) + or not callable(getattr(port, "discover_vector", None)) or not callable(getattr(port, "discover_exact_phrase", None)) or not callable(getattr(port, "observe_publication", None)) ): @@ -363,6 +372,49 @@ def _discover_materialized_exact_phrase( return candidates +def _discover_materialized_vector( + session: MaterializedProjectionSession, + query_embedding: tuple[float, ...], + limit: int, + *, + source_refs: tuple[str, ...] | None = None, + resource_refs: tuple[str, ...] | None = None, +) -> tuple[CandidateRef, ...]: + """Discover bounded content-free ANN lineage in the retained transaction.""" + + _require_active_materialized_projection_session(session) + if type(query_embedding) is not tuple or not query_embedding: + raise ValueError("vector discovery requires a nonempty query embedding") + if type(limit) is not int or limit <= 0: + raise ValueError("vector discovery requires a positive exact limit") + for field_name, refs in ( + ("source_refs", source_refs), + ("resource_refs", resource_refs), + ): + if refs is not None and ( + type(refs) is not tuple + or not refs + or any(type(ref) is not str or not ref for ref in refs) + ): + raise ValueError(f"vector discovery {field_name} must be nonempty refs") + candidates = session._port.discover_vector( + query_embedding, + limit, + source_refs, + resource_refs, + ) + if ( + type(candidates) is not tuple + or len(candidates) > limit + or any(type(candidate) is not CandidateRef for candidate in candidates) + ): + raise TypeError( + "materialized vector discovery must return bounded exact CandidateRef " + "values" + ) + return candidates + + def _observe_materialized_publication( session: MaterializedProjectionSession, candidate_ref: CandidateRef, diff --git a/engine/supply/embeddings.py b/engine/supply/embeddings.py index e9615bee..621f28a9 100644 --- a/engine/supply/embeddings.py +++ b/engine/supply/embeddings.py @@ -1,4 +1,4 @@ -"""Supply-only embedding contracts for immutable Fragment publication.""" +"""Embedding contracts for Fragment publication and content-free discovery.""" from __future__ import annotations @@ -30,7 +30,7 @@ class EmbeddingProviderUnavailable(RuntimeError): class EmbeddingProvider(Protocol): - """Batch embedding seam used only by Supply publication.""" + """Batch embedding seam shared by Supply publication and query discovery.""" @property def profile(self) -> EmbeddingProfile: ... diff --git a/tests/integration/test_pgvector_candidate_index.py b/tests/integration/test_pgvector_candidate_index.py new file mode 100644 index 00000000..03b46eac --- /dev/null +++ b/tests/integration/test_pgvector_candidate_index.py @@ -0,0 +1,797 @@ +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Event +from typing import Any, cast +from uuid import UUID, uuid4 + +import pytest +from fastapi.testclient import TestClient +from httpx import Response +from sqlalchemy import Engine, text + +import engine.persistence.membership_context as membership_context_module +from adapters.embeddings import DeterministicEmbeddingTwin +from adapters.http.app import HEALTH_RESPONSE, create_app +from adapters.pgvector import ( + DEFAULT_VECTOR_CANDIDATE_LIMIT, + PostgreSQLVectorCandidateIndex, +) +from engine.persistence import ( + DatabaseConfiguration, + PostgreSQLAccessPolicyControl, + PostgreSQLMembershipAuthority, + ResourceAccessRevocation, + create_database_engine, +) +from engine.persistence.membership_context import _VECTOR_CANDIDATE_SQL +from engine.runtime.construction import Runtime, required_kernel_dependencies +from engine.runtime.content_io import CandidateIndex +from engine.runtime.contracts import Acquire +from engine.runtime.evidence import CandidateRef +from engine.runtime.materialized import MaterializedProjectionSession +from engine.runtime.package_digest import QueryDigestKeyring +from engine.supply import EmbeddingProfile, EmbeddingProviderUnavailable +from tests.integration.test_file_import_tracer import ( + _ExactScopeAuthority, + _OrganizationAuthority, + _RuntimeAuthenticator, +) +from tests.integration.test_zz_file_revision_replacement import _scenario_user_id +from tests.support.file_imports import ( + NOW, + FileImportScenario, + delete_file_import_scenario, + prepare_file_import_scenario, + run_file_import, +) +from tests.support.releases import ( + clear_test_runtime_release, + ensure_test_runtime_release, +) + +pytestmark = pytest.mark.integration +QUERY = "ContextEngine delivers context." +NARROWED_TARGET_CONTENT = "The target handbook passage remains authorized." +TOKEN = "runtime-secret" +_ANN_DISTRACTOR_COUNT = 96 +_NARROWING_DISTRACTOR_COUNT = DEFAULT_VECTOR_CANDIDATE_LIMIT + 4 + + +class _RecordingVectorCandidateIndex: + def __init__(self) -> None: + self.inner = PostgreSQLVectorCandidateIndex(DeterministicEmbeddingTwin()) + self.calls: list[tuple[CandidateRef, ...]] = [] + + def discover( + self, + request: Acquire, + projection_session: MaterializedProjectionSession, + ) -> tuple[CandidateRef, ...]: + candidates = self.inner.discover(request, projection_session) + self.calls.append(candidates) + return candidates + + +class _BlockingVectorCandidateIndex: + def __init__(self) -> None: + self.inner = PostgreSQLVectorCandidateIndex(DeterministicEmbeddingTwin()) + self.discovered = Event() + self.release = Event() + self.calls: list[tuple[CandidateRef, ...]] = [] + + def discover( + self, + request: Acquire, + projection_session: MaterializedProjectionSession, + ) -> tuple[CandidateRef, ...]: + candidates = self.inner.discover(request, projection_session) + self.calls.append(candidates) + self.discovered.set() + if not self.release.wait(timeout=10): + raise RuntimeError("vector candidate barrier timed out") + return candidates + + +class _UnavailableEmbeddingProvider: + profile = EmbeddingProfile(384) + + def embed(self, inputs: tuple[str, ...]) -> tuple[tuple[float, ...], ...]: + del inputs + raise EmbeddingProviderUnavailable("provider detail must not escape") + + +def _published_scenario( + request: pytest.FixtureRequest, + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, + *, + label: str, + payload: bytes | None = b"# Handbook\n\nContextEngine delivers context.\n", +) -> tuple[FileImportScenario, CandidateRef, UUID]: + root = tmp_path / label + root.mkdir() + scenario = prepare_file_import_scenario( + root, + migration_configuration, + guarded_control_engine, + payload=payload, + ) + request.addfinalizer( + lambda: _delete_published_scenario( + migration_configuration, scenario.organization_id + ) + ) + assert scenario.token is not None + published = run_file_import( + scenario, + scenario.prepared, + scenario.token, + guarded_worker_engine, + ) + ensure_test_runtime_release( + scenario.organization_id, + active_revision_refs=(published.candidate_ref.revision_ref,), + ) + return ( + scenario, + published.candidate_ref, + _scenario_user_id(scenario, migration_configuration), + ) + + +def _delete_published_scenario( + migration_configuration: DatabaseConfiguration, + organization_id: UUID, +) -> None: + clear_test_runtime_release(organization_id) + delete_file_import_scenario(migration_configuration, organization_id) + + +def _add_cross_organization_vector_distractors( + migration_configuration: DatabaseConfiguration, + scenario: FileImportScenario, + candidate: CandidateRef, +) -> None: + embedding = DeterministicEmbeddingTwin().embed((QUERY,))[0] + parameters = [ + { + "organization_id": scenario.organization_id, + "resource_ref": candidate.resource_ref, + "revision_id": UUID(candidate.revision_ref), + "fragment_ref": f"fragment:vector-distractor:{ordinal:03d}", + "ordinal": ordinal, + "content": QUERY, + "embedding": "[" + ",".join(repr(value) for value in embedding) + "]", + } + for ordinal in range(1, _ANN_DISTRACTOR_COUNT + 1) + ] + engine = create_database_engine(migration_configuration) + try: + with engine.begin() as connection: + connection.execute( + text( + """ + INSERT INTO context_fragment ( + organization_id, resource_ref, revision_id, + fragment_ref, ordinal, content, projection_kind, + embedding + ) VALUES ( + :organization_id, :resource_ref, :revision_id, + :fragment_ref, :ordinal, :content, 'body', + CAST(:embedding AS vector) + ) + """ + ), + parameters, + ) + finally: + engine.dispose() + + +def _add_same_organization_narrowing_distractors( + migration_configuration: DatabaseConfiguration, + scenario: FileImportScenario, +) -> None: + embedding = DeterministicEmbeddingTwin().embed((QUERY,))[0] + parameters = [ + { + "organization_id": scenario.organization_id, + "resource_ref": f"resource:vector-narrowing-distractor:{ordinal:03d}", + "source_ref": str(scenario.source_ref.value), + "revision_id": uuid4(), + "fragment_ref": f"fragment:vector-narrowing-distractor:{ordinal:03d}", + "content": QUERY, + "embedding": "[" + ",".join(repr(value) for value in embedding) + "]", + "membership_id": scenario.membership_id, + } + for ordinal in range(_NARROWING_DISTRACTOR_COUNT) + ] + engine = create_database_engine(migration_configuration) + try: + with engine.begin() as connection: + connection.execute(text("SET CONSTRAINTS ALL DEFERRED")) + connection.execute( + text( + """ + INSERT INTO context_resource ( + organization_id, resource_ref, source_ref, + active_revision_id, tombstoned + ) VALUES ( + :organization_id, :resource_ref, :source_ref, + :revision_id, false + ) + """ + ), + parameters, + ) + connection.execute( + text( + """ + INSERT INTO context_revision ( + organization_id, resource_ref, revision_id + ) VALUES ( + :organization_id, :resource_ref, :revision_id + ) + """ + ), + parameters, + ) + connection.execute( + text( + """ + INSERT INTO context_fragment ( + organization_id, resource_ref, revision_id, + fragment_ref, ordinal, content, projection_kind, + embedding + ) VALUES ( + :organization_id, :resource_ref, :revision_id, + :fragment_ref, 0, :content, 'body', + CAST(:embedding AS vector) + ) + """ + ), + parameters, + ) + connection.execute( + text( + """ + INSERT INTO resource_access_policy ( + organization_id, resource_ref, principal_ref, + access_version, access_state, revoked_at + ) VALUES ( + :organization_id, :resource_ref, 'principal:file-reader', + 1, 'allowed', NULL + ) + """ + ), + parameters, + ) + connection.execute( + text( + """ + INSERT INTO membership_resource_field_right ( + organization_id, membership_id, membership_version, + resource_ref, field_ref + ) VALUES ( + :organization_id, :membership_id, 1, + :resource_ref, 'body' + ) + """ + ), + parameters, + ) + finally: + engine.dispose() + + +def _actor_settings( + scenario: FileImportScenario, + user_id: UUID, +) -> dict[str, str]: + return { + "app.actor_kind": "user", + "app.authentication_binding_ref": "binding:file-tracer", + "app.checked_at": NOW.isoformat().replace("+00:00", "Z"), + "app.membership_id": str(scenario.membership_id), + "app.membership_version": "1", + "app.organization_id": str(scenario.organization_id), + "app.principal_ref": "principal:file-reader", + "app.request_id": "issue-101-vector-plan", + "app.user_id": str(user_id), + } + + +def _ann_plan_and_exact_result( + guarded_runtime_engine: Engine, + scenario: FileImportScenario, + user_id: UUID, + *, + source_refs: tuple[str, ...] | None = None, + resource_refs: tuple[str, ...] | None = None, +) -> tuple[ + str, + tuple[tuple[object, ...], ...], + tuple[tuple[object, ...], ...], +]: + embedding = DeterministicEmbeddingTwin().embed((QUERY,))[0] + parameters = { + "query_embedding": "[" + + ",".join(repr(value) for value in embedding) + + "]", + "limit": 1, + "source_refs": list(source_refs) if source_refs is not None else None, + "resource_refs": list(resource_refs) if resource_refs is not None else None, + } + with guarded_runtime_engine.begin() as connection: + for name, value in _actor_settings(scenario, user_id).items(): + connection.execute( + text("SELECT set_config(:name, :value, true)"), + {"name": name, "value": value}, + ) + for name, value in ( + ("hnsw.iterative_scan", "strict_order"), + ("hnsw.max_scan_tuples", "20000"), + ("enable_seqscan", "off"), + ("enable_sort", "off"), + ): + connection.execute( + text("SELECT set_config(:name, :value, true)"), + {"name": name, "value": value}, + ) + plan = "\n".join( + str(line) + for line in connection.execute( + text( + "EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) " + + _VECTOR_CANDIDATE_SQL + ), + parameters, + ).scalars() + ) + approximate: tuple[tuple[object, ...], ...] = tuple( + tuple(row) + for row in connection.execute( + text(_VECTOR_CANDIDATE_SQL), + parameters, + ) + ) + connection.execute( + text("SELECT set_config('enable_indexscan', 'off', true)") + ) + exact: tuple[tuple[object, ...], ...] = tuple( + tuple(row) + for row in connection.execute( + text(_VECTOR_CANDIDATE_SQL), + parameters, + ) + ) + return plan, approximate, exact + + +def _client( + scenario: FileImportScenario, + candidate: CandidateRef, + user_id: UUID, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, + index: CandidateIndex, + *, + request_id: str, +) -> TestClient: + return TestClient( + create_app( + authenticator=_RuntimeAuthenticator( + scenario.organization_id, + user_id, + scenario.membership_id, + ), + organization_authority=_OrganizationAuthority(), + membership_authority=PostgreSQLMembershipAuthority( + guarded_runtime_engine + ), + scope_authority=_ExactScopeAuthority( + candidate.source_ref, + candidate.resource_ref, + ), + runtime=Runtime( + required_kernel_dependencies(), + candidate_index=index, + clock=lambda: NOW, + query_digest_keyring=query_digest_keyring, + ), + clock=lambda: NOW, + request_id_factory=lambda: request_id, + ) + ) + + +def _resolve( + client: TestClient, + *, + source_refs: tuple[str, ...] | None = None, + resource_refs: tuple[str, ...] | None = None, +) -> Response: + body: dict[str, object] = {"kind": "acquire", "need": {"query": QUERY}} + if source_refs is not None or resource_refs is not None: + body["requestNarrowing"] = { + key: value + for key, value in ( + ("sourceRefs", source_refs), + ("resourceRefs", resource_refs), + ) + if value is not None + } + return client.post( + "/v0/resolve", + headers={ + "Authorization": f"Bearer {TOKEN}", + "X-Context-Request-Id": "issue-101-vector-http", + }, + json=body, + ) + + +def _stable_empty(package: dict[str, Any]) -> dict[str, Any]: + return { + "blocks": package["blocks"], + "evidence": package["evidence"], + "gaps": package["gaps"], + "coverage": package["coverage"], + "budgetUsage": package["budgetUsage"], + } + + +def _assert_empty(response: Response) -> dict[str, Any]: + assert response.status_code == 200 + package = cast(dict[str, Any], response.json()["package"]) + assert package["blocks"] == package["evidence"] == package["gaps"] == [] + assert package["coverage"] == { + "status": "empty", + "reason": "no_authorized_evidence", + } + return _stable_empty(package) + + +def test_vector_candidate_http_chain_is_rls_scoped_content_free_and_bounded( + request: pytest.FixtureRequest, + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, +) -> None: + org_a, candidate_a, user_a = _published_scenario( + request, + tmp_path, + migration_configuration, + guarded_control_engine, + guarded_worker_engine, + label="org-a", + payload=f"# Handbook\n\n{NARROWED_TARGET_CONTENT}\n".encode(), + ) + org_b, candidate_b, _user_b = _published_scenario( + request, + tmp_path, + migration_configuration, + guarded_control_engine, + guarded_worker_engine, + label="org-b", + ) + _add_cross_organization_vector_distractors( + migration_configuration, + org_b, + candidate_b, + ) + _add_same_organization_narrowing_distractors( + migration_configuration, + org_a, + ) + index = _RecordingVectorCandidateIndex() + response = _resolve( + _client( + org_a, + candidate_a, + user_a, + guarded_runtime_engine, + query_digest_keyring, + index, + request_id="issue-101-vector-authorized", + ), + resource_refs=(candidate_a.resource_ref,), + ) + + assert response.status_code == 200 + package = response.json()["package"] + assert [block["text"] for block in package["blocks"]] == [ + NARROWED_TARGET_CONTENT + ] + assert len(package["evidence"]) == 1 + evidence = package["evidence"][0] + assert evidence["sourceRef"] == candidate_a.source_ref + assert evidence["resourceRef"] == candidate_a.resource_ref + assert evidence["revisionRef"] == candidate_a.revision_ref + assert evidence["fragmentRef"] == candidate_a.fragment_ref + assert index.calls == [(candidate_a,)] + assert candidate_b not in index.calls[0] + assert len(index.calls[0]) <= DEFAULT_VECTOR_CANDIDATE_LIMIT + assert set(CandidateRef.__dataclass_fields__) == { + "organization_id", + "source_ref", + "resource_ref", + "revision_ref", + "fragment_ref", + } + for forbidden in ( + str(org_b.organization_id), + candidate_b.source_ref, + candidate_b.resource_ref, + candidate_b.revision_ref, + ): + assert forbidden not in response.text + _unfiltered_plan, unfiltered_approximate, unfiltered_exact = ( + _ann_plan_and_exact_result( + guarded_runtime_engine, + org_a, + user_a, + ) + ) + target_row = ( + candidate_a.organization_id, + candidate_a.source_ref, + candidate_a.resource_ref, + UUID(candidate_a.revision_ref), + candidate_a.fragment_ref, + ) + assert target_row not in unfiltered_approximate + assert target_row not in unfiltered_exact + plan, approximate, exact = _ann_plan_and_exact_result( + guarded_runtime_engine, + org_a, + user_a, + resource_refs=(candidate_a.resource_ref,), + ) + assert "Index Scan using ix_context_fragment_embedding_hnsw" in plan + assert approximate == exact + assert len(approximate) == 1 + assert approximate[0][:5] == target_row + + +def test_vector_candidate_denials_have_one_non_enumerating_empty_shape( + request: pytest.FixtureRequest, + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, +) -> None: + scenario, candidate, user_id = _published_scenario( + request, + tmp_path, + migration_configuration, + guarded_control_engine, + guarded_worker_engine, + label="denials", + ) + index = _RecordingVectorCandidateIndex() + client = _client( + scenario, + candidate, + user_id, + guarded_runtime_engine, + query_digest_keyring, + index, + request_id="issue-101-vector-denials", + ) + unknown_candidate = CandidateRef( + organization_id=scenario.organization_id, + source_ref=candidate.source_ref, + resource_ref=f"resource:missing:{uuid4()}", + revision_ref=str(uuid4()), + fragment_ref="fragment:missing", + ) + unknown_shape = _assert_empty( + _resolve( + _client( + scenario, + unknown_candidate, + user_id, + guarded_runtime_engine, + query_digest_keyring, + index, + request_id="issue-101-vector-unknown", + ) + ) + ) + assert index.calls == [(candidate,)] + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE context_resource SET tombstoned = true " + "WHERE organization_id = :organization_id " + "AND resource_ref = :resource_ref" + ), + { + "organization_id": scenario.organization_id, + "resource_ref": candidate.resource_ref, + }, + ) + finally: + migration_engine.dispose() + + tombstoned_shape = _assert_empty(_resolve(client)) + assert index.calls == [(candidate,), ()] + assert tombstoned_shape == unknown_shape + + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE membership SET status = 'revoked' " + "WHERE organization_id = :organization_id " + "AND membership_id = :membership_id" + ), + { + "organization_id": scenario.organization_id, + "membership_id": scenario.membership_id, + }, + ) + finally: + migration_engine.dispose() + + revoked = _resolve(client) + assert revoked.status_code == 401 + assert revoked.json() == {"code": "authentication_failed"} + assert index.calls == [(candidate,), ()] + for forbidden in ( + candidate.source_ref, + candidate.resource_ref, + candidate.revision_ref, + candidate.fragment_ref, + ): + assert forbidden not in tombstoned_shape + assert forbidden not in revoked.text + + +def test_vector_candidate_stale_epoch_vetoes_already_discovered_evidence( + request: pytest.FixtureRequest, + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + control_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scenario, candidate, user_id = _published_scenario( + request, + tmp_path, + migration_configuration, + guarded_control_engine, + guarded_worker_engine, + label="stale-epoch", + ) + index = _BlockingVectorCandidateIndex() + final_read_reached = Event() + release_final_read = Event() + original_read = ( + membership_context_module._PostgreSQLPolicyEpochPort.read_current_epoch + ) + reads = 0 + + def block_final_epoch_read( + port: membership_context_module._PostgreSQLPolicyEpochPort, + organization_id: UUID, + ) -> object: + nonlocal reads + reads += 1 + if reads == 3: + final_read_reached.set() + if not release_final_read.wait(timeout=10): + raise RuntimeError("final Policy Epoch read was not released") + return original_read(port, organization_id) + + monkeypatch.setattr( + membership_context_module._PostgreSQLPolicyEpochPort, + "read_current_epoch", + block_final_epoch_read, + ) + client = _client( + scenario, + candidate, + user_id, + guarded_runtime_engine, + query_digest_keyring, + index, + request_id="issue-101-vector-stale-epoch", + ) + control_engine = create_database_engine(control_configuration) + try: + with ThreadPoolExecutor(max_workers=1) as executor: + pending = executor.submit(_resolve, client) + assert index.discovered.wait(timeout=10) + index.release.set() + assert final_read_reached.wait(timeout=10) + epoch = PostgreSQLAccessPolicyControl(control_engine).change_access( + ResourceAccessRevocation( + organization_id=scenario.organization_id, + resource_ref=candidate.resource_ref, + principal_ref="principal:file-reader", + expected_access_version=1, + ) + ) + assert epoch.value == 2 + release_final_read.set() + response = pending.result(timeout=10) + finally: + index.release.set() + release_final_read.set() + control_engine.dispose() + + _assert_empty(response) + assert index.calls == [(candidate,)] + assert reads == 3 + for forbidden in ( + candidate.source_ref, + candidate.resource_ref, + candidate.revision_ref, + candidate.fragment_ref, + ): + assert forbidden not in response.text + + +def test_vector_embedding_outage_is_one_content_free_service_unavailable_response( + request: pytest.FixtureRequest, + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, + guarded_worker_engine: Engine, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, +) -> None: + scenario, candidate, user_id = _published_scenario( + request, + tmp_path, + migration_configuration, + guarded_control_engine, + guarded_worker_engine, + label="provider-outage", + ) + client = _client( + scenario, + candidate, + user_id, + guarded_runtime_engine, + query_digest_keyring, + PostgreSQLVectorCandidateIndex(_UnavailableEmbeddingProvider()), + request_id="issue-101-vector-provider-outage", + ) + + response = _resolve(client) + + assert response.status_code == 503 + assert response.json() == {"code": "service_unavailable"} + assert "provider detail" not in response.text + for forbidden in ( + candidate.source_ref, + candidate.resource_ref, + candidate.revision_ref, + candidate.fragment_ref, + ): + assert forbidden not in response.text + + +def test_default_composition_keeps_vector_retrieval_not_active() -> None: + response = TestClient(create_app()).get("/health") + assert response.status_code == 200 + assert response.json()["runtime_delivery"] == "NOT_ACTIVE" + assert HEALTH_RESPONSE["runtime_delivery"] == "NOT_ACTIVE" diff --git a/tests/unit/test_materialized_projection.py b/tests/unit/test_materialized_projection.py index 208fa3ab..664d1dc6 100644 --- a/tests/unit/test_materialized_projection.py +++ b/tests/unit/test_materialized_projection.py @@ -17,6 +17,7 @@ MaterializedProjectionSession, _close_materialized_projection_scope, _construct_materialized_projection_session, + _discover_materialized_vector, _locate_materialized_fragment, _open_materialized_projection_scope, _project_materialized_fragment, @@ -54,6 +55,16 @@ def discover_exact_phrase(self, phrase_digest: str) -> tuple[CandidateRef, ...]: del phrase_digest return () + def discover_vector( + self, + query_embedding: tuple[float, ...], + limit: int, + source_refs: tuple[str, ...] | None, + resource_refs: tuple[str, ...] | None, + ) -> tuple[CandidateRef, ...]: + del query_embedding, limit, source_refs, resource_refs + return () + def source_is_active(self, source_ref: UUID) -> bool: del source_ref return True @@ -95,6 +106,7 @@ def locate(self, *args: object) -> None: del args project = locate + discover_vector = locate discover_exact_phrase = locate observe_publication = locate @@ -157,6 +169,21 @@ def test_locator_is_content_free_and_body_projection_is_a_separate_operation( _close_materialized_projection_scope(scope) +def test_vector_discovery_is_bounded_and_lifetime_bound() -> None: + scope = _open_materialized_projection_scope() + port = RecordingProjectionPort() + session = _construct_materialized_projection_session( + authority_scope=scope, + port=cast(MaterializedProjectionPort, port), + ) + + assert _discover_materialized_vector(session, (0.25,), 1) == () + _close_materialized_projection_scope(scope) + + with pytest.raises(ValueError, match="active materialized projection scope"): + _discover_materialized_vector(session, (0.25,), 1) + + def test_materialized_projection_rejects_more_than_the_public_field_bound() -> None: fields = tuple( MaterializedFieldValue( diff --git a/tests/unit/test_membership_context.py b/tests/unit/test_membership_context.py index 367e49b2..7a559a50 100644 --- a/tests/unit/test_membership_context.py +++ b/tests/unit/test_membership_context.py @@ -22,6 +22,7 @@ MembershipRejectionCategory, ) from engine.runtime.construction import PolicyEpochGate +from engine.runtime.evidence import CandidateRef from engine.runtime.materialized import ( MaterializedFragmentLocator, _require_active_materialized_projection_session, @@ -114,6 +115,21 @@ def execute( return _ProjectionRowsResult((self._row,)) +class _VectorConnection: + def __init__(self, rows: tuple[SimpleNamespace, ...]) -> None: + self._rows = rows + self.calls: list[tuple[str, dict[str, object] | None]] = [] + + def execute( + self, + statement: object, + parameters: dict[str, object] | None = None, + ) -> tuple[SimpleNamespace, ...]: + sql = str(statement) + self.calls.append((sql, parameters)) + return () if "set_config('hnsw.iterative_scan'" in sql else self._rows + + class _ReleaseConnection: def __init__(self, row: _ActiveReleaseRow) -> None: self._row = row @@ -330,6 +346,87 @@ def identity() -> MembershipIdentity: ) +def test_postgres_vector_discovery_uses_one_content_free_bounded_ann_query() -> None: + organization_id = identity().organization_id + connection = _VectorConnection( + ( + SimpleNamespace( + organization_id=organization_id, + source_ref="source:vector", + resource_ref="resource:vector", + revision_id=UUID("05b82c43-4e8f-49ae-a286-a40289a3413e"), + fragment_ref="fragment:vector", + ), + ) + ) + port = membership_context_module._PostgreSQLMaterializedProjectionPort( + cast(Connection, connection) + ) + + candidates = port.discover_vector( + (0.25, -0.5), + 1, + ("source:vector",), + ("resource:vector",), + ) + + assert candidates == ( + CandidateRef( + organization_id=organization_id, + source_ref="source:vector", + resource_ref="resource:vector", + revision_ref="05b82c43-4e8f-49ae-a286-a40289a3413e", + fragment_ref="fragment:vector", + ), + ) + assert len(connection.calls) == 2 + setting_statement, setting_parameters = connection.calls[0] + assert "set_config('hnsw.iterative_scan'" in setting_statement + assert "set_config('hnsw.max_scan_tuples'" in setting_statement + assert setting_parameters == { + "iterative_scan": "strict_order", + "max_scan_tuples": "20000", + } + statement, parameters = connection.calls[1] + assert "fragment.embedding <=> CAST(:query_embedding AS vector)" in statement + assert "resource.active_revision_id = fragment.revision_id" in statement + assert "resource.tombstoned IS FALSE" in statement + assert "fragment.embedding IS NOT NULL" in statement + assert "resource.source_ref = ANY(CAST(:source_refs AS text[]))" in statement + assert "fragment.resource_ref = ANY(CAST(:resource_refs AS text[]))" in statement + assert "LIMIT :limit" in statement + assert "content" not in statement + assert "path" not in statement + assert "title" not in statement + assert "score" not in statement + assert parameters == { + "query_embedding": "[0.25,-0.5]", + "limit": 1, + "source_refs": ["source:vector"], + "resource_refs": ["resource:vector"], + } + + +def test_postgres_vector_discovery_rejects_invalid_revision_lineage() -> None: + connection = _VectorConnection( + ( + SimpleNamespace( + organization_id=identity().organization_id, + source_ref="source:vector", + resource_ref="resource:vector", + revision_id="not-a-database-uuid", + fragment_ref="fragment:vector", + ), + ) + ) + port = membership_context_module._PostgreSQLMaterializedProjectionPort( + cast(Connection, connection) + ) + + with pytest.raises(TypeError, match="invalid revision lineage"): + port.discover_vector((0.25, -0.5), 1, None, None) + + @pytest.mark.parametrize("invalid_value", (None, "", " \t")) def test_postgres_projection_absorbs_malformed_structured_field_values( invalid_value: object, diff --git a/tests/unit/test_membership_field_projection.py b/tests/unit/test_membership_field_projection.py index ea85db09..448fee39 100644 --- a/tests/unit/test_membership_field_projection.py +++ b/tests/unit/test_membership_field_projection.py @@ -44,6 +44,16 @@ def discover_exact_phrase(self, phrase_digest: str) -> tuple[()]: del phrase_digest return () + def discover_vector( + self, + query_embedding: tuple[float, ...], + limit: int, + source_refs: tuple[str, ...] | None, + resource_refs: tuple[str, ...] | None, + ) -> tuple[()]: + del query_embedding, limit, source_refs, resource_refs + return () + def source_is_active(self, source_ref: UUID) -> bool: del source_ref return True diff --git a/tests/unit/test_pgvector_candidate_index.py b/tests/unit/test_pgvector_candidate_index.py new file mode 100644 index 00000000..6a27d3b2 --- /dev/null +++ b/tests/unit/test_pgvector_candidate_index.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +from typing import cast +from uuid import UUID + +import pytest + +from adapters.embeddings import DeterministicEmbeddingTwin +from adapters.pgvector import ( + MAX_VECTOR_CANDIDATE_LIMIT, + PostgreSQLVectorCandidateIndex, + VectorCandidateIndexUnavailable, +) +from engine.runtime.contracts import Acquire, ContextNeed, RequestNarrowing +from engine.runtime.evidence import CandidateRef +from engine.runtime.materialized import ( + MaterializedProjectionPort, + _close_materialized_projection_scope, + _construct_materialized_projection_session, + _open_materialized_projection_scope, +) +from engine.supply import EmbeddingProfile, EmbeddingProviderUnavailable + + +class _RecordingPort: + def __init__(self, candidates: tuple[CandidateRef, ...]) -> None: + self.candidates = candidates + self.calls: list[ + tuple[ + tuple[float, ...], + int, + tuple[str, ...] | None, + tuple[str, ...] | None, + ] + ] = [] + + def discover_vector( + self, + query_embedding: tuple[float, ...], + limit: int, + source_refs: tuple[str, ...] | None, + resource_refs: tuple[str, ...] | None, + ) -> tuple[CandidateRef, ...]: + self.calls.append((query_embedding, limit, source_refs, resource_refs)) + return self.candidates[:limit] + + def discover_exact_phrase(self, phrase_digest: str) -> tuple[()]: + del phrase_digest + return () + + def source_is_active(self, source_ref: UUID) -> bool: + del source_ref + return True + + def observe_publication(self, candidate_ref: CandidateRef) -> None: + del candidate_ref + + def locate(self, candidate_ref: CandidateRef) -> None: + del candidate_ref + + def project(self, locator: object) -> None: + del locator + + +class _UnavailableProvider: + profile = EmbeddingProfile(384) + + def embed(self, inputs: tuple[str, ...]) -> tuple[tuple[float, ...], ...]: + del inputs + raise EmbeddingProviderUnavailable("provider detail") + + +def _candidate() -> CandidateRef: + return CandidateRef( + organization_id=UUID("81e18bca-86a1-478a-937d-7675c6fe69b0"), + source_ref="source:vector", + resource_ref="resource:vector", + revision_ref="05b82c43-4e8f-49ae-a286-a40289a3413e", + fragment_ref="fragment:vector", + ) + + +def test_vector_index_embeds_query_and_returns_only_bounded_candidate_refs() -> None: + port = _RecordingPort((_candidate(),)) + scope = _open_materialized_projection_scope() + session = _construct_materialized_projection_session( + authority_scope=scope, + port=cast(MaterializedProjectionPort, port), + ) + try: + candidates = PostgreSQLVectorCandidateIndex( + DeterministicEmbeddingTwin(), + limit=1, + ).discover( + Acquire(need=ContextNeed(query="semantic query")), + session, + ) + finally: + _close_materialized_projection_scope(scope) + + assert candidates == (_candidate(),) + assert len(port.calls) == 1 + query_embedding, limit, source_refs, resource_refs = port.calls[0] + assert len(query_embedding) == 384 + assert limit == 1 + assert source_refs is None + assert resource_refs is None + assert set(CandidateRef.__dataclass_fields__) == { + "organization_id", + "source_ref", + "resource_ref", + "revision_ref", + "fragment_ref", + } + + +def test_vector_index_applies_request_narrowing_before_ann_limit() -> None: + port = _RecordingPort((_candidate(),)) + scope = _open_materialized_projection_scope() + session = _construct_materialized_projection_session( + authority_scope=scope, + port=cast(MaterializedProjectionPort, port), + ) + try: + PostgreSQLVectorCandidateIndex( + DeterministicEmbeddingTwin(), + limit=1, + ).discover( + Acquire( + need=ContextNeed(query="semantic query"), + narrowing=RequestNarrowing( + source_refs=("source:vector",), + resource_refs=("resource:vector",), + ), + ), + session, + ) + finally: + _close_materialized_projection_scope(scope) + + assert port.calls[0][2:] == ( + ("source:vector",), + ("resource:vector",), + ) + + +def test_vector_index_genericizes_query_embedding_failure_before_database_io() -> None: + port = _RecordingPort((_candidate(),)) + scope = _open_materialized_projection_scope() + session = _construct_materialized_projection_session( + authority_scope=scope, + port=cast(MaterializedProjectionPort, port), + ) + try: + with pytest.raises( + VectorCandidateIndexUnavailable, + match="Vector candidate discovery is unavailable", + ) as failure: + PostgreSQLVectorCandidateIndex(_UnavailableProvider()).discover( + Acquire(need=ContextNeed(query="semantic query")), + session, + ) + finally: + _close_materialized_projection_scope(scope) + + assert failure.value.__cause__ is None + assert port.calls == [] + + +@pytest.mark.parametrize("limit", [0, MAX_VECTOR_CANDIDATE_LIMIT + 1, True]) +def test_vector_index_refuses_unbounded_limits(limit: int) -> None: + with pytest.raises(ValueError, match="limit is not available"): + PostgreSQLVectorCandidateIndex(DeterministicEmbeddingTwin(), limit=limit) diff --git a/tests/unit/test_runtime_authorized_evidence.py b/tests/unit/test_runtime_authorized_evidence.py index 622f7cdb..56c01b14 100644 --- a/tests/unit/test_runtime_authorized_evidence.py +++ b/tests/unit/test_runtime_authorized_evidence.py @@ -175,6 +175,16 @@ def discover_exact_phrase(self, phrase_digest: str) -> tuple[()]: del phrase_digest return () + def discover_vector( + self, + query_embedding: tuple[float, ...], + limit: int, + source_refs: tuple[str, ...] | None, + resource_refs: tuple[str, ...] | None, + ) -> tuple[()]: + del query_embedding, limit, source_refs, resource_refs + return () + def source_is_active(self, source_ref: UUID) -> bool: del source_ref return True