From d30276dc67e460d488886e552f94a0733fee7ae0 Mon Sep 17 00:00:00 2001 From: rica-v3 Date: Tue, 25 Aug 2026 17:32:04 +0900 Subject: [PATCH 1/2] Add deterministic RAG projection lifecycle Implement bounded revision-aware projection, truthful failure and checkpoint behavior, destructive rebuild support, deterministic tests, clean import and artifact checks, and the standalone caller integration guide.\n\nRefs #3 --- .github/workflows/ci.yml | 3 +- PYTHON_MODULE_INDEX.md | 65 +- README.md | 135 ++- docs/api.md | 228 ++-- docs/lifecycle.md | 74 +- docs/projection.md | 257 +++++ docs/security-and-privacy.md | 134 ++- src/generic_rag/contracts.py | 245 ++++ src/generic_rag/ports.py | 10 + src/generic_rag/projection.py | 811 +++++++++++++ tests/support/clean_import_probe.py | 33 + tests/support/verify_artifacts.py | 1 + tests/test_package_boundaries.py | 28 + tests/test_ports.py | 32 +- tests/test_projection.py | 1661 +++++++++++++++++++++++++++ tests/test_projection_contracts.py | 545 ++++++++- 16 files changed, 4065 insertions(+), 197 deletions(-) create mode 100644 docs/projection.md create mode 100644 src/generic_rag/projection.py create mode 100644 tests/test_projection.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 329b6c2..90a7532 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -162,7 +162,8 @@ jobs: generic_rag \ generic_rag.errors \ generic_rag.contracts \ - generic_rag.ports + generic_rag.ports \ + generic_rag.projection do ( cd "$rag_probe_dir" diff --git a/PYTHON_MODULE_INDEX.md b/PYTHON_MODULE_INDEX.md index 5a4d342..08092cc 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: 4 -- Indexed production units: 4 -- Source/index parity: 4/4 +- Importable production units: 5 +- Indexed production units: 5 +- Source/index parity: 5/5 - Package data: `src/generic_rag/py.typed` - Locked verification owner: `.github/workflows/ci.yml` (supporting workflow, not an importable unit) @@ -41,9 +41,9 @@ - Owned state or external resources: none. - Material side effects: none. - Verification: `tests/test_errors.py`, `tests/test_package_boundaries.py`, - `tests/support/clean_import_probe.py`, and the locked CI import and boundary - checks. -- Documentation: `docs/api.md`. + `tests/test_projection.py`, `tests/support/clean_import_probe.py`, and the + locked CI import and boundary checks. +- Documentation: `docs/api.md` and `docs/projection.md`. ## `generic_rag.contracts` @@ -53,9 +53,12 @@ - Supported public imports: `DocumentKey`, `DocumentIdentity`, `Document`, `FragmentIdentity`, `Fragment`, `EmbeddingIdentity`, `EmbeddingVector`, `VectorRecord`, `ProjectionIdentity`, `ProjectionCheckpoint`, - `ProjectionOutcome`, `ProjectionReceipt`, `RetrievalQuery`, - `RetrievalOutcome`, `RetrievalHit`, and `RetrievalResult` from - `generic_rag.contracts`. + `ProjectionOutcome`, `ProjectionReceipt`, `ChunkingPolicy`, + `ProjectionLimits`, `ProjectionRequest`, `ProjectionManifestEntry`, + `ProjectionManifest`, `ProjectionStateAvailability`, + `ProjectionStateSnapshot`, `ProjectionStateStatus`, `ProjectionResult`, + `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`. @@ -63,11 +66,12 @@ values. - Material side effects: none. - Verification: `tests/test_contract_values.py`, - `tests/test_projection_contracts.py`, `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` and `docs/security-and-privacy.md`. + `tests/test_projection_contracts.py`, `tests/test_projection.py`, + `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 + `docs/security-and-privacy.md`. ## `generic_rag.ports` @@ -75,7 +79,8 @@ - Responsibility: define synchronous injected collaborator interfaces and explicit caller-owned borrowing semantics. - Supported public imports: `Borrowed`, `Embedder`, `VectorIndexWriter`, - `VectorIndexReader`, and `LexicalRetriever` from `generic_rag.ports`. + `VectorIndexResetter`, `VectorIndexReader`, and `LexicalRetriever` from + `generic_rag.ports`. - Re-exports: exactly the names in the module's `__all__`; none from the package root. - Direct internal dependencies: `generic_rag.contracts`. @@ -83,7 +88,33 @@ owns, acquires, releases, closes, or shuts down the resource. - Material side effects: none. - Verification: `tests/test_ports.py`, `tests/test_package_boundaries.py`, - `tests/support/clean_import_probe.py`, and the locked CI import and boundary + `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`. + +## `generic_rag.projection` + +- Source: `src/generic_rag/projection.py` +- Responsibility: deterministically plan and synchronously execute bounded, + revision-aware document projection against caller-supplied state. +- Supported public imports: `ProjectionFailureStage`, `ProjectionStateError`, + `ProjectionOperationError`, `project_documents`, and `rebuild_projection` + from `generic_rag.projection`. +- 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`. +- 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`. +- Material side effects: none at import time. At explicit workflow call time it + may invoke the borrowed embedder and vector writer, and full rebuild may + invoke the borrowed corpus resetter; it performs no persistence, network, + retry, acquisition, release, or lifecycle action itself. +- Verification: `tests/test_projection.py`, `tests/test_package_boundaries.py`, + `tests/support/clean_import_probe.py`, `tests/support/verify_artifacts.py`, and + the locked CI test, lint, type, build, clean-install, import, and artifact checks. -- Documentation: `docs/api.md`, `docs/lifecycle.md`, and +- Documentation: `docs/projection.md`, `docs/api.md`, `docs/lifecycle.md`, and `docs/security-and-privacy.md`. diff --git a/README.md b/README.md index 9471b69..1c7bd03 100644 --- a/README.md +++ b/README.md @@ -3,9 +3,10 @@ `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, and explicit caller-owned borrowing. +collaborator protocols, deterministic bounded projection orchestration, and +explicit caller-owned borrowing. -No projection or retrieval algorithm is implemented in 0.1.0. The package has +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. @@ -21,78 +22,113 @@ python -m pip install . The installed package has no runtime dependencies. Build and development tools are separate locked dependency groups. -## Use the contracts in application code +## Project approved documents The host application remains responsible for authorization and policy checks. -After approving a source and query, application code can construct generic -values and accept an application-owned provider through a protocol: +After approving source content, construct a complete bounded target, wrap +application-owned adapters in `Borrowed`, and call a projection workflow: ```python from generic_rag.contracts import ( + ChunkingPolicy, Document, DocumentIdentity, DocumentKey, + EmbeddingIdentity, EmbeddingVector, - RetrievalQuery, + ProjectionIdentity, + ProjectionLimits, + ProjectionRequest, + ProjectionStateAvailability, + ProjectionStateSnapshot, ) -from generic_rag.ports import Borrowed, Embedder +from generic_rag.ports import Borrowed +from generic_rag.projection import rebuild_projection -# Construct these values only after application-specific authorization. -approved_document = Document( - identity=DocumentIdentity( - key=DocumentKey(corpus_id="corpus-a", document_id="document-1"), - revision_id="revision-3", +class ExampleEmbedder: + identity = EmbeddingIdentity("example-model", 2) + + def embed(self, texts, /): + return tuple(EmbeddingVector((float(len(text)), 0.0)) for text in texts) + + +class ExampleWriter: + def replace_document(self, document, records, /): + # Replace the complete projection for this stable document key. + return None + + def delete_document(self, document, /): + return None + + +class ExampleResetter: + def reset_corpus(self, corpus_id, /): + # Remove every projected document for this corpus. + return None + + +request = ProjectionRequest( + "corpus-a", + ProjectionIdentity("schema-v1", ExampleEmbedder.identity), + ChunkingPolicy(max_fragment_codepoints=800, overlap_codepoints=80), + ProjectionLimits( + max_documents=100, + max_document_codepoints=100_000, + max_embedding_batch_size=32, + ), + ( + Document( + DocumentIdentity( + DocumentKey("corpus-a", "document-1"), + "revision-3", + ), + "Approved source text", + (("classification", "internal"),), + ), ), - text="Approved source text", - attributes=(("classification", "internal"),), -) -query = RetrievalQuery( - corpus_id=approved_document.identity.key.corpus_id, - text="What does the source say?", - hit_limit=5, - candidate_limit=20, ) +# Bootstrap and recovery are explicit and destructive: reset, then replace. +result = rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + Borrowed(ExampleEmbedder()), + Borrowed(ExampleWriter()), + Borrowed(ExampleResetter()), +) -def application_embed_query( - provider: Embedder, - request: RetrievalQuery, -) -> EmbeddingVector: - with Borrowed(provider) as embedder: - vectors = embedder.embed((request.text,)) - if len(vectors) != 1: - raise ValueError("the provider violated the Embedder contract") - return vectors[0] +# Persist result.manifest in application-owned state only after success. ``` -This is application orchestration, not a package retrieval workflow. Version -0.1.0 defines the boundary that provider implementations and later generic -workflows will use; it does not construct providers or call them on a user's -behalf. +For normal updates, load that manifest into a present +`ProjectionStateSnapshot` and call `project_documents`; it changes only added, +updated, removed, or rechunked documents. Use `rebuild_projection` only when an +explicit corpus-wide reset is intended. See the [projection guide](docs/projection.md) +for the complete lifecycle, state matrix, adapter obligations, and failure +behavior. Public values must be imported from their owning modules: - `generic_rag.contracts` - `generic_rag.errors` - `generic_rag.ports` +- `generic_rag.projection` The package root intentionally has no re-exports: `generic_rag.__all__ == ()`. See the [API reference](docs/api.md) for every supported name and invariant. -## Planned RAG flow +## Application RAG flow The package itself has no concept of a user or agent. A consuming application decides which sources a user may approve, which queries may be submitted, which provider implementations receive data, and whether retrieved fragments are shown to a user or supplied to a downstream tool or agent. -- [Issue #3](https://github.com/Kims-DeveloperGroup/generic-rag/issues/3) is - planned to add generic projection orchestration. Its intended responsibility - is to accept caller-approved documents and explicitly injected collaborators, - derive fragments under a defined chunking policy, embed ordered fragment - text, replace or delete complete document projections, and report truthful - checkpoints and receipts. Its precise API and failure behavior are not part - of 0.1.0. +- Projection accepts caller-approved documents and explicitly injected + collaborators. It derives deterministic fragments under a bounded chunking + 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 @@ -100,8 +136,11 @@ shown to a user or supplied to a downstream tool or agent. deduplication, fusion, limiting, and outcome behavior. Provider rank will be the input; raw provider scores are not represented or assumed comparable. -The caller/provider ownership model remains explicit throughout this plan. See -[resource lifecycle](docs/lifecycle.md) and +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). ## Compatibility @@ -111,9 +150,10 @@ 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 four +The distribution includes `py.typed`. The wheel contains exactly the five importable modules `generic_rag`, `generic_rag.errors`, -`generic_rag.contracts`, and `generic_rag.ports`, plus the typing marker. +`generic_rag.contracts`, `generic_rag.ports`, and `generic_rag.projection`, plus +the typing marker. ## Development verification @@ -140,6 +180,5 @@ uv run --frozen python tests/support/verify_artifacts.py "$rag_dist_dir" CI is configured to run the tests on Python 3.11 and 3.14. On Python 3.11 it also runs lint, format, strict type, compilation, artifact, source-rebuild, -clean-install, and isolated-import checks. A local Python 3.14.2 run currently -contains 53 passing tests; the CI matrix is the authoritative cross-version -result. +clean-install, and isolated-import checks. The CI matrix is the authoritative +cross-version result. diff --git a/docs/api.md b/docs/api.md index 863105e..115c259 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1,10 +1,10 @@ # Public API -Version 0.1.0 exposes immutable values, typed error categories, and synchronous -collaborator protocols. It does not expose projection or retrieval algorithms. -See the [project overview](../README.md), [resource lifecycle](lifecycle.md), -and [security and privacy boundary](security-and-privacy.md) for the surrounding -usage contract. +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. ## Import boundary @@ -33,6 +33,15 @@ public symbols. Import names from their owning modules instead. - `ProjectionCheckpoint` - `ProjectionOutcome` - `ProjectionReceipt` +- `ChunkingPolicy` +- `ProjectionLimits` +- `ProjectionRequest` +- `ProjectionManifestEntry` +- `ProjectionManifest` +- `ProjectionStateAvailability` +- `ProjectionStateSnapshot` +- `ProjectionStateStatus` +- `ProjectionResult` - `RetrievalQuery` - `RetrievalOutcome` - `RetrievalHit` @@ -43,10 +52,20 @@ public symbols. Import names from their owning modules instead. - `Borrowed` - `Embedder` - `VectorIndexWriter` +- `VectorIndexResetter` - `VectorIndexReader` - `LexicalRetriever` -The package does not support importing public values from the package root. +`generic_rag.projection` exports exactly: + +- `ProjectionFailureStage` +- `ProjectionStateError` +- `ProjectionOperationError` +- `project_documents` +- `rebuild_projection` + +The package does not support importing any of these names from the package +root. ## Shared value rules @@ -72,23 +91,41 @@ and pair order and duplicates are preserved. ## Errors -`GenericRagError` is the base for the three public categories: +The base error relationships used by projection are: ```text GenericRagError ├── ContractValidationError ├── CollaborationError +│ └── ProjectionOperationError └── StateCompatibilityError + └── ProjectionStateError ``` -- `ContractValidationError` reports a violated public value invariant. -- `CollaborationError` is reserved for a workflow that translates a - collaborator operation failure. -- `StateCompatibilityError` is reserved for a workflow that detects derived - state with an incompatible projection identity. +- `ContractValidationError` reports a violated public value invariant or an + invalid top-level workflow input. +- `ProjectionStateError(status: ProjectionStateStatus)` reports state that an + incremental operation cannot use. Its exact `status` field gives the reason; + its message contains no state identifier. +- `ProjectionOperationError(stage, affected_document, receipt)` translates an + ordinary collaborator failure or invalid collaborator return. Its message is + content-free. `affected_document` is a stable key for document-specific + failures and otherwise `None`; `receipt` is truthful `FAILED` or `PARTIAL` + progress, or `None` when there were zero document attempts. -Version 0.1.0 has no projection or retrieval workflow that raises the latter -two categories. `Borrowed` also leaves provider exceptions unchanged. +The closed string enum `ProjectionFailureStage` has exact values: + +| Member | String value | +| --- | --- | +| `EMBEDDER_IDENTITY` | `"embedder_identity"` | +| `EMBEDDING` | `"embedding"` | +| `REPLACEMENT` | `"replacement"` | +| `DELETION` | `"deletion"` | +| `RESET` | `"reset"` | + +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. ## Documents and fragments @@ -104,31 +141,65 @@ A fragment range is half-open, `[start, end)`, in Python Unicode code points. It is not measured in bytes or user-perceived grapheme clusters. For example, `"😀"` has one code point while `"e\u0301"` has two. -The package checks that fragment text length equals the range width. Version -0.1.0 does not retain an authoritative `Document` beside a `Fragment`, so it -cannot verify that the text equals the indicated source slice. The caller, or -a future projection workflow, must establish that correspondence. +The contract checks that fragment text length equals the range width. A +standalone fragment cannot prove that its text equals the indicated source +slice; projection establishes that correspondence for fragments it derives. ## Embeddings and vector records | Type | Fields | Construction rules | | --- | --- | --- | | `EmbeddingIdentity` | `model_id: str`, `dimensions: int` | The model ID is nonblank and opaque; dimensions is a positive exact integer. | -| `EmbeddingVector` | `values: tuple[float, ...]` | The tuple is exact and nonempty. Every coordinate is an exact `int` or `float`, excluding `bool`, and must convert to a finite float without overflow. Oversized integers that cannot be represented as finite floats are rejected; accepted coordinates are stored canonically as floats. | +| `EmbeddingVector` | `values: tuple[float, ...]` | The tuple is exact and nonempty. Every coordinate is an exact `int` or `float`, excluding `bool`, and must convert to a finite float without overflow. Accepted coordinates are stored canonically as floats. | | `VectorRecord` | `fragment: Fragment`, `embedding: EmbeddingVector` | Both fields require their exact contract classes. | -A standalone `EmbeddingVector` does not carry an `EmbeddingIdentity`. Version -0.1.0 therefore does not compare the vector length with an identity's declared -`dimensions`; a provider and future orchestration must satisfy that semantic -relationship. +A standalone `EmbeddingVector` does not carry an `EmbeddingIdentity` and does +not itself compare length with declared dimensions. Projection validates the +embedder identity and every returned vector's exact dimensionality and +canonical finite-float representation. -## Projection state +## Projection request and manifest values | Type | Fields | Construction rules | | --- | --- | --- | | `ProjectionIdentity` | `schema_id: str`, `embedding: EmbeddingIdentity` | The schema ID is nonblank and opaque; embedding requires its exact class. | -| `ProjectionCheckpoint` | `corpus_id: str`, `projection: ProjectionIdentity`, `token: str` | Corpus and token are nonblank opaque strings; projection requires its exact class. | -| `ProjectionReceipt` | `corpus_id: str`, `projection: ProjectionIdentity`, `outcome: ProjectionOutcome`, `attempted_documents: int`, `completed_documents: int`, `checkpoint: ProjectionCheckpoint \| None` | Nested values require their exact classes; counts are nonnegative exact integers and completed cannot exceed attempted. | +| `ChunkingPolicy` | `max_fragment_codepoints: int`, `overlap_codepoints: int` | Maximum is positive; overlap is nonnegative and smaller than maximum. | +| `ProjectionLimits` | `max_documents: int`, `max_document_codepoints: int`, `max_embedding_batch_size: int` | All three values are positive exact integers. | +| `ProjectionRequest` | `corpus_id: str`, `projection: ProjectionIdentity`, `chunking: ChunkingPolicy`, `limits: ProjectionLimits`, `documents: tuple[Document, ...]` | Documents must match the corpus, have unique stable keys, and remain within the count and per-document text caps. They are canonicalized by opaque `document_id`. | +| `ProjectionManifestEntry` | `document: DocumentIdentity`, `source_digest: str`, `fragment_count: int` | Digest must be lowercase `sha256:<64hex>` and fragment count is nonnegative. | +| `ProjectionCheckpoint` | `corpus_id: str`, `projection: ProjectionIdentity`, `token: str` | Corpus and token are nonblank; projection requires its exact class. | +| `ProjectionManifest` | `corpus_id: str`, `projection: ProjectionIdentity`, `chunking: ChunkingPolicy`, `entries: tuple[ProjectionManifestEntry, ...]`, `checkpoint: ProjectionCheckpoint` | Entries match the corpus, have unique stable keys, and are canonicalized by `document_id`; checkpoint corpus and projection match the manifest. | + +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). + +## Projection state and results + +`ProjectionStateAvailability` is a closed string enum: + +| Member | String value | Manifest rule | +| --- | --- | --- | +| `MISSING` | `"missing"` | Must be `None` | +| `PRESENT` | `"present"` | Must be an exact `ProjectionManifest` | +| `CORRUPT` | `"corrupt"` | Must be `None` | + +`ProjectionStateSnapshot(availability, manifest)` stores that caller-supplied +state. Projection evaluates it to a closed `ProjectionStateStatus`: + +| Member | String value | Meaning | +| --- | --- | --- | +| `MISSING` | `"missing"` | No state is available. | +| `CURRENT` | `"current"` | Valid state exactly matches the complete target. | +| `STALE` | `"stale"` | Valid compatible state requires mutations. | +| `CORRUPT` | `"corrupt"` | State is declared corrupt or fails integrity/consistency checks. | +| `SCHEMA_MISMATCH` | `"schema_mismatch"` | The schema ID differs. | +| `EMBEDDING_MISMATCH` | `"embedding_mismatch"` | The embedding identity differs. | + +`ProjectionResult(status_before, receipt, manifest)` represents only complete +success. Its receipt and manifest have the same corpus, projection, and +checkpoint. An `UNCHANGED` result requires `CURRENT` state and zero attempted +and completed documents. `ProjectionOutcome` is a closed string enum with these exact member values: @@ -139,8 +210,9 @@ relationship. | `PARTIAL` | `"partial"` | | `FAILED` | `"failed"` | -Unknown enum values raise `ContractValidationError`. A receipt can represent -only the following truthful combinations: +`ProjectionReceipt` has fields `corpus_id`, `projection`, `outcome`, +`attempted_documents`, `completed_documents`, and `checkpoint`. It permits only +these truthful combinations: | Outcome | Counts | Checkpoint | | --- | --- | --- | @@ -150,7 +222,43 @@ only the following truthful combinations: | `FAILED` | `attempted_documents > 0` and `completed_documents == 0` | Forbidden | Any supplied checkpoint must have exactly the receipt's `corpus_id` and -`projection`. +`projection`. A successful workflow returns `ProjectionResult`; a failed +workflow exposes a failed or partial receipt only through +`ProjectionOperationError`. + +## Projection workflows + +Both public functions are synchronous and all parameters are positional-only: + +```python +def project_documents( + request: ProjectionRequest, + state: ProjectionStateSnapshot, + embedder: Borrowed[Embedder], + writer: Borrowed[VectorIndexWriter], + /, +) -> ProjectionResult: ... + +def rebuild_projection( + request: ProjectionRequest, + state: ProjectionStateSnapshot, + embedder: Borrowed[Embedder], + writer: Borrowed[VectorIndexWriter], + resetter: Borrowed[VectorIndexResetter], + /, +) -> ProjectionResult: ... +``` + +`project_documents` returns without collaborator effects when state is +`CURRENT`, applies only the canonical delta when state is `STALE`, and raises +`ProjectionStateError` before effects for every other status. + +`rebuild_projection` accepts every state status and always calls the resetter. +For a nonempty target it verifies embedder identity before reset, then replaces +every target document. For an empty target it resets without accessing the +embedder or writer. It is intentionally destructive and supplies no rollback +or retry. See the [projection guide](projection.md) for the full state and +failure matrices. ## Retrieval values @@ -160,60 +268,46 @@ Any supplied checkpoint must have exactly the receipt's `corpus_id` and | `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`. | -`RetrievalOutcome` is a closed string enum with these exact member values: - -| Member | String value | -| --- | --- | -| `COMPLETE` | `"complete"` | -| `PARTIAL` | `"partial"` | -| `UNAVAILABLE` | `"unavailable"` | -| `STALE` | `"stale"` | -| `FAILED` | `"failed"` | - -Unknown enum values raise `ContractValidationError`. Within every result, ranks -must be contiguous starting at one, fragment identities must be unique, and -every fragment's corpus must match the query corpus. Fragment attributes do not -make two otherwise identical fragment identities distinct. - -The outcome matrix is: - -| Outcome | Hits | `truncated` | -| --- | --- | --- | -| `COMPLETE` | Zero through `query.hit_limit` | Either boolean | -| `PARTIAL` | One through `query.hit_limit` | Either boolean | -| `UNAVAILABLE` | None | `False` | -| `STALE` | None | `False` | -| `FAILED` | None | `False` | +`RetrievalOutcome` is a closed string enum with exact values `"complete"`, +`"partial"`, `"unavailable"`, `"stale"`, and `"failed"`. Within every result, +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. It does not imply that -`len(hits) == query.hit_limit`; a bounded `COMPLETE` result may therefore still -be truncated. Hits and reader ports are score-free. Raw scores from different -providers are neither represented nor promised to be comparable. +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. ## Collaborator ports The protocols are synchronous, injected, structurally typed, and decorated -with `runtime_checkable`. Runtime protocol checks establish structural presence, -not the behavioral obligations below. Version 0.1.0 provides no implementation, -adapter, factory, provider discovery, or provider-behavior enforcement. +with `runtime_checkable`. Runtime protocol checks establish structural +presence, not the behavioral obligations below. Version 0.1.0 provides no +adapter, factory, or provider discovery. | Port | Exact public operation | Semantic obligation | | --- | --- | --- | | `Embedder` | `identity: EmbeddingIdentity` | Identify the exact model used for produced vectors. | | `Embedder` | `embed(texts: tuple[str, ...], /) -> tuple[EmbeddingVector, ...]` | Return one same-order vector per input text, each with `identity.dimensions` coordinates; empty input returns an empty tuple. | -| `VectorIndexWriter` | `replace_document(document: DocumentIdentity, records: tuple[VectorRecord, ...], /) -> None` | Replace all derived vectors for the document's stable key. Every record carries the supplied full `DocumentIdentity`; an empty record tuple is valid. | +| `VectorIndexWriter` | `replace_document(document: DocumentIdentity, records: tuple[VectorRecord, ...], /) -> None` | Replace all derived vectors for the document's stable key. Every record carries the supplied full identity; an empty record tuple is valid. | | `VectorIndexWriter` | `delete_document(document: DocumentKey, /) -> None` | Delete every derived revision for the stable document key. | +| `VectorIndexResetter` | `reset_corpus(corpus_id: str, /) -> None` | Remove the complete derived vector projection for the corpus. | | `VectorIndexReader` | `search(query: RetrievalQuery, embedding: EmbeddingVector, /) -> tuple[Fragment, ...]` | Return fragments from the requested corpus, in provider rank order, with at most `query.candidate_limit` entries. | | `LexicalRetriever` | `search(query: RetrievalQuery, /) -> tuple[Fragment, ...]` | Return fragments from the requested corpus, in provider rank order, with at most `query.candidate_limit` entries. | +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. + `Borrowed[T]` is the companion ownership marker, not a provider port. Its exact behavior is documented in [resource lifecycle](lifecycle.md). -## Planned workflows +## Planned retrieval workflow -Projection orchestration is planned for Issue #3. Retrieval and composition are -planned for Issue #4. Those future workflows are expected to accept protocol- -compatible collaborators explicitly, but their algorithms, APIs, compatibility -checks, exception translation, and outcome mapping are not implemented or -promised by version 0.1.0. +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. diff --git a/docs/lifecycle.md b/docs/lifecycle.md index c362a41..bac169f 100644 --- a/docs/lifecycle.md +++ b/docs/lifecycle.md @@ -1,21 +1,30 @@ # Resource Lifecycle Version 0.1.0 uses an explicit caller-owned lifecycle. The package defines -collaborator protocols and `Borrowed[T]`; it does not acquire or own provider -resources. See the [API reference](api.md) for the exact port signatures. +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. ## Ownership rule The caller or provider integration owns every lifecycle decision: -1. Acquire and configure the resource. -2. Establish any required synchronization or exclusive access. -3. Pass the resource explicitly to application code or a future generic - workflow. -4. Reset, release, close, or shut down the resource according to the provider's - rules after use. - -`generic-rag` does none of those steps implicitly. +1. Acquire and configure the embedder, vector index, projection-state store, + 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 + `project_documents` for a compatible incremental update, or deliberately + call `rebuild_projection` for bootstrap or destructive recovery. +4. On complete success, publish the returned manifest through caller-owned + persistence and synchronization. +5. Release, close, or shut down the real resources according to their provider + contracts. + +`generic-rag` performs none of the acquisition, manifest persistence, +publication, synchronization, or release steps implicitly. It provides no +transaction across collaborators and the manifest store. ## `Borrowed[T]` @@ -31,9 +40,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. +manager or lifecycle methods. Projection itself translates ordinary method +failures into `ProjectionOperationError`; that workflow behavior does not +change `Borrowed` semantics. -## Successful scope +## Successful borrowed scope The resource that leaves the scope is still the application-owned object: @@ -50,7 +61,7 @@ assert active_provider is provider # Borrowed did not close or replace it. Application code decides when and how to release the real provider afterward. -## Exceptional scope +## Exceptional borrowed scope An exception leaves the borrowed scope unchanged and unsuppressed: @@ -68,15 +79,38 @@ except RuntimeError as caught: assert caught is failure ``` -If a caller needs cleanup after either success or failure, it must arrange that +If a caller needs cleanup after success or failure, it must arrange that cleanup around the borrowed scope according to the provider's contract. -## Future workflows +## Projection call scope + +`project_documents` and `rebuild_projection` receive exact `Borrowed` wrappers. +They enter only those no-op wrappers; they never enter, close, or shut down the +underlying adapter objects. + +The caller must keep each adapter alive for the complete synchronous call. +Providers that require sessions, transactions, locks, or thread affinity must +be prepared before wrapping and must remain valid until the call returns or +raises. + +An incremental call may perform multiple ordered document mutations. A rebuild +performs a corpus reset before its document replacements. The package does not +roll back or retry earlier effects if a later operation fails. The +`ProjectionOperationError.receipt` reports completed document operations but +never supplies a checkpoint for partial work. Keep partially updated state out +of service and recover under application-owned coordination. + +The caller should persist `ProjectionResult.manifest` only after complete +success. The package does not retain it, and a vector index without the matching +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 -Projection and retrieval orchestration are planned for Issues #3 and #4. Their -collaborators are intended to remain explicitly injected and caller-owned. The -precise future APIs are not part of version 0.1.0, and `Borrowed` must not be -read as a promise that a workflow already exists. +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. Review the [security and privacy boundary](security-and-privacy.md) before -passing content to a provider implementation. +passing content to any adapter implementation. diff --git a/docs/projection.md b/docs/projection.md new file mode 100644 index 0000000..e407b6a --- /dev/null +++ b/docs/projection.md @@ -0,0 +1,257 @@ +# Projection + +Version 0.1.0 can turn a complete, caller-approved document set into a +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). + +## Required adapters and state + +Implement application-specific objects that structurally satisfy these public +ports: + +- `Embedder` exposes an `EmbeddingIdentity` and returns one same-order vector + of exactly that dimensionality for every input text. +- `VectorIndexWriter` replaces the complete record set for one stable document + key or deletes every revision of that key. Both commands must return `None`. +- `VectorIndexResetter` removes every projected document for one corpus and + returns `None`. It is required only by `rebuild_projection`. + +The package does not provide adapters or a manifest store. Persist the +successful `ProjectionResult.manifest` in application-owned state, then load it +as a `ProjectionStateSnapshot` for the next operation. The stored manifest and +the vector index are one logical projection: publish them under application- +controlled synchronization so another operation cannot observe an unintended +combination. + +All workflow parameters are positional-only. Each collaborator must be wrapped +in the exact `Borrowed` class; borrowing never acquires, closes, resets, or +otherwise owns the wrapped resource. + +## Bootstrap, then update incrementally + +The first projection has no valid present manifest, so bootstrap it through the +explicitly destructive rebuild path: + +```python +from generic_rag.contracts import ( + ChunkingPolicy, + Document, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + EmbeddingVector, + ProjectionIdentity, + ProjectionLimits, + ProjectionRequest, + ProjectionStateAvailability, + ProjectionStateSnapshot, +) +from generic_rag.ports import Borrowed +from generic_rag.projection import project_documents, rebuild_projection + + +class ApplicationEmbedder: + identity = EmbeddingIdentity("embedding-model-v1", 2) + + def embed(self, texts, /): + return tuple(EmbeddingVector((float(len(text)), 0.0)) for text in texts) + + +class ApplicationVectorIndex: + def replace_document(self, document, records, /): + # Replace all derived records for document.key in one adapter operation. + return None + + def delete_document(self, document, /): + # Delete every projected revision of this stable document key. + return None + + def reset_corpus(self, corpus_id, /): + # Delete the complete vector projection for this corpus. + return None + + +def target(revision_id, text): + return ProjectionRequest( + "approved-corpus", + ProjectionIdentity("schema-v1", ApplicationEmbedder.identity), + ChunkingPolicy(800, 80), + ProjectionLimits(100, 100_000, 32), + ( + Document( + DocumentIdentity( + DocumentKey("approved-corpus", "document-1"), + revision_id, + ), + text, + (("source", "application-authorized"),), + ), + ), + ) + + +embedder = ApplicationEmbedder() +index = ApplicationVectorIndex() +initial_request = target("revision-1", "First approved source text") + +initial_result = rebuild_projection( + initial_request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + Borrowed(embedder), + Borrowed(index), + Borrowed(index), +) +# Persist initial_result.manifest only after the call succeeds. + +updated_request = target("revision-2", "Updated approved source text") +updated_result = project_documents( + updated_request, + ProjectionStateSnapshot( + ProjectionStateAvailability.PRESENT, + initial_result.manifest, + ), + Borrowed(embedder), + Borrowed(index), +) +# Atomically publish updated_result.manifest as the new application-owned state. +``` + +The example uses one object for the writer and resetter ports, but separate +objects are equally valid. Production adapters must implement the stated +complete replacement, deletion, reset, persistence, and synchronization +semantics; the example deliberately omits storage. + +## Incremental lifecycle + +`project_documents(request, state, embedder, writer)` applies a complete target +to compatible state: + +1. It validates the exact request, snapshot, and `Borrowed` wrapper types and + prepares the complete target before calling a collaborator. +2. It evaluates the supplied manifest against the target. `current` returns an + `UNCHANGED` result with zero attempts and no collaborator calls. Only + `stale` proceeds; every other status raises `ProjectionStateError` before + effects. +3. It sorts mutations by opaque `document_id`. Added documents, new revisions, + and documents affected by a chunking change are replaced; absent target + documents are deleted; unchanged entries are skipped. A source change under + an unchanged full document identity is corrupt state, not an incremental + replacement. +4. If any replacement is needed, it validates the embedder identity once, + embeds nonempty ordered batches no larger than + `max_embedding_batch_size`, and calls the writer once per document. A + delete-only update does not access the embedder. +5. After every mutation succeeds, it returns `COMPLETED` with the complete + target manifest and checkpoint. The caller may then publish that manifest. + +An empty document produces no fragments and is still replaced with an explicit +empty record tuple. An empty target incrementally deletes every document in the +previous valid manifest. + +## Destructive rebuild lifecycle + +`rebuild_projection(request, state, embedder, writer, resetter)` is the explicit +bootstrap and recovery operation. It accepts every state status, including +`current`, but always resets the requested corpus and recreates the complete +target: + +1. It validates inputs and computes `status_before` without effects. +2. For a nonempty target, it validates the embedder identity once *before* the + reset. A mismatch or invalid identity therefore leaves reset and writer + untouched. +3. It calls `reset_corpus`, then embeds and replaces every target document in + canonical `document_id` order. An empty target calls only the resetter. +4. Complete success returns a `COMPLETED` result, even when the target has zero + documents. + +Reset is intentionally destructive. The package provides no transaction, +rollback, retry, or two-phase publication across the resetter, writer, and +caller-owned manifest store. If reset succeeds and a later replacement fails, +the vector index can contain an incomplete rebuild and no successful +checkpoint is issued. Keep the prior manifest out of service and run an +application-controlled recovery, normally another full rebuild. + +## State matrix + +`ProjectionStateAvailability` describes what the caller could load. +`ProjectionStateStatus` is the workflow's evaluation of that snapshot against +the requested complete target. + +| Supplied state | Evaluated status | `project_documents` | `rebuild_projection` | +| --- | --- | --- | --- | +| `MISSING` with no manifest | `MISSING` | Raises before effects | Resets and builds target | +| `CORRUPT` with no manifest | `CORRUPT` | Raises before effects | Resets and builds target | +| Valid present manifest exactly matching target | `CURRENT` | Returns `UNCHANGED`; no effects | Resets and rebuilds target | +| Valid present manifest with compatible target differences | `STALE` | Applies incremental mutations | Resets and rebuilds target | +| Present manifest with invalid checkpoint, wrong corpus, or same-revision source/count inconsistency | `CORRUPT` | Raises before effects | Resets and rebuilds target | +| Present manifest with another schema ID | `SCHEMA_MISMATCH` | Raises before effects | Resets and rebuilds target | +| Present manifest with another embedding identity | `EMBEDDING_MISMATCH` | Raises before effects | Resets and rebuilds target | + +A revision change is a normal stale update. For the same full document +identity, changing the source digest—or the fragment count under unchanged +chunking—is treated as corrupt state rather than an unannounced rewrite. + +## Deterministic projection values + +Documents and manifest entries are canonicalized by opaque `document_id`. +Fragments use half-open Python Unicode code-point ranges and copy the source's +ordered attributes. Each fragment contains at most +`max_fragment_codepoints`; consecutive fragments overlap by +`overlap_codepoints`. Attribute order and duplicates remain significant. + +Source digests, fragment IDs, and checkpoint tokens use lowercase +`sha256:<64hex>` values. The current algorithms serialize tagged fields as +UTF-8 with `surrogatepass`, prefix each encoded field with its unsigned +eight-byte big-endian length, and hash them under these versioned domains: + +| Value | v1 domain | Bound inputs | +| --- | --- | --- | +| Source digest | `generic-rag:projection-source:v1` | Exact document text and ordered attributes | +| Fragment ID | `generic-rag:fragment-id:v1` | Corpus, document, revision, and fragment range | +| Checkpoint token | `generic-rag:projection-checkpoint:v1` | Corpus, projection and embedding identities, chunking policy, and ordered manifest entries | + +These values are reproducible for the same inputs and current v1 algorithm, +including across clean processes. The v1 domain names do not promise that a +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. + +Hashes are deterministic comparison and identity values, not encryption, +authorization, or a proof of source ownership. See [security and +privacy](security-and-privacy.md). + +## Failures and receipts + +Invalid public values raise `ContractValidationError`. Incremental state that +is not `current` or `stale` raises `ProjectionStateError`, whose `status` gives +the evaluated reason. Both cases are detected before collaborator effects. + +An ordinary collaborator exception, or an invalid collaborator return, raises +`ProjectionOperationError` with: + +- `stage`: `EMBEDDER_IDENTITY`, `EMBEDDING`, `REPLACEMENT`, `DELETION`, or + `RESET`; +- `affected_document`: the stable key for a document-specific failure, else + `None`; and +- `receipt`: `FAILED` when no planned document completed, `PARTIAL` after one + or more but not all planned documents completed, or `None` when there were + zero document attempts. + +Failure receipts never contain a checkpoint. `attempted_documents` is the +total mutation or rebuild-document count; `completed_documents` counts only +fully completed document operations. The original ordinary exception is +chained as the cause. A structurally invalid identity, vector result, or +non-`None` writer/resetter return has no internal cause. `KeyboardInterrupt` +and `SystemExit` are neither translated nor retried. + +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. diff --git a/docs/security-and-privacy.md b/docs/security-and-privacy.md index 7147817..91abad7 100644 --- a/docs/security-and-privacy.md +++ b/docs/security-and-privacy.md @@ -1,64 +1,114 @@ # Security and Privacy -Version 0.1.0 defines in-process values and collaborator boundaries. By itself, -the package performs no persistence, network transmission, provider discovery, -credential loading, telemetry, or background work. Contract objects do retain -caller-supplied values in process memory. +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. ## Caller responsibility -Before constructing a `Document`, `Fragment`, or `RetrievalQuery`, the caller -must perform its own authorization and policy checks. The caller also controls: +Before constructing a `ProjectionRequest`, the caller must authorize every +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. -- which source content and metadata enter the contracts; -- which provider implementations receive document text, fragment text, - embeddings, attributes, identifiers, or queries; +The caller also controls: + +- corpus and tenant isolation; +- which adapter implementations receive document text, fragment text, + embeddings, attributes, identifiers, or future queries; - provider account, region, transport, and credential configuration; -- retention, replacement, deletion, backup, and recovery behavior; -- logging, tracing, metrics, redaction, and incident response; and -- whether retrieved fragments are displayed, persisted, or supplied to another - tool or agent. +- 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. -Do not place credentials or other secrets in attributes, opaque identifiers, or -checkpoint tokens. These fields deliberately preserve caller input and do not -apply redaction, escaping, access control, or tenant isolation. +Do not place credentials or other secrets in attributes, opaque identifiers, +or checkpoint tokens. These fields preserve caller input and do not apply +redaction, escaping, authorization, or tenant isolation. -## Provider effects +## Adapter effects -The protocols describe operations that an external implementation may perform. -Calling an embedder, vector index, or lexical retriever can store or transmit -data according to that implementation. Version 0.1.0 supplies no such provider -and does not call one on the caller's behalf. +Calling `project_documents` can invoke the supplied embedder and writer. +Calling `rebuild_projection` can additionally reset all projected data for the +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. -`Borrowed` does not reduce this responsibility. It only marks the wrapped -resource as caller-owned and does not acquire, close, reset, authenticate, or -synchronize it. See [resource lifecycle](lifecycle.md). +`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 +caller-owned manifest store. See [resource lifecycle](lifecycle.md). -## Sensitive derived data +## Sensitive source and derived data Treat all of the following as potentially sensitive: -- document and query text; +- document and future query text; - ordered attributes and opaque identities; - fragments and their source ranges; -- embeddings and vector records; and -- projection checkpoint tokens. - -Fragments and embeddings are derived data but may reveal information from the -source. Deleting an authoritative source does not automatically delete copies -or derived values held by an application or provider. +- embeddings and vector records; +- source digests, fragment IDs, manifests, and checkpoint tokens; and +- adapter exceptions and 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. + +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 +by the supplied vector-index adapter. + +## Limits and resource policy + +`ProjectionLimits` rejects a request that exceeds its configured document count +or per-document code-point cap and bounds each embedding batch. +`ChunkingPolicy` bounds fragment size and overlap. These checks prevent one +accepted request from exceeding caller-selected values; they are not global +quotas, rate limits, memory isolation, provider billing controls, timeouts, or +admission control. + +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 +external cancellation, concurrency, cost, and capacity controls where needed. + +## Errors and logging + +`ProjectionStateError` and the direct message of +`ProjectionOperationError` are content-free. Ordinary adapter exceptions are +preserved as chained causes, so rendering the full exception chain may expose +an adapter's message or fields. Adapter implementations must avoid placing +document text, fragments, embeddings, credentials, or sensitive identifiers in +exceptions and logs. Applications should apply redaction before exporting +traces or error reports. + +The package does not retry collaborator operations. A failure can leave earlier +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. ## Trust boundary -The package does not own, authenticate, or prove an authoritative source or -revision. A fragment range checks only a half-open code-point width against the -fragment text length; it does not verify the text against a source document. -Ranges and attributes are not a citation or provenance-verification mechanism. +A fragment range is a half-open Python code-point range. Standalone contracts +check its width but do not prove that text came from the indicated source. +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. Version 0.1.0 provides no built-in encryption, authentication, authorization, -ACL, content filter, persistence, network security, citation validation, or -vendor guarantee. A consuming application must select and assess those controls -for its environment. - -The complete public value and provider boundaries are listed in the -[API reference](api.md). +ACL, content filter, persistence security, network security, citation +validation, secret management, vendor guarantee, retrieval workflow, 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). diff --git a/src/generic_rag/contracts.py b/src/generic_rag/contracts.py index a8cfb66..978cbc5 100644 --- a/src/generic_rag/contracts.py +++ b/src/generic_rag/contracts.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from dataclasses import dataclass from enum import StrEnum from math import isfinite @@ -21,12 +22,23 @@ "ProjectionCheckpoint", "ProjectionOutcome", "ProjectionReceipt", + "ChunkingPolicy", + "ProjectionLimits", + "ProjectionRequest", + "ProjectionManifestEntry", + "ProjectionManifest", + "ProjectionStateAvailability", + "ProjectionStateSnapshot", + "ProjectionStateStatus", + "ProjectionResult", "RetrievalQuery", "RetrievalOutcome", "RetrievalHit", "RetrievalResult", ) +_SOURCE_DIGEST = re.compile(r"sha256:[0-9a-f]{64}") + def _require_exact_type(name: str, value: object, expected: type[object]) -> None: if type(value) is not expected: @@ -310,6 +322,239 @@ def __post_init__(self) -> None: ) +@dataclass(frozen=True, slots=True) +class ChunkingPolicy: + """Code-point chunk size and overlap used by one projection.""" + + max_fragment_codepoints: int + overlap_codepoints: int + + def __post_init__(self) -> None: + maximum = _require_positive_integer( + "max_fragment_codepoints", + self.max_fragment_codepoints, + ) + overlap = _require_nonnegative_integer( + "overlap_codepoints", + self.overlap_codepoints, + ) + if overlap >= maximum: + raise ContractValidationError( + "overlap_codepoints must be smaller than max_fragment_codepoints" + ) + + +@dataclass(frozen=True, slots=True) +class ProjectionLimits: + """Independent document, text, and embedding-batch projection bounds.""" + + max_documents: int + max_document_codepoints: int + max_embedding_batch_size: int + + def __post_init__(self) -> None: + _require_positive_integer("max_documents", self.max_documents) + _require_positive_integer( + "max_document_codepoints", + self.max_document_codepoints, + ) + _require_positive_integer( + "max_embedding_batch_size", + self.max_embedding_batch_size, + ) + + +@dataclass(frozen=True, slots=True) +class ProjectionRequest: + """A complete bounded canonical target projection for one corpus.""" + + corpus_id: str + projection: ProjectionIdentity + chunking: ChunkingPolicy + limits: ProjectionLimits + documents: tuple[Document, ...] + + def __post_init__(self) -> None: + _require_nonblank_string("corpus_id", self.corpus_id) + _require_exact_type("projection", self.projection, ProjectionIdentity) + _require_exact_type("chunking", self.chunking, ChunkingPolicy) + _require_exact_type("limits", self.limits, ProjectionLimits) + _require_exact_type("documents", self.documents, tuple) + if len(self.documents) > self.limits.max_documents: + raise ContractValidationError("documents must not exceed max_documents") + + keys: set[DocumentKey] = set() + canonical: list[Document] = [] + for index, document in enumerate(self.documents): + _require_exact_type(f"documents[{index}]", document, Document) + if document.identity.key.corpus_id != self.corpus_id: + raise ContractValidationError( + "every document corpus_id must match the request corpus_id" + ) + if len(document.text) > self.limits.max_document_codepoints: + raise ContractValidationError( + "document text must not exceed max_document_codepoints" + ) + if document.identity.key in keys: + raise ContractValidationError( + "documents must have unique stable document keys" + ) + keys.add(document.identity.key) + canonical.append(document) + canonical.sort(key=lambda document: document.identity.key.document_id) + object.__setattr__(self, "documents", tuple(canonical)) + + +@dataclass(frozen=True, slots=True) +class ProjectionManifestEntry: + """One projected document revision, source digest, and fragment count.""" + + document: DocumentIdentity + source_digest: str + fragment_count: int + + def __post_init__(self) -> None: + _require_exact_type("document", self.document, DocumentIdentity) + _require_exact_type("source_digest", self.source_digest, str) + if _SOURCE_DIGEST.fullmatch(self.source_digest) is None: + raise ContractValidationError( + "source_digest must be lowercase sha256:<64hex>" + ) + _require_nonnegative_integer("fragment_count", self.fragment_count) + + +@dataclass(frozen=True, slots=True) +class ProjectionManifest: + """A complete canonical successful projection checkpoint manifest.""" + + corpus_id: str + projection: ProjectionIdentity + chunking: ChunkingPolicy + entries: tuple[ProjectionManifestEntry, ...] + checkpoint: ProjectionCheckpoint + + def __post_init__(self) -> None: + _require_nonblank_string("corpus_id", self.corpus_id) + _require_exact_type("projection", self.projection, ProjectionIdentity) + _require_exact_type("chunking", self.chunking, ChunkingPolicy) + _require_exact_type("entries", self.entries, tuple) + _require_exact_type("checkpoint", self.checkpoint, ProjectionCheckpoint) + if self.checkpoint.corpus_id != self.corpus_id: + raise ContractValidationError( + "checkpoint corpus_id must match the manifest corpus_id" + ) + if self.checkpoint.projection != self.projection: + raise ContractValidationError( + "checkpoint projection must match the manifest projection" + ) + + keys: set[DocumentKey] = set() + canonical: list[ProjectionManifestEntry] = [] + for index, entry in enumerate(self.entries): + _require_exact_type(f"entries[{index}]", entry, ProjectionManifestEntry) + if entry.document.key.corpus_id != self.corpus_id: + raise ContractValidationError( + "every manifest entry corpus_id must match the manifest corpus_id" + ) + if entry.document.key in keys: + raise ContractValidationError( + "manifest entries must have unique stable document keys" + ) + keys.add(entry.document.key) + canonical.append(entry) + canonical.sort(key=lambda entry: entry.document.key.document_id) + object.__setattr__(self, "entries", tuple(canonical)) + + +class ProjectionStateAvailability(StrEnum): + """Structural availability of a caller-supplied projection snapshot.""" + + MISSING = "missing" + PRESENT = "present" + CORRUPT = "corrupt" + + @classmethod + def _missing_(cls, value: object) -> None: + raise ContractValidationError(f"{cls.__name__} value is not a defined member") + + +@dataclass(frozen=True, slots=True) +class ProjectionStateSnapshot: + """Caller-supplied projection state without package-owned persistence.""" + + availability: ProjectionStateAvailability + manifest: ProjectionManifest | None + + def __post_init__(self) -> None: + _require_exact_type( + "availability", + self.availability, + ProjectionStateAvailability, + ) + if self.availability is ProjectionStateAvailability.PRESENT: + _require_exact_type("manifest", self.manifest, ProjectionManifest) + elif self.manifest is not None: + raise ContractValidationError( + "missing and corrupt state snapshots must not contain a manifest" + ) + + +class ProjectionStateStatus(StrEnum): + """Compatibility of supplied state with one complete projection target.""" + + MISSING = "missing" + CURRENT = "current" + STALE = "stale" + CORRUPT = "corrupt" + SCHEMA_MISMATCH = "schema_mismatch" + EMBEDDING_MISMATCH = "embedding_mismatch" + + @classmethod + def _missing_(cls, value: object) -> None: + raise ContractValidationError(f"{cls.__name__} value is not a defined member") + + +@dataclass(frozen=True, slots=True) +class ProjectionResult: + """One completely successful projection result and authoritative manifest.""" + + status_before: ProjectionStateStatus + receipt: ProjectionReceipt + manifest: ProjectionManifest + + def __post_init__(self) -> None: + _require_exact_type("status_before", self.status_before, ProjectionStateStatus) + _require_exact_type("receipt", self.receipt, ProjectionReceipt) + _require_exact_type("manifest", self.manifest, ProjectionManifest) + if self.receipt.outcome not in ( + ProjectionOutcome.COMPLETED, + ProjectionOutcome.UNCHANGED, + ): + raise ContractValidationError( + "projection results require a completed or unchanged receipt" + ) + if self.receipt.corpus_id != self.manifest.corpus_id: + raise ContractValidationError( + "receipt corpus_id must match the result manifest" + ) + if self.receipt.projection != self.manifest.projection: + raise ContractValidationError( + "receipt projection must match the result manifest" + ) + if self.receipt.checkpoint != self.manifest.checkpoint: + raise ContractValidationError( + "receipt checkpoint must equal the result manifest checkpoint" + ) + if self.receipt.outcome is ProjectionOutcome.UNCHANGED and ( + self.status_before is not ProjectionStateStatus.CURRENT + or self.receipt.attempted_documents != 0 + or self.receipt.completed_documents != 0 + ): + raise ContractValidationError( + "unchanged results require current state and zero document attempts" + ) + + @dataclass(frozen=True, slots=True) class RetrievalQuery: """A bounded retrieval request for one opaque corpus.""" diff --git a/src/generic_rag/ports.py b/src/generic_rag/ports.py index fdc931b..0eb7105 100644 --- a/src/generic_rag/ports.py +++ b/src/generic_rag/ports.py @@ -20,6 +20,7 @@ "Borrowed", "Embedder", "VectorIndexWriter", + "VectorIndexResetter", "VectorIndexReader", "LexicalRetriever", ) @@ -81,6 +82,15 @@ def delete_document(self, document: DocumentKey, /) -> None: ... +@runtime_checkable +class VectorIndexResetter(Protocol): + """Synchronously removes every projected document for one corpus.""" + + def reset_corpus(self, corpus_id: str, /) -> None: + """Remove the complete derived vector projection for the corpus.""" + ... + + @runtime_checkable class VectorIndexReader(Protocol): """Returns best-first vector candidates without exposing raw scores.""" diff --git a/src/generic_rag/projection.py b/src/generic_rag/projection.py new file mode 100644 index 0000000..ac7035f --- /dev/null +++ b/src/generic_rag/projection.py @@ -0,0 +1,811 @@ +"""Deterministic bounded document projection orchestration.""" + +from __future__ import annotations + +import hashlib +from collections.abc import Callable +from dataclasses import dataclass +from enum import StrEnum +from math import isfinite +from typing import NoReturn, cast + +from .contracts import ( + ChunkingPolicy, + Document, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + EmbeddingVector, + Fragment, + FragmentIdentity, + ProjectionCheckpoint, + ProjectionIdentity, + ProjectionLimits, + ProjectionManifest, + ProjectionManifestEntry, + ProjectionOutcome, + ProjectionReceipt, + ProjectionRequest, + ProjectionResult, + ProjectionStateAvailability, + ProjectionStateSnapshot, + ProjectionStateStatus, + VectorRecord, +) +from .errors import ( + CollaborationError, + ContractValidationError, + StateCompatibilityError, +) +from .ports import Borrowed, Embedder, VectorIndexResetter, VectorIndexWriter + +__all__ = ( + "ProjectionFailureStage", + "ProjectionStateError", + "ProjectionOperationError", + "project_documents", + "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.""" + + EMBEDDER_IDENTITY = "embedder_identity" + EMBEDDING = "embedding" + REPLACEMENT = "replacement" + DELETION = "deletion" + RESET = "reset" + + @classmethod + def _missing_(cls, value: object) -> None: + raise ContractValidationError(f"{cls.__name__} value is not a defined member") + + +class ProjectionStateError(StateCompatibilityError): + """Raised before effects when incremental projection cannot use state.""" + + status: ProjectionStateStatus + + def __init__(self, status: ProjectionStateStatus) -> None: + _require_exact_type("status", status, ProjectionStateStatus) + self.status = status + super().__init__("projection state is not compatible with incremental update") + + +class ProjectionOperationError(CollaborationError): + """A content-free translation of one collaborator operation failure.""" + + stage: ProjectionFailureStage + affected_document: DocumentKey | None + receipt: ProjectionReceipt | None + + def __init__( + self, + stage: ProjectionFailureStage, + affected_document: DocumentKey | None, + receipt: ProjectionReceipt | None, + ) -> None: + _require_exact_type("stage", stage, ProjectionFailureStage) + if affected_document is not None: + _validate_document_key(affected_document) + if receipt is not None: + _require_exact_type("receipt", receipt, ProjectionReceipt) + if receipt.outcome not in ( + ProjectionOutcome.FAILED, + ProjectionOutcome.PARTIAL, + ): + raise ContractValidationError( + "operation error receipt must be failed or partial" + ) + self.stage = stage + self.affected_document = affected_document + self.receipt = receipt + super().__init__(f"projection collaborator failed during {stage.value}") + + +@dataclass(frozen=True, slots=True) +class _PreparedDocument: + source: Document + fragments: tuple[Fragment, ...] + entry: ProjectionManifestEntry + + +@dataclass(frozen=True, slots=True) +class _PreparedTarget: + request: ProjectionRequest + documents: tuple[_PreparedDocument, ...] + manifest: ProjectionManifest + + +@dataclass(frozen=True, slots=True) +class _Mutation: + key: DocumentKey + replacement: _PreparedDocument | None + + +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_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_limits(value: object) -> ProjectionLimits: + _require_exact_type("limits", value, ProjectionLimits) + assert isinstance(value, ProjectionLimits) + return ProjectionLimits( + value.max_documents, + value.max_document_codepoints, + value.max_embedding_batch_size, + ) + + +def _validate_request(value: object) -> ProjectionRequest: + _require_exact_type("request", value, ProjectionRequest) + assert isinstance(value, ProjectionRequest) + _require_exact_type("request documents", value.documents, tuple) + documents = tuple(_validate_document(document) for document in value.documents) + canonical = ProjectionRequest( + value.corpus_id, + _validate_projection_identity(value.projection), + _validate_chunking(value.chunking), + _validate_limits(value.limits), + documents, + ) + if documents != canonical.documents: + raise ContractValidationError("request documents must be in canonical order") + return canonical + + +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_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 + text_length = len(document.text) + while start < text_length: + end = min(start + chunking.max_fragment_codepoints, text_length) + identity = FragmentIdentity( + document.identity, + _fragment_id(document.identity, start, end), + start, + end, + ) + fragments.append( + Fragment( + identity, + document.text[start:end], + document.attributes, + ) + ) + if end == text_length: + break + start = end - chunking.overlap_codepoints + return tuple(fragments) + + +def _prepare_target(request: ProjectionRequest) -> _PreparedTarget: + prepared: list[_PreparedDocument] = [] + for document in request.documents: + fragments = _fragments(document, request.chunking) + entry = ProjectionManifestEntry( + document.identity, + _source_digest(document), + len(fragments), + ) + prepared.append(_PreparedDocument(document, fragments, entry)) + entries = tuple(item.entry for item in prepared) + checkpoint = ProjectionCheckpoint( + request.corpus_id, + request.projection, + _checkpoint_token( + request.corpus_id, + request.projection, + request.chunking, + entries, + ), + ) + manifest = ProjectionManifest( + request.corpus_id, + request.projection, + request.chunking, + entries, + checkpoint, + ) + 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, +) -> ProjectionStateStatus: + if state.availability is ProjectionStateAvailability.MISSING: + return ProjectionStateStatus.MISSING + if state.availability is ProjectionStateAvailability.CORRUPT: + return ProjectionStateStatus.CORRUPT + + manifest = state.manifest + assert manifest is not None + if not _has_valid_checkpoint(manifest): + return ProjectionStateStatus.CORRUPT + if manifest.corpus_id != target.request.corpus_id: + return ProjectionStateStatus.CORRUPT + if manifest.projection.schema_id != target.request.projection.schema_id: + return ProjectionStateStatus.SCHEMA_MISMATCH + if manifest.projection.embedding != target.request.projection.embedding: + return ProjectionStateStatus.EMBEDDING_MISMATCH + + target_by_key = {item.entry.document.key: item.entry for item in target.documents} + for previous in manifest.entries: + current = target_by_key.get(previous.document.key) + if current is None or previous.document != current.document: + continue + if previous.source_digest != current.source_digest: + return ProjectionStateStatus.CORRUPT + if ( + manifest.chunking == target.request.chunking + and previous.fragment_count != current.fragment_count + ): + return ProjectionStateStatus.CORRUPT + + if ( + manifest.chunking == target.request.chunking + and manifest.entries == target.manifest.entries + ): + return ProjectionStateStatus.CURRENT + return ProjectionStateStatus.STALE + + +def _incremental_plan( + manifest: ProjectionManifest, + target: _PreparedTarget, +) -> tuple[_Mutation, ...]: + previous_by_key = {entry.document.key: entry for entry in manifest.entries} + target_by_key = {item.entry.document.key: item for item in target.documents} + mutations: list[_Mutation] = [] + + for key, item in target_by_key.items(): + previous = previous_by_key.get(key) + if manifest.chunking != target.request.chunking or previous != item.entry: + mutations.append(_Mutation(key, item)) + for key in previous_by_key.keys() - target_by_key.keys(): + mutations.append(_Mutation(key, None)) + mutations.sort(key=lambda mutation: mutation.key.document_id) + return tuple(mutations) + + +def _rebuild_plan(target: _PreparedTarget) -> tuple[_Mutation, ...]: + return tuple(_Mutation(item.entry.document.key, item) for item in target.documents) + + +def _failure_receipt( + target: _PreparedTarget, + attempted: int, + completed: int, +) -> ProjectionReceipt | None: + if attempted == 0: + return None + outcome = ProjectionOutcome.FAILED if completed == 0 else ProjectionOutcome.PARTIAL + return ProjectionReceipt( + target.request.corpus_id, + target.request.projection, + outcome, + attempted, + completed, + None, + ) + + +def _raise_operation_error( + stage: ProjectionFailureStage, + affected_document: DocumentKey | None, + receipt: ProjectionReceipt | None, + cause: Exception | None = None, +) -> NoReturn: + error = ProjectionOperationError(stage, affected_document, receipt) + if cause is None: + raise error + raise error from cause + + +def _require_embedder_identity( + embedder: Embedder, + expected: EmbeddingIdentity, + receipt: ProjectionReceipt | None, +) -> None: + try: + identity = embedder.identity + except Exception as exc: + _raise_operation_error( + ProjectionFailureStage.EMBEDDER_IDENTITY, + None, + receipt, + exc, + ) + try: + validated = _validate_embedding_identity(identity) + except ContractValidationError: + _raise_operation_error( + ProjectionFailureStage.EMBEDDER_IDENTITY, + None, + receipt, + ) + if validated != expected: + _raise_operation_error( + ProjectionFailureStage.EMBEDDER_IDENTITY, + None, + receipt, + ) + + +def _validate_embeddings( + value: object, + expected_count: int, + expected_dimensions: int, +) -> tuple[EmbeddingVector, ...]: + if type(value) is not tuple or len(value) != expected_count: + raise ContractValidationError( + "embedder output must be an exact same-count tuple" + ) + assert isinstance(value, tuple) + for vector in value: + _require_exact_type("embedding", vector, EmbeddingVector) + assert isinstance(vector, EmbeddingVector) + if ( + type(vector.values) is not tuple + or len(vector.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 vector.values + ): + raise ContractValidationError( + "embedding vector coordinates must be canonical finite floats" + ) + return value + + +def _records_for_document( + embedder: Embedder, + item: _PreparedDocument, + batch_size: int, + target: _PreparedTarget, + attempted: int, + completed: int, +) -> tuple[VectorRecord, ...]: + records: list[VectorRecord] = [] + for start in range(0, len(item.fragments), batch_size): + fragments = item.fragments[start : start + batch_size] + texts = tuple(fragment.text for fragment in fragments) + try: + output = embedder.embed(texts) + except Exception as exc: + _raise_operation_error( + ProjectionFailureStage.EMBEDDING, + item.entry.document.key, + _failure_receipt(target, attempted, completed), + exc, + ) + try: + vectors = _validate_embeddings( + output, + len(fragments), + target.request.projection.embedding.dimensions, + ) + except ContractValidationError: + _raise_operation_error( + ProjectionFailureStage.EMBEDDING, + item.entry.document.key, + _failure_receipt(target, attempted, completed), + ) + records.extend( + VectorRecord(fragment, vector) + for fragment, vector in zip(fragments, vectors, strict=True) + ) + return tuple(records) + + +def _execute_prevalidated_mutations( + target: _PreparedTarget, + mutations: tuple[_Mutation, ...], + embedder: Embedder | None, + writer_scope: Borrowed[VectorIndexWriter], +) -> None: + if not mutations: + return + attempted = len(mutations) + completed = 0 + with writer_scope as writer: + for mutation in mutations: + if mutation.replacement is None: + try: + delete_document = cast( + Callable[[DocumentKey], object], + writer.delete_document, + ) + command_result = delete_document(mutation.key) + except Exception as exc: + _raise_operation_error( + ProjectionFailureStage.DELETION, + mutation.key, + _failure_receipt(target, attempted, completed), + exc, + ) + if command_result is not None: + _raise_operation_error( + ProjectionFailureStage.DELETION, + mutation.key, + _failure_receipt(target, attempted, completed), + ) + else: + assert embedder is not None + records = _records_for_document( + embedder, + mutation.replacement, + target.request.limits.max_embedding_batch_size, + target, + attempted, + completed, + ) + try: + replace_document = cast( + Callable[ + [DocumentIdentity, tuple[VectorRecord, ...]], + object, + ], + writer.replace_document, + ) + command_result = replace_document( + mutation.replacement.entry.document, + records, + ) + except Exception as exc: + _raise_operation_error( + ProjectionFailureStage.REPLACEMENT, + mutation.key, + _failure_receipt(target, attempted, completed), + exc, + ) + if command_result is not None: + _raise_operation_error( + ProjectionFailureStage.REPLACEMENT, + mutation.key, + _failure_receipt(target, attempted, completed), + ) + completed += 1 + + +def _execute_mutations( + target: _PreparedTarget, + mutations: tuple[_Mutation, ...], + embedder_scope: Borrowed[Embedder], + writer_scope: Borrowed[VectorIndexWriter], +) -> None: + needs_embedder = any(mutation.replacement is not None for mutation in mutations) + if not needs_embedder: + _execute_prevalidated_mutations(target, mutations, None, writer_scope) + return + + with embedder_scope as embedder: + _require_embedder_identity( + embedder, + target.request.projection.embedding, + _failure_receipt(target, len(mutations), 0), + ) + _execute_prevalidated_mutations(target, mutations, embedder, writer_scope) + + +def _reset_projection( + target: _PreparedTarget, + attempted: int, + resetter_scope: Borrowed[VectorIndexResetter], +) -> None: + try: + with resetter_scope as resetter: + reset_corpus = cast( + Callable[[str], object], + resetter.reset_corpus, + ) + command_result = reset_corpus(target.request.corpus_id) + except Exception as exc: + _raise_operation_error( + ProjectionFailureStage.RESET, + None, + _failure_receipt(target, attempted, 0), + exc, + ) + if command_result is not None: + _raise_operation_error( + ProjectionFailureStage.RESET, + None, + _failure_receipt(target, attempted, 0), + ) + + +def _successful_result( + target: _PreparedTarget, + status_before: ProjectionStateStatus, + outcome: ProjectionOutcome, + attempted: int, +) -> ProjectionResult: + receipt = ProjectionReceipt( + target.request.corpus_id, + target.request.projection, + outcome, + attempted, + attempted, + target.manifest.checkpoint, + ) + return ProjectionResult(status_before, receipt, target.manifest) + + +def project_documents( + request: ProjectionRequest, + state: ProjectionStateSnapshot, + embedder: Borrowed[Embedder], + writer: Borrowed[VectorIndexWriter], + /, +) -> ProjectionResult: + """Incrementally project a valid present snapshot to a complete target.""" + + canonical_request = _validate_request(request) + canonical_state = _validate_state(state) + _validate_borrowed("embedder", embedder) + _validate_borrowed("writer", writer) + target = _prepare_target(canonical_request) + status = _state_status(canonical_state, target) + if status is ProjectionStateStatus.CURRENT: + return _successful_result(target, status, ProjectionOutcome.UNCHANGED, 0) + if status is not ProjectionStateStatus.STALE: + raise ProjectionStateError(status) + + manifest = canonical_state.manifest + assert manifest is not None + mutations = _incremental_plan(manifest, target) + _execute_mutations(target, mutations, embedder, writer) + return _successful_result( + target, + status, + ProjectionOutcome.COMPLETED, + len(mutations), + ) + + +def rebuild_projection( + request: ProjectionRequest, + state: ProjectionStateSnapshot, + embedder: Borrowed[Embedder], + writer: Borrowed[VectorIndexWriter], + resetter: Borrowed[VectorIndexResetter], + /, +) -> ProjectionResult: + """Reset one corpus and write its complete canonical target projection.""" + + canonical_request = _validate_request(request) + canonical_state = _validate_state(state) + _validate_borrowed("embedder", embedder) + _validate_borrowed("writer", writer) + _validate_borrowed("resetter", resetter) + target = _prepare_target(canonical_request) + status = _state_status(canonical_state, target) + mutations = _rebuild_plan(target) + + if mutations: + with embedder as embedder_resource: + _require_embedder_identity( + embedder_resource, + target.request.projection.embedding, + _failure_receipt(target, len(mutations), 0), + ) + _reset_projection(target, len(mutations), resetter) + _execute_prevalidated_mutations( + target, + mutations, + embedder_resource, + writer, + ) + else: + _reset_projection(target, 0, resetter) + return _successful_result( + target, + status, + ProjectionOutcome.COMPLETED, + len(mutations), + ) diff --git a/tests/support/clean_import_probe.py b/tests/support/clean_import_probe.py index d0db065..e61cdaa 100644 --- a/tests/support/clean_import_probe.py +++ b/tests/support/clean_import_probe.py @@ -161,6 +161,32 @@ def _assert_forbidden_paths_absent(paths: list[str]) -> None: ) +def _bootstrap_source_root(raw_source_root: str | None) -> Path | None: + if raw_source_root is None: + return None + source_root = Path(raw_source_root).resolve(strict=True) + if not source_root.is_dir(): + raise AssertionError(f"source root is not a directory: {source_root}") + sys.path.insert(0, os.fspath(source_root)) + return source_root + + +def _assert_import_origin(imported: ModuleType, source_root: Path | None) -> None: + if source_root is None: + return + raw_origin = getattr(imported, "__file__", None) + if not isinstance(raw_origin, str): + raise AssertionError(f"source import has no file origin: {imported.__name__}") + origin = Path(raw_origin).resolve(strict=True) + expected_package = source_root / _PACKAGE_ROOT + try: + origin.relative_to(expected_package) + except ValueError: + raise AssertionError( + f"{imported.__name__} came from {origin}, not {expected_package}" + ) from None + + def _parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("module") @@ -170,6 +196,10 @@ def _parse_arguments() -> argparse.Namespace: default=[], help="Fail when this path or one of its children occurs on sys.path.", ) + parser.add_argument( + "--source-root", + help="Explicit resolved src root for an isolated source-tree probe.", + ) return parser.parse_args() @@ -177,14 +207,17 @@ def main() -> int: arguments = _parse_arguments() module_name = cast(str, arguments.module) forbidden_paths = cast(list[str], arguments.forbid_path) + raw_source_root = cast(str | None, arguments.source_root) if module_name != _PACKAGE_ROOT and not module_name.startswith(f"{_PACKAGE_ROOT}."): raise ValueError(f"probe is restricted to {_PACKAGE_ROOT} modules") _assert_forbidden_paths_absent(forbidden_paths) + source_root = _bootstrap_source_root(raw_source_root) before = set(sys.modules) imported = _import_with_guards(module_name) after = set(sys.modules) _assert_no_external_imports(before, after) + _assert_import_origin(imported, source_root) if imported.__name__ != module_name: raise AssertionError(f"requested {module_name}, imported {imported.__name__}") diff --git a/tests/support/verify_artifacts.py b/tests/support/verify_artifacts.py index bf78892..ba6ea5f 100644 --- a/tests/support/verify_artifacts.py +++ b/tests/support/verify_artifacts.py @@ -16,6 +16,7 @@ "generic_rag/contracts.py", "generic_rag/errors.py", "generic_rag/ports.py", + "generic_rag/projection.py", } _PACKAGE_DATA = {"generic_rag/py.typed"} diff --git a/tests/test_package_boundaries.py b/tests/test_package_boundaries.py index eedc054..a5b4eef 100644 --- a/tests/test_package_boundaries.py +++ b/tests/test_package_boundaries.py @@ -16,6 +16,7 @@ import generic_rag.contracts as contracts import generic_rag.errors as errors import generic_rag.ports as ports +import generic_rag.projection as projection _PROJECT_ROOT = Path(__file__).resolve().parents[1] _SOURCE_ROOT = _PROJECT_ROOT / "src" @@ -26,12 +27,18 @@ "generic_rag.contracts": "src/generic_rag/contracts.py", "generic_rag.errors": "src/generic_rag/errors.py", "generic_rag.ports": "src/generic_rag/ports.py", + "generic_rag.projection": "src/generic_rag/projection.py", } _EXPECTED_DEPENDENCIES = { "generic_rag": set(), "generic_rag.contracts": {"generic_rag.errors"}, "generic_rag.errors": set(), "generic_rag.ports": {"generic_rag.contracts"}, + "generic_rag.projection": { + "generic_rag.contracts", + "generic_rag.errors", + "generic_rag.ports", + }, } _EXPECTED_EXPORTS = { "generic_rag": (), @@ -54,6 +61,15 @@ "ProjectionCheckpoint", "ProjectionOutcome", "ProjectionReceipt", + "ChunkingPolicy", + "ProjectionLimits", + "ProjectionRequest", + "ProjectionManifestEntry", + "ProjectionManifest", + "ProjectionStateAvailability", + "ProjectionStateSnapshot", + "ProjectionStateStatus", + "ProjectionResult", "RetrievalQuery", "RetrievalOutcome", "RetrievalHit", @@ -63,9 +79,17 @@ "Borrowed", "Embedder", "VectorIndexWriter", + "VectorIndexResetter", "VectorIndexReader", "LexicalRetriever", ), + "generic_rag.projection": ( + "ProjectionFailureStage", + "ProjectionStateError", + "ProjectionOperationError", + "project_documents", + "rebuild_projection", + ), } @@ -150,6 +174,7 @@ def test_supported_exports_are_exact_and_owned(self) -> None: "generic_rag.errors": errors, "generic_rag.contracts": contracts, "generic_rag.ports": ports, + "generic_rag.projection": projection, } for module_name, expected_exports in _EXPECTED_EXPORTS.items(): @@ -164,6 +189,7 @@ def test_supported_exports_are_exact_and_owned(self) -> None: *_EXPECTED_EXPORTS["generic_rag.errors"], *_EXPECTED_EXPORTS["generic_rag.contracts"], *_EXPECTED_EXPORTS["generic_rag.ports"], + *_EXPECTED_EXPORTS["generic_rag.projection"], ): with self.subTest(root_reexport=name): self.assertFalse(hasattr(generic_rag, name)) @@ -339,6 +365,8 @@ def test_each_module_imports_in_an_isolated_side_effect_guarded_process( "-B", str(_CLEAN_IMPORT_PROBE), module_name, + "--source-root", + str(_SOURCE_ROOT), ), cwd=working_directory, env=environment, diff --git a/tests/test_ports.py b/tests/test_ports.py index d3bc6f3..4033810 100644 --- a/tests/test_ports.py +++ b/tests/test_ports.py @@ -4,9 +4,10 @@ import inspect import unittest +from collections.abc import Callable from dataclasses import FrozenInstanceError, fields from types import TracebackType -from typing import get_type_hints +from typing import cast, get_type_hints import generic_rag.ports as ports from generic_rag.contracts import ( @@ -24,6 +25,7 @@ Embedder, LexicalRetriever, VectorIndexReader, + VectorIndexResetter, VectorIndexWriter, ) @@ -107,6 +109,14 @@ def delete_document(self, document: DocumentKey, /) -> None: self.deletions.append(document) +class _FakeVectorResetter: + def __init__(self) -> None: + self.corpora: list[str] = [] + + def reset_corpus(self, corpus_id: str, /) -> None: + self.corpora.append(corpus_id) + + class _FakeVectorReader: def __init__(self, ranked: tuple[Fragment, ...]) -> None: self.ranked = ranked @@ -189,6 +199,7 @@ def test_exports_are_exact_and_owned_by_the_module(self) -> None: "Borrowed", "Embedder", "VectorIndexWriter", + "VectorIndexResetter", "VectorIndexReader", "LexicalRetriever", ) @@ -201,12 +212,14 @@ def test_exports_are_exact_and_owned_by_the_module(self) -> None: def test_protocols_are_runtime_checkable_structural_shapes(self) -> None: self.assertIsInstance(_FakeEmbedder(), Embedder) self.assertIsInstance(_FakeVectorWriter(), VectorIndexWriter) + self.assertIsInstance(_FakeVectorResetter(), VectorIndexResetter) self.assertIsInstance(_FakeVectorReader(()), VectorIndexReader) self.assertIsInstance(_FakeLexicalRetriever(()), LexicalRetriever) missing = _MissingMethods() self.assertNotIsInstance(missing, Embedder) self.assertNotIsInstance(missing, VectorIndexWriter) + self.assertNotIsInstance(missing, VectorIndexResetter) self.assertNotIsInstance(missing, VectorIndexReader) self.assertNotIsInstance(missing, LexicalRetriever) @@ -235,6 +248,7 @@ def test_protocol_methods_are_synchronous_and_positional_only(self) -> None: ("self", "document", "records"), ), (VectorIndexWriter.delete_document, ("self", "document")), + (VectorIndexResetter.reset_corpus, ("self", "corpus_id")), ( VectorIndexReader.search, ("self", "query", "embedding"), @@ -279,6 +293,13 @@ def test_protocol_annotations_keep_scores_and_lifecycle_out(self) -> None: "return": type(None), }, ) + self.assertEqual( + get_type_hints(VectorIndexResetter.reset_corpus), + { + "corpus_id": str, + "return": type(None), + }, + ) self.assertEqual( get_type_hints(VectorIndexReader.search), { @@ -298,6 +319,7 @@ def test_protocol_annotations_keep_scores_and_lifecycle_out(self) -> None: for protocol in ( Embedder, VectorIndexWriter, + VectorIndexResetter, VectorIndexReader, LexicalRetriever, ): @@ -340,6 +362,14 @@ def test_writer_witness_replaces_complete_sets_and_accepts_empty(self) -> None: with self.assertRaises(AssertionError): writer.replace_document(wrong_document, records) + def test_resetter_witness_removes_one_exact_corpus_and_returns_none(self) -> None: + resetter = _FakeVectorResetter() + + result = cast(Callable[[str], object], resetter.reset_corpus)(" Corpus/../A ") + + self.assertIsNone(result) + self.assertEqual(resetter.corpora, [" Corpus/../A "]) + def test_reader_witnesses_preserve_rank_and_enforce_candidate_bound(self) -> None: ranked = ( _fragment("first"), diff --git a/tests/test_projection.py b/tests/test_projection.py new file mode 100644 index 0000000..700f2df --- /dev/null +++ b/tests/test_projection.py @@ -0,0 +1,1661 @@ +"""Deterministic projection orchestration, failure, and lifecycle tests.""" + +from __future__ import annotations + +import hashlib +import inspect +import json +import os +import subprocess +import sys +import textwrap +import unittest +from collections.abc import Callable +from pathlib import Path +from typing import cast + +import generic_rag.projection as projection_module +from generic_rag.contracts import ( + ChunkingPolicy, + Document, + DocumentIdentity, + DocumentKey, + EmbeddingIdentity, + EmbeddingVector, + ProjectionCheckpoint, + ProjectionIdentity, + ProjectionLimits, + ProjectionManifest, + ProjectionManifestEntry, + ProjectionOutcome, + ProjectionReceipt, + ProjectionRequest, + ProjectionStateAvailability, + ProjectionStateSnapshot, + ProjectionStateStatus, + VectorRecord, +) +from generic_rag.errors import ( + CollaborationError, + ContractValidationError, + StateCompatibilityError, +) +from generic_rag.ports import ( + Borrowed, + Embedder, + VectorIndexResetter, + VectorIndexWriter, +) +from generic_rag.projection import ( + ProjectionFailureStage, + ProjectionOperationError, + ProjectionStateError, + project_documents, + rebuild_projection, +) + +_SOURCE_ROOT = Path(__file__).resolve().parents[1] / "src" + + +def _document( + document_id: str, + *, + revision_id: str = "revision-1", + text: str = "abcdefgh", + corpus_id: str = "corpus", + attributes: tuple[tuple[str, str], ...] = (), +) -> Document: + return Document( + DocumentIdentity(DocumentKey(corpus_id, document_id), revision_id), + text, + attributes, + ) + + +def _identity( + *, + schema_id: str = "schema-v1", + model_id: str = "model-v1", + dimensions: int = 2, +) -> ProjectionIdentity: + return ProjectionIdentity( + schema_id, + EmbeddingIdentity(model_id, dimensions), + ) + + +def _request( + documents: tuple[Document, ...], + *, + projection: ProjectionIdentity | None = None, + chunking: ChunkingPolicy | None = None, + batch_size: int = 2, + corpus_id: str = "corpus", +) -> ProjectionRequest: + return ProjectionRequest( + corpus_id, + projection or _identity(), + chunking or ChunkingPolicy(4, 1), + ProjectionLimits(max(1, len(documents)), 100, batch_size), + documents, + ) + + +def _hash_fields(fields_to_hash: tuple[str, ...]) -> str: + """Independent length-delimited SHA-256 oracle from the public contract.""" + + digest = hashlib.sha256() + for value in fields_to_hash: + encoded = value.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: + 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)) + return _hash_fields(tuple(source_fields)) + + +def _fragment_id(document: DocumentIdentity, start: int, end: int) -> str: + return _hash_fields( + ( + "generic-rag:fragment-id:v1", + "corpus_id", + document.key.corpus_id, + "document_id", + document.key.document_id, + "revision_id", + document.revision_id, + "start", + str(start), + "end", + str(end), + ) + ) + + +def _expected_ranges( + text: str, chunking: ChunkingPolicy +) -> tuple[tuple[int, int], ...]: + ranges: list[tuple[int, int]] = [] + start = 0 + while start < len(text): + end = min(start + chunking.max_fragment_codepoints, len(text)) + ranges.append((start, end)) + if end == len(text): + break + start = end - chunking.overlap_codepoints + return tuple(ranges) + + +def _checkpoint_token( + request: ProjectionRequest, + entries: tuple[ProjectionManifestEntry, ...], +) -> str: + token_fields = [ + "generic-rag:projection-checkpoint:v1", + "corpus_id", + request.corpus_id, + "schema_id", + request.projection.schema_id, + "embedding_model_id", + request.projection.embedding.model_id, + "embedding_dimensions", + str(request.projection.embedding.dimensions), + "max_fragment_codepoints", + str(request.chunking.max_fragment_codepoints), + "overlap_codepoints", + str(request.chunking.overlap_codepoints), + "entry_count", + str(len(entries)), + ] + for entry in entries: + token_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 _hash_fields(tuple(token_fields)) + + +def _manifest(request: ProjectionRequest) -> ProjectionManifest: + entries = tuple( + ProjectionManifestEntry( + document.identity, + _source_digest(document), + len(_expected_ranges(document.text, request.chunking)), + ) + for document in request.documents + ) + checkpoint = ProjectionCheckpoint( + request.corpus_id, + request.projection, + _checkpoint_token(request, entries), + ) + return ProjectionManifest( + request.corpus_id, + request.projection, + request.chunking, + entries, + checkpoint, + ) + + +def _present(request: ProjectionRequest) -> ProjectionStateSnapshot: + return ProjectionStateSnapshot( + ProjectionStateAvailability.PRESENT, + _manifest(request), + ) + + +class _FakeEmbedder: + def __init__( + self, + expected_identity: EmbeddingIdentity, + events: list[str], + *, + identity_value: object | None = None, + identity_failure: BaseException | None = None, + output: Callable[[tuple[str, ...]], object] | None = None, + embed_failure_at: int | None = None, + embed_failure: BaseException | None = None, + ) -> None: + self.expected_identity = expected_identity + self.events = events + self.identity_value = ( + expected_identity if identity_value is None else identity_value + ) + self.identity_failure = identity_failure + self.output = output + self.embed_failure_at = embed_failure_at + self.embed_failure = embed_failure or RuntimeError("embedding failed") + self.identity_calls = 0 + self.embed_calls: list[tuple[str, ...]] = [] + self.lifecycle_calls: list[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 + return cast(EmbeddingIdentity, self.identity_value) + + def embed(self, texts: tuple[str, ...], /) -> tuple[EmbeddingVector, ...]: + call_index = len(self.embed_calls) + self.embed_calls.append(texts) + self.events.append("embed:" + "|".join(texts)) + if self.embed_failure_at == call_index: + raise self.embed_failure + if self.output is not None: + return cast(tuple[EmbeddingVector, ...], self.output(texts)) + return tuple( + EmbeddingVector((float(len(text)), float(index))) + for index, text in enumerate(texts) + ) + + def __enter__(self) -> _FakeEmbedder: + self.lifecycle_calls.append("enter") + return self + + def __exit__(self, *arguments: object) -> bool: + del arguments + 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 _FakeWriter: + def __init__( + self, + events: list[str], + *, + replace_failure_at: int | None = None, + replace_failure: BaseException | None = None, + replace_result: object = None, + delete_failure_at: int | None = None, + delete_failure: BaseException | None = None, + delete_result: object = None, + ) -> None: + self.events = events + self.replace_failure_at = replace_failure_at + self.replace_failure = replace_failure or RuntimeError("replacement failed") + self.replace_result = replace_result + self.delete_failure_at = delete_failure_at + self.delete_failure = delete_failure or RuntimeError("deletion failed") + self.delete_result = delete_result + self.replacements: list[tuple[DocumentIdentity, tuple[VectorRecord, ...]]] = [] + self.deletions: list[DocumentKey] = [] + self.lifecycle_calls: list[str] = [] + + def replace_document( + self, + document: DocumentIdentity, + records: tuple[VectorRecord, ...], + /, + ) -> None: + call_index = len(self.replacements) + self.replacements.append((document, records)) + self.events.append(f"replace:{document.key.document_id}") + if self.replace_failure_at == call_index: + raise self.replace_failure + return cast(None, self.replace_result) + + def delete_document(self, document: DocumentKey, /) -> None: + call_index = len(self.deletions) + self.deletions.append(document) + self.events.append(f"delete:{document.document_id}") + if self.delete_failure_at == call_index: + raise self.delete_failure + return cast(None, self.delete_result) + + def __enter__(self) -> _FakeWriter: + self.lifecycle_calls.append("enter") + return self + + def __exit__(self, *arguments: object) -> bool: + del arguments + 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 _FakeResetter: + def __init__( + self, + events: list[str], + *, + failure: BaseException | None = None, + result: object = None, + ) -> None: + self.events = events + self.failure = failure + self.result = result + self.corpora: list[str] = [] + self.lifecycle_calls: list[str] = [] + + def reset_corpus(self, corpus_id: str, /) -> None: + self.corpora.append(corpus_id) + self.events.append(f"reset:{corpus_id}") + if self.failure is not None: + raise self.failure + return cast(None, self.result) + + def __enter__(self) -> _FakeResetter: + self.lifecycle_calls.append("enter") + return self + + def __exit__(self, *arguments: object) -> bool: + del arguments + 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") + + +def _borrow_embedder(embedder: _FakeEmbedder) -> Borrowed[Embedder]: + return Borrowed(cast(Embedder, embedder)) + + +def _borrow_writer(writer: _FakeWriter) -> Borrowed[VectorIndexWriter]: + return Borrowed(cast(VectorIndexWriter, writer)) + + +def _borrow_resetter(resetter: _FakeResetter) -> Borrowed[VectorIndexResetter]: + return Borrowed(cast(VectorIndexResetter, resetter)) + + +def _constant_output(value: object) -> Callable[[tuple[str, ...]], object]: + def output(texts: tuple[str, ...]) -> object: + del texts + return value + + return output + + +def _collaborators( + request: ProjectionRequest, +) -> tuple[_FakeEmbedder, _FakeWriter, _FakeResetter, list[str]]: + events: list[str] = [] + return ( + _FakeEmbedder(request.projection.embedding, events), + _FakeWriter(events), + _FakeResetter(events), + events, + ) + + +class ProjectionPublicApiTests(unittest.TestCase): + def test_exports_are_exact_and_owned_by_projection_module(self) -> None: + expected = ( + "ProjectionFailureStage", + "ProjectionStateError", + "ProjectionOperationError", + "project_documents", + "rebuild_projection", + ) + + self.assertEqual(projection_module.__all__, expected) + for name in expected: + self.assertEqual( + getattr(projection_module, name).__module__, + "generic_rag.projection", + ) + + def test_public_workflows_are_synchronous_and_positional_only(self) -> None: + expected_parameters = { + project_documents: ("request", "state", "embedder", "writer"), + rebuild_projection: ( + "request", + "state", + "embedder", + "writer", + "resetter", + ), + } + for function, expected in expected_parameters.items(): + with self.subTest(function=function.__name__): + self.assertFalse(inspect.iscoroutinefunction(function)) + parameters = tuple( + inspect.signature( + cast(Callable[..., object], function) + ).parameters.values() + ) + self.assertEqual(tuple(item.name for item in parameters), expected) + self.assertTrue( + all( + item.kind is inspect.Parameter.POSITIONAL_ONLY + for item in parameters + ) + ) + + def test_failure_stage_is_an_exact_closed_string_enum(self) -> None: + self.assertEqual( + tuple((member.name, member.value) for member in ProjectionFailureStage), + ( + ("EMBEDDER_IDENTITY", "embedder_identity"), + ("EMBEDDING", "embedding"), + ("REPLACEMENT", "replacement"), + ("DELETION", "deletion"), + ("RESET", "reset"), + ), + ) + for value in ("EMBEDDING", "unknown", "", None, 1, object()): + with self.subTest(value=value): + with self.assertRaises(ContractValidationError): + ProjectionFailureStage(cast(str, value)) + + def test_state_error_is_typed_content_free_and_requires_exact_status(self) -> None: + error = ProjectionStateError(ProjectionStateStatus.SCHEMA_MISMATCH) + + self.assertIsInstance(error, StateCompatibilityError) + self.assertIs(error.status, ProjectionStateStatus.SCHEMA_MISMATCH) + self.assertNotIn("schema", str(error).lower()) + with self.assertRaises(ContractValidationError): + ProjectionStateError(cast(ProjectionStateStatus, "schema_mismatch")) + + def test_operation_error_fields_and_receipt_are_exact_and_truthful(self) -> None: + request = _request((_document("alpha"),)) + receipt = ProjectionReceipt( + request.corpus_id, + request.projection, + ProjectionOutcome.FAILED, + 1, + 0, + None, + ) + key = request.documents[0].identity.key + error = ProjectionOperationError( + ProjectionFailureStage.REPLACEMENT, + key, + receipt, + ) + + self.assertIsInstance(error, CollaborationError) + self.assertEqual( + ProjectionOperationError.__annotations__, + { + "stage": "ProjectionFailureStage", + "affected_document": "DocumentKey | None", + "receipt": "ProjectionReceipt | None", + }, + ) + self.assertIs(error.stage, ProjectionFailureStage.REPLACEMENT) + self.assertEqual(error.affected_document, key) + self.assertIs(error.receipt, receipt) + for stage, affected, value_receipt in ( + ("replacement", key, receipt), + (ProjectionFailureStage.REPLACEMENT, object(), receipt), + ( + ProjectionFailureStage.REPLACEMENT, + key, + ProjectionReceipt( + request.corpus_id, + request.projection, + ProjectionOutcome.COMPLETED, + 1, + 1, + _manifest(request).checkpoint, + ), + ), + (ProjectionFailureStage.REPLACEMENT, key, object()), + ): + with self.subTest(stage=stage, affected=affected): + with self.assertRaises(ContractValidationError): + ProjectionOperationError( + cast(ProjectionFailureStage, stage), + cast(DocumentKey, affected), + cast(ProjectionReceipt, value_receipt), + ) + + def test_invalid_top_level_inputs_are_rejected_before_any_effect(self) -> None: + request = _request((_document("alpha"),)) + state = _present(_request((), projection=request.projection)) + embedder, writer, resetter, events = _collaborators(request) + invalid_calls: tuple[Callable[[], object], ...] = ( + lambda: project_documents( + cast(ProjectionRequest, object()), + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ), + lambda: project_documents( + request, + cast(ProjectionStateSnapshot, object()), + _borrow_embedder(embedder), + _borrow_writer(writer), + ), + lambda: project_documents( + request, + state, + cast(Borrowed[Embedder], object()), + _borrow_writer(writer), + ), + lambda: project_documents( + request, + state, + _borrow_embedder(embedder), + cast(Borrowed[VectorIndexWriter], object()), + ), + lambda: rebuild_projection( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + cast(Borrowed[VectorIndexResetter], object()), + ), + ) + + for call in invalid_calls: + with self.subTest(call=call): + with self.assertRaises(ContractValidationError): + call() + self.assertEqual(events, []) + + +class ProjectionDeterminismTests(unittest.TestCase): + def test_unicode_codepoint_chunks_overlap_and_preserve_source_attributes( + self, + ) -> None: + document = _document( + "unicode", + text="A😀e\u0301한Z", + attributes=(("", ""), ("tag", "one"), ("tag", "one")), + ) + request = _request( + (document,), + chunking=ChunkingPolicy(3, 1), + batch_size=8, + ) + previous = _request( + (), projection=request.projection, chunking=request.chunking + ) + embedder, writer, _, _ = _collaborators(request) + + result = project_documents( + request, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + records = writer.replacements[0][1] + self.assertEqual( + tuple( + (record.fragment.identity.start, record.fragment.identity.end) + for record in records + ), + ((0, 3), (2, 5), (4, 6)), + ) + self.assertEqual( + tuple(record.fragment.text for record in records), + ("A😀e", "e\u0301한", "한Z"), + ) + self.assertTrue( + all(record.fragment.attributes == document.attributes for record in records) + ) + self.assertTrue( + all( + record.fragment.identity.document == document.identity + for record in records + ) + ) + self.assertEqual(result.manifest.entries[0].fragment_count, 3) + + def test_fragment_ids_source_digest_and_checkpoint_match_independent_oracle( + self, + ) -> None: + document = _document( + "doc/../A", + revision_id=" rev e\u0301 ", + text="abcdef", + attributes=(("k", "v"), ("k", "v"), ("", "")), + ) + request = _request((document,), chunking=ChunkingPolicy(4, 1)) + embedder, writer, resetter, _ = _collaborators(request) + + result = rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + expected_manifest = _manifest(request) + self.assertEqual(result.manifest, expected_manifest) + self.assertEqual( + tuple( + record.fragment.identity.fragment_id + for record in writer.replacements[0][1] + ), + tuple( + _fragment_id(document.identity, start, end) + for start, end in _expected_ranges(document.text, request.chunking) + ), + ) + self.assertEqual( + result.manifest.entries[0].source_digest, + _source_digest(document), + ) + self.assertEqual( + result.manifest.checkpoint.token, + _checkpoint_token(request, expected_manifest.entries), + ) + + def test_source_digest_distinguishes_attribute_order_duplicates_and_text( + self, + ) -> None: + base = _document("doc", text="same", attributes=(("a", "1"), ("b", "2"))) + variants = ( + _document("doc", text="same", attributes=(("b", "2"), ("a", "1"))), + _document("doc", text="same", attributes=(("a", "1"), ("a", "1"))), + _document("doc", text="same!", attributes=(("a", "1"), ("b", "2"))), + ) + + self.assertEqual(len({_source_digest(base), *map(_source_digest, variants)}), 4) + + def test_embedding_batches_are_bounded_ordered_and_never_empty(self) -> None: + document = _document("doc", text="abcdefghijklmn") + request = _request( + (document,), + chunking=ChunkingPolicy(3, 0), + batch_size=2, + ) + previous = _request( + (), projection=request.projection, chunking=request.chunking + ) + embedder, writer, _, _ = _collaborators(request) + + project_documents( + request, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertEqual( + embedder.embed_calls, + [("abc", "def"), ("ghi", "jkl"), ("mn",)], + ) + self.assertTrue(all(batch for batch in embedder.embed_calls)) + self.assertEqual( + tuple(record.fragment.text for record in writer.replacements[0][1]), + ("abc", "def", "ghi", "jkl", "mn"), + ) + + def test_empty_document_is_replaced_by_an_explicit_empty_record_tuple(self) -> None: + document = _document("doc", revision_id="revision-2", text="") + request = _request((document,)) + previous = _request( + (_document("doc", revision_id="revision-1", text="old"),), + projection=request.projection, + chunking=request.chunking, + ) + embedder, writer, _, _ = _collaborators(request) + + result = project_documents( + request, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertEqual(embedder.embed_calls, []) + self.assertEqual(writer.replacements, [(document.identity, ())]) + self.assertEqual(result.manifest.entries[0].fragment_count, 0) + + def test_document_and_mutation_order_is_canonical_by_opaque_document_id( + self, + ) -> None: + request = _request( + ( + _document("c", revision_id="new"), + _document("a", revision_id="new"), + ) + ) + previous = _request( + ( + _document("d", revision_id="old"), + _document("b", revision_id="old"), + ), + projection=request.projection, + chunking=request.chunking, + ) + embedder, writer, _, events = _collaborators(request) + + project_documents( + request, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + mutation_events = [ + event for event in events if event.startswith(("replace:", "delete:")) + ] + self.assertEqual( + mutation_events, + ["replace:a", "delete:b", "replace:c", "delete:d"], + ) + + def test_fixed_projection_is_identical_in_clean_processes(self) -> None: + script = textwrap.dedent( + f""" + import json + import sys + sys.path.insert(0, {os.fspath(_SOURCE_ROOT)!r}) + from generic_rag.contracts import * + from generic_rag.ports import Borrowed + from generic_rag.projection import rebuild_projection + + class Embedder: + identity = EmbeddingIdentity('model-v1', 2) + def embed(self, texts, /): + return tuple( + EmbeddingVector((len(text), index)) + for index, text in enumerate(texts) + ) + class Writer: + def __init__(self): self.records = () + def replace_document(self, document, records, /): self.records = records + def delete_document(self, document, /): return None + class Resetter: + def reset_corpus(self, corpus_id, /): return None + + document = Document( + DocumentIdentity(DocumentKey('corpus', 'doc/../A'), ' rev e\\u0301 '), + 'abcdef', + (('k', 'v'), ('k', 'v'), ('', '')), + ) + request = ProjectionRequest( + 'corpus', + ProjectionIdentity('schema-v1', EmbeddingIdentity('model-v1', 2)), + ChunkingPolicy(4, 1), + ProjectionLimits(1, 100, 2), + (document,), + ) + writer = Writer() + result = rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + Borrowed(Embedder()), + Borrowed(writer), + Borrowed(Resetter()), + ) + print(json.dumps({{ + 'source': result.manifest.entries[0].source_digest, + 'checkpoint': result.manifest.checkpoint.token, + 'fragments': [ + record.fragment.identity.fragment_id + for record in writer.records + ], + }}, sort_keys=True)) + """ + ) + outputs: list[dict[str, object]] = [] + for _ in range(2): + completed = subprocess.run( + (sys.executable, "-I", "-B", "-c", script), + check=False, + capture_output=True, + text=True, + timeout=10, + ) + self.assertEqual(completed.returncode, 0, completed.stderr) + outputs.append(cast(dict[str, object], json.loads(completed.stdout))) + + request = _request( + ( + _document( + "doc/../A", + revision_id=" rev e\u0301 ", + text="abcdef", + attributes=(("k", "v"), ("k", "v"), ("", "")), + ), + ), + chunking=ChunkingPolicy(4, 1), + ) + expected = _manifest(request) + self.assertEqual(outputs[0], outputs[1]) + self.assertEqual(outputs[0]["source"], expected.entries[0].source_digest) + self.assertEqual(outputs[0]["checkpoint"], expected.checkpoint.token) + self.assertEqual( + outputs[0]["fragments"], + [ + _fragment_id(request.documents[0].identity, start, end) + for start, end in _expected_ranges( + request.documents[0].text, + request.chunking, + ) + ], + ) + + +class IncrementalProjectionTests(unittest.TestCase): + def test_current_state_is_unchanged_without_touching_collaborators(self) -> None: + request = _request((_document("alpha"), _document("beta"))) + embedder, writer, _, events = _collaborators(request) + embedder.identity_failure = AssertionError("identity must not be read") + writer.replace_failure_at = 0 + + result = project_documents( + request, + _present(request), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertIs(result.status_before, ProjectionStateStatus.CURRENT) + self.assertIs(result.receipt.outcome, ProjectionOutcome.UNCHANGED) + self.assertEqual(result.receipt.attempted_documents, 0) + self.assertEqual(result.receipt.completed_documents, 0) + self.assertEqual(result.manifest, _manifest(request)) + self.assertEqual(events, []) + + def test_incompatible_state_statuses_fail_before_collaborator_effects(self) -> None: + request = _request((_document("alpha"),)) + schema_request = _request( + request.documents, + projection=_identity(schema_id="other-schema"), + ) + embedding_request = _request( + request.documents, + projection=_identity(model_id="other-model"), + ) + corpus_request = _request( + (_document("alpha", corpus_id="other"),), + projection=request.projection, + corpus_id="other", + ) + cases = ( + ( + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + ProjectionStateStatus.MISSING, + ), + ( + ProjectionStateSnapshot(ProjectionStateAvailability.CORRUPT, None), + ProjectionStateStatus.CORRUPT, + ), + (_present(schema_request), ProjectionStateStatus.SCHEMA_MISMATCH), + (_present(embedding_request), ProjectionStateStatus.EMBEDDING_MISMATCH), + (_present(corpus_request), ProjectionStateStatus.CORRUPT), + ) + + for state, expected_status in cases: + embedder, writer, _, events = _collaborators(request) + with self.subTest(status=expected_status): + with self.assertRaises(ProjectionStateError) as raised: + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + self.assertIs(raised.exception.status, expected_status) + self.assertEqual(events, []) + + def test_invalid_checkpoint_is_corrupt_before_any_effect(self) -> None: + request = _request((_document("alpha"),)) + valid = _manifest(request) + corrupt = ProjectionManifest( + valid.corpus_id, + valid.projection, + valid.chunking, + valid.entries, + ProjectionCheckpoint(valid.corpus_id, valid.projection, "wrong-token"), + ) + embedder, writer, _, events = _collaborators(request) + + with self.assertRaises(ProjectionStateError) as raised: + project_documents( + request, + ProjectionStateSnapshot( + ProjectionStateAvailability.PRESENT, + corrupt, + ), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertIs(raised.exception.status, ProjectionStateStatus.CORRUPT) + self.assertEqual(events, []) + + def test_same_revision_digest_or_fragment_count_drift_is_corrupt(self) -> None: + request = _request((_document("alpha"),)) + valid = _manifest(request) + original = valid.entries[0] + changed_entries = ( + ( + ProjectionManifestEntry( + original.document, + "sha256:" + "f" * 64, + original.fragment_count, + ), + ), + ( + ProjectionManifestEntry( + original.document, + original.source_digest, + original.fragment_count + 1, + ), + ), + ) + + for entries in changed_entries: + previous_request = _request( + request.documents, + projection=request.projection, + chunking=request.chunking, + ) + manifest = ProjectionManifest( + request.corpus_id, + request.projection, + request.chunking, + entries, + ProjectionCheckpoint( + request.corpus_id, + request.projection, + _checkpoint_token(previous_request, entries), + ), + ) + embedder, writer, _, events = _collaborators(request) + with self.subTest(entries=entries): + with self.assertRaises(ProjectionStateError) as raised: + project_documents( + request, + ProjectionStateSnapshot( + ProjectionStateAvailability.PRESENT, + manifest, + ), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + self.assertIs( + raised.exception.status, + ProjectionStateStatus.CORRUPT, + ) + self.assertEqual(events, []) + + def test_revision_and_chunking_changes_are_stale_and_replaced(self) -> None: + target = _request( + (_document("alpha", revision_id="revision-2", text="new text"),), + chunking=ChunkingPolicy(4, 1), + ) + previous_requests = ( + _request( + (_document("alpha", revision_id="revision-1", text="new text"),), + projection=target.projection, + chunking=target.chunking, + ), + _request( + target.documents, + projection=target.projection, + chunking=ChunkingPolicy(5, 0), + ), + ) + + for previous in previous_requests: + embedder, writer, _, _ = _collaborators(target) + with self.subTest(previous=previous): + result = project_documents( + target, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + self.assertIs(result.status_before, ProjectionStateStatus.STALE) + self.assertEqual( + [identity for identity, _ in writer.replacements], + [target.documents[0].identity], + ) + + def test_same_revision_source_change_is_corrupt_not_stale(self) -> None: + target = _request( + (_document("alpha", revision_id="revision-2", text="new text"),) + ) + previous = _request( + (_document("alpha", revision_id="revision-2", text="old text"),), + projection=target.projection, + chunking=target.chunking, + ) + embedder, writer, _, events = _collaborators(target) + + with self.assertRaises(ProjectionStateError) as raised: + project_documents( + target, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertIs(raised.exception.status, ProjectionStateStatus.CORRUPT) + self.assertEqual(events, []) + + def test_delete_only_plan_does_not_access_embedder(self) -> None: + target = _request(()) + previous = _request( + (_document("zeta"), _document("alpha")), + projection=target.projection, + chunking=target.chunking, + ) + embedder, writer, _, events = _collaborators(target) + embedder.identity_failure = AssertionError("delete-only must not embed") + + result = project_documents( + target, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertEqual( + writer.deletions, + [DocumentKey("corpus", "alpha"), DocumentKey("corpus", "zeta")], + ) + self.assertEqual(embedder.identity_calls, 0) + self.assertFalse(any(event.startswith("embed:") for event in events)) + self.assertEqual(result.receipt.attempted_documents, 2) + self.assertEqual(result.receipt.completed_documents, 2) + + def test_success_receipt_counts_only_changed_and_removed_documents(self) -> None: + unchanged = _document("same", revision_id="r1", text="same") + target = _request( + ( + unchanged, + _document("added", revision_id="r1", text="added"), + _document("changed", revision_id="r2", text="new"), + ) + ) + previous = _request( + ( + unchanged, + _document("changed", revision_id="r1", text="old"), + _document("removed", revision_id="r1", text="gone"), + ), + projection=target.projection, + chunking=target.chunking, + ) + embedder, writer, _, _ = _collaborators(target) + + result = project_documents( + target, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertIs(result.receipt.outcome, ProjectionOutcome.COMPLETED) + self.assertEqual(result.receipt.attempted_documents, 3) + self.assertEqual(result.receipt.completed_documents, 3) + self.assertEqual(result.receipt.checkpoint, result.manifest.checkpoint) + self.assertNotIn(unchanged.identity, [item[0] for item in writer.replacements]) + + +class RebuildProjectionTests(unittest.TestCase): + def test_nonempty_rebuild_validates_identity_once_before_reset_and_reuses_it( + self, + ) -> None: + request = _request((_document("alpha"), _document("beta"))) + embedder, writer, resetter, events = _collaborators(request) + + result = rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + self.assertEqual(embedder.identity_calls, 1) + self.assertEqual(resetter.corpora, ["corpus"]) + self.assertEqual( + [identity.key.document_id for identity, _ in writer.replacements], + ["alpha", "beta"], + ) + self.assertEqual(events[0:2], ["identity", "reset:corpus"]) + self.assertEqual(sum(event == "identity" for event in events), 1) + self.assertIs(result.status_before, ProjectionStateStatus.MISSING) + self.assertEqual(result.receipt.attempted_documents, 2) + + def test_wrong_embedder_identity_fails_before_reset_without_a_cause(self) -> None: + request = _request((_document("alpha"),)) + events: list[str] = [] + embedder = _FakeEmbedder( + request.projection.embedding, + events, + identity_value=EmbeddingIdentity("other-model", 2), + ) + writer = _FakeWriter(events) + resetter = _FakeResetter(events) + + with self.assertRaises(ProjectionOperationError) as raised: + rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + self.assertIs( + raised.exception.stage, + ProjectionFailureStage.EMBEDDER_IDENTITY, + ) + self.assertIsNone(raised.exception.affected_document) + self.assertIsNone(raised.exception.__cause__) + self.assertIsNotNone(raised.exception.receipt) + assert raised.exception.receipt is not None + self.assertIs(raised.exception.receipt.outcome, ProjectionOutcome.FAILED) + self.assertEqual(resetter.corpora, []) + self.assertEqual(writer.replacements, []) + + def test_identity_exception_is_preserved_as_cause_before_reset(self) -> None: + request = _request((_document("alpha"),)) + failure = RuntimeError("identity provider failed") + events: list[str] = [] + embedder = _FakeEmbedder( + request.projection.embedding, + events, + identity_failure=failure, + ) + writer = _FakeWriter(events) + resetter = _FakeResetter(events) + + with self.assertRaises(ProjectionOperationError) as raised: + rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + self.assertIs(raised.exception.__cause__, failure) + self.assertEqual(resetter.corpora, []) + + def test_empty_rebuild_resets_only_without_embedder_or_writer_access(self) -> None: + request = _request(()) + embedder, writer, resetter, events = _collaborators(request) + embedder.identity_failure = AssertionError("empty rebuild must not embed") + writer.replace_failure_at = 0 + + result = rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.CORRUPT, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + self.assertEqual(events, ["reset:corpus"]) + self.assertEqual(embedder.identity_calls, 0) + self.assertEqual(writer.replacements, []) + self.assertIs(result.status_before, ProjectionStateStatus.CORRUPT) + self.assertEqual(result.receipt.attempted_documents, 0) + self.assertEqual(result.receipt.completed_documents, 0) + self.assertIs(result.receipt.outcome, ProjectionOutcome.COMPLETED) + + def test_rebuild_accepts_every_state_status_and_reports_it_truthfully(self) -> None: + request = _request((_document("alpha"),)) + other_schema = _request( + request.documents, + projection=_identity(schema_id="other"), + ) + other_embedding = _request( + request.documents, + projection=_identity(model_id="other"), + ) + stale = _request( + (_document("alpha", revision_id="old"),), + projection=request.projection, + ) + states = ( + ( + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + ProjectionStateStatus.MISSING, + ), + ( + ProjectionStateSnapshot(ProjectionStateAvailability.CORRUPT, None), + ProjectionStateStatus.CORRUPT, + ), + (_present(request), ProjectionStateStatus.CURRENT), + (_present(stale), ProjectionStateStatus.STALE), + (_present(other_schema), ProjectionStateStatus.SCHEMA_MISMATCH), + (_present(other_embedding), ProjectionStateStatus.EMBEDDING_MISMATCH), + ) + + for state, status in states: + embedder, writer, resetter, _ = _collaborators(request) + with self.subTest(status=status): + result = rebuild_projection( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + self.assertIs(result.status_before, status) + self.assertEqual(resetter.corpora, ["corpus"]) + self.assertIs(result.receipt.outcome, ProjectionOutcome.COMPLETED) + + def test_reset_exception_has_failed_receipt_and_preserves_cause(self) -> None: + request = _request((_document("alpha"), _document("beta"))) + failure = RuntimeError("reset failed") + events: list[str] = [] + embedder = _FakeEmbedder(request.projection.embedding, events) + writer = _FakeWriter(events) + resetter = _FakeResetter(events, failure=failure) + + with self.assertRaises(ProjectionOperationError) as raised: + rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + error = raised.exception + self.assertIs(error.stage, ProjectionFailureStage.RESET) + self.assertIsNone(error.affected_document) + self.assertIs(error.__cause__, failure) + self.assertIsNotNone(error.receipt) + assert error.receipt is not None + self.assertIs(error.receipt.outcome, ProjectionOutcome.FAILED) + self.assertEqual(error.receipt.attempted_documents, 2) + self.assertEqual(error.receipt.completed_documents, 0) + self.assertIsNone(error.receipt.checkpoint) + self.assertEqual(writer.replacements, []) + + def test_non_none_reset_result_is_failure_without_cause(self) -> None: + for documents, expected_receipt in ( + ((_document("alpha"),), True), + ((), False), + ): + request = _request(documents) + events: list[str] = [] + embedder = _FakeEmbedder(request.projection.embedding, events) + writer = _FakeWriter(events) + resetter = _FakeResetter(events, result=False) + with self.subTest(documents=documents): + with self.assertRaises(ProjectionOperationError) as raised: + rebuild_projection( + request, + ProjectionStateSnapshot( + ProjectionStateAvailability.MISSING, + None, + ), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + self.assertIs(raised.exception.stage, ProjectionFailureStage.RESET) + self.assertIsNone(raised.exception.__cause__) + self.assertEqual( + raised.exception.receipt is not None, + expected_receipt, + ) + + +class ProjectionFailureTests(unittest.TestCase): + def _incremental_target( + self, + documents: tuple[Document, ...], + *, + batch_size: int = 2, + ) -> tuple[ProjectionRequest, ProjectionStateSnapshot]: + target = _request(documents, batch_size=batch_size) + previous = _request( + (), + projection=target.projection, + chunking=target.chunking, + batch_size=batch_size, + ) + return target, _present(previous) + + def test_embedding_exception_is_called_once_and_preserved_as_cause(self) -> None: + request, state = self._incremental_target((_document("alpha", text="x"),)) + failure = RuntimeError("provider failed") + events: list[str] = [] + embedder = _FakeEmbedder( + request.projection.embedding, + events, + embed_failure_at=0, + embed_failure=failure, + ) + writer = _FakeWriter(events) + + with self.assertRaises(ProjectionOperationError) as raised: + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + error = raised.exception + self.assertIs(error.stage, ProjectionFailureStage.EMBEDDING) + self.assertEqual(error.affected_document, DocumentKey("corpus", "alpha")) + self.assertIs(error.__cause__, failure) + self.assertEqual(embedder.embed_calls, [("x",)]) + self.assertEqual(writer.replacements, []) + self.assertIsNotNone(error.receipt) + assert error.receipt is not None + self.assertIs(error.receipt.outcome, ProjectionOutcome.FAILED) + self.assertEqual(error.receipt.completed_documents, 0) + self.assertIsNone(error.receipt.checkpoint) + + def test_malformed_embedding_outputs_fail_without_internal_cause(self) -> None: + request, state = self._incremental_target((_document("alpha", text="x"),)) + nonfinite = EmbeddingVector((1.0, 2.0)) + object.__setattr__(nonfinite, "values", (float("nan"), 2.0)) + noncanonical = EmbeddingVector((1.0, 2.0)) + object.__setattr__(noncanonical, "values", (1, 2.0)) + outputs: tuple[object, ...] = ( + [EmbeddingVector((1.0, 2.0))], + (), + (object(),), + (EmbeddingVector((1.0,)),), + (EmbeddingVector((1.0, 2.0, 3.0)),), + (nonfinite,), + (noncanonical,), + ) + + for output in outputs: + events: list[str] = [] + embedder = _FakeEmbedder( + request.projection.embedding, + events, + output=_constant_output(output), + ) + writer = _FakeWriter(events) + with self.subTest(output=output): + with self.assertRaises(ProjectionOperationError) as raised: + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + self.assertIs( + raised.exception.stage, + ProjectionFailureStage.EMBEDDING, + ) + self.assertIsNone(raised.exception.__cause__) + self.assertEqual(len(embedder.embed_calls), 1) + self.assertEqual(writer.replacements, []) + + def test_second_document_embedding_failure_reports_partial_completion(self) -> None: + request, state = self._incremental_target( + (_document("alpha"), _document("beta")), + batch_size=2, + ) + events: list[str] = [] + failure = RuntimeError("second document failed") + embedder = _FakeEmbedder( + request.projection.embedding, + events, + embed_failure_at=2, + embed_failure=failure, + ) + writer = _FakeWriter(events) + + with self.assertRaises(ProjectionOperationError) as raised: + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + error = raised.exception + self.assertEqual(error.affected_document, DocumentKey("corpus", "beta")) + self.assertIsNotNone(error.receipt) + assert error.receipt is not None + self.assertIs(error.receipt.outcome, ProjectionOutcome.PARTIAL) + self.assertEqual(error.receipt.attempted_documents, 2) + self.assertEqual(error.receipt.completed_documents, 1) + self.assertEqual( + [identity.key.document_id for identity, _ in writer.replacements], + ["alpha"], + ) + self.assertIs(error.__cause__, failure) + + def test_replacement_exception_has_matching_failed_receipt_and_cause(self) -> None: + request, state = self._incremental_target((_document("alpha", text="x"),)) + events: list[str] = [] + failure = RuntimeError("writer failed") + embedder = _FakeEmbedder(request.projection.embedding, events) + writer = _FakeWriter( + events, + replace_failure_at=0, + replace_failure=failure, + ) + + with self.assertRaises(ProjectionOperationError) as raised: + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + error = raised.exception + self.assertIs(error.stage, ProjectionFailureStage.REPLACEMENT) + self.assertEqual(error.affected_document, DocumentKey("corpus", "alpha")) + self.assertIs(error.__cause__, failure) + self.assertIsNotNone(error.receipt) + assert error.receipt is not None + self.assertIs(error.receipt.outcome, ProjectionOutcome.FAILED) + self.assertEqual(len(writer.replacements), 1) + + def test_non_none_replacement_result_is_failure_without_cause(self) -> None: + request, state = self._incremental_target((_document("alpha", text="x"),)) + events: list[str] = [] + embedder = _FakeEmbedder(request.projection.embedding, events) + writer = _FakeWriter(events, replace_result=False) + + with self.assertRaises(ProjectionOperationError) as raised: + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertIs(raised.exception.stage, ProjectionFailureStage.REPLACEMENT) + self.assertIsNone(raised.exception.__cause__) + self.assertIsNotNone(raised.exception.receipt) + assert raised.exception.receipt is not None + self.assertEqual(raised.exception.receipt.completed_documents, 0) + + def test_deletion_exception_after_replacement_reports_partial_receipt(self) -> None: + target = _request((_document("alpha", text="x"),)) + previous = _request( + (_document("beta", text="x"),), + projection=target.projection, + chunking=target.chunking, + ) + events: list[str] = [] + failure = RuntimeError("delete failed") + embedder = _FakeEmbedder(target.projection.embedding, events) + writer = _FakeWriter( + events, + delete_failure_at=0, + delete_failure=failure, + ) + + with self.assertRaises(ProjectionOperationError) as raised: + project_documents( + target, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + error = raised.exception + self.assertIs(error.stage, ProjectionFailureStage.DELETION) + self.assertEqual(error.affected_document, DocumentKey("corpus", "beta")) + self.assertIs(error.__cause__, failure) + self.assertIsNotNone(error.receipt) + assert error.receipt is not None + self.assertIs(error.receipt.outcome, ProjectionOutcome.PARTIAL) + self.assertEqual(error.receipt.attempted_documents, 2) + self.assertEqual(error.receipt.completed_documents, 1) + self.assertIsNone(error.receipt.checkpoint) + self.assertEqual( + [event for event in events if event.startswith(("replace:", "delete:"))], + ["replace:alpha", "delete:beta"], + ) + + def test_non_none_deletion_result_is_failed_without_cause(self) -> None: + target = _request(()) + previous = _request( + (_document("alpha", text="x"),), + projection=target.projection, + chunking=target.chunking, + ) + events: list[str] = [] + embedder = _FakeEmbedder(target.projection.embedding, events) + writer = _FakeWriter(events, delete_result=0) + + with self.assertRaises(ProjectionOperationError) as raised: + project_documents( + target, + _present(previous), + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + + self.assertIs(raised.exception.stage, ProjectionFailureStage.DELETION) + self.assertIsNone(raised.exception.__cause__) + self.assertIsNotNone(raised.exception.receipt) + assert raised.exception.receipt is not None + self.assertIs(raised.exception.receipt.outcome, ProjectionOutcome.FAILED) + + def test_rebuild_replacement_failure_never_claims_a_checkpoint(self) -> None: + request = _request((_document("alpha"), _document("beta"))) + events: list[str] = [] + embedder = _FakeEmbedder(request.projection.embedding, events) + writer = _FakeWriter(events, replace_failure_at=1) + resetter = _FakeResetter(events) + + with self.assertRaises(ProjectionOperationError) as raised: + rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.CORRUPT, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + error = raised.exception + self.assertIs(error.stage, ProjectionFailureStage.REPLACEMENT) + self.assertEqual(error.affected_document, DocumentKey("corpus", "beta")) + self.assertIsNotNone(error.receipt) + assert error.receipt is not None + self.assertIs(error.receipt.outcome, ProjectionOutcome.PARTIAL) + self.assertEqual(error.receipt.completed_documents, 1) + self.assertIsNone(error.receipt.checkpoint) + self.assertEqual(resetter.corpora, ["corpus"]) + + def test_control_flow_exceptions_are_never_translated_or_retried(self) -> None: + for failure in (KeyboardInterrupt("interrupt"), SystemExit("exit")): + request, state = self._incremental_target((_document("alpha", text="x"),)) + events: list[str] = [] + embedder = _FakeEmbedder( + request.projection.embedding, + events, + embed_failure_at=0, + embed_failure=failure, + ) + writer = _FakeWriter(events) + with self.subTest(failure=type(failure).__name__): + with self.assertRaises(type(failure)) as raised: + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + self.assertIs(raised.exception, failure) + self.assertEqual(len(embedder.embed_calls), 1) + self.assertEqual(writer.replacements, []) + + def test_reset_control_flow_exception_is_not_translated_or_retried(self) -> None: + request = _request((_document("alpha"),)) + failure = KeyboardInterrupt("reset interrupted") + events: list[str] = [] + embedder = _FakeEmbedder(request.projection.embedding, events) + writer = _FakeWriter(events) + resetter = _FakeResetter(events, failure=failure) + + with self.assertRaises(KeyboardInterrupt) as raised: + rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + self.assertIs(raised.exception, failure) + self.assertEqual(resetter.corpora, ["corpus"]) + self.assertEqual(writer.replacements, []) + + def test_workflows_never_enter_close_shutdown_or_suppress_resources(self) -> None: + request, state = self._incremental_target((_document("alpha", text="x"),)) + embedder, writer, resetter, _ = _collaborators(request) + + project_documents( + request, + state, + _borrow_embedder(embedder), + _borrow_writer(writer), + ) + rebuild_projection( + request, + ProjectionStateSnapshot(ProjectionStateAvailability.MISSING, None), + _borrow_embedder(embedder), + _borrow_writer(writer), + _borrow_resetter(resetter), + ) + + self.assertEqual(embedder.lifecycle_calls, []) + self.assertEqual(writer.lifecycle_calls, []) + self.assertEqual(resetter.lifecycle_calls, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_projection_contracts.py b/tests/test_projection_contracts.py index 7f59932..e4c2cb1 100644 --- a/tests/test_projection_contracts.py +++ b/tests/test_projection_contracts.py @@ -1,4 +1,4 @@ -"""Contract tests for projection identities, checkpoints, and receipts.""" +"""Contract tests for immutable projection values and truthful state.""" from __future__ import annotations @@ -7,11 +7,23 @@ from typing import cast from generic_rag.contracts import ( + ChunkingPolicy, + Document, + DocumentIdentity, + DocumentKey, EmbeddingIdentity, ProjectionCheckpoint, ProjectionIdentity, + ProjectionLimits, + ProjectionManifest, + ProjectionManifestEntry, ProjectionOutcome, ProjectionReceipt, + ProjectionRequest, + ProjectionResult, + ProjectionStateAvailability, + ProjectionStateSnapshot, + ProjectionStateStatus, ) from generic_rag.errors import ContractValidationError @@ -49,6 +61,60 @@ def _checkpoint( ) +def _document( + document_id: str = "document", + *, + corpus_id: str = "corpus", + revision_id: str = "revision", + text: str = "text", +) -> Document: + return Document( + DocumentIdentity(DocumentKey(corpus_id, document_id), revision_id), + text, + ) + + +def _entry( + document_id: str = "document", + *, + corpus_id: str = "corpus", + revision_id: str = "revision", + digest_digit: str = "0", + fragment_count: int = 1, +) -> ProjectionManifestEntry: + return ProjectionManifestEntry( + _document( + document_id, + corpus_id=corpus_id, + revision_id=revision_id, + ).identity, + f"sha256:{digest_digit * 64}", + fragment_count, + ) + + +def _manifest( + *, + corpus_id: str = "corpus", + projection: ProjectionIdentity | None = None, + chunking: ChunkingPolicy | None = None, + entries: tuple[ProjectionManifestEntry, ...] | None = None, + token: str = "checkpoint", +) -> ProjectionManifest: + resolved_projection = projection or _projection() + return ProjectionManifest( + corpus_id, + resolved_projection, + chunking or ChunkingPolicy(4, 1), + (_entry(corpus_id=corpus_id),) if entries is None else entries, + _checkpoint( + corpus_id=corpus_id, + projection=resolved_projection, + token=token, + ), + ) + + class ProjectionContractTests(unittest.TestCase): def test_projection_fields_are_exact_frozen_and_slotted(self) -> None: self.assertEqual( @@ -372,5 +438,482 @@ def test_receipt_rejects_inexact_nested_and_discrete_types(self) -> None: ) +class ProjectionWorkflowContractTests(unittest.TestCase): + def test_new_projection_values_have_exact_frozen_slotted_fields(self) -> None: + expected_fields = { + ChunkingPolicy: ("max_fragment_codepoints", "overlap_codepoints"), + ProjectionLimits: ( + "max_documents", + "max_document_codepoints", + "max_embedding_batch_size", + ), + ProjectionRequest: ( + "corpus_id", + "projection", + "chunking", + "limits", + "documents", + ), + ProjectionManifestEntry: ( + "document", + "source_digest", + "fragment_count", + ), + ProjectionManifest: ( + "corpus_id", + "projection", + "chunking", + "entries", + "checkpoint", + ), + ProjectionStateSnapshot: ("availability", "manifest"), + ProjectionResult: ("status_before", "receipt", "manifest"), + } + projection = _projection() + manifest = _manifest(projection=projection) + instances = ( + ChunkingPolicy(4, 1), + ProjectionLimits(2, 20, 2), + ProjectionRequest( + "corpus", + projection, + ChunkingPolicy(4, 1), + ProjectionLimits(2, 20, 2), + (_document(),), + ), + manifest.entries[0], + manifest, + ProjectionStateSnapshot(ProjectionStateAvailability.PRESENT, manifest), + ProjectionResult( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "corpus", + projection, + ProjectionOutcome.COMPLETED, + 1, + 1, + manifest.checkpoint, + ), + manifest, + ), + ) + + for instance in instances: + with self.subTest(value=type(instance).__name__): + self.assertEqual( + tuple(field.name for field in fields(type(instance))), + expected_fields[type(instance)], + ) + self.assertFalse(hasattr(instance, "__dict__")) + with self.assertRaises(FrozenInstanceError): + setattr(instance, expected_fields[type(instance)][0], object()) + self.assertIsInstance(hash(instance), int) + + def test_chunking_policy_uses_positive_codepoint_size_and_bounded_overlap( + self, + ) -> None: + self.assertEqual(ChunkingPolicy(4, 0), ChunkingPolicy(4, 0)) + self.assertEqual(ChunkingPolicy(4, 3).overlap_codepoints, 3) + + for maximum, overlap in ( + (0, 0), + (-1, 0), + (4, -1), + (4, 4), + (4, 5), + (True, 0), + (4, True), + (4.0, 0), + (4, 1.0), + (_IntegerSubclass(4), 0), + ): + with self.subTest(maximum=maximum, overlap=overlap): + with self.assertRaises(ContractValidationError): + ChunkingPolicy(cast(int, maximum), cast(int, overlap)) + + def test_projection_limits_require_three_positive_exact_integers(self) -> None: + self.assertEqual(ProjectionLimits(1, 1, 1), ProjectionLimits(1, 1, 1)) + + for field_index in range(3): + for invalid in (0, -1, True, 1.0, _IntegerSubclass(1)): + values: list[object] = [2, 20, 3] + values[field_index] = invalid + with self.subTest(field=field_index, value=invalid): + with self.assertRaises(ContractValidationError): + ProjectionLimits( + cast(int, values[0]), + cast(int, values[1]), + cast(int, values[2]), + ) + + def test_request_preserves_opaque_values_and_canonicalizes_document_order( + self, + ) -> None: + documents = ( + _document("zeta", text=""), + _document("alpha", revision_id=" revision/../2 ", text="e\u0301"), + ) + request = ProjectionRequest( + "corpus", + _projection(), + ChunkingPolicy(2, 1), + ProjectionLimits(2, 2, 1), + documents, + ) + + self.assertEqual( + tuple(document.identity.key.document_id for document in request.documents), + ("alpha", "zeta"), + ) + self.assertEqual(request.documents[0].identity.revision_id, " revision/../2 ") + self.assertEqual(request.documents[0].text, "e\u0301") + self.assertEqual( + ProjectionRequest( + "corpus", + _projection(), + ChunkingPolicy(2, 1), + ProjectionLimits(2, 2, 1), + (), + ).documents, + (), + ) + + def test_request_rejects_inexact_fields_and_document_bound_violations( + self, + ) -> None: + projection = _projection() + chunking = ChunkingPolicy(4, 1) + limits = ProjectionLimits(1, 4, 2) + valid = _document(text="four") + invalid_requests = ( + ("", projection, chunking, limits, (valid,)), + (_StringSubclass("corpus"), projection, chunking, limits, (valid,)), + ("corpus", object(), chunking, limits, (valid,)), + ("corpus", projection, object(), limits, (valid,)), + ("corpus", projection, chunking, object(), (valid,)), + ("corpus", projection, chunking, limits, [valid]), + ("corpus", projection, chunking, limits, (object(),)), + ("corpus", projection, chunking, limits, (valid, _document("two"))), + ("corpus", projection, chunking, limits, (_document(text="12345"),)), + ( + "corpus", + projection, + chunking, + limits, + (_document(corpus_id="other"),), + ), + ( + "corpus", + projection, + chunking, + ProjectionLimits(2, 4, 2), + (_document(revision_id="r1"), _document(revision_id="r2")), + ), + ) + for values in invalid_requests: + with self.subTest(values=values): + with self.assertRaises(ContractValidationError): + ProjectionRequest( + values[0], + cast(ProjectionIdentity, values[1]), + cast(ChunkingPolicy, values[2]), + cast(ProjectionLimits, values[3]), + cast(tuple[Document, ...], values[4]), + ) + + def test_manifest_entry_requires_lowercase_sha256_and_nonnegative_count( + self, + ) -> None: + entry = _entry(fragment_count=0) + self.assertEqual(entry.source_digest, "sha256:" + "0" * 64) + self.assertEqual(entry.fragment_count, 0) + + for digest in ( + "", + "0" * 64, + "sha256:" + "A" * 64, + "sha256:" + "0" * 63, + "sha256:" + "0" * 65, + _StringSubclass("sha256:" + "0" * 64), + ): + with self.subTest(digest=digest): + with self.assertRaises(ContractValidationError): + ProjectionManifestEntry( + _document().identity, + digest, + 1, + ) + for count in (-1, True, 1.0, _IntegerSubclass(1)): + with self.subTest(count=count): + with self.assertRaises(ContractValidationError): + ProjectionManifestEntry( + _document().identity, + "sha256:" + "0" * 64, + cast(int, count), + ) + with self.assertRaises(ContractValidationError): + ProjectionManifestEntry( + cast(DocumentIdentity, object()), + "sha256:" + "0" * 64, + 1, + ) + + def test_manifest_canonicalizes_entries_and_preserves_checkpoint(self) -> None: + projection = _projection() + checkpoint = _checkpoint(projection=projection, token=" token/../x ") + zeta = _entry("zeta", digest_digit="1") + alpha = _entry("alpha", digest_digit="2") + manifest = ProjectionManifest( + "corpus", + projection, + ChunkingPolicy(4, 1), + (zeta, alpha), + checkpoint, + ) + + self.assertEqual(manifest.entries, (alpha, zeta)) + self.assertIs(manifest.checkpoint, checkpoint) + self.assertEqual(manifest.checkpoint.token, " token/../x ") + + def test_manifest_rejects_mismatched_inexact_or_duplicate_content(self) -> None: + projection = _projection() + chunking = ChunkingPolicy(4, 1) + checkpoint = _checkpoint(projection=projection) + entry = _entry() + invalid_values = ( + ("", projection, chunking, (entry,), checkpoint), + ("corpus", object(), chunking, (entry,), checkpoint), + ("corpus", projection, object(), (entry,), checkpoint), + ("corpus", projection, chunking, [entry], checkpoint), + ("corpus", projection, chunking, (object(),), checkpoint), + ("corpus", projection, chunking, (entry,), object()), + ( + "corpus", + projection, + chunking, + (_entry(corpus_id="other"),), + checkpoint, + ), + ("corpus", projection, chunking, (entry, entry), checkpoint), + ( + "corpus", + projection, + chunking, + (entry,), + _checkpoint(corpus_id="other", projection=projection), + ), + ( + "corpus", + projection, + chunking, + (entry,), + _checkpoint(projection=_projection(schema_id="other")), + ), + ) + for values in invalid_values: + with self.subTest(values=values): + with self.assertRaises(ContractValidationError): + ProjectionManifest( + values[0], + cast(ProjectionIdentity, values[1]), + cast(ChunkingPolicy, values[2]), + cast(tuple[ProjectionManifestEntry, ...], values[3]), + cast(ProjectionCheckpoint, values[4]), + ) + + def test_state_availability_is_an_exact_closed_string_enum(self) -> None: + self.assertEqual( + tuple( + (member.name, member.value) for member in ProjectionStateAvailability + ), + (("MISSING", "missing"), ("PRESENT", "present"), ("CORRUPT", "corrupt")), + ) + for invalid in ("MISSING", "", "unknown", None, 1, object()): + with self.subTest(value=invalid): + with self.assertRaises(ContractValidationError): + ProjectionStateAvailability(cast(str, invalid)) + + def test_state_snapshot_requires_manifest_only_when_present(self) -> None: + manifest = _manifest() + self.assertIs( + ProjectionStateSnapshot( + ProjectionStateAvailability.PRESENT, + manifest, + ).manifest, + manifest, + ) + for availability in ( + ProjectionStateAvailability.MISSING, + ProjectionStateAvailability.CORRUPT, + ): + self.assertIsNone(ProjectionStateSnapshot(availability, None).manifest) + with self.assertRaises(ContractValidationError): + ProjectionStateSnapshot(availability, manifest) + with self.assertRaises(ContractValidationError): + ProjectionStateSnapshot(ProjectionStateAvailability.PRESENT, None) + with self.assertRaises(ContractValidationError): + ProjectionStateSnapshot( + cast(ProjectionStateAvailability, "present"), + manifest, + ) + + def test_state_status_is_an_exact_closed_string_enum(self) -> None: + self.assertEqual( + tuple((member.name, member.value) for member in ProjectionStateStatus), + ( + ("MISSING", "missing"), + ("CURRENT", "current"), + ("STALE", "stale"), + ("CORRUPT", "corrupt"), + ("SCHEMA_MISMATCH", "schema_mismatch"), + ("EMBEDDING_MISMATCH", "embedding_mismatch"), + ), + ) + for invalid in ("CURRENT", "", "unknown", None, 1, object()): + with self.subTest(value=invalid): + with self.assertRaises(ContractValidationError): + ProjectionStateStatus(cast(str, invalid)) + + def test_result_accepts_matching_completed_and_truthful_unchanged_values( + self, + ) -> None: + manifest = _manifest() + completed = ProjectionResult( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.COMPLETED, + 2, + 2, + manifest.checkpoint, + ), + manifest, + ) + unchanged = ProjectionResult( + ProjectionStateStatus.CURRENT, + ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.UNCHANGED, + 0, + 0, + manifest.checkpoint, + ), + manifest, + ) + + self.assertEqual(completed.status_before, ProjectionStateStatus.STALE) + self.assertEqual(unchanged.receipt.attempted_documents, 0) + + def test_result_rejects_inexact_mismatched_or_untruthful_values(self) -> None: + manifest = _manifest() + completed = ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.COMPLETED, + 1, + 1, + manifest.checkpoint, + ) + invalid_values = ( + ("stale", completed, manifest), + (ProjectionStateStatus.STALE, object(), manifest), + (ProjectionStateStatus.STALE, completed, object()), + ( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "other", + manifest.projection, + ProjectionOutcome.COMPLETED, + 1, + 1, + _checkpoint(corpus_id="other", projection=manifest.projection), + ), + manifest, + ), + ( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.PARTIAL, + 2, + 1, + None, + ), + manifest, + ), + ( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.FAILED, + 1, + 0, + None, + ), + manifest, + ), + ( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "corpus", + _projection(schema_id="other"), + ProjectionOutcome.COMPLETED, + 1, + 1, + _checkpoint(projection=_projection(schema_id="other")), + ), + manifest, + ), + ( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.COMPLETED, + 1, + 1, + _checkpoint(projection=manifest.projection, token="other"), + ), + manifest, + ), + ( + ProjectionStateStatus.STALE, + ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.UNCHANGED, + 0, + 0, + manifest.checkpoint, + ), + manifest, + ), + ( + ProjectionStateStatus.CURRENT, + ProjectionReceipt( + "corpus", + manifest.projection, + ProjectionOutcome.UNCHANGED, + 1, + 1, + manifest.checkpoint, + ), + manifest, + ), + ) + for status, receipt, value_manifest in invalid_values: + with self.subTest(status=status, receipt=receipt): + with self.assertRaises(ContractValidationError): + ProjectionResult( + cast(ProjectionStateStatus, status), + cast(ProjectionReceipt, receipt), + cast(ProjectionManifest, value_manifest), + ) + + if __name__ == "__main__": unittest.main() From 2470b52c16774e01ef4dad2f1541517b3a187669 Mon Sep 17 00:00:00 2001 From: rica-v3 Date: Tue, 25 Aug 2026 17:39:26 +0900 Subject: [PATCH 2/2] Clarify incremental projection state handling Document that only compatible present state is projected and that incompatible snapshots raise before collaborator effects.\n\nRefs #3 --- src/generic_rag/projection.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/generic_rag/projection.py b/src/generic_rag/projection.py index ac7035f..a8ac690 100644 --- a/src/generic_rag/projection.py +++ b/src/generic_rag/projection.py @@ -743,7 +743,10 @@ def project_documents( writer: Borrowed[VectorIndexWriter], /, ) -> ProjectionResult: - """Incrementally project a valid present snapshot to a complete target.""" + """Project a compatible present state to the complete target. + + Incompatible state raises ``ProjectionStateError`` before collaborator effects. + """ canonical_request = _validate_request(request) canonical_state = _validate_state(state)