diff --git a/PYTHON_MODULE_INDEX.md b/PYTHON_MODULE_INDEX.md index 08092cc..d2d0c18 100644 --- a/PYTHON_MODULE_INDEX.md +++ b/PYTHON_MODULE_INDEX.md @@ -4,9 +4,9 @@ - Declared source root: `src` - Packaging source of truth: `pyproject.toml` -- Importable production units: 5 -- Indexed production units: 5 -- Source/index parity: 5/5 +- Importable production units: 7 +- Indexed production units: 7 +- Source/index parity: 7/7 - Package data: `src/generic_rag/py.typed` - Locked verification owner: `.github/workflows/ci.yml` (supporting workflow, not an importable unit) @@ -57,8 +57,8 @@ `ProjectionLimits`, `ProjectionRequest`, `ProjectionManifestEntry`, `ProjectionManifest`, `ProjectionStateAvailability`, `ProjectionStateSnapshot`, `ProjectionStateStatus`, `ProjectionResult`, - `RetrievalQuery`, `RetrievalOutcome`, `RetrievalHit`, and `RetrievalResult` - from `generic_rag.contracts`. + `RetrievalLimits`, `RetrievalQuery`, `RetrievalOutcome`, `RetrievalHit`, and + `RetrievalResult` from `generic_rag.contracts`. - Re-exports: exactly the names in the module's `__all__`; none from the package root. - Direct internal dependencies: `generic_rag.errors`. @@ -70,7 +70,7 @@ `tests/test_retrieval_contracts.py`, `tests/test_package_boundaries.py`, `tests/support/clean_import_probe.py`, `tests/support/verify_artifacts.py`, and the locked CI import, boundary, and artifact checks. -- Documentation: `docs/api.md`, `docs/projection.md`, and +- Documentation: `docs/api.md`, `docs/projection.md`, `docs/retrieval.md`, and `docs/security-and-privacy.md`. ## `generic_rag.ports` @@ -90,8 +90,8 @@ - Verification: `tests/test_ports.py`, `tests/test_package_boundaries.py`, `tests/test_projection.py`, `tests/support/clean_import_probe.py`, and the locked CI import and boundary checks. -- Documentation: `docs/api.md`, `docs/lifecycle.md`, `docs/projection.md`, and - `docs/security-and-privacy.md`. +- Documentation: `docs/api.md`, `docs/lifecycle.md`, `docs/projection.md`, + `docs/retrieval.md`, and `docs/security-and-privacy.md`. ## `generic_rag.projection` @@ -104,7 +104,10 @@ - Re-exports: exactly the names in the module's `__all__`; none from the package root. - Direct internal dependencies: `generic_rag.contracts`, `generic_rag.errors`, - and `generic_rag.ports`. + `generic_rag.ports`, and `generic_rag.projection_integrity`. +- Integrity ownership: delegates deterministic source-digest, fragment-ID, and + checkpoint-token derivation and checkpoint validation to + `generic_rag.projection_integrity`. - Owned state or external resources: none; planning state is immutable and local to each call, while every embedder, writer, and resetter remains caller-owned through `Borrowed`. @@ -118,3 +121,59 @@ checks. - Documentation: `docs/projection.md`, `docs/api.md`, `docs/lifecycle.md`, and `docs/security-and-privacy.md`. + +## `generic_rag.projection_integrity` + +- Source: `src/generic_rag/projection_integrity.py` +- Responsibility: own deterministic v1 derivation and validation algorithms + for projection integrity values. +- Supported public imports: `derive_source_digest`, `derive_fragment_id`, + `derive_projection_checkpoint_token`, and + `has_valid_projection_checkpoint` from `generic_rag.projection_integrity`. +- Re-exports: exactly the names in the module's `__all__`; none from the package + root. +- Direct internal dependencies: `generic_rag.contracts` and + `generic_rag.errors`. +- Owned state or external resources: none; each call derives or checks an + immutable value from exact caller-supplied contracts. +- Material side effects: none at import or call time. +- Verification: `tests/test_projection_integrity.py`, + `tests/test_projection.py`, `tests/test_retrieval.py`, + `tests/test_package_boundaries.py`, `tests/support/clean_import_probe.py`, + `tests/support/verify_artifacts.py`, and the locked CI test, lint, strict + type, compile, build, clean-install, import, and artifact checks. +- Documentation: `docs/api.md`, `docs/projection.md`, `docs/retrieval.md`, and + `docs/security-and-privacy.md`. + +## `generic_rag.retrieval` + +- Source: `src/generic_rag/retrieval.py` +- Responsibility: deterministically compose bounded semantic and hybrid + retrieval results from published caller-owned projection state. +- Supported public imports: `retrieve_semantic` and `retrieve_hybrid` from + `generic_rag.retrieval`. +- Re-exports: exactly the names in the module's `__all__`; none from the package + root. +- Direct internal dependencies: `generic_rag.contracts`, `generic_rag.errors`, + `generic_rag.ports`, and `generic_rag.projection_integrity`. +- Owned state or external resources: none; ranking state is local to each call, + while every embedder, vector reader, and lexical retriever remains + caller-owned through `Borrowed`. +- Material side effects: none at import time. At explicit workflow call time it + may read one borrowed embedder identity, embed one query, search one borrowed + vector reader, and for hybrid retrieval search one borrowed lexical + retriever; it performs no persistence, network, retry, acquisition, release, + logging, authorization, citation, or lifecycle action itself. +- Retrieval semantics: validates and deduplicates exact fragment identities, + filters candidates against the full current published document revision, + and preserves semantic provider order or fuses original semantic and lexical + ranks deterministically without comparing raw scores. Returned fragments + remain non-authoritative; this module does not authorize a query or source, + validate authoritative source text, or create citations. +- Verification: `tests/test_retrieval.py`, + `tests/test_retrieval_contracts.py`, `tests/test_ports.py`, + `tests/test_package_boundaries.py`, `tests/support/clean_import_probe.py`, + `tests/support/verify_artifacts.py`, and the locked CI test, lint, strict + type, compile, build, clean-install, import, and artifact checks. +- Documentation: `docs/retrieval.md`, `docs/api.md`, `docs/lifecycle.md`, + `docs/security-and-privacy.md`, and `README.md`. diff --git a/README.md b/README.md index 1c7bd03..ec47813 100644 --- a/README.md +++ b/README.md @@ -3,12 +3,13 @@ `generic-rag` is a provider-neutral, runtime-dependency-free foundation for retrieval-augmented generation (RAG). Version 0.1.0 requires Python 3.11 or later and provides immutable contracts, typed error categories, synchronous -collaborator protocols, deterministic bounded projection orchestration, and -explicit caller-owned borrowing. +collaborator protocols, deterministic bounded projection orchestration, +semantic retrieval, rank-based hybrid retrieval, and explicit caller-owned +borrowing. -Retrieval and result composition are not implemented in 0.1.0. The package has -no built-in adapter, provider, factory, persistence, network client, -configuration system, authentication, citation mechanism, or CLI. +The package has no built-in adapter, provider, factory, persistence, network +client, configuration system, authentication, authorization, citation +mechanism, or CLI. ## Install from a checkout @@ -107,12 +108,23 @@ explicit corpus-wide reset is intended. See the [projection guide](docs/projecti for the complete lifecycle, state matrix, adapter obligations, and failure behavior. +Load the corresponding published manifest before retrieval, reauthorize each +query and source in the host, and call `retrieve_semantic` or +`retrieve_hybrid` with borrowed provider implementations. Returned fragment +text is non-authoritative: resolve each identity against the still-authorized +source revision and create host-owned citations before showing results to a +user or injecting bounded context into an agent. See the [retrieval +guide](docs/retrieval.md) for an executable independent-consumer example, +deterministic fusion behavior, outcome handling, and the complete host flow. + Public values must be imported from their owning modules: - `generic_rag.contracts` - `generic_rag.errors` - `generic_rag.ports` - `generic_rag.projection` +- `generic_rag.projection_integrity` +- `generic_rag.retrieval` The package root intentionally has no re-exports: `generic_rag.__all__ == ()`. See the [API reference](docs/api.md) for every supported name and invariant. @@ -129,19 +141,20 @@ shown to a user or supplied to a downstream tool or agent. policy, embeds ordered fragment text, replaces or deletes complete document projections, and returns a manifest and truthful receipt for caller-owned persistence. -- [Issue #4](https://github.com/Kims-DeveloperGroup/generic-rag/issues/4) is - planned to add retrieval and composition. Its intended responsibility is to - use an injected `Embedder` and `VectorIndexReader` for semantic candidates - and an injected `LexicalRetriever` for lexical candidates, then define - deduplication, fusion, limiting, and outcome behavior. Provider rank will be - the input; raw provider scores are not represented or assumed comparable. - -There is no end-user or agent query workflow yet. A consuming application can -project data now, but must wait for or implement a separate reviewed retrieval -layer before supplying retrieved context to users, tools, or agents. The -caller/provider ownership model remains explicit throughout. See [resource -lifecycle](docs/lifecycle.md) and -[security and privacy](docs/security-and-privacy.md). +- Retrieval accepts the matching caller-loaded published state and injected + semantic or lexical providers. It validates and filters current-revision + candidates, deduplicates exact identities, and returns bounded score-free + hits. Hybrid retrieval fuses provider ranks deterministically; it does not + compare raw provider scores. +- The host reauthorizes every query, resolves each returned fragment identity + against authoritative source data, verifies the exact source slice, and + creates citations. It may then show cited results to a user or inject bounded + cited context into an agent; the agent must retain those citations. + +The package does not decide provider selection, authentication, authorization, +prompt or tool policy, retry, display, logging, or resource lifecycle. See the +[retrieval guide](docs/retrieval.md), [resource lifecycle](docs/lifecycle.md), +and [security and privacy](docs/security-and-privacy.md). ## Compatibility @@ -150,10 +163,11 @@ assume compatibility across minor releases. For this release, direct imports from the documented owning modules are the supported public paths; root-level imports are not. -The distribution includes `py.typed`. The wheel contains exactly the five +The distribution includes `py.typed`. The wheel contains exactly the seven importable modules `generic_rag`, `generic_rag.errors`, -`generic_rag.contracts`, `generic_rag.ports`, and `generic_rag.projection`, plus -the typing marker. +`generic_rag.contracts`, `generic_rag.ports`, `generic_rag.projection`, +`generic_rag.projection_integrity`, and `generic_rag.retrieval`, plus the typing +marker. ## Development verification diff --git a/docs/api.md b/docs/api.md index 115c259..47d3c7c 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1,10 +1,11 @@ # Public API Version 0.1.0 exposes immutable values, typed error categories, synchronous -collaborator protocols, and deterministic bounded document projection. It does -not expose retrieval or result-composition orchestration. See the [projection -guide](projection.md), [resource lifecycle](lifecycle.md), and [security and -privacy boundary](security-and-privacy.md) for the surrounding usage contract. +collaborator protocols, deterministic bounded document projection, and +score-free semantic and hybrid retrieval. See the [projection guide](projection.md), +[retrieval guide](retrieval.md), [resource lifecycle](lifecycle.md), and +[security and privacy boundary](security-and-privacy.md) for the surrounding +usage contract. ## Import boundary @@ -42,6 +43,7 @@ public symbols. Import names from their owning modules instead. - `ProjectionStateSnapshot` - `ProjectionStateStatus` - `ProjectionResult` +- `RetrievalLimits` - `RetrievalQuery` - `RetrievalOutcome` - `RetrievalHit` @@ -64,6 +66,18 @@ public symbols. Import names from their owning modules instead. - `project_documents` - `rebuild_projection` +`generic_rag.projection_integrity` exports exactly: + +- `derive_source_digest` +- `derive_fragment_id` +- `derive_projection_checkpoint_token` +- `has_valid_projection_checkpoint` + +`generic_rag.retrieval` exports exactly: + +- `retrieve_semantic` +- `retrieve_hybrid` + The package does not support importing any of these names from the package root. @@ -127,6 +141,15 @@ An ordinary collaborator exception is chained as the operation error's cause. An invalid identity, malformed vector result, or non-`None` command result has no internal cause. `KeyboardInterrupt` and `SystemExit` pass through unchanged. +Retrieval defines no workflow-specific public exception. Invalid top-level +workflow inputs, including a query that exceeds `RetrievalLimits`, raise +`ContractValidationError` before collaborator effects. After a collaborator +boundary is entered, ordinary `Exception` failures and malformed collaborator +returns contribute a content-free failure state. A failed branch contributes no +fragment or exception text; a `PARTIAL` result can contain independently +validated hits from another branch. No exception cause crosses the result +boundary. `BaseException` subclasses pass through unchanged. + ## Documents and fragments | Type | Fields | Construction rules | @@ -174,6 +197,38 @@ The package produces manifests; the caller owns their persistence. Source digests, fragment IDs, and checkpoint tokens are deterministic under explicit v1 domains described in the [projection guide](projection.md#deterministic-projection-values). +The public integrity functions are synchronous and positional-only: + +```python +def derive_source_digest(document: Document, /) -> str: ... + +def derive_fragment_id( + document: DocumentIdentity, + start: int, + end: int, + /, +) -> str: ... + +def derive_projection_checkpoint_token( + corpus_id: str, + projection: ProjectionIdentity, + chunking: ChunkingPolicy, + entries: tuple[ProjectionManifestEntry, ...], + /, +) -> str: ... + +def has_valid_projection_checkpoint( + manifest: ProjectionManifest, + /, +) -> bool: ... +``` + +They reproduce the same v1 integrity values used by projection and retrieval. +They validate exact public contract shapes and canonical manifest-entry order; +invalid inputs raise `ContractValidationError`. Checkpoint validation returns +whether the supplied token equals the derived token. It does not inspect a +provider index or establish authorization. + ## Projection state and results `ProjectionStateAvailability` is a closed string enum: @@ -264,6 +319,7 @@ failure matrices. | Type | Fields | Construction rules | | --- | --- | --- | +| `RetrievalLimits` | `max_query_codepoints: int` | The query-text cap is a positive exact integer. | | `RetrievalQuery` | `corpus_id: str`, `text: str`, `hit_limit: int`, `candidate_limit: int` | Corpus and text are nonblank exact strings; limits are positive exact integers and `hit_limit <= candidate_limit`. Values are preserved exactly. | | `RetrievalHit` | `fragment: Fragment`, `rank: int` | Fragment requires its exact class and rank is a positive exact integer. There is no score field. | | `RetrievalResult` | `query: RetrievalQuery`, `outcome: RetrievalOutcome`, `hits: tuple[RetrievalHit, ...]`, `truncated: bool` | Nested values, the hit tuple, and the boolean require exact types. Hit count cannot exceed `query.hit_limit`. | @@ -274,12 +330,12 @@ ranks are contiguous from one, fragment identities are unique, and every fragment belongs to the query corpus. `PARTIAL` requires at least one hit; `UNAVAILABLE`, `STALE`, and `FAILED` require no hits and `truncated=False`. -`truncated=True` is the caller's explicit assertion that otherwise valid work -or results were cut by the query budget. Hits and reader ports are score-free; -raw provider scores are neither represented nor promised comparable. - -These are value contracts only. Version 0.1.0 has no package retrieval, -composition, citation, user, tool, or agent workflow. +The workflows set `truncated=True` exactly when validated, unique, +current-revision candidates exceed `query.hit_limit`. Hits and reader ports are +score-free; raw provider scores are neither represented nor promised +comparable. When constructing a `RetrievalResult` directly, callers remain +responsible for supplying a truthful `truncated` value because the value +contract cannot reconstruct discarded candidates. ## Collaborator ports @@ -302,12 +358,50 @@ Projection enforces the embedder result rules and requires each writer or resetter command to return exactly `None`. It cannot enforce external storage, atomicity, authorization, concurrency, or lifecycle behavior. +Retrieval checks its embedder identity and output, provider tuple types and +candidate bounds, fragment integrity, corpus and published revisions, and +cross-provider identity consistency. The caller still owns provider selection, +authorization, persistence, concurrency, retries, and authoritative source +validation. + `Borrowed[T]` is the companion ownership marker, not a provider port. Its exact behavior is documented in [resource lifecycle](lifecycle.md). -## Planned retrieval workflow +## Retrieval workflows + +Both public functions are synchronous and all parameters are positional-only: + +```python +def retrieve_semantic( + query: RetrievalQuery, + state: ProjectionStateSnapshot, + limits: RetrievalLimits, + embedder: Borrowed[Embedder], + vector_reader: Borrowed[VectorIndexReader], + /, +) -> RetrievalResult: ... + +def retrieve_hybrid( + query: RetrievalQuery, + state: ProjectionStateSnapshot, + limits: RetrievalLimits, + embedder: Borrowed[Embedder], + vector_reader: Borrowed[VectorIndexReader], + lexical_retriever: Borrowed[LexicalRetriever], + /, +) -> RetrievalResult: ... +``` -Retrieval and composition remain planned for Issue #4. The existing query, -result, reader, and lexical contracts do not promise an implemented workflow, -fusion algorithm, compatibility check, exception mapping, citation policy, or -user/agent integration. +Semantic retrieval embeds the query once, validates at most +`candidate_limit` vector candidates, preserves provider order through +current-revision filtering, and returns at most `hit_limit` hits. Hybrid +retrieval also obtains at most `candidate_limit` lexical candidates, preserves +each provider's original ranks, and fuses exact identities using deterministic +reciprocal rank fusion with offset 60. It uses provider ranks rather than raw +scores and applies opaque identity ordering to ties. + +The workflows return `complete`, `partial`, `unavailable`, `stale`, or `failed` +according to published-state and collaborator results. They do not authorize, +cite, persist, log, retry, or manage collaborator resources. See the +[retrieval guide](retrieval.md) for candidate validation, exact outcome +handling, deterministic fusion, and the required user and agent host flow. diff --git a/docs/lifecycle.md b/docs/lifecycle.md index bac169f..167d06a 100644 --- a/docs/lifecycle.md +++ b/docs/lifecycle.md @@ -3,15 +3,16 @@ Version 0.1.0 uses an explicit caller-owned lifecycle. The package defines collaborator protocols and `Borrowed[T]`; it does not acquire, configure, discover, persist, synchronize, or release provider resources. Projection calls -borrowed collaborators only during an explicit workflow invocation. See the -[API reference](api.md) for exact port signatures. +and retrieval calls use borrowed collaborators only during an explicit +workflow invocation. See the [API reference](api.md) for exact port signatures. ## Ownership rule The caller or provider integration owns every lifecycle decision: -1. Acquire and configure the embedder, vector index, projection-state store, - and any required synchronization. +1. Select, acquire, and configure the embedder, vector index, lexical + retriever, projection-state store, credentials, network clients, and any + required synchronization. 2. Authorize the complete source set and construct a bounded `ProjectionRequest`. 3. Wrap the application-owned collaborators in `Borrowed` and call @@ -26,6 +27,12 @@ The caller or provider integration owns every lifecycle decision: publication, synchronization, or release steps implicitly. It provides no transaction across collaborators and the manifest store. +For retrieval, the caller also loads the manifest paired with the index, +reauthorizes the requester and source set for each query, applies retry and +timeout policy around package calls, resolves returned identities against +authoritative source revisions, creates citations, and decides whether cited +results may be shown to a user or supplied to an agent. + ## `Borrowed[T]` `Borrowed` is a frozen, slotted context manager that retains a reference. It @@ -40,9 +47,11 @@ has exactly these context semantics: provider health checks. These guarantees still apply if the wrapped provider defines its own context -manager or lifecycle methods. Projection itself translates ordinary method -failures into `ProjectionOperationError`; that workflow behavior does not -change `Borrowed` semantics. +manager or lifecycle methods. Projection translates ordinary method failures +into `ProjectionOperationError`; retrieval represents an ordinary method +failure as a content-free failed branch and may return independent validated +hits from another branch. Those workflow behaviors do not change `Borrowed` +semantics. ## Successful borrowed scope @@ -106,11 +115,41 @@ published manifest cannot be used safely by the next incremental operation. See the [projection guide](projection.md) for the state matrix and destructive rebuild ordering. -## Retrieval lifecycle is not implemented - -The query and reader contracts do not create a package retrieval workflow. -Retrieval and composition remain planned for Issue #4, and no user or agent -query lifecycle is implied by the current projection API. +## Retrieval call scope + +`retrieve_semantic` and `retrieve_hybrid` receive exact `Borrowed` wrappers and +a caller-loaded `ProjectionStateSnapshot`. As with projection, they enter only +the no-op wrappers and never enter, close, or shut down the underlying provider +objects. Keep every provider alive for the complete synchronous call. + +Semantic retrieval reads the embedder identity, embeds one query, and then +searches the vector reader. Hybrid retrieval completes that semantic branch +and then calls the lexical retriever; an ordinary semantic failure does not +prevent the lexical call. The package does not retry. Ordinary provider +exceptions and malformed returns produce a failed branch: the overall result +is `failed` without hits or `partial` when the other branch supplies validated +hits. An identity or revision mismatch similarly produces `stale` without hits +or `partial` alongside another branch's hits. `BaseException` subclasses +propagate through the no-op borrowed scope. + +The host owns the complete operating lifecycle around those calls: + +- persist and publish each manifest with the provider index it describes; +- prevent projection, retrieval, rebuild, disable, and purge operations from + observing incompatible manifest/index combinations; +- select and acquire providers, credentials, and network resources and apply + provider-specific retry, timeout, and fallback policy; +- reauthorize each query and authoritative source, then revalidate returned + fragment identities and create citations; +- interpret `complete`, `partial`, `unavailable`, `stale`, and `failed` under + application policy and record any required audit events; and +- coordinate in-flight calls before disabling, rebuilding, purging, or shutting + down caller-owned state and resources. + +The package provides no daemon, background worker, registry, persistence, +dynamic loading, network discovery, lock, transaction, purge command, or +shutdown hook. See the [retrieval guide](retrieval.md) for the exact query flow +and outcome meanings. Review the [security and privacy boundary](security-and-privacy.md) before passing content to any adapter implementation. diff --git a/docs/projection.md b/docs/projection.md index e407b6a..933721b 100644 --- a/docs/projection.md +++ b/docs/projection.md @@ -5,9 +5,11 @@ deterministic vector projection. It supplies orchestration, contracts, and failure reporting; the caller supplies and owns the embedder, vector index, projection-state persistence, authorization policy, and synchronization. -Projection does not make content retrievable through this package. Retrieval, -result composition, and user or agent integration remain planned for [Issue -#4](https://github.com/Kims-DeveloperGroup/generic-rag/issues/4). +Projection establishes the published manifest and derived vector index that +retrieval validates. The host must publish and load those as one matching +logical state, then reauthorize queries and authoritative sources. See the +[retrieval guide](retrieval.md) for semantic and hybrid use; neither workflow +implements authentication, authorization, citations, or user/agent policy. ## Required adapters and state @@ -221,6 +223,14 @@ future package version will retain the same algorithm or accept an old manifest. Consumers that persist projection state should pin and review the package version and use explicit rebuild for an incompatible upgrade. +The positional-only public helpers `derive_source_digest`, +`derive_fragment_id`, `derive_projection_checkpoint_token`, and +`has_valid_projection_checkpoint` live in +`generic_rag.projection_integrity`. Projection and retrieval call the same +helpers; hosts may use them to construct or validate published state. They +validate contract structure but do not inspect provider storage. See the [API +reference](api.md) for their exact signatures. + Hashes are deterministic comparison and identity values, not encryption, authorization, or a proof of source ownership. See [security and privacy](security-and-privacy.md). @@ -253,5 +263,6 @@ The public error messages do not include document or vector content. Adapter exception messages remain reachable through exception chaining, so adapters and application logging must avoid disclosing sensitive values. -See the [API reference](api.md) for exact signatures and value invariants and -[resource lifecycle](lifecycle.md) for ownership details. +See the [API reference](api.md) for exact signatures and value invariants, +[resource lifecycle](lifecycle.md) for ownership details, and the [retrieval +guide](retrieval.md) for consuming a successfully published projection. diff --git a/docs/retrieval.md b/docs/retrieval.md new file mode 100644 index 0000000..3e25f21 --- /dev/null +++ b/docs/retrieval.md @@ -0,0 +1,324 @@ +# Retrieval + +`generic-rag` provides synchronous semantic and hybrid retrieval over a +host-published projection. The package composes caller-supplied collaborators; +it does not select, load, configure, retry, or close a provider. + +The package root intentionally exports nothing. Import retrieval contracts from +`generic_rag.contracts`, borrowed collaborator protocols from +`generic_rag.ports`, integrity helpers from +`generic_rag.projection_integrity`, and workflows from +`generic_rag.retrieval`. + +## Independent consumer example + +This complete example constructs a valid published state, injects local fake +providers, invokes both workflows, and resolves returned fragment identities +against a still-authorized authoritative source before creating citations. + +```python +from generic_rag.contracts import ( + ChunkingPolicy, + Document, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + EmbeddingVector, + Fragment, + FragmentIdentity, + ProjectionCheckpoint, + ProjectionIdentity, + ProjectionManifest, + ProjectionManifestEntry, + ProjectionStateAvailability, + ProjectionStateSnapshot, + RetrievalLimits, + RetrievalOutcome, + RetrievalQuery, +) +from generic_rag.ports import Borrowed +from generic_rag.projection_integrity import ( + derive_fragment_id, + derive_projection_checkpoint_token, + derive_source_digest, +) +from generic_rag.retrieval import retrieve_hybrid, retrieve_semantic + +corpus_id = "corpus-a" +source_document = Document( + identity=DocumentIdentity( + key=DocumentKey(corpus_id=corpus_id, document_id="document-a"), + revision_id="revision-1", + ), + text="Approved source text.", + attributes=(("classification", "public"),), +) +chunking = ChunkingPolicy(max_fragment_codepoints=128, overlap_codepoints=0) +embedding_identity = EmbeddingIdentity(model_id="example-embedding", dimensions=2) +projection_identity = ProjectionIdentity( + schema_id="schema-v1", + embedding=embedding_identity, +) +entries = ( + ProjectionManifestEntry( + document=source_document.identity, + source_digest=derive_source_digest(source_document), + fragment_count=1, + ), +) +checkpoint_token = derive_projection_checkpoint_token( + corpus_id, + projection_identity, + chunking, + entries, +) +published_manifest = ProjectionManifest( + corpus_id=corpus_id, + projection=projection_identity, + chunking=chunking, + entries=entries, + checkpoint=ProjectionCheckpoint( + corpus_id=corpus_id, + projection=projection_identity, + token=checkpoint_token, + ), +) +published_state = ProjectionStateSnapshot( + availability=ProjectionStateAvailability.PRESENT, + manifest=published_manifest, +) + +start, end = 0, len(source_document.text) +fragment = Fragment( + identity=FragmentIdentity( + document=source_document.identity, + fragment_id=derive_fragment_id(source_document.identity, start, end), + start=start, + end=end, + ), + text=source_document.text[start:end], + attributes=source_document.attributes, +) + + +class ExampleEmbedder: + @property + def identity(self): + return embedding_identity + + def embed(self, texts, /): + assert texts == ("Where is the approved source?",) + return (EmbeddingVector(values=(1.0, 0.0)),) + + +class ExampleVectorReader: + def search(self, query, embedding, /): + assert embedding == EmbeddingVector(values=(1.0, 0.0)) + return (fragment,) + + +class ExampleLexicalRetriever: + def search(self, query, /): + return (fragment,) + + +query = RetrievalQuery( + corpus_id=corpus_id, + text="Where is the approved source?", + hit_limit=2, + candidate_limit=4, +) +limits = RetrievalLimits(max_query_codepoints=200) + +# The host reauthorizes the corpus and documents for this query before calling +# retrieval. The published state must describe the projection being searched. +authorized_revisions = {source_document.identity: source_document} + +semantic_result = retrieve_semantic( + query, + published_state, + limits, + Borrowed(ExampleEmbedder()), + Borrowed(ExampleVectorReader()), +) +hybrid_result = retrieve_hybrid( + query, + published_state, + limits, + Borrowed(ExampleEmbedder()), + Borrowed(ExampleVectorReader()), + Borrowed(ExampleLexicalRetriever()), +) + + +def host_citations(result): + if result.outcome not in ( + RetrievalOutcome.COMPLETE, + RetrievalOutcome.PARTIAL, + ): + return () + + citations = [] + for hit in result.hits: + identity = hit.fragment.identity + authoritative = authorized_revisions.get(identity.document) + if authoritative is None: + continue + authoritative_text = authoritative.text[identity.start : identity.end] + if authoritative_text != hit.fragment.text: + continue + citations.append( + ( + identity.document.key.document_id, + identity.document.revision_id, + identity.start, + identity.end, + ) + ) + return tuple(citations) + + +assert semantic_result.outcome is RetrievalOutcome.COMPLETE +assert hybrid_result.outcome is RetrievalOutcome.COMPLETE +assert host_citations(semantic_result) == (("document-a", "revision-1", 0, 21),) +assert host_citations(hybrid_result) == (("document-a", "revision-1", 0, 21),) +``` + +`authorized_revisions` represents a host authorization decision made for this +query; it is not package state. A real host also verifies that the requesting +user or agent may use the corpus before invoking retrieval. + +## Inputs and limits + +Both workflows require: + +- a `RetrievalQuery` with a nonblank corpus and query, a positive `hit_limit`, + and a positive `candidate_limit` where `hit_limit <= candidate_limit`; +- the matching `ProjectionStateSnapshot` loaded by the host; +- a `RetrievalLimits` whose positive `max_query_codepoints` bounds the query + before any collaborator call; and +- caller-owned collaborators wrapped in `Borrowed`. + +The limits are caller-selected work budgets, not tenant quotas, rate limits, +authorization rules, or guarantees about a provider's resource use. Each +provider may return at most `candidate_limit` candidates. The package validates +that bound but does not configure the provider or restrict work hidden behind +its interface. + +A missing state produces `unavailable`. A corrupt state or invalid checkpoint +produces `failed`. A state for another corpus produces `unavailable`. A valid +published state with no entries produces an empty `complete` result without +calling a provider. + +Every provider return must be an exact tuple with no more than +`candidate_limit` entries. Each entry must be an exact `Fragment` for the query +corpus, stay within the manifest's fragment-width bound, and carry the +deterministic fragment ID for its document identity and range. Retrieval looks +up each stable `DocumentKey` in the manifest and keeps it only when the full +published `DocumentIdentity`, including revision, matches. Missing and old +revisions are filtered as stale. These checks still cannot prove that candidate +text equals its claimed authoritative source slice. + +## Semantic retrieval + +`retrieve_semantic` performs these steps in order: + +1. Validate the top-level query, state, limits, and borrowed handles without + collaborator effects. +2. Check that the borrowed embedder identity exactly matches the manifest's + embedding identity. A mismatch produces `stale` without embedding or vector + search. +3. Embed the exact one-element tuple `(query.text,)` and require exactly one + finite vector with the published dimensions. +4. Ask the vector reader for up to `candidate_limit` fragments. +5. Validate, deduplicate, and filter the candidates against current manifest + revisions, then return at most `hit_limit` hits in provider order. + +Provider ranks used during validation and fusion are the candidates' original +one-based positions in the provider tuple. Filtering or exact-identity +deduplication does not close gaps in those source ranks. Semantic retrieval +preserves the remaining provider order, while public `RetrievalHit.rank` values +are final contiguous ranks from one. The first identical occurrence wins. +Conflicting payloads for one exact fragment identity fail that provider branch. + +## Hybrid retrieval + +`retrieve_hybrid` executes the semantic branch and then the lexical branch. +Semantic failure or staleness does not prevent the lexical call. Each provider +returns at most `candidate_limit` candidates, and its original ranks are +preserved. + +Validated current-revision candidates are fused by reciprocal rank fusion: + +```text +fused score = sum(1 / (60 + provider rank)) +``` + +Raw provider scores are neither accepted nor returned. Exact fragment +identities are deduplicated across providers. Identical payloads contribute +both ranks; conflicting payloads produce a content-free `failed` result. Hits +sort by descending fused score, then by this opaque identity tuple: + +```text +(corpus_id, document_id, revision_id, fragment_id, start, end) +``` + +That final comparison is deterministic raw string/integer ordering, not text +normalization or semantic relevance. `truncated` is true only when the number +of validated, unique, current-revision fragments exceeds `hit_limit`. + +## Outcomes + +| Outcome | Meaning and host response | +| --- | --- | +| `complete` | All required branches completed. Hits may be empty. Resolve every returned identity against current authorized source data before use. | +| `partial` | At least one validated hit is available, but a branch was stale or failed. Revalidate and cite usable hits; apply host policy before showing or injecting them. | +| `unavailable` | State is missing or the loaded manifest belongs to another corpus. Do not treat this as an empty authoritative answer. | +| `stale` | No hits are usable and a provider identity or candidate revision is stale. Reconcile or rebuild the projection. | +| `failed` | No hits are usable because state, collaborator output, or an ordinary collaborator call failed. Apply host retry, fallback, and audit policy. | + +If stale and failed conditions coexist without hits, `failed` takes precedence. +Ordinary `Exception` failures at collaborator boundaries contribute only a +content-free failure state: the failed branch contributes no fragment or +exception text. A terminal `failed` result is empty; a `partial` result contains +only independently validated hits from another branch. `BaseException` +subclasses such as cancellation signals propagate and must be handled by the +host runtime. + +## User and agent utilization + +The required host flow is: + +1. Authorize documents before projection and publish the resulting manifest + and provider index under host-controlled persistence and concurrency. +2. Reauthorize the user or agent, corpus, and documents for every query. +3. Load the published manifest that matches the provider index being searched. +4. Invoke semantic or hybrid retrieval with caller-owned collaborators. +5. Resolve every returned `FragmentIdentity` against the still-authorized, + authoritative source revision. Verify the exact source slice still equals + the returned text, then create a host citation. +6. Show cited results to the authorized user, or inject only bounded cited + context into an agent. Agents must retain the host citations in any derived + answer or artifact. + +Fragment text is derived, non-authoritative data. A collaborator candidate can +prove neither that its text equals the authoritative source slice nor that the +requester remains authorized. Matching deterministic identities and the +published revision is necessary but insufficient; host revalidation is +mandatory. + +The package never authenticates or authorizes users, creates citations, logs +queries or results, loads providers, opens network connections, owns credentials +or provider resources, or decides prompt, tool, display, retry, or fallback +policy. See [Lifecycle and ownership](lifecycle.md) and +[Security and privacy](security-and-privacy.md) for host responsibilities. + +## Operational ownership + +The host owns provider selection and acquisition, network and credential +handling, retry and timeout policy, manifest and index persistence, concurrency +control, audit, disable/rebuild/purge procedures, and orderly shutdown. +`Borrowed` is only a no-op ownership marker: retrieval neither enters nor exits +the underlying collaborator. See [Projection](projection.md) for how to create +the published state and [API reference](api.md) for exact imports and +signatures. diff --git a/docs/security-and-privacy.md b/docs/security-and-privacy.md index 91abad7..179e188 100644 --- a/docs/security-and-privacy.md +++ b/docs/security-and-privacy.md @@ -1,11 +1,12 @@ # Security and Privacy -Version 0.1.0 defines in-process values, collaborator boundaries, and an -explicit projection workflow. The package itself performs no persistence, -network setup, provider discovery, credential loading, telemetry, or background -work. A projection call does pass derived fragment text and metadata to the -caller-supplied embedder and vector writer, whose effects are outside the -package. +Version 0.1.0 defines in-process values, collaborator boundaries, and explicit +projection and retrieval workflows. The package itself performs no +persistence, network setup, provider discovery, credential loading, telemetry, +or background work. Projection passes derived fragment text and metadata to +caller-supplied providers. Retrieval passes query data to caller-supplied +providers and accepts fragment candidates from them. Those collaborator +effects are outside the package. ## Caller responsibility @@ -14,19 +15,26 @@ source, revision, attribute, and intended destination. The package does not authenticate an authoritative source or decide whether a user, tool, or agent may project it. +Before every retrieval, the caller must reauthorize the requester, corpus, and +source revisions, then load the published manifest paired with the index being +searched. After retrieval, it must resolve each fragment identity against +still-authorized authoritative data, verify the exact source slice, and create +a citation before display or agent use. An agent must retain those host +citations in derived output. + The caller also controls: - corpus and tenant isolation; - which adapter implementations receive document text, fragment text, - embeddings, attributes, identifiers, or future queries; + embeddings, attributes, identifiers, or queries; - provider account, region, transport, and credential configuration; - vector-index and manifest-store access control, retention, replacement, deletion, backup, and recovery; - synchronization between vector mutations and manifest publication; - logging, tracing, metrics, exception rendering, redaction, and incident response; and -- whether future retrieved fragments are displayed, persisted, or supplied to - another tool or agent. +- whether retrieved fragments are displayed, persisted, or supplied to another + tool or agent, and the prompt/tool policy applied to them. Do not place credentials or other secrets in attributes, opaque identifiers, or checkpoint tokens. These fields preserve caller input and do not apply @@ -40,6 +48,13 @@ requested corpus. Those adapters may persist or transmit data according to their implementations. Review their transport, storage, subprocess, network, credential, and deletion behavior before use. +Calling `retrieve_semantic` passes the exact query text to the embedder, then +passes the query and derived query embedding to the vector reader. Calling +`retrieve_hybrid` also passes the query to the lexical retriever. A provider may +transmit, retain, correlate, or log those values according to its +implementation. The package has no hidden network or runtime dependency, but +injected collaborators can have both. + `Borrowed` only marks resources as caller-owned. It does not acquire, close, authenticate, synchronize, sandbox, or reduce the privileges of an adapter. The package supplies no transaction or rollback across the vector index and @@ -49,18 +64,24 @@ caller-owned manifest store. See [resource lifecycle](lifecycle.md). Treat all of the following as potentially sensitive: -- document and future query text; +- document and query text; - ordered attributes and opaque identities; - fragments and their source ranges; - embeddings and vector records; -- source digests, fragment IDs, manifests, and checkpoint tokens; and -- adapter exceptions and logs. +- source digests, fragment IDs, manifests, and checkpoint tokens; +- retrieval ranks and outcomes; and +- adapter exceptions and host logs. Fragments and embeddings may reveal source information. Deterministic IDs and digests may allow equality correlation or guessing attacks against predictable content. Their `sha256:` representation provides neither encryption nor access control and should not be used as proof of source ownership. +Retrieval results expose no raw provider score, but their ranks, identities, +fragment text, truncation flag, and outcome can still reveal content, +correlation, or availability information. Do not treat the score-free boundary +as anonymization. + Deleting an authoritative source does not automatically delete copies, backups, logs, embeddings, or derived records held by an application or provider. Incremental deletion and corpus reset cover only the behavior promised @@ -75,6 +96,12 @@ accepted request from exceeding caller-selected values; they are not global quotas, rate limits, memory isolation, provider billing controls, timeouts, or admission control. +`RetrievalLimits.max_query_codepoints` bounds query text before collaborator +effects. `candidate_limit` bounds the tuple accepted from each provider, and +`hit_limit` bounds returned hits. These are likewise caller-selected work +budgets, not quotas, authorization, cost controls, or proof that a provider did +only bounded internal work. + Choose limits from trusted application policy rather than untrusted request parameters. Account for the fact that a small fragment size and large permitted document set can still produce many fragments and provider operations. Supply @@ -95,6 +122,18 @@ document mutations in place, and a rebuild failure can occur after the corpus was reset. Do not publish a failed or partial receipt as a completed checkpoint; isolate the affected projection and recover under caller-owned policy. +Retrieval maps ordinary collaborator exceptions and malformed collaborator +returns to a failed provider branch, producing `failed` when no validated hit +remains or `partial` when another branch supplies one. A `failed` result is +content-free, and a `partial` result contains only independently validated +hits; provider exception text and rejected candidates do not cross the result +boundary. Stale provider identity or revision data is reported separately as +`stale`, or as `partial` when another branch supplies a validated hit. The host +decides whether to retry, fall back, audit, or suppress a result. +`BaseException` subclasses propagate unchanged, so the host runtime must handle +its own cancellation and shutdown signals. The package does not log queries, +fragments, provider failures, or outcomes. + ## Trust boundary A fragment range is a half-open Python code-point range. Standalone contracts @@ -103,12 +142,19 @@ Projection derives its own fragment text from the supplied document, but the package still cannot prove that the supplied document or revision was authoritative or authorized. +Retrieval candidates are collaborator-supplied derived data. Deterministic +fragment identity and a current manifest revision do not prove that candidate +text equals the authoritative source slice or that the requester remains +authorized. Host reauthorization, source resolution, exact-slice validation, +and citation creation are mandatory before use. + Version 0.1.0 provides no built-in encryption, authentication, authorization, ACL, content filter, persistence security, network security, citation -validation, secret management, vendor guarantee, retrieval workflow, or +creation or validation, secret management, vendor guarantee, logging, or user/agent policy. A consuming application must select and assess those controls for its environment. The complete public value and collaborator boundaries are listed in the [API reference](api.md), and deterministic projection behavior is documented in the -[projection guide](projection.md). +[projection guide](projection.md). Retrieval behavior and the required host +flow are documented in the [retrieval guide](retrieval.md). diff --git a/src/generic_rag/contracts.py b/src/generic_rag/contracts.py index 978cbc5..077349f 100644 --- a/src/generic_rag/contracts.py +++ b/src/generic_rag/contracts.py @@ -31,6 +31,7 @@ "ProjectionStateSnapshot", "ProjectionStateStatus", "ProjectionResult", + "RetrievalLimits", "RetrievalQuery", "RetrievalOutcome", "RetrievalHit", @@ -555,6 +556,16 @@ def __post_init__(self) -> None: ) +@dataclass(frozen=True, slots=True) +class RetrievalLimits: + """Independent bound on caller-supplied retrieval query text.""" + + max_query_codepoints: int + + def __post_init__(self) -> None: + _require_positive_integer("max_query_codepoints", self.max_query_codepoints) + + @dataclass(frozen=True, slots=True) class RetrievalQuery: """A bounded retrieval request for one opaque corpus.""" diff --git a/src/generic_rag/projection.py b/src/generic_rag/projection.py index a8ac690..019d79c 100644 --- a/src/generic_rag/projection.py +++ b/src/generic_rag/projection.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib from collections.abc import Callable from dataclasses import dataclass from enum import StrEnum @@ -38,6 +37,12 @@ StateCompatibilityError, ) from .ports import Borrowed, Embedder, VectorIndexResetter, VectorIndexWriter +from .projection_integrity import ( + derive_fragment_id, + derive_projection_checkpoint_token, + derive_source_digest, + has_valid_projection_checkpoint, +) __all__ = ( "ProjectionFailureStage", @@ -47,10 +52,6 @@ "rebuild_projection", ) -_SOURCE_DIGEST_DOMAIN = "generic-rag:projection-source:v1" -_FRAGMENT_ID_DOMAIN = "generic-rag:fragment-id:v1" -_CHECKPOINT_DOMAIN = "generic-rag:projection-checkpoint:v1" - class ProjectionFailureStage(StrEnum): """The collaborator stage at which a projection operation failed.""" @@ -264,85 +265,6 @@ def _validate_borrowed(name: str, value: object) -> None: _require_exact_type(name, value, Borrowed) -def _sha256_fields(fields: tuple[str, ...]) -> str: - digest = hashlib.sha256() - for field in fields: - encoded = field.encode("utf-8", "surrogatepass") - digest.update(len(encoded).to_bytes(8, "big", signed=False)) - digest.update(encoded) - return f"sha256:{digest.hexdigest()}" - - -def _source_digest(document: Document) -> str: - fields = [ - _SOURCE_DIGEST_DOMAIN, - "text", - document.text, - "attributes_count", - str(len(document.attributes)), - ] - for key, value in document.attributes: - fields.extend(("attribute_key", key, "attribute_value", value)) - return _sha256_fields(tuple(fields)) - - -def _fragment_id(document: DocumentIdentity, start: int, end: int) -> str: - return _sha256_fields( - ( - _FRAGMENT_ID_DOMAIN, - "corpus_id", - document.key.corpus_id, - "document_id", - document.key.document_id, - "revision_id", - document.revision_id, - "start", - str(start), - "end", - str(end), - ) - ) - - -def _checkpoint_token( - corpus_id: str, - projection: ProjectionIdentity, - chunking: ChunkingPolicy, - entries: tuple[ProjectionManifestEntry, ...], -) -> str: - fields = [ - _CHECKPOINT_DOMAIN, - "corpus_id", - corpus_id, - "schema_id", - projection.schema_id, - "embedding_model_id", - projection.embedding.model_id, - "embedding_dimensions", - str(projection.embedding.dimensions), - "max_fragment_codepoints", - str(chunking.max_fragment_codepoints), - "overlap_codepoints", - str(chunking.overlap_codepoints), - "entry_count", - str(len(entries)), - ] - for entry in entries: - fields.extend( - ( - "document_id", - entry.document.key.document_id, - "revision_id", - entry.document.revision_id, - "source_digest", - entry.source_digest, - "fragment_count", - str(entry.fragment_count), - ) - ) - return _sha256_fields(tuple(fields)) - - def _fragments(document: Document, chunking: ChunkingPolicy) -> tuple[Fragment, ...]: fragments: list[Fragment] = [] start = 0 @@ -351,7 +273,7 @@ def _fragments(document: Document, chunking: ChunkingPolicy) -> tuple[Fragment, end = min(start + chunking.max_fragment_codepoints, text_length) identity = FragmentIdentity( document.identity, - _fragment_id(document.identity, start, end), + derive_fragment_id(document.identity, start, end), start, end, ) @@ -374,7 +296,7 @@ def _prepare_target(request: ProjectionRequest) -> _PreparedTarget: fragments = _fragments(document, request.chunking) entry = ProjectionManifestEntry( document.identity, - _source_digest(document), + derive_source_digest(document), len(fragments), ) prepared.append(_PreparedDocument(document, fragments, entry)) @@ -382,7 +304,7 @@ def _prepare_target(request: ProjectionRequest) -> _PreparedTarget: checkpoint = ProjectionCheckpoint( request.corpus_id, request.projection, - _checkpoint_token( + derive_projection_checkpoint_token( request.corpus_id, request.projection, request.chunking, @@ -399,16 +321,6 @@ def _prepare_target(request: ProjectionRequest) -> _PreparedTarget: return _PreparedTarget(request, tuple(prepared), manifest) -def _has_valid_checkpoint(manifest: ProjectionManifest) -> bool: - expected = _checkpoint_token( - manifest.corpus_id, - manifest.projection, - manifest.chunking, - manifest.entries, - ) - return manifest.checkpoint.token == expected - - def _state_status( state: ProjectionStateSnapshot, target: _PreparedTarget, @@ -420,7 +332,7 @@ def _state_status( manifest = state.manifest assert manifest is not None - if not _has_valid_checkpoint(manifest): + if not has_valid_projection_checkpoint(manifest): return ProjectionStateStatus.CORRUPT if manifest.corpus_id != target.request.corpus_id: return ProjectionStateStatus.CORRUPT diff --git a/src/generic_rag/projection_integrity.py b/src/generic_rag/projection_integrity.py new file mode 100644 index 0000000..f03d30b --- /dev/null +++ b/src/generic_rag/projection_integrity.py @@ -0,0 +1,267 @@ +"""Deterministic integrity values shared by projection and retrieval.""" + +from __future__ import annotations + +import hashlib + +from .contracts import ( + ChunkingPolicy, + Document, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + ProjectionCheckpoint, + ProjectionIdentity, + ProjectionManifest, + ProjectionManifestEntry, +) +from .errors import ContractValidationError + +__all__ = ( + "derive_source_digest", + "derive_fragment_id", + "derive_projection_checkpoint_token", + "has_valid_projection_checkpoint", +) + +_SOURCE_DIGEST_DOMAIN = "generic-rag:projection-source:v1" +_FRAGMENT_ID_DOMAIN = "generic-rag:fragment-id:v1" +_CHECKPOINT_DOMAIN = "generic-rag:projection-checkpoint:v1" + + +def _require_exact_type(name: str, value: object, expected: type[object]) -> None: + if type(value) is not expected: + raise ContractValidationError( + f"{name} must be exactly {expected.__name__}, not {type(value).__name__}" + ) + + +def _require_nonblank_string(name: str, value: object) -> str: + _require_exact_type(name, value, str) + assert isinstance(value, str) + if not value or value.isspace(): + raise ContractValidationError(f"{name} must not be empty or whitespace-only") + return value + + +def _require_nonnegative_integer(name: str, value: object) -> int: + _require_exact_type(name, value, int) + assert isinstance(value, int) + if value < 0: + raise ContractValidationError(f"{name} must be nonnegative") + return value + + +def _validate_document_key(value: object) -> DocumentKey: + _require_exact_type("document key", value, DocumentKey) + assert isinstance(value, DocumentKey) + return DocumentKey(value.corpus_id, value.document_id) + + +def _validate_document_identity(value: object) -> DocumentIdentity: + _require_exact_type("document identity", value, DocumentIdentity) + assert isinstance(value, DocumentIdentity) + return DocumentIdentity( + _validate_document_key(value.key), + value.revision_id, + ) + + +def _validate_document(value: object) -> Document: + _require_exact_type("document", value, Document) + assert isinstance(value, Document) + return Document( + _validate_document_identity(value.identity), + value.text, + value.attributes, + ) + + +def _validate_embedding_identity(value: object) -> EmbeddingIdentity: + _require_exact_type("embedding identity", value, EmbeddingIdentity) + assert isinstance(value, EmbeddingIdentity) + return EmbeddingIdentity(value.model_id, value.dimensions) + + +def _validate_projection_identity(value: object) -> ProjectionIdentity: + _require_exact_type("projection identity", value, ProjectionIdentity) + assert isinstance(value, ProjectionIdentity) + return ProjectionIdentity( + value.schema_id, + _validate_embedding_identity(value.embedding), + ) + + +def _validate_chunking(value: object) -> ChunkingPolicy: + _require_exact_type("chunking", value, ChunkingPolicy) + assert isinstance(value, ChunkingPolicy) + return ChunkingPolicy( + value.max_fragment_codepoints, + value.overlap_codepoints, + ) + + +def _validate_manifest_entry(value: object) -> ProjectionManifestEntry: + _require_exact_type("manifest entry", value, ProjectionManifestEntry) + assert isinstance(value, ProjectionManifestEntry) + return ProjectionManifestEntry( + _validate_document_identity(value.document), + value.source_digest, + value.fragment_count, + ) + + +def _validate_checkpoint(value: object) -> ProjectionCheckpoint: + _require_exact_type("checkpoint", value, ProjectionCheckpoint) + assert isinstance(value, ProjectionCheckpoint) + return ProjectionCheckpoint( + value.corpus_id, + _validate_projection_identity(value.projection), + value.token, + ) + + +def _validate_manifest(value: object) -> ProjectionManifest: + _require_exact_type("manifest", value, ProjectionManifest) + assert isinstance(value, ProjectionManifest) + _require_exact_type("manifest entries", value.entries, tuple) + entries = tuple(_validate_manifest_entry(entry) for entry in value.entries) + canonical = ProjectionManifest( + value.corpus_id, + _validate_projection_identity(value.projection), + _validate_chunking(value.chunking), + entries, + _validate_checkpoint(value.checkpoint), + ) + if entries != canonical.entries: + raise ContractValidationError("manifest entries must be in canonical order") + return canonical + + +def _sha256_fields(fields: tuple[str, ...]) -> str: + digest = hashlib.sha256() + for field in fields: + encoded = field.encode("utf-8", "surrogatepass") + digest.update(len(encoded).to_bytes(8, "big", signed=False)) + digest.update(encoded) + return f"sha256:{digest.hexdigest()}" + + +def derive_source_digest(document: Document, /) -> str: + """Derive the stable v1 digest of exact source text and attributes.""" + + canonical = _validate_document(document) + fields = [ + _SOURCE_DIGEST_DOMAIN, + "text", + canonical.text, + "attributes_count", + str(len(canonical.attributes)), + ] + for key, value in canonical.attributes: + fields.extend(("attribute_key", key, "attribute_value", value)) + return _sha256_fields(tuple(fields)) + + +def derive_fragment_id( + document: DocumentIdentity, + start: int, + end: int, + /, +) -> str: + """Derive the stable v1 identity for one valid half-open source range.""" + + canonical = _validate_document_identity(document) + canonical_start = _require_nonnegative_integer("start", start) + canonical_end = _require_nonnegative_integer("end", end) + if canonical_start >= canonical_end: + raise ContractValidationError("fragment range must satisfy start < end") + return _sha256_fields( + ( + _FRAGMENT_ID_DOMAIN, + "corpus_id", + canonical.key.corpus_id, + "document_id", + canonical.key.document_id, + "revision_id", + canonical.revision_id, + "start", + str(canonical_start), + "end", + str(canonical_end), + ) + ) + + +def derive_projection_checkpoint_token( + corpus_id: str, + projection: ProjectionIdentity, + chunking: ChunkingPolicy, + entries: tuple[ProjectionManifestEntry, ...], + /, +) -> str: + """Derive the stable v1 token for one canonical projection manifest.""" + + canonical_corpus_id = _require_nonblank_string("corpus_id", corpus_id) + canonical_projection = _validate_projection_identity(projection) + canonical_chunking = _validate_chunking(chunking) + _require_exact_type("entries", entries, tuple) + canonical_entries = tuple(_validate_manifest_entry(entry) for entry in entries) + ordering_witness = ProjectionManifest( + canonical_corpus_id, + canonical_projection, + canonical_chunking, + canonical_entries, + ProjectionCheckpoint( + canonical_corpus_id, + canonical_projection, + "integrity-validation", + ), + ) + if canonical_entries != ordering_witness.entries: + raise ContractValidationError("manifest entries must be in canonical order") + + fields = [ + _CHECKPOINT_DOMAIN, + "corpus_id", + canonical_corpus_id, + "schema_id", + canonical_projection.schema_id, + "embedding_model_id", + canonical_projection.embedding.model_id, + "embedding_dimensions", + str(canonical_projection.embedding.dimensions), + "max_fragment_codepoints", + str(canonical_chunking.max_fragment_codepoints), + "overlap_codepoints", + str(canonical_chunking.overlap_codepoints), + "entry_count", + str(len(canonical_entries)), + ] + for entry in canonical_entries: + fields.extend( + ( + "document_id", + entry.document.key.document_id, + "revision_id", + entry.document.revision_id, + "source_digest", + entry.source_digest, + "fragment_count", + str(entry.fragment_count), + ) + ) + return _sha256_fields(tuple(fields)) + + +def has_valid_projection_checkpoint(manifest: ProjectionManifest, /) -> bool: + """Return whether a structurally valid manifest has its exact v1 token.""" + + canonical = _validate_manifest(manifest) + expected = derive_projection_checkpoint_token( + canonical.corpus_id, + canonical.projection, + canonical.chunking, + canonical.entries, + ) + return canonical.checkpoint.token == expected diff --git a/src/generic_rag/retrieval.py b/src/generic_rag/retrieval.py new file mode 100644 index 0000000..6341631 --- /dev/null +++ b/src/generic_rag/retrieval.py @@ -0,0 +1,517 @@ +"""Bounded semantic retrieval and deterministic rank-based hybrid fusion.""" + +from __future__ import annotations + +from dataclasses import dataclass +from fractions import Fraction +from math import isfinite + +from .contracts import ( + ChunkingPolicy, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + EmbeddingVector, + Fragment, + FragmentIdentity, + ProjectionCheckpoint, + ProjectionIdentity, + ProjectionManifest, + ProjectionManifestEntry, + ProjectionStateAvailability, + ProjectionStateSnapshot, + RetrievalHit, + RetrievalLimits, + RetrievalOutcome, + RetrievalQuery, + RetrievalResult, +) +from .errors import ContractValidationError +from .ports import Borrowed, Embedder, LexicalRetriever, VectorIndexReader +from .projection_integrity import ( + derive_fragment_id, + has_valid_projection_checkpoint, +) + +__all__ = ("retrieve_semantic", "retrieve_hybrid") + +_RRF_OFFSET = 60 + + +@dataclass(frozen=True, slots=True) +class _RankedFragment: + fragment: Fragment + source_rank: int + + +@dataclass(frozen=True, slots=True) +class _BranchResult: + all_candidates: tuple[_RankedFragment, ...] + current_candidates: tuple[_RankedFragment, ...] + stale: bool + failed: bool + + +def _require_exact_type(name: str, value: object, expected: type[object]) -> None: + if type(value) is not expected: + raise ContractValidationError( + f"{name} must be exactly {expected.__name__}, not {type(value).__name__}" + ) + + +def _validate_document_key(value: object) -> DocumentKey: + _require_exact_type("document key", value, DocumentKey) + assert isinstance(value, DocumentKey) + return DocumentKey(value.corpus_id, value.document_id) + + +def _validate_document_identity(value: object) -> DocumentIdentity: + _require_exact_type("document identity", value, DocumentIdentity) + assert isinstance(value, DocumentIdentity) + return DocumentIdentity( + _validate_document_key(value.key), + value.revision_id, + ) + + +def _validate_embedding_identity(value: object) -> EmbeddingIdentity: + _require_exact_type("embedding identity", value, EmbeddingIdentity) + assert isinstance(value, EmbeddingIdentity) + return EmbeddingIdentity(value.model_id, value.dimensions) + + +def _validate_projection_identity(value: object) -> ProjectionIdentity: + _require_exact_type("projection identity", value, ProjectionIdentity) + assert isinstance(value, ProjectionIdentity) + return ProjectionIdentity( + value.schema_id, + _validate_embedding_identity(value.embedding), + ) + + +def _validate_chunking(value: object) -> ChunkingPolicy: + _require_exact_type("chunking", value, ChunkingPolicy) + assert isinstance(value, ChunkingPolicy) + return ChunkingPolicy( + value.max_fragment_codepoints, + value.overlap_codepoints, + ) + + +def _validate_manifest_entry(value: object) -> ProjectionManifestEntry: + _require_exact_type("manifest entry", value, ProjectionManifestEntry) + assert isinstance(value, ProjectionManifestEntry) + return ProjectionManifestEntry( + _validate_document_identity(value.document), + value.source_digest, + value.fragment_count, + ) + + +def _validate_checkpoint(value: object) -> ProjectionCheckpoint: + _require_exact_type("checkpoint", value, ProjectionCheckpoint) + assert isinstance(value, ProjectionCheckpoint) + return ProjectionCheckpoint( + value.corpus_id, + _validate_projection_identity(value.projection), + value.token, + ) + + +def _validate_manifest(value: object) -> ProjectionManifest: + _require_exact_type("manifest", value, ProjectionManifest) + assert isinstance(value, ProjectionManifest) + _require_exact_type("manifest entries", value.entries, tuple) + entries = tuple(_validate_manifest_entry(entry) for entry in value.entries) + canonical = ProjectionManifest( + value.corpus_id, + _validate_projection_identity(value.projection), + _validate_chunking(value.chunking), + entries, + _validate_checkpoint(value.checkpoint), + ) + if entries != canonical.entries: + raise ContractValidationError("manifest entries must be in canonical order") + return canonical + + +def _validate_state(value: object) -> ProjectionStateSnapshot: + _require_exact_type("state", value, ProjectionStateSnapshot) + assert isinstance(value, ProjectionStateSnapshot) + _require_exact_type( + "state availability", + value.availability, + ProjectionStateAvailability, + ) + manifest = None if value.manifest is None else _validate_manifest(value.manifest) + return ProjectionStateSnapshot(value.availability, manifest) + + +def _validate_query(value: object) -> RetrievalQuery: + _require_exact_type("query", value, RetrievalQuery) + assert isinstance(value, RetrievalQuery) + return RetrievalQuery( + value.corpus_id, + value.text, + value.hit_limit, + value.candidate_limit, + ) + + +def _validate_limits(value: object) -> RetrievalLimits: + _require_exact_type("limits", value, RetrievalLimits) + assert isinstance(value, RetrievalLimits) + return RetrievalLimits(value.max_query_codepoints) + + +def _validate_borrowed(name: str, value: object) -> None: + _require_exact_type(name, value, Borrowed) + + +def _validate_embedding( + value: object, + expected_dimensions: int, +) -> EmbeddingVector: + _require_exact_type("embedding", value, EmbeddingVector) + assert isinstance(value, EmbeddingVector) + _require_exact_type("embedding values", value.values, tuple) + if len(value.values) != expected_dimensions: + raise ContractValidationError( + "embedding vector must have the requested dimensions" + ) + if any( + type(coordinate) is not float or not isfinite(coordinate) + for coordinate in value.values + ): + raise ContractValidationError( + "embedding vector coordinates must be canonical finite floats" + ) + return EmbeddingVector(value.values) + + +def _validate_fragment( + value: object, + query: RetrievalQuery, + chunking: ChunkingPolicy, +) -> Fragment: + _require_exact_type("candidate", value, Fragment) + assert isinstance(value, Fragment) + identity_value = value.identity + _require_exact_type("candidate identity", identity_value, FragmentIdentity) + assert isinstance(identity_value, FragmentIdentity) + identity = FragmentIdentity( + _validate_document_identity(identity_value.document), + identity_value.fragment_id, + identity_value.start, + identity_value.end, + ) + canonical = Fragment(identity, value.text, value.attributes) + if canonical.identity.document.key.corpus_id != query.corpus_id: + raise ContractValidationError("candidate corpus_id must match the query") + if len(canonical.text) > chunking.max_fragment_codepoints: + raise ContractValidationError( + "candidate width must not exceed the published chunk bound" + ) + if canonical.identity.fragment_id != derive_fragment_id( + canonical.identity.document, + canonical.identity.start, + canonical.identity.end, + ): + raise ContractValidationError( + "candidate fragment_id must match its deterministic identity" + ) + return canonical + + +def _validate_provider_candidates( + value: object, + query: RetrievalQuery, + manifest: ProjectionManifest, +) -> _BranchResult: + if type(value) is not tuple: + raise ContractValidationError("retriever output must be exactly tuple") + assert isinstance(value, tuple) + if len(value) > query.candidate_limit: + raise ContractValidationError( + "retriever output must not exceed query.candidate_limit" + ) + + manifest_by_key = {entry.document.key: entry for entry in manifest.entries} + first_by_identity: dict[FragmentIdentity, _RankedFragment] = {} + ordered: list[_RankedFragment] = [] + for source_rank, candidate in enumerate(value, start=1): + fragment = _validate_fragment(candidate, query, manifest.chunking) + previous = first_by_identity.get(fragment.identity) + if previous is not None: + if previous.fragment != fragment: + raise ContractValidationError( + "one provider returned conflicting candidate payloads" + ) + continue + ranked = _RankedFragment(fragment, source_rank) + first_by_identity[fragment.identity] = ranked + ordered.append(ranked) + + current: list[_RankedFragment] = [] + stale = False + for candidate in ordered: + published = manifest_by_key.get(candidate.fragment.identity.document.key) + if ( + published is None + or candidate.fragment.identity.document != published.document + ): + stale = True + continue + current.append(candidate) + return _BranchResult(tuple(ordered), tuple(current), stale, False) + + +def _failed_branch() -> _BranchResult: + return _BranchResult((), (), False, True) + + +def _stale_branch() -> _BranchResult: + return _BranchResult((), (), True, False) + + +def _semantic_branch( + query: RetrievalQuery, + manifest: ProjectionManifest, + embedder: Borrowed[Embedder], + vector_reader: Borrowed[VectorIndexReader], +) -> _BranchResult: + try: + with embedder as embedder_resource: + identity_value = embedder_resource.identity + identity = _validate_embedding_identity(identity_value) + if identity != manifest.projection.embedding: + return _stale_branch() + embedding_values = embedder_resource.embed((query.text,)) + if type(embedding_values) is not tuple or len(embedding_values) != 1: + raise ContractValidationError( + "embedder output must be an exact one-vector tuple" + ) + embedding = _validate_embedding( + embedding_values[0], + identity.dimensions, + ) + except Exception: + return _failed_branch() + + try: + with vector_reader as reader_resource: + candidates = reader_resource.search(query, embedding) + return _validate_provider_candidates(candidates, query, manifest) + except Exception: + return _failed_branch() + + +def _lexical_branch( + query: RetrievalQuery, + manifest: ProjectionManifest, + lexical_retriever: Borrowed[LexicalRetriever], +) -> _BranchResult: + try: + with lexical_retriever as lexical_resource: + candidates = lexical_resource.search(query) + return _validate_provider_candidates(candidates, query, manifest) + except Exception: + return _failed_branch() + + +def _terminal_result( + query: RetrievalQuery, + outcome: RetrievalOutcome, +) -> RetrievalResult: + return RetrievalResult(query, outcome, (), False) + + +def _state_manifest_or_result( + query: RetrievalQuery, + state: ProjectionStateSnapshot, +) -> ProjectionManifest | RetrievalResult: + if state.availability is ProjectionStateAvailability.MISSING: + return _terminal_result(query, RetrievalOutcome.UNAVAILABLE) + if state.availability is ProjectionStateAvailability.CORRUPT: + return _terminal_result(query, RetrievalOutcome.FAILED) + + manifest = state.manifest + assert manifest is not None + if manifest.corpus_id != query.corpus_id: + return _terminal_result(query, RetrievalOutcome.UNAVAILABLE) + if not has_valid_projection_checkpoint(manifest): + return _terminal_result(query, RetrievalOutcome.FAILED) + if not manifest.entries: + return _terminal_result(query, RetrievalOutcome.COMPLETE) + return manifest + + +def _outcome_for_hits( + has_hits: bool, + *, + stale: bool, + failed: bool, +) -> RetrievalOutcome: + if has_hits: + if stale or failed: + return RetrievalOutcome.PARTIAL + return RetrievalOutcome.COMPLETE + if failed: + return RetrievalOutcome.FAILED + if stale: + return RetrievalOutcome.STALE + return RetrievalOutcome.COMPLETE + + +def _ranked_result( + query: RetrievalQuery, + fragments: tuple[Fragment, ...], + *, + stale: bool, + failed: bool, +) -> RetrievalResult: + truncated = len(fragments) > query.hit_limit + selected = fragments[: query.hit_limit] + hits = tuple( + RetrievalHit(fragment, rank) for rank, fragment in enumerate(selected, start=1) + ) + outcome = _outcome_for_hits(bool(hits), stale=stale, failed=failed) + return RetrievalResult(query, outcome, hits, truncated) + + +def _identity_order(fragment: Fragment) -> tuple[str, str, str, str, int, int]: + identity = fragment.identity + return ( + identity.document.key.corpus_id, + identity.document.key.document_id, + identity.document.revision_id, + identity.fragment_id, + identity.start, + identity.end, + ) + + +def _has_cross_provider_conflict( + semantic: _BranchResult, + lexical: _BranchResult, +) -> bool: + semantic_by_identity = { + candidate.fragment.identity: candidate.fragment + for candidate in semantic.all_candidates + } + for candidate in lexical.all_candidates: + previous = semantic_by_identity.get(candidate.fragment.identity) + if previous is not None and previous != candidate.fragment: + return True + return False + + +def _fuse_candidates( + semantic: _BranchResult, + lexical: _BranchResult, +) -> tuple[Fragment, ...]: + fragments: dict[FragmentIdentity, Fragment] = {} + scores: dict[FragmentIdentity, Fraction] = {} + for branch in (semantic, lexical): + for candidate in branch.current_candidates: + identity = candidate.fragment.identity + fragments.setdefault(identity, candidate.fragment) + scores[identity] = scores.get(identity, Fraction()) + Fraction( + 1, + _RRF_OFFSET + candidate.source_rank, + ) + ranked_identities = sorted( + scores, + key=lambda identity: ( + -scores[identity], + _identity_order(fragments[identity]), + ), + ) + return tuple(fragments[identity] for identity in ranked_identities) + + +def retrieve_semantic( + query: RetrievalQuery, + state: ProjectionStateSnapshot, + limits: RetrievalLimits, + embedder: Borrowed[Embedder], + vector_reader: Borrowed[VectorIndexReader], + /, +) -> RetrievalResult: + """Retrieve bounded current fragments from one borrowed semantic index.""" + + canonical_query = _validate_query(query) + canonical_state = _validate_state(state) + canonical_limits = _validate_limits(limits) + _validate_borrowed("embedder", embedder) + _validate_borrowed("vector_reader", vector_reader) + if len(canonical_query.text) > canonical_limits.max_query_codepoints: + raise ContractValidationError( + "query text must not exceed limits.max_query_codepoints" + ) + + state_result = _state_manifest_or_result(canonical_query, canonical_state) + if type(state_result) is RetrievalResult: + return state_result + assert isinstance(state_result, ProjectionManifest) + branch = _semantic_branch( + canonical_query, + state_result, + embedder, + vector_reader, + ) + fragments = tuple(candidate.fragment for candidate in branch.current_candidates) + return _ranked_result( + canonical_query, + fragments, + stale=branch.stale, + failed=branch.failed, + ) + + +def retrieve_hybrid( + query: RetrievalQuery, + state: ProjectionStateSnapshot, + limits: RetrievalLimits, + embedder: Borrowed[Embedder], + vector_reader: Borrowed[VectorIndexReader], + lexical_retriever: Borrowed[LexicalRetriever], + /, +) -> RetrievalResult: + """Fuse bounded semantic and lexical ranks without comparing raw scores.""" + + canonical_query = _validate_query(query) + canonical_state = _validate_state(state) + canonical_limits = _validate_limits(limits) + _validate_borrowed("embedder", embedder) + _validate_borrowed("vector_reader", vector_reader) + _validate_borrowed("lexical_retriever", lexical_retriever) + if len(canonical_query.text) > canonical_limits.max_query_codepoints: + raise ContractValidationError( + "query text must not exceed limits.max_query_codepoints" + ) + + state_result = _state_manifest_or_result(canonical_query, canonical_state) + if type(state_result) is RetrievalResult: + return state_result + assert isinstance(state_result, ProjectionManifest) + semantic = _semantic_branch( + canonical_query, + state_result, + embedder, + vector_reader, + ) + lexical = _lexical_branch( + canonical_query, + state_result, + lexical_retriever, + ) + if _has_cross_provider_conflict(semantic, lexical): + return _terminal_result(canonical_query, RetrievalOutcome.FAILED) + fragments = _fuse_candidates(semantic, lexical) + return _ranked_result( + canonical_query, + fragments, + stale=semantic.stale or lexical.stale, + failed=semantic.failed or lexical.failed, + ) diff --git a/tests/support/verify_artifacts.py b/tests/support/verify_artifacts.py index ba6ea5f..667753d 100644 --- a/tests/support/verify_artifacts.py +++ b/tests/support/verify_artifacts.py @@ -17,6 +17,8 @@ "generic_rag/errors.py", "generic_rag/ports.py", "generic_rag/projection.py", + "generic_rag/projection_integrity.py", + "generic_rag/retrieval.py", } _PACKAGE_DATA = {"generic_rag/py.typed"} diff --git a/tests/test_package_boundaries.py b/tests/test_package_boundaries.py index a5b4eef..5ac439d 100644 --- a/tests/test_package_boundaries.py +++ b/tests/test_package_boundaries.py @@ -17,6 +17,8 @@ import generic_rag.errors as errors import generic_rag.ports as ports import generic_rag.projection as projection +import generic_rag.projection_integrity as projection_integrity +import generic_rag.retrieval as retrieval _PROJECT_ROOT = Path(__file__).resolve().parents[1] _SOURCE_ROOT = _PROJECT_ROOT / "src" @@ -28,6 +30,8 @@ "generic_rag.errors": "src/generic_rag/errors.py", "generic_rag.ports": "src/generic_rag/ports.py", "generic_rag.projection": "src/generic_rag/projection.py", + "generic_rag.projection_integrity": ("src/generic_rag/projection_integrity.py"), + "generic_rag.retrieval": "src/generic_rag/retrieval.py", } _EXPECTED_DEPENDENCIES = { "generic_rag": set(), @@ -38,6 +42,17 @@ "generic_rag.contracts", "generic_rag.errors", "generic_rag.ports", + "generic_rag.projection_integrity", + }, + "generic_rag.projection_integrity": { + "generic_rag.contracts", + "generic_rag.errors", + }, + "generic_rag.retrieval": { + "generic_rag.contracts", + "generic_rag.errors", + "generic_rag.ports", + "generic_rag.projection_integrity", }, } _EXPECTED_EXPORTS = { @@ -70,6 +85,7 @@ "ProjectionStateSnapshot", "ProjectionStateStatus", "ProjectionResult", + "RetrievalLimits", "RetrievalQuery", "RetrievalOutcome", "RetrievalHit", @@ -90,6 +106,16 @@ "project_documents", "rebuild_projection", ), + "generic_rag.projection_integrity": ( + "derive_source_digest", + "derive_fragment_id", + "derive_projection_checkpoint_token", + "has_valid_projection_checkpoint", + ), + "generic_rag.retrieval": ( + "retrieve_semantic", + "retrieve_hybrid", + ), } @@ -166,6 +192,7 @@ def test_production_module_inventory_is_exact(self) -> None: for path in _production_sources() } + self.assertEqual(len(actual), 7) self.assertEqual(actual, _EXPECTED_SOURCES) def test_supported_exports_are_exact_and_owned(self) -> None: @@ -175,6 +202,8 @@ def test_supported_exports_are_exact_and_owned(self) -> None: "generic_rag.contracts": contracts, "generic_rag.ports": ports, "generic_rag.projection": projection, + "generic_rag.projection_integrity": projection_integrity, + "generic_rag.retrieval": retrieval, } for module_name, expected_exports in _EXPECTED_EXPORTS.items(): @@ -190,6 +219,8 @@ def test_supported_exports_are_exact_and_owned(self) -> None: *_EXPECTED_EXPORTS["generic_rag.contracts"], *_EXPECTED_EXPORTS["generic_rag.ports"], *_EXPECTED_EXPORTS["generic_rag.projection"], + *_EXPECTED_EXPORTS["generic_rag.projection_integrity"], + *_EXPECTED_EXPORTS["generic_rag.retrieval"], ): with self.subTest(root_reexport=name): self.assertFalse(hasattr(generic_rag, name)) @@ -210,6 +241,7 @@ def test_internal_dependency_graph_is_exact_and_acyclic(self) -> None: for module_name, relative_path in _EXPECTED_SOURCES.items() } + self.assertEqual(sum(map(len, actual.values())), 12) self.assertEqual(actual, _EXPECTED_DEPENDENCIES) root_tree = _syntax_tree(_PROJECT_ROOT / _EXPECTED_SOURCES["generic_rag"]) root_imports = [ diff --git a/tests/test_projection_integrity.py b/tests/test_projection_integrity.py new file mode 100644 index 0000000..5e3c67a --- /dev/null +++ b/tests/test_projection_integrity.py @@ -0,0 +1,464 @@ +"""Independent v1 goldens and validation for projection integrity values.""" + +from __future__ import annotations + +import hashlib +import inspect +import unittest +from collections.abc import Callable +from typing import cast + +import generic_rag.projection_integrity as integrity +from generic_rag.contracts import ( + ChunkingPolicy, + Document, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + EmbeddingVector, + ProjectionCheckpoint, + ProjectionIdentity, + ProjectionLimits, + ProjectionManifest, + ProjectionManifestEntry, + ProjectionRequest, + ProjectionStateAvailability, + ProjectionStateSnapshot, + VectorRecord, +) +from generic_rag.errors import ContractValidationError +from generic_rag.ports import Borrowed, Embedder, VectorIndexResetter, VectorIndexWriter +from generic_rag.projection import rebuild_projection +from generic_rag.projection_integrity import ( + derive_fragment_id, + derive_projection_checkpoint_token, + derive_source_digest, + has_valid_projection_checkpoint, +) + +_CORPUS_ID = " Corpus/../e\u0301 " +_SOURCE_GOLDEN = ( + "sha256:19390f5d911704efab45a06821296d2309454cfb8bdca63799d1c451afa1a9ea" +) +_FRAGMENT_GOLDEN = ( + "sha256:49d28767699855dbf6778396488946303a62dc8394c46adcdd22bef653950293" +) +_CHECKPOINT_GOLDEN = ( + "sha256:f62fca95057be15ab4e4bab8eebde94b7f92eb87773ae2d82d849664c80ba59f" +) + + +class _TupleSubclass(tuple[object, ...]): + pass + + +def _golden_document() -> Document: + return Document( + DocumentIdentity( + DocumentKey(_CORPUS_ID, "doc/\0😀"), + "rev/한", + ), + "A\0😀e\u0301\n", + (("k", "v"), ("k", "v"), ("", "한")), + ) + + +def _projection() -> ProjectionIdentity: + return ProjectionIdentity( + "schema/😀", + EmbeddingIdentity("model/e\u0301", 3), + ) + + +def _checkpoint_entries() -> tuple[ProjectionManifestEntry, ...]: + return ( + ProjectionManifestEntry( + DocumentIdentity(DocumentKey(_CORPUS_ID, "alpha/e\u0301"), "r1😀"), + _SOURCE_GOLDEN, + 2, + ), + ProjectionManifestEntry( + DocumentIdentity(DocumentKey(_CORPUS_ID, "한"), "r2\0"), + "sha256:" + "b" * 64, + 0, + ), + ) + + +def _manifest(token: str = _CHECKPOINT_GOLDEN) -> ProjectionManifest: + projection = _projection() + return ProjectionManifest( + _CORPUS_ID, + projection, + ChunkingPolicy(4, 1), + _checkpoint_entries(), + ProjectionCheckpoint(_CORPUS_ID, projection, token), + ) + + +def _independent_hash(fields: tuple[str, ...]) -> str: + digest = hashlib.sha256() + for field in fields: + encoded = field.encode("utf-8", "surrogatepass") + digest.update(len(encoded).to_bytes(8, "big", signed=False)) + digest.update(encoded) + return f"sha256:{digest.hexdigest()}" + + +class _ProjectionEmbedder: + identity = _projection().embedding + + def embed(self, texts: tuple[str, ...], /) -> tuple[EmbeddingVector, ...]: + return tuple(EmbeddingVector((float(len(text)), 0.0, 1.0)) for text in texts) + + +class _ProjectionWriter: + def __init__(self) -> None: + self.records: tuple[VectorRecord, ...] = () + + def replace_document( + self, + document: DocumentIdentity, + records: tuple[VectorRecord, ...], + /, + ) -> None: + del document + self.records = records + + def delete_document(self, document: DocumentKey, /) -> None: + del document + + +class _ProjectionResetter: + def reset_corpus(self, corpus_id: str, /) -> None: + del corpus_id + + +class ProjectionIntegrityPublicTests(unittest.TestCase): + def test_exports_are_exact_owned_synchronous_and_positional_only(self) -> None: + expected = ( + "derive_source_digest", + "derive_fragment_id", + "derive_projection_checkpoint_token", + "has_valid_projection_checkpoint", + ) + expected_parameters = { + derive_source_digest: ("document",), + derive_fragment_id: ("document", "start", "end"), + derive_projection_checkpoint_token: ( + "corpus_id", + "projection", + "chunking", + "entries", + ), + has_valid_projection_checkpoint: ("manifest",), + } + + self.assertEqual(integrity.__all__, expected) + for function, parameter_names in expected_parameters.items(): + with self.subTest(function=function.__name__): + self.assertEqual( + function.__module__, "generic_rag.projection_integrity" + ) + self.assertFalse(inspect.iscoroutinefunction(function)) + parameters = tuple( + inspect.signature( + cast(Callable[..., object], function) + ).parameters.values() + ) + self.assertEqual( + tuple(parameter.name for parameter in parameters), + parameter_names, + ) + self.assertTrue( + all( + parameter.kind is inspect.Parameter.POSITIONAL_ONLY + for parameter in parameters + ) + ) + + def test_fixed_unicode_nul_and_duplicate_attribute_v1_goldens(self) -> None: + document = _golden_document() + entries = _checkpoint_entries() + + source_fields = [ + "generic-rag:projection-source:v1", + "text", + document.text, + "attributes_count", + str(len(document.attributes)), + ] + for key, value in document.attributes: + source_fields.extend(("attribute_key", key, "attribute_value", value)) + self.assertEqual(_independent_hash(tuple(source_fields)), _SOURCE_GOLDEN) + self.assertEqual(derive_source_digest(document), _SOURCE_GOLDEN) + + fragment_fields = ( + "generic-rag:fragment-id:v1", + "corpus_id", + document.identity.key.corpus_id, + "document_id", + document.identity.key.document_id, + "revision_id", + document.identity.revision_id, + "start", + "1", + "end", + "5", + ) + self.assertEqual(_independent_hash(fragment_fields), _FRAGMENT_GOLDEN) + self.assertEqual(derive_fragment_id(document.identity, 1, 5), _FRAGMENT_GOLDEN) + + checkpoint_fields = [ + "generic-rag:projection-checkpoint:v1", + "corpus_id", + _CORPUS_ID, + "schema_id", + _projection().schema_id, + "embedding_model_id", + _projection().embedding.model_id, + "embedding_dimensions", + "3", + "max_fragment_codepoints", + "4", + "overlap_codepoints", + "1", + "entry_count", + "2", + ] + for entry in entries: + checkpoint_fields.extend( + ( + "document_id", + entry.document.key.document_id, + "revision_id", + entry.document.revision_id, + "source_digest", + entry.source_digest, + "fragment_count", + str(entry.fragment_count), + ) + ) + self.assertEqual( + _independent_hash(tuple(checkpoint_fields)), _CHECKPOINT_GOLDEN + ) + self.assertEqual( + derive_projection_checkpoint_token( + _CORPUS_ID, + _projection(), + ChunkingPolicy(4, 1), + entries, + ), + _CHECKPOINT_GOLDEN, + ) + + def test_checkpoint_validation_is_exact_and_does_not_mutate_manifest(self) -> None: + manifest = _manifest() + before = repr(manifest) + + self.assertTrue(has_valid_projection_checkpoint(manifest)) + self.assertFalse(has_valid_projection_checkpoint(_manifest("wrong-token"))) + self.assertEqual(repr(manifest), before) + + def test_projection_workflow_preserves_the_extracted_v1_algorithms(self) -> None: + document = _golden_document() + projection = _projection() + chunking = ChunkingPolicy(4, 1) + request = ProjectionRequest( + _CORPUS_ID, + projection, + chunking, + ProjectionLimits(1, len(document.text), 4), + (document,), + ) + writer = _ProjectionWriter() + + result = rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + Borrowed(cast(Embedder, _ProjectionEmbedder())), + Borrowed(cast(VectorIndexWriter, writer)), + Borrowed(cast(VectorIndexResetter, _ProjectionResetter())), + ) + + entry = result.manifest.entries[0] + self.assertEqual(entry.source_digest, derive_source_digest(document)) + self.assertEqual( + tuple(record.fragment.identity.fragment_id for record in writer.records), + tuple( + derive_fragment_id(document.identity, start, end) + for start, end in ((0, 4), (3, 6)) + ), + ) + self.assertEqual( + result.manifest.checkpoint.token, + derive_projection_checkpoint_token( + request.corpus_id, + request.projection, + request.chunking, + result.manifest.entries, + ), + ) + + +class ProjectionIntegrityValidationTests(unittest.TestCase): + def assert_contract_failure(self, operation: Callable[[], object]) -> None: + with self.assertRaises(ContractValidationError) as raised: + operation() + self.assertIs(type(raised.exception), ContractValidationError) + + def test_top_level_values_require_exact_public_types(self) -> None: + operations: tuple[Callable[[], object], ...] = ( + lambda: derive_source_digest(cast(Document, object())), + lambda: derive_fragment_id( + cast(DocumentIdentity, object()), + 0, + 1, + ), + lambda: derive_projection_checkpoint_token( + cast(str, object()), + _projection(), + ChunkingPolicy(4, 1), + (), + ), + lambda: derive_projection_checkpoint_token( + _CORPUS_ID, + cast(ProjectionIdentity, object()), + ChunkingPolicy(4, 1), + (), + ), + lambda: derive_projection_checkpoint_token( + _CORPUS_ID, + _projection(), + cast(ChunkingPolicy, object()), + (), + ), + lambda: derive_projection_checkpoint_token( + _CORPUS_ID, + _projection(), + ChunkingPolicy(4, 1), + cast(tuple[ProjectionManifestEntry, ...], []), + ), + lambda: has_valid_projection_checkpoint(cast(ProjectionManifest, object())), + ) + + for operation in operations: + with self.subTest(operation=operation): + self.assert_contract_failure(operation) + + def test_corrupted_frozen_nested_values_fail_as_contract_errors(self) -> None: + malformed_documents = (_golden_document(), _golden_document()) + object.__setattr__(malformed_documents[0], "identity", object()) + object.__setattr__(malformed_documents[1], "attributes", [("k", "v")]) + + malformed_identity = _golden_document().identity + object.__setattr__(malformed_identity, "key", object()) + + malformed_projection = _projection() + object.__setattr__(malformed_projection.embedding, "dimensions", True) + + malformed_chunking = ChunkingPolicy(4, 1) + object.__setattr__(malformed_chunking, "overlap_codepoints", 4) + + malformed_entry = _checkpoint_entries()[0] + object.__setattr__(malformed_entry, "fragment_count", -1) + + operations: tuple[Callable[[], object], ...] = ( + *( + lambda value=value: derive_source_digest(value) + for value in malformed_documents + ), + lambda: derive_fragment_id(malformed_identity, 0, 1), + lambda: derive_projection_checkpoint_token( + _CORPUS_ID, + malformed_projection, + ChunkingPolicy(4, 1), + (), + ), + lambda: derive_projection_checkpoint_token( + _CORPUS_ID, + _projection(), + malformed_chunking, + (), + ), + lambda: derive_projection_checkpoint_token( + _CORPUS_ID, + _projection(), + ChunkingPolicy(4, 1), + (malformed_entry,), + ), + ) + + for operation in operations: + with self.subTest(operation=operation): + self.assert_contract_failure(operation) + + def test_fragment_ranges_are_nonnegative_ordered_exact_integers(self) -> None: + document = _golden_document().identity + for start, end in ( + (-1, 1), + (0, 0), + (2, 1), + (True, 1), + (0, False), + (0.0, 1), + (0, 1.0), + ): + with self.subTest(start=start, end=end): + self.assert_contract_failure( + lambda: derive_fragment_id( + document, + cast(int, start), + cast(int, end), + ) + ) + + def test_checkpoint_entries_must_be_an_exact_canonical_tuple(self) -> None: + entries = _checkpoint_entries() + wrong_corpus = ProjectionManifestEntry( + DocumentIdentity(DocumentKey("other", "document"), "revision"), + _SOURCE_GOLDEN, + 1, + ) + duplicate = ProjectionManifestEntry( + entries[0].document, + "sha256:" + "c" * 64, + 1, + ) + malformed_cases: tuple[object, ...] = ( + list(entries), + _TupleSubclass(entries), + tuple(reversed(entries)), + (entries[0], duplicate), + (wrong_corpus,), + (object(),), + ) + + for malformed in malformed_cases: + with self.subTest(container=type(malformed).__name__, value=malformed): + self.assert_contract_failure( + lambda: derive_projection_checkpoint_token( + _CORPUS_ID, + _projection(), + ChunkingPolicy(4, 1), + cast(tuple[ProjectionManifestEntry, ...], malformed), + ) + ) + + def test_checkpoint_validation_rejects_corrupted_manifest_shapes(self) -> None: + malformed_manifests = (_manifest(), _manifest(), _manifest()) + object.__setattr__( + malformed_manifests[0], "entries", list(_checkpoint_entries()) + ) + object.__setattr__(malformed_manifests[1], "checkpoint", object()) + object.__setattr__(malformed_manifests[2].projection, "schema_id", 1) + + for malformed in malformed_manifests: + with self.subTest(manifest=malformed): + self.assert_contract_failure( + lambda: has_valid_projection_checkpoint(malformed) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_retrieval.py b/tests/test_retrieval.py new file mode 100644 index 0000000..5f66da1 --- /dev/null +++ b/tests/test_retrieval.py @@ -0,0 +1,1312 @@ +"""Behavioral tests for bounded semantic and deterministic hybrid retrieval.""" + +from __future__ import annotations + +import inspect +import itertools +import json +import os +import subprocess +import sys +import textwrap +import unittest +from collections.abc import Callable +from fractions import Fraction +from pathlib import Path +from types import TracebackType +from typing import cast + +import generic_rag.retrieval as retrieval_module +from generic_rag.contracts import ( + ChunkingPolicy, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + EmbeddingVector, + Fragment, + FragmentIdentity, + ProjectionCheckpoint, + ProjectionIdentity, + ProjectionManifest, + ProjectionManifestEntry, + ProjectionStateAvailability, + ProjectionStateSnapshot, + RetrievalHit, + RetrievalLimits, + RetrievalOutcome, + RetrievalQuery, + RetrievalResult, +) +from generic_rag.errors import ContractValidationError +from generic_rag.ports import Borrowed, Embedder, LexicalRetriever, VectorIndexReader +from generic_rag.projection_integrity import ( + derive_fragment_id, + derive_projection_checkpoint_token, +) +from generic_rag.retrieval import retrieve_hybrid, retrieve_semantic + +_SOURCE_ROOT = Path(__file__).resolve().parents[1] / "src" +_DEFAULT = object() + + +class _TupleSubclass(tuple[object, ...]): + pass + + +class _ControlFlow(BaseException): + pass + + +def _projection( + *, + model_id: str = "model-v1", + dimensions: int = 2, +) -> ProjectionIdentity: + return ProjectionIdentity( + "schema-v1", + EmbeddingIdentity(model_id, dimensions), + ) + + +def _document( + document_id: str, + *, + corpus_id: str = "corpus", + revision_id: str = "revision-1", +) -> DocumentIdentity: + return DocumentIdentity(DocumentKey(corpus_id, document_id), revision_id) + + +def _manifest( + documents: tuple[DocumentIdentity, ...], + *, + corpus_id: str = "corpus", + projection: ProjectionIdentity | None = None, + chunking: ChunkingPolicy | None = None, + token: str | None = None, +) -> ProjectionManifest: + selected_projection = projection or _projection() + selected_chunking = chunking or ChunkingPolicy(4, 1) + entries = tuple( + sorted( + ( + ProjectionManifestEntry( + document, + "sha256:" + f"{index:064x}", + 1, + ) + for index, document in enumerate(documents, start=1) + ), + key=lambda entry: entry.document.key.document_id, + ) + ) + checkpoint_token = token or derive_projection_checkpoint_token( + corpus_id, + selected_projection, + selected_chunking, + entries, + ) + return ProjectionManifest( + corpus_id, + selected_projection, + selected_chunking, + entries, + ProjectionCheckpoint(corpus_id, selected_projection, checkpoint_token), + ) + + +def _present(manifest: ProjectionManifest) -> ProjectionStateSnapshot: + return ProjectionStateSnapshot(ProjectionStateAvailability.PRESENT, manifest) + + +def _fragment( + document: DocumentIdentity, + *, + text: str = "x", + start: int = 0, + fragment_id: str | None = None, + attributes: tuple[tuple[str, str], ...] = (), +) -> Fragment: + end = start + len(text) + identity = FragmentIdentity( + document, + fragment_id or derive_fragment_id(document, start, end), + start, + end, + ) + return Fragment(identity, text, attributes) + + +def _query( + *, + corpus_id: str = "corpus", + text: str = "query", + hit_limit: int = 4, + candidate_limit: int = 8, +) -> RetrievalQuery: + return RetrievalQuery(corpus_id, text, hit_limit, candidate_limit) + + +def _limits(max_query_codepoints: int = 100) -> RetrievalLimits: + return RetrievalLimits(max_query_codepoints) + + +class _LifecycleResource: + def __init__(self, events: list[str]) -> None: + self.events = events + self.lifecycle_calls: list[str] = [] + + def __enter__(self) -> _LifecycleResource: + self.lifecycle_calls.append("enter") + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> bool: + del exc_type, exc_value, traceback + self.lifecycle_calls.append("exit") + return True + + def close(self) -> None: + self.lifecycle_calls.append("close") + + def shutdown(self) -> None: + self.lifecycle_calls.append("shutdown") + + +class _FakeEmbedder(_LifecycleResource): + def __init__( + self, + expected_identity: EmbeddingIdentity, + events: list[str], + *, + identity_value: object = _DEFAULT, + output: object = _DEFAULT, + identity_failure: BaseException | None = None, + embed_failure: BaseException | None = None, + ) -> None: + super().__init__(events) + self.expected_identity = expected_identity + self.identity_value = identity_value + self.output = output + self.identity_failure = identity_failure + self.embed_failure = embed_failure + self.identity_calls = 0 + self.embed_calls: list[tuple[str, ...]] = [] + + @property + def identity(self) -> EmbeddingIdentity: + self.identity_calls += 1 + self.events.append("identity") + if self.identity_failure is not None: + raise self.identity_failure + if self.identity_value is _DEFAULT: + return self.expected_identity + return cast(EmbeddingIdentity, self.identity_value) + + def embed(self, texts: tuple[str, ...], /) -> tuple[EmbeddingVector, ...]: + self.embed_calls.append(texts) + self.events.append("embed:" + "|".join(texts)) + if self.embed_failure is not None: + raise self.embed_failure + if self.output is _DEFAULT: + return (EmbeddingVector((1.0, 2.0)),) + return cast(tuple[EmbeddingVector, ...], self.output) + + +class _FakeVectorReader(_LifecycleResource): + def __init__( + self, + events: list[str], + output: object = (), + *, + failure: BaseException | None = None, + ) -> None: + super().__init__(events) + self.output = output + self.failure = failure + self.calls: list[tuple[RetrievalQuery, EmbeddingVector]] = [] + + def search( + self, + query: RetrievalQuery, + embedding: EmbeddingVector, + /, + ) -> tuple[Fragment, ...]: + self.calls.append((query, embedding)) + self.events.append("vector") + if self.failure is not None: + raise self.failure + return cast(tuple[Fragment, ...], self.output) + + +class _FakeLexicalRetriever(_LifecycleResource): + def __init__( + self, + events: list[str], + output: object = (), + *, + failure: BaseException | None = None, + ) -> None: + super().__init__(events) + self.output = output + self.failure = failure + self.calls: list[RetrievalQuery] = [] + + def search(self, query: RetrievalQuery, /) -> tuple[Fragment, ...]: + self.calls.append(query) + self.events.append("lexical") + if self.failure is not None: + raise self.failure + return cast(tuple[Fragment, ...], self.output) + + +def _borrow_embedder(value: _FakeEmbedder) -> Borrowed[Embedder]: + return Borrowed(cast(Embedder, value)) + + +def _borrow_vector(value: _FakeVectorReader) -> Borrowed[VectorIndexReader]: + return Borrowed(cast(VectorIndexReader, value)) + + +def _borrow_lexical(value: _FakeLexicalRetriever) -> Borrowed[LexicalRetriever]: + return Borrowed(cast(LexicalRetriever, value)) + + +def _hit_fragments(result: RetrievalResult) -> tuple[Fragment, ...]: + return tuple(hit.fragment for hit in result.hits) + + +def _identity_order(fragment: Fragment) -> tuple[str, str, str, str, int, int]: + identity = fragment.identity + return ( + identity.document.key.corpus_id, + identity.document.key.document_id, + identity.document.revision_id, + identity.fragment_id, + identity.start, + identity.end, + ) + + +def _independent_fusion( + semantic: tuple[Fragment, ...], + lexical: tuple[Fragment, ...], +) -> tuple[Fragment, ...]: + fragments: dict[FragmentIdentity, Fragment] = {} + scores: dict[FragmentIdentity, Fraction] = {} + for candidates in (semantic, lexical): + seen: set[FragmentIdentity] = set() + for rank, fragment in enumerate(candidates, start=1): + if fragment.identity in seen: + continue + seen.add(fragment.identity) + fragments.setdefault(fragment.identity, fragment) + scores[fragment.identity] = scores.get( + fragment.identity, + Fraction(), + ) + Fraction(1, 60 + rank) + identities = sorted( + scores, + key=lambda identity: ( + -scores[identity], + _identity_order(fragments[identity]), + ), + ) + return tuple(fragments[identity] for identity in identities) + + +class RetrievalPublicAndInputTests(unittest.TestCase): + def test_exports_and_function_shapes_are_exact_and_owned(self) -> None: + expected = ("retrieve_semantic", "retrieve_hybrid") + signatures = { + retrieve_semantic: ( + "query", + "state", + "limits", + "embedder", + "vector_reader", + ), + retrieve_hybrid: ( + "query", + "state", + "limits", + "embedder", + "vector_reader", + "lexical_retriever", + ), + } + + self.assertEqual(retrieval_module.__all__, expected) + for function, names in signatures.items(): + with self.subTest(function=function.__name__): + self.assertEqual(function.__module__, "generic_rag.retrieval") + self.assertFalse(inspect.iscoroutinefunction(function)) + parameters = tuple( + inspect.signature( + cast(Callable[..., object], function) + ).parameters.values() + ) + self.assertEqual(tuple(item.name for item in parameters), names) + self.assertTrue( + all( + item.kind is inspect.Parameter.POSITIONAL_ONLY + for item in parameters + ) + ) + + def test_invalid_top_level_inputs_fail_before_any_collaborator_effect(self) -> None: + document = _document("document") + state = _present(_manifest((document,))) + query = _query(hit_limit=1, candidate_limit=1) + events: list[str] = [] + embedder = _FakeEmbedder(_projection().embedding, events) + vector = _FakeVectorReader(events, (_fragment(document),)) + operations: tuple[Callable[[], object], ...] = ( + lambda: retrieve_semantic( + cast(RetrievalQuery, object()), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ), + lambda: retrieve_semantic( + query, + cast(ProjectionStateSnapshot, object()), + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ), + lambda: retrieve_semantic( + query, + state, + cast(RetrievalLimits, object()), + _borrow_embedder(embedder), + _borrow_vector(vector), + ), + lambda: retrieve_semantic( + query, + state, + _limits(), + cast(Borrowed[Embedder], object()), + _borrow_vector(vector), + ), + lambda: retrieve_semantic( + query, + state, + _limits(), + _borrow_embedder(embedder), + cast(Borrowed[VectorIndexReader], object()), + ), + lambda: retrieve_hybrid( + query, + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + cast(Borrowed[LexicalRetriever], object()), + ), + ) + + for operation in operations: + with self.subTest(operation=operation): + with self.assertRaises(ContractValidationError): + operation() + self.assertEqual(events, []) + + def test_corrupted_frozen_query_and_state_fail_before_effects(self) -> None: + document = _document("document") + query = _query(hit_limit=1, candidate_limit=1) + object.__setattr__(query, "candidate_limit", True) + state = _present(_manifest((document,))) + malformed_state = _present(_manifest((document,))) + assert malformed_state.manifest is not None + object.__setattr__(malformed_state.manifest, "entries", []) + + for malformed_query, malformed_snapshot in ( + (query, state), + (_query(hit_limit=1, candidate_limit=1), malformed_state), + ): + events: list[str] = [] + embedder = _FakeEmbedder(_projection().embedding, events) + vector = _FakeVectorReader(events) + with self.subTest(query=malformed_query, state=malformed_snapshot): + with self.assertRaises(ContractValidationError): + retrieve_semantic( + malformed_query, + malformed_snapshot, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + self.assertEqual(events, []) + + def test_query_codepoint_cap_is_independent_and_checked_before_effects( + self, + ) -> None: + document = _document("document") + fragment = _fragment(document) + state = _present(_manifest((document,))) + exact_text = "😀e\u0301\0" + self.assertEqual(len(exact_text), 4) + + events: list[str] = [] + embedder = _FakeEmbedder(_projection().embedding, events) + vector = _FakeVectorReader(events, (fragment,)) + result = retrieve_semantic( + _query(text=exact_text, hit_limit=1, candidate_limit=1), + state, + RetrievalLimits(4), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + self.assertIs(result.outcome, RetrievalOutcome.COMPLETE) + self.assertEqual(embedder.embed_calls, [(exact_text,)]) + + blocked_events: list[str] = [] + blocked_embedder = _FakeEmbedder(_projection().embedding, blocked_events) + blocked_vector = _FakeVectorReader(blocked_events, (fragment,)) + with self.assertRaises(ContractValidationError): + retrieve_semantic( + _query(text=exact_text + "Z", hit_limit=1, candidate_limit=1), + state, + RetrievalLimits(4), + _borrow_embedder(blocked_embedder), + _borrow_vector(blocked_vector), + ) + self.assertEqual(blocked_events, []) + + +class RetrievalStateAndEffectTests(unittest.TestCase): + def test_terminal_state_outcomes_never_touch_collaborators(self) -> None: + document = _document("document") + valid = _manifest((document,)) + invalid_checkpoint = ProjectionManifest( + valid.corpus_id, + valid.projection, + valid.chunking, + valid.entries, + ProjectionCheckpoint(valid.corpus_id, valid.projection, "wrong-token"), + ) + other_document = _document("document", corpus_id="other") + cases = ( + ( + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + RetrievalOutcome.UNAVAILABLE, + ), + ( + ProjectionStateSnapshot(ProjectionStateAvailability.CORRUPT, None), + RetrievalOutcome.FAILED, + ), + ( + _present(_manifest((other_document,), corpus_id="other")), + RetrievalOutcome.UNAVAILABLE, + ), + (_present(invalid_checkpoint), RetrievalOutcome.FAILED), + (_present(_manifest(())), RetrievalOutcome.COMPLETE), + ) + + for state, expected in cases: + events: list[str] = [] + embedder = _FakeEmbedder( + _projection().embedding, + events, + identity_failure=AssertionError("must not run"), + ) + vector = _FakeVectorReader( + events, + failure=AssertionError("must not run"), + ) + lexical = _FakeLexicalRetriever( + events, + failure=AssertionError("must not run"), + ) + with self.subTest(outcome=expected): + result = retrieve_hybrid( + _query(), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + _borrow_lexical(lexical), + ) + self.assertIs(result.outcome, expected) + self.assertEqual(result.hits, ()) + self.assertFalse(result.truncated) + self.assertEqual(events, []) + + def test_semantic_and_hybrid_effect_order_and_call_bounds_are_exact(self) -> None: + semantic_document = _document("semantic") + lexical_document = _document("lexical") + state = _present(_manifest((semantic_document, lexical_document))) + query = _query(hit_limit=2, candidate_limit=2) + + semantic_events: list[str] = [] + semantic_embedder = _FakeEmbedder(_projection().embedding, semantic_events) + semantic_vector = _FakeVectorReader( + semantic_events, + (_fragment(semantic_document),), + ) + semantic_result = retrieve_semantic( + query, + state, + _limits(), + _borrow_embedder(semantic_embedder), + _borrow_vector(semantic_vector), + ) + self.assertEqual(semantic_events, ["identity", "embed:query", "vector"]) + self.assertEqual(semantic_embedder.identity_calls, 1) + self.assertEqual(semantic_embedder.embed_calls, [(query.text,)]) + self.assertEqual(len(semantic_vector.calls), 1) + self.assertEqual(semantic_vector.calls[0][0], query) + self.assertEqual(semantic_vector.calls[0][1], EmbeddingVector((1.0, 2.0))) + self.assertIs(semantic_result.outcome, RetrievalOutcome.COMPLETE) + + hybrid_events: list[str] = [] + hybrid_embedder = _FakeEmbedder(_projection().embedding, hybrid_events) + hybrid_vector = _FakeVectorReader( + hybrid_events, + (_fragment(semantic_document),), + ) + hybrid_lexical = _FakeLexicalRetriever( + hybrid_events, + (_fragment(lexical_document),), + ) + hybrid_result = retrieve_hybrid( + query, + state, + _limits(), + _borrow_embedder(hybrid_embedder), + _borrow_vector(hybrid_vector), + _borrow_lexical(hybrid_lexical), + ) + self.assertEqual( + hybrid_events, + ["identity", "embed:query", "vector", "lexical"], + ) + self.assertEqual(hybrid_embedder.identity_calls, 1) + self.assertEqual(hybrid_embedder.embed_calls, [(query.text,)]) + self.assertEqual(len(hybrid_vector.calls), 1) + self.assertEqual(hybrid_lexical.calls, [query]) + self.assertIs(hybrid_result.outcome, RetrievalOutcome.COMPLETE) + self.assertEqual(tuple(hit.rank for hit in hybrid_result.hits), (1, 2)) + + +class RetrievalProviderValidationTests(unittest.TestCase): + def test_embedder_identity_and_vector_outputs_are_strictly_validated(self) -> None: + document = _document("document") + state = _present(_manifest((document,))) + malformed_identity = EmbeddingIdentity("model-v1", 2) + object.__setattr__(malformed_identity, "dimensions", True) + wrong_dimensions = EmbeddingVector((1.0,)) + noncanonical_number = EmbeddingVector((1.0, 2.0)) + object.__setattr__(noncanonical_number, "values", (1, 2.0)) + nonfinite = EmbeddingVector((1.0, 2.0)) + object.__setattr__(nonfinite, "values", (float("nan"), 2.0)) + malformed_container = EmbeddingVector((1.0, 2.0)) + object.__setattr__(malformed_container, "values", [1.0, 2.0]) + cases = ( + ("identity type", object(), _DEFAULT), + ("identity shape", malformed_identity, _DEFAULT), + ("output list", _DEFAULT, [EmbeddingVector((1.0, 2.0))]), + ("output empty", _DEFAULT, ()), + ( + "output count", + _DEFAULT, + (EmbeddingVector((1.0, 2.0)), EmbeddingVector((3.0, 4.0))), + ), + ("vector type", _DEFAULT, (object(),)), + ("dimensions", _DEFAULT, (wrong_dimensions,)), + ("coordinate type", _DEFAULT, (noncanonical_number,)), + ("nonfinite", _DEFAULT, (nonfinite,)), + ("values container", _DEFAULT, (malformed_container,)), + ) + + for label, identity_value, output in cases: + events: list[str] = [] + embedder = _FakeEmbedder( + _projection().embedding, + events, + identity_value=identity_value, + output=output, + ) + vector = _FakeVectorReader(events, (_fragment(document),)) + with self.subTest(case=label): + result = retrieve_semantic( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + self.assertIs(result.outcome, RetrievalOutcome.FAILED) + self.assertEqual(result.hits, ()) + self.assertFalse(result.truncated) + self.assertLessEqual(embedder.identity_calls, 1) + self.assertLessEqual(len(embedder.embed_calls), 1) + self.assertEqual(vector.calls, []) + + def test_vector_and_lexical_candidate_outputs_share_strict_validation(self) -> None: + documents = tuple(_document(name) for name in ("a", "b", "c")) + state = _present(_manifest(documents)) + valid = tuple(_fragment(document) for document in documents) + wrong_corpus = _fragment(_document("wrong", corpus_id="other")) + too_wide = _fragment(documents[0], text="abcde") + wrong_id = _fragment(documents[0], fragment_id="wrong-fragment-id") + malformed_outputs: tuple[tuple[str, object], ...] = ( + ("list", list(valid[:1])), + ("tuple subclass", _TupleSubclass(valid[:1])), + ("candidate count", valid), + ("candidate type", (object(),)), + ("wrong corpus", (wrong_corpus,)), + ("fragment width", (too_wide,)), + ("fragment id", (wrong_id,)), + ) + query = _query(hit_limit=2, candidate_limit=2) + + for pathway in ("vector", "lexical"): + for label, malformed in malformed_outputs: + events: list[str] = [] + embedder = _FakeEmbedder(_projection().embedding, events) + vector = _FakeVectorReader( + events, + malformed if pathway == "vector" else (), + ) + lexical = _FakeLexicalRetriever( + events, + malformed if pathway == "lexical" else (), + ) + with self.subTest(pathway=pathway, case=label): + if pathway == "vector": + result = retrieve_semantic( + query, + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + else: + result = retrieve_hybrid( + query, + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + _borrow_lexical(lexical), + ) + self.assertIs(result.outcome, RetrievalOutcome.FAILED) + self.assertEqual(result.hits, ()) + self.assertFalse(result.truncated) + if pathway == "vector": + self.assertEqual(len(vector.calls), 1) + self.assertEqual(lexical.calls, []) + else: + self.assertEqual(len(vector.calls), 1) + self.assertEqual(len(lexical.calls), 1) + + def test_revision_and_document_membership_staleness_preserve_current_hits( + self, + ) -> None: + published_a = _document("a", revision_id="current") + published_c = _document("c", revision_id="current") + stale_revision = _fragment(_document("a", revision_id="old")) + absent_document = _fragment(_document("b", revision_id="current")) + current = _fragment(published_c) + state = _present(_manifest((published_a, published_c))) + events: list[str] = [] + embedder = _FakeEmbedder(_projection().embedding, events) + vector = _FakeVectorReader( + events, + (stale_revision, absent_document, current), + ) + + result = retrieve_semantic( + _query(hit_limit=2, candidate_limit=3), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + + self.assertIs(result.outcome, RetrievalOutcome.PARTIAL) + self.assertEqual(_hit_fragments(result), (current,)) + self.assertEqual(result.hits[0], RetrievalHit(current, 1)) + self.assertFalse(result.truncated) + + def test_embedder_identity_mismatch_is_stale_without_embedding_or_vector_read( + self, + ) -> None: + document = _document("document") + state = _present(_manifest((document,))) + events: list[str] = [] + embedder = _FakeEmbedder( + _projection().embedding, + events, + identity_value=EmbeddingIdentity("other-model", 2), + ) + vector = _FakeVectorReader(events, (_fragment(document),)) + + result = retrieve_semantic( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + + self.assertIs(result.outcome, RetrievalOutcome.STALE) + self.assertEqual(events, ["identity"]) + self.assertEqual(embedder.embed_calls, []) + self.assertEqual(vector.calls, []) + + +class RetrievalDeduplicationAndFusionTests(unittest.TestCase): + def test_identical_provider_duplicates_are_deduplicated_without_truncation( + self, + ) -> None: + documents = (_document("a"), _document("b")) + first, second = tuple(_fragment(document) for document in documents) + events: list[str] = [] + result = retrieve_semantic( + _query(hit_limit=2, candidate_limit=3), + _present(_manifest(documents)), + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, events)), + _borrow_vector(_FakeVectorReader(events, (first, first, second))), + ) + + self.assertIs(result.outcome, RetrievalOutcome.COMPLETE) + self.assertEqual(_hit_fragments(result), (first, second)) + self.assertFalse(result.truncated) + + def test_within_and_cross_provider_payload_conflicts_fail_content_free( + self, + ) -> None: + document = _document("document") + original = _fragment(document, attributes=(("variant", "one"),)) + conflict = _fragment(document, attributes=(("sentinel-secret", "two"),)) + state = _present(_manifest((document,))) + + events: list[str] = [] + within = retrieve_semantic( + _query(hit_limit=2, candidate_limit=2), + state, + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, events)), + _borrow_vector(_FakeVectorReader(events, (original, conflict))), + ) + self.assertIs(within.outcome, RetrievalOutcome.FAILED) + self.assertEqual(within.hits, ()) + self.assertNotIn("sentinel-secret", repr(within)) + + cross_events: list[str] = [] + cross = retrieve_hybrid( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, cross_events)), + _borrow_vector(_FakeVectorReader(cross_events, (original,))), + _borrow_lexical(_FakeLexicalRetriever(cross_events, (conflict,))), + ) + self.assertIs(cross.outcome, RetrievalOutcome.FAILED) + self.assertEqual(cross.hits, ()) + self.assertFalse(cross.truncated) + self.assertNotIn("sentinel-secret", repr(cross)) + + def test_exact_cross_provider_duplicate_is_one_fused_hit(self) -> None: + document = _document("document") + fragment = _fragment(document) + events: list[str] = [] + result = retrieve_hybrid( + _query(hit_limit=1, candidate_limit=1), + _present(_manifest((document,))), + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, events)), + _borrow_vector(_FakeVectorReader(events, (fragment,))), + _borrow_lexical(_FakeLexicalRetriever(events, (fragment,))), + ) + + self.assertIs(result.outcome, RetrievalOutcome.COMPLETE) + self.assertEqual(_hit_fragments(result), (fragment,)) + self.assertFalse(result.truncated) + + def test_duplicate_gaps_retain_original_provider_ranks(self) -> None: + documents = tuple(_document(name) for name in ("a", "b", "c", "d")) + by_id = { + document.key.document_id: _fragment(document) for document in documents + } + semantic = (by_id["d"], by_id["d"], by_id["b"]) + lexical = (by_id["a"], by_id["c"]) + events: list[str] = [] + result = retrieve_hybrid( + _query(hit_limit=4, candidate_limit=4), + _present(_manifest(documents)), + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, events)), + _borrow_vector(_FakeVectorReader(events, semantic)), + _borrow_lexical(_FakeLexicalRetriever(events, lexical)), + ) + + self.assertEqual(_hit_fragments(result), _independent_fusion(semantic, lexical)) + document_ids = tuple( + hit.fragment.identity.document.key.document_id for hit in result.hits + ) + self.assertLess(document_ids.index("c"), document_ids.index("b")) + + def test_rrf_uses_exact_fraction_offset_sixty_witness(self) -> None: + semantic_ids = ( + "A", + "s02", + "s03", + "s04", + "B", + "s06", + "s07", + "s08", + "s09", + "s10", + "s11", + "s12", + "s13", + ) + lexical_ids = ( + "l01", + "l02", + "l03", + "l04", + "l05", + "l06", + "l07", + "B", + "l09", + "l10", + "l11", + "l12", + "A", + ) + all_ids = tuple(sorted(set((*semantic_ids, *lexical_ids)))) + documents = tuple(_document(document_id) for document_id in all_ids) + fragments = { + document.key.document_id: _fragment(document) for document in documents + } + semantic = tuple(fragments[document_id] for document_id in semantic_ids) + lexical = tuple(fragments[document_id] for document_id in lexical_ids) + events: list[str] = [] + result = retrieve_hybrid( + _query(hit_limit=13, candidate_limit=13), + _present(_manifest(documents)), + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, events)), + _borrow_vector(_FakeVectorReader(events, semantic)), + _borrow_lexical(_FakeLexicalRetriever(events, lexical)), + ) + + expected = _independent_fusion(semantic, lexical) + self.assertEqual(_hit_fragments(result), expected[:13]) + self.assertEqual( + tuple( + hit.fragment.identity.document.key.document_id + for hit in result.hits[:2] + ), + ("A", "B"), + ) + self.assertGreater( + Fraction(1, 60 + 1) + Fraction(1, 60 + 13), + Fraction(1, 60 + 5) + Fraction(1, 60 + 8), + ) + self.assertLess( + Fraction(1, 61 + 1) + Fraction(1, 61 + 13), + Fraction(1, 61 + 5) + Fraction(1, 61 + 8), + ) + self.assertTrue(result.truncated) + + def test_unicode_ties_and_branch_permutation_use_identity_order(self) -> None: + decomposed = _document("e\u0301") + composed = _document("é") + first = _fragment(decomposed) + second = _fragment(composed) + state = _present(_manifest((decomposed, composed))) + + outputs: list[tuple[Fragment, ...]] = [] + for semantic, lexical in (((first,), (second,)), ((second,), (first,))): + events: list[str] = [] + result = retrieve_hybrid( + _query(hit_limit=2, candidate_limit=2), + state, + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, events)), + _borrow_vector(_FakeVectorReader(events, semantic)), + _borrow_lexical(_FakeLexicalRetriever(events, lexical)), + ) + outputs.append(_hit_fragments(result)) + + self.assertEqual(outputs, [(first, second), (first, second)]) + + def test_all_small_provider_permutations_match_fraction_oracle(self) -> None: + documents = tuple(_document(name) for name in ("a", "b", "c")) + fragments = tuple(_fragment(document) for document in documents) + state = _present(_manifest(documents)) + + for semantic in itertools.permutations(fragments): + for lexical in itertools.permutations(fragments): + events: list[str] = [] + with self.subTest( + semantic=tuple( + item.identity.document.key.document_id for item in semantic + ), + lexical=tuple( + item.identity.document.key.document_id for item in lexical + ), + ): + result = retrieve_hybrid( + _query(hit_limit=3, candidate_limit=3), + state, + _limits(), + _borrow_embedder( + _FakeEmbedder(_projection().embedding, events) + ), + _borrow_vector(_FakeVectorReader(events, semantic)), + _borrow_lexical(_FakeLexicalRetriever(events, lexical)), + ) + self.assertEqual( + _hit_fragments(result), + _independent_fusion(semantic, lexical), + ) + + def test_truncation_reflects_post_validation_unique_current_candidates( + self, + ) -> None: + documents = tuple(_document(name) for name in ("a", "b", "c")) + current = tuple(_fragment(document) for document in documents) + + events: list[str] = [] + truncated = retrieve_semantic( + _query(hit_limit=2, candidate_limit=3), + _present(_manifest(documents)), + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, events)), + _borrow_vector(_FakeVectorReader(events, current)), + ) + self.assertEqual(len(truncated.hits), 2) + self.assertTrue(truncated.truncated) + + stale = _fragment(_document("b", revision_id="old")) + filtered_events: list[str] = [] + filtered = retrieve_semantic( + _query(hit_limit=1, candidate_limit=3), + _present(_manifest(documents[:2])), + _limits(), + _borrow_embedder(_FakeEmbedder(_projection().embedding, filtered_events)), + _borrow_vector( + _FakeVectorReader(filtered_events, (current[0], current[0], stale)) + ), + ) + self.assertIs(filtered.outcome, RetrievalOutcome.PARTIAL) + self.assertEqual(_hit_fragments(filtered), (current[0],)) + self.assertFalse(filtered.truncated) + + +class RetrievalFailureAndLifecycleTests(unittest.TestCase): + def assert_no_lifecycle(self, *resources: _LifecycleResource) -> None: + for resource in resources: + self.assertEqual(resource.lifecycle_calls, []) + + def test_failure_and_stale_precedence_is_truthful(self) -> None: + document = _document("document") + fragment = _fragment(document) + state = _present(_manifest((document,))) + wrong_identity = EmbeddingIdentity("other-model", 2) + cases = ( + (wrong_identity, None, (), RetrievalOutcome.STALE), + ( + wrong_identity, + RuntimeError("lexical sentinel"), + (), + RetrievalOutcome.FAILED, + ), + (wrong_identity, None, (fragment,), RetrievalOutcome.PARTIAL), + ( + RuntimeError("identity sentinel"), + None, + (fragment,), + RetrievalOutcome.PARTIAL, + ), + (_DEFAULT, None, (), RetrievalOutcome.COMPLETE), + ) + + for semantic_state, lexical_failure, lexical_output, expected in cases: + events: list[str] = [] + embedder = _FakeEmbedder( + _projection().embedding, + events, + identity_value=( + semantic_state + if isinstance(semantic_state, EmbeddingIdentity) + else _DEFAULT + ), + identity_failure=( + semantic_state + if isinstance(semantic_state, BaseException) + else None + ), + ) + vector = _FakeVectorReader(events) + lexical = _FakeLexicalRetriever( + events, + lexical_output, + failure=lexical_failure, + ) + with self.subTest(expected=expected, semantic=semantic_state): + result = retrieve_hybrid( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + _borrow_lexical(lexical), + ) + self.assertIs(result.outcome, expected) + self.assertEqual( + bool(result.hits), expected is RetrievalOutcome.PARTIAL + ) + self.assertNotIn("sentinel", repr(result)) + + def test_ordinary_exceptions_are_content_free_and_never_retried(self) -> None: + document = _document("document") + state = _present(_manifest((document,))) + stages = ("identity", "embedding", "vector", "lexical") + expected_calls = { + "identity": (1, 0, 0, 0), + "embedding": (1, 1, 0, 0), + "vector": (1, 1, 1, 0), + "lexical": (1, 1, 1, 1), + } + + for stage in stages: + failure = RuntimeError(f"secret-{stage}-sentinel") + events: list[str] = [] + embedder = _FakeEmbedder( + _projection().embedding, + events, + identity_failure=failure if stage == "identity" else None, + embed_failure=failure if stage == "embedding" else None, + ) + vector = _FakeVectorReader( + events, + failure=failure if stage == "vector" else None, + ) + lexical = _FakeLexicalRetriever( + events, + failure=failure if stage == "lexical" else None, + ) + with self.subTest(stage=stage): + if stage == "lexical": + result = retrieve_hybrid( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + _borrow_lexical(lexical), + ) + else: + result = retrieve_semantic( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + self.assertIs(result.outcome, RetrievalOutcome.FAILED) + self.assertEqual(result.hits, ()) + self.assertFalse(result.truncated) + self.assertNotIn(str(failure), repr(result)) + self.assertEqual( + ( + embedder.identity_calls, + len(embedder.embed_calls), + len(vector.calls), + len(lexical.calls), + ), + expected_calls[stage], + ) + self.assert_no_lifecycle(embedder, vector, lexical) + + def test_base_exceptions_propagate_exactly_once_without_lifecycle_actions( + self, + ) -> None: + document = _document("document") + state = _present(_manifest((document,))) + expected_calls = { + "identity": (1, 0, 0, 0), + "embedding": (1, 1, 0, 0), + "vector": (1, 1, 1, 0), + "lexical": (1, 1, 1, 1), + } + for stage in ("identity", "embedding", "vector", "lexical"): + failure = _ControlFlow(f"control-{stage}") + events: list[str] = [] + embedder = _FakeEmbedder( + _projection().embedding, + events, + identity_failure=failure if stage == "identity" else None, + embed_failure=failure if stage == "embedding" else None, + ) + vector = _FakeVectorReader( + events, + failure=failure if stage == "vector" else None, + ) + lexical = _FakeLexicalRetriever( + events, + failure=failure if stage == "lexical" else None, + ) + with self.subTest(stage=stage): + with self.assertRaises(_ControlFlow) as raised: + if stage == "lexical": + retrieve_hybrid( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + _borrow_lexical(lexical), + ) + else: + retrieve_semantic( + _query(hit_limit=1, candidate_limit=1), + state, + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + ) + self.assertIs(raised.exception, failure) + self.assertEqual( + ( + embedder.identity_calls, + len(embedder.embed_calls), + len(vector.calls), + len(lexical.calls), + ), + expected_calls[stage], + ) + self.assert_no_lifecycle(embedder, vector, lexical) + + def test_success_never_enters_closes_or_shuts_down_borrowed_resources(self) -> None: + document = _document("document") + fragment = _fragment(document) + events: list[str] = [] + embedder = _FakeEmbedder(_projection().embedding, events) + vector = _FakeVectorReader(events, (fragment,)) + lexical = _FakeLexicalRetriever(events, (fragment,)) + + result = retrieve_hybrid( + _query(hit_limit=1, candidate_limit=1), + _present(_manifest((document,))), + _limits(), + _borrow_embedder(embedder), + _borrow_vector(vector), + _borrow_lexical(lexical), + ) + + self.assertIs(result.outcome, RetrievalOutcome.COMPLETE) + self.assert_no_lifecycle(embedder, vector, lexical) + + +class RetrievalProcessDeterminismTests(unittest.TestCase): + def test_hash_seed_does_not_change_hybrid_rank_order(self) -> None: + script = textwrap.dedent( + f""" + import json + import sys + sys.path.insert(0, {os.fspath(_SOURCE_ROOT)!r}) + from generic_rag.contracts import ( + ChunkingPolicy, DocumentIdentity, DocumentKey, + EmbeddingIdentity, EmbeddingVector, Fragment, FragmentIdentity, + ProjectionCheckpoint, ProjectionIdentity, ProjectionManifest, + ProjectionManifestEntry, ProjectionStateAvailability, + ProjectionStateSnapshot, RetrievalLimits, RetrievalQuery, + ) + from generic_rag.ports import Borrowed + from generic_rag.projection_integrity import ( + derive_fragment_id, derive_projection_checkpoint_token, + ) + from generic_rag.retrieval import retrieve_hybrid + + projection = ProjectionIdentity( + 'schema-v1', EmbeddingIdentity('model-v1', 2) + ) + chunking = ChunkingPolicy(4, 1) + documents = tuple( + DocumentIdentity(DocumentKey('corpus', name), 'revision-1') + for name in ('e\\u0301', 'é', '한') + ) + entries = tuple( + ProjectionManifestEntry( + document, 'sha256:' + format(index, '064x'), 1 + ) + for index, document in enumerate(documents, start=1) + ) + token = derive_projection_checkpoint_token( + 'corpus', projection, chunking, entries + ) + manifest = ProjectionManifest( + 'corpus', projection, chunking, entries, + ProjectionCheckpoint('corpus', projection, token), + ) + fragments = tuple( + Fragment( + FragmentIdentity( + document, derive_fragment_id(document, 0, 1), 0, 1 + ), + 'x', + ) + for document in documents + ) + + class E: + identity = projection.embedding + def embed(self, texts, /): + return (EmbeddingVector((1.0, 2.0)),) + class V: + def search(self, query, embedding, /): + return (fragments[2], fragments[0], fragments[1]) + class L: + def search(self, query, /): + return (fragments[1], fragments[0], fragments[2]) + + result = retrieve_hybrid( + RetrievalQuery('corpus', 'query', 3, 3), + ProjectionStateSnapshot(ProjectionStateAvailability.PRESENT, manifest), + RetrievalLimits(100), Borrowed(E()), Borrowed(V()), Borrowed(L()), + ) + print(json.dumps([ + hit.fragment.identity.document.key.document_id + for hit in result.hits + ], ensure_ascii=False)) + """ + ) + outputs: list[list[str]] = [] + for seed in ("1", "17", "987654"): + environment = dict(os.environ) + environment["PYTHONHASHSEED"] = seed + completed = subprocess.run( + (sys.executable, "-B", "-c", script), + env=environment, + check=False, + capture_output=True, + text=True, + timeout=10, + ) + self.assertEqual(completed.returncode, 0, completed.stderr) + outputs.append(cast(list[str], json.loads(completed.stdout))) + + self.assertEqual(outputs[0], outputs[1]) + self.assertEqual(outputs[1], outputs[2]) + self.assertEqual(outputs[0], ["é", "한", "e\u0301"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_retrieval_contracts.py b/tests/test_retrieval_contracts.py index 1798385..2fe397c 100644 --- a/tests/test_retrieval_contracts.py +++ b/tests/test_retrieval_contracts.py @@ -13,6 +13,7 @@ Fragment, FragmentIdentity, RetrievalHit, + RetrievalLimits, RetrievalOutcome, RetrievalQuery, RetrievalResult, @@ -71,6 +72,10 @@ def _query( class RetrievalContractTests(unittest.TestCase): def test_retrieval_fields_are_exact_frozen_and_slotted(self) -> None: + self.assertEqual( + tuple(field.name for field in fields(RetrievalLimits)), + ("max_query_codepoints",), + ) self.assertEqual( tuple(field.name for field in fields(RetrievalQuery)), ("corpus_id", "text", "hit_limit", "candidate_limit"), @@ -84,6 +89,7 @@ def test_retrieval_fields_are_exact_frozen_and_slotted(self) -> None: ("query", "outcome", "hits", "truncated"), ) + limits = RetrievalLimits(100) query = _query() hit = RetrievalHit(_fragment("fragment"), 1) result = RetrievalResult( @@ -93,6 +99,7 @@ def test_retrieval_fields_are_exact_frozen_and_slotted(self) -> None: False, ) for instance, field_name in ( + (limits, "max_query_codepoints"), (query, "text"), (hit, "rank"), (result, "truncated"), @@ -103,6 +110,20 @@ def test_retrieval_fields_are_exact_frozen_and_slotted(self) -> None: setattr(instance, field_name, object()) self.assertIsInstance(hash(instance), int) + def test_retrieval_limits_are_independent_positive_exact_integers(self) -> None: + limits = RetrievalLimits(max_query_codepoints=7) + + self.assertEqual(limits.max_query_codepoints, 7) + self.assertNotIn("hit_limit", inspect.signature(RetrievalLimits).parameters) + self.assertNotIn( + "candidate_limit", + inspect.signature(RetrievalLimits).parameters, + ) + for invalid in (0, -1, True, 1.0, _IntegerSubclass(1)): + with self.subTest(value=invalid): + with self.assertRaises(ContractValidationError): + RetrievalLimits(cast(int, invalid)) + def test_retrieval_outcomes_are_exact_closed_string_enums(self) -> None: self.assertEqual( tuple((member.name, member.value) for member in RetrievalOutcome),