feat(core): SPI split — RetrievalBackend / IndexBackend (Step 1.1b)

TRACKER.md

@@ -10,7 +10,7 @@
 
 **Last updated:** 2026-05-24
 **Current phase:** Phase 1 — Ingestion + Knowledge Store
-**Next action:** Phase 1 Step 1.1b — SPI split (Retrieval/Index, bulk + streaming + ID-only methods, IndexHint)
+**Next action:** Phase 1 Step 1.1c — PolicyEngine package (`rag-policy`): `PolicyEngine` SPI + noop impl as single PDP for ACL/PII/quotas/redaction
 
 > **Refactor window (Steps 1.1a–1.1f):** Before resuming the connectors framework (1.2), we insert a six-step refactor that locks in architecture + optimization decisions which are very expensive to retrofit later (PolicyEngine PDP, RequestContext-threaded SPIs, split Retrieval/Index backends, bulk + streaming + ID-only methods, Pipeline + Batcher primitives, three-way cache split, hot-path discipline). See [docs/adr/ADR-0005…0009] and [docs/architecture/policy-engine.md], [request-context.md], [caching.md], [performance.md].
 
@@ -32,14 +32,14 @@
 | Phase | Title | Steps | ✅ Done | Remaining |
 |-------|-------|------:|-------:|----------:|
 | 0 | Foundation | 13 | **13** | 0 |
-| 1 | Ingestion + Knowledge Store | 16 | **2** | 14 |
+| 1 | Ingestion + Knowledge Store | 16 | **3** | 13 |
 | 2 | Retrieval Engine | 11 | 0 | 11 |
 | 3 | Gateway & Agent Runtime | 11 | 0 | 11 |
 | 4 | Reliability | 6 | 0 | 6 |
 | 5 | Eval & Observability | 7 | 0 | 7 |
 | 6 | Governance & Tenancy | 10 | 0 | 10 |
 | 7 | Pilot, Harden, GA | 10 | 0 | 10 |
-| **Total** | | **84** | **15** | **69** |
+| **Total** | | **84** | **16** | **68** |
 
 ---
 
@@ -69,7 +69,7 @@
 |------|-------|--------|--------|----|-----------------|
 | 1.1 | Storage backends | ✅ | `build/phase-1/step-1.1-storage-backends` | [#40](https://github.com/officialCodeWork/AgentContextOS/pull/40) | `rag-backends` package: `PgVectorStore` (asyncpg + pgvector, ivfflat), `QdrantVectorStore` (query_points API), `RedisCache`, `S3Storage` (aioboto3, MinIO-compatible), `LocalFileStorage`; integration tests (skip-if-no-service); MinIO added to dev stack; `task test-integration` + `task test-backends`; ADR-0004 |
 | 1.1a | Core type & SPI refactor | ✅ | `build/phase-1/step-1.1a-core-type-spi-refactor` | _pending_ | `RequestContext` frozen model threaded through every SPI; `tenant_id` + `acl_labels` typed required on `Chunk`/`Embedding` (not metadata dict); `trust_level` on `Chunk` for prompt-injection defense; `dtype` on `Embedding` (float32/int8/binary); `BlobRef` for lazy chunk text; `QueryPlan` + `ChunkRef` + `Cost` + `PlanNode` types; typed `StageEvent`. `tests/contract/spi_signature.py` linter (RequestContext-first); rag-backends (`PgVectorStore`, `QdrantVectorStore`, `RedisCache`, `S3Storage`, `LocalFileStorage`) migrated; conformance + integration tests updated; `Budget.spend()` for agent-loop sub-turn budgets; schemas regenerated. ADR-0005 / ADR-0007 / ADR-0008 / ADR-0009 referenced. |
-| 1.1b | SPI split — Retrieval/Index, bulk + streaming + ID-only | ⏳ | 1.1a | — | Split `VectorStore`/`KeywordStore`/`GraphStore` into `RetrievalBackend` (read) + `IndexBackend` (write); add `bulk_embed`/`bulk_index`/`bulk_delete`; add async-iterator ingest variants; add `retrieve_ids` + `hydrate` pair; add `IndexHint` parameter for scale-tier index selection. ADR-0009 (vector index strategy) authored. |
+| 1.1b | SPI split — Retrieval/Index, bulk + streaming + ID-only | ✅ | `build/phase-1/step-1.1b-spi-split-retrieval-index` | _pending_ | Split `VectorStore`/`KeywordStore`/`GraphStore` into `*RetrievalBackend` (read) + `*IndexBackend` (write) composite ABCs. `retrieve_ids` returns `list[ChunkRef]`; `hydrate` lives on the retrieval side (keyword full-Chunk; vector pass-through). Bulk: `bulk_index`/`bulk_delete` (+ graph bulk node/edge variants); streaming: `stream_index` async-iterator default that batches into `bulk_index`. `Embedder` split into single `embed` + canonical `bulk_embed`. New `IndexHint` + `WriteVolume` types passed to writes (per ADR-0009). Noop impls, `PgVectorStore`, `QdrantVectorStore` migrated; conformance + integration tests updated; `spi_signature.py` extended to enforce the split. Schemas regenerated (`IndexHint.json`). |
 | 1.1c | PolicyEngine package | ⏳ | 1.1a | — | New `packages/policy/` (`rag-policy`): `PolicyEngine` SPI + noop impl; single decision point consulted by every retrieval/ingest path for ACL, PII, quotas, redaction. Replaces scattered checks across Steps 1.7, 4.5, 6.3, 6.5. ADR-0005 finalized. |
 | 1.1d | Pipeline + Batcher primitives | ⏳ | 1.1a | — | `Pipeline` primitive in `rag-core`: async DAG with bounded queues, per-stage worker counts, backpressure (used by Step 1.10 write path). `Batcher[Req, Resp]` middleware (DataLoader pattern) coalescing concurrent SPI calls into batched provider calls; sits under Embedder/Reranker SPIs. |
 | 1.1e | Cache SPI split + perf discipline + async telemetry | ⏳ | 1.1a | — | Split `Cache` into `EmbeddingCache` (key model_id+text_hash), `RetrievalCache` (plan_hash+corpus_version), `AnswerCache` (plan_hash+corpus_version+policy_version). Each has distinct invalidation. Hot-path convention doc (Pydantic at SPI boundary, `model_construct`/msgspec inside). Async telemetry path with bounded buffer + drop-on-overflow counter. |

docs/architecture/storage-backends.md

@@ -86,7 +86,7 @@ stack.
 To add a new VectorStore backend:
 
 1. Create `packages/backends/src/rag_backends/vector/<name>.py`
-2. Implement `VectorStore` ABC (upsert, query, delete, health)
+2. Implement `VectorStore` ABC — `bulk_index`, `retrieve_ids`, `bulk_delete`, `health` (the inherited `stream_index` and `hydrate` defaults are usually fine). See Step 1.1b notes in [docs/reference/rag-core.md](../reference/rag-core.md#read--write-spi-split-step-11b).
 3. Add an `initialize()` method for schema/collection setup
 4. Add an integration test in `tests/integration/test_<name>.py`
 5. Export from `rag_backends.vector.__init__` and `rag_backends.__init__`

docs/reference/rag-core.md

@@ -30,6 +30,7 @@ don't mutate.
 | `StageEvent` | Typed cross-stage observation | Step 1.1a (new) |
 | `TrustLevel` | Chunk provenance for prompt-injection defense | Step 1.1a (new) |
 | `EmbeddingDtype` | float32 / int8 / binary quantization | Step 1.1a (new) — ADR-0009 |
+| `IndexHint` / `WriteVolume` | Scale-tier hint passed to `IndexBackend` writes | Step 1.1b (new) — ADR-0009 |
 | `PiiAction` | block / redact / mask / encrypt / tag_only / allow | Step 1.1a (new) |
 
 ---
@@ -73,13 +74,57 @@ Every public SPI method takes `ctx` as its first argument.  Tenant scoping is
 derived from `ctx.tenant_id`:
 
 ```python
-await vector_store.upsert(ctx, embeddings)
-results = await vector_store.query(ctx, query_vec, top_k=10, corpus_ids=[])
+await vector_store.bulk_index(ctx, embeddings, hint=IndexHint(estimated_size=50_000))
+refs = await vector_store.retrieve_ids(ctx, query_vec, top_k=10, corpus_ids=[])
 ```
 
 The `tests/contract/spi_signature.py` linter fails CI if an SPI method is
 added without `ctx: RequestContext` as its first argument.
 
+### Read / write SPI split (Step 1.1b)
+
+`VectorStore`, `KeywordStore`, and `GraphStore` are now thin composite ABCs
+over two narrower roles:
+
+| Role | Methods | Used by |
+|---|---|---|
+| `*RetrievalBackend` | `retrieve_ids`, `hydrate` (where applicable), `query` (graph) | Gateway query path, rerankers |
+| `*IndexBackend`     | `bulk_index`, `stream_index`, `bulk_delete` (+ graph node/edge bulk variants) | Ingest pipeline |
+
+Backends serving both sides keep inheriting from the composite class
+(`VectorStore`, `KeywordStore`, `GraphStore`) — nothing changes for
+implementers.  Consumers should depend on the narrower role for clarity.
+
+```python
+# Hot-path discipline: retrieve cheap refs, rerank, hydrate the survivors
+refs = await vector_backend.retrieve_ids(ctx, qvec, top_k=200, corpus_ids=[])
+top  = await reranker.rerank(ctx, refs, query_text)
+hot  = await keyword_backend.hydrate(ctx, top[:20])
+```
+
+`Embedder` similarly exposes `bulk_embed(ctx, texts, chunk_ids)` (canonical
+batch entry) and a thin `embed(ctx, text, chunk_id)` for single-item callers.
+See [performance.md](../architecture/performance.md) for the p99 budgets per
+method.
+
+### Choosing an `IndexHint`
+
+The hint is consulted at write time (and at `initialize()` time for backends
+that need to pick an index implementation).  Reasonable starting values:
+
+```python
+IndexHint(
+    estimated_size=50_000,          # current + projected vector count
+    recall_target=0.95,             # default; raise to 0.99 for legal/medical
+    latency_target_ms=50.0,         # p99 query budget
+    write_volume=WriteVolume.low,   # bump to medium/high for streaming ingest
+)
+```
+
+PgVector and Qdrant currently honour `hint` only at `initialize()` time; the
+per-call writes ignore it.  Per ADR-0009, future backends will use it to pick
+between flat / ivfflat / HNSW / IVF-PQ index types.
+
 ### Lazy chunk text via `BlobRef`
 
 `Chunk.content` and `Chunk.content_ref` are mutually exclusive at the

packages/backends/src/rag_backends/vector/pgvector.py

@@ -4,8 +4,8 @@
 
     store = PgVectorStore(dsn="postgresql://rag:rag@localhost:5432/rag")
     await store.initialize()   # creates table + index once
-    await store.upsert(ctx, embeddings)
-    results = await store.query(ctx, query_vector, top_k=10, corpus_ids=[])
+    await store.bulk_index(ctx, embeddings)
+    refs = await store.retrieve_ids(ctx, query_vector, top_k=10, corpus_ids=[])
     await store.close()
 
 The table ``rag_vector_store`` is created on ``initialize()`` and is separate
@@ -22,7 +22,15 @@
 import numpy as np
 from pgvector.asyncpg import register_vector
 from rag_core.spi.vector_store import VectorStore
-from rag_core.types import ChunkId, CorpusId, Embedding, RequestContext
+from rag_core.types import (
+    ChunkId,
+    ChunkRef,
+    CorpusId,
+    Embedding,
+    IndexHint,
+    RequestContext,
+    TenantId,
+)
 from rag_observability.logging import get_logger
 
 _log = get_logger(__name__)
@@ -65,6 +73,7 @@
 
 _SQL_QUERY_BASE = """
 SELECT chunk_id,
+       corpus_id,
        1.0 - (vector <=> $2) AS score
 FROM   {table}
 WHERE  tenant_id = $1
@@ -145,10 +154,18 @@ async def close(self) -> None:
             self._pool = None
 
     # ------------------------------------------------------------------
-    # VectorStore SPI
+    # VectorStore SPI (Step 1.1b split: bulk_index / retrieve_ids / bulk_delete)
     # ------------------------------------------------------------------
 
-    async def upsert(self, ctx: RequestContext, embeddings: list[Embedding]) -> None:
+    async def bulk_index(
+        self,
+        ctx: RequestContext,
+        embeddings: list[Embedding],
+        *,
+        hint: IndexHint | None = None,
+    ) -> None:
+        # ``hint`` is consulted only at ``initialize()`` time (index choice);
+        # per-call writes ignore it.  See ADR-0009.
         if not embeddings:
             return
         pool = await self._get_pool()
@@ -167,14 +184,14 @@ async def upsert(self, ctx: RequestContext, embeddings: list[Embedding]) -> None
         async with pool.acquire() as conn:
             await conn.executemany(_SQL_UPSERT.format(table=self._table), rows)
 
-    async def query(
+    async def retrieve_ids(
         self,
         ctx: RequestContext,
         vector: list[float],
         top_k: int,
         corpus_ids: list[CorpusId],
         filters: dict[str, Any] | None = None,
-    ) -> list[tuple[ChunkId, float]]:
+    ) -> list[ChunkRef]:
         pool = await self._get_pool()
         qvec = _to_np(vector)
 
@@ -195,9 +212,18 @@ async def query(
         async with pool.acquire() as conn:
             rows = await conn.fetch(sql, *args)
 
-        return [(ChunkId(row["chunk_id"]), float(row["score"])) for row in rows]
+        tenant = TenantId(ctx.tenant_id)
+        return [
+            ChunkRef(
+                chunk_id=ChunkId(row["chunk_id"]),
+                tenant_id=tenant,
+                score=float(row["score"]),
+                corpus_id=CorpusId(row["corpus_id"]) if row.get("corpus_id") else None,
+            )
+            for row in rows
+        ]
 
-    async def delete(self, ctx: RequestContext, chunk_ids: list[ChunkId]) -> None:
+    async def bulk_delete(self, ctx: RequestContext, chunk_ids: list[ChunkId]) -> None:
         if not chunk_ids:
             return
         pool = await self._get_pool()

packages/backends/src/rag_backends/vector/qdrant.py

@@ -4,8 +4,8 @@
 
     store = QdrantVectorStore(url="http://localhost:6333")
     await store.initialize(dimension=1536)   # create collection once
-    await store.upsert(ctx, embeddings)
-    results = await store.query(ctx, vector, top_k=10, corpus_ids=[])
+    await store.bulk_index(ctx, embeddings)
+    refs = await store.retrieve_ids(ctx, vector, top_k=10, corpus_ids=[])
 
 A single Qdrant collection (``rag_embeddings`` by default) stores all tenants.
 Tenant isolation is enforced via a ``tenant_id`` payload filter on every query
@@ -29,7 +29,15 @@
     VectorParams,
 )
 from rag_core.spi.vector_store import VectorStore
-from rag_core.types import ChunkId, CorpusId, Embedding, RequestContext
+from rag_core.types import (
+    ChunkId,
+    ChunkRef,
+    CorpusId,
+    Embedding,
+    IndexHint,
+    RequestContext,
+    TenantId,
+)
 from rag_observability.logging import get_logger
 
 _log = get_logger(__name__)
@@ -101,10 +109,18 @@ async def close(self) -> None:
         await self._client.close()
 
     # ------------------------------------------------------------------
-    # VectorStore SPI
+    # VectorStore SPI (Step 1.1b split: bulk_index / retrieve_ids / bulk_delete)
     # ------------------------------------------------------------------
 
-    async def upsert(self, ctx: RequestContext, embeddings: list[Embedding]) -> None:
+    async def bulk_index(
+        self,
+        ctx: RequestContext,
+        embeddings: list[Embedding],
+        *,
+        hint: IndexHint | None = None,
+    ) -> None:
+        # Qdrant collection-level HNSW parameters are configured at
+        # ``initialize()`` time; ``hint`` is reserved for that path (ADR-0009).
         if not embeddings:
             return
 
@@ -124,14 +140,14 @@ async def upsert(self, ctx: RequestContext, embeddings: list[Embedding]) -> None
         ]
         await self._client.upsert(collection_name=self._collection, points=points)
 
-    async def query(
+    async def retrieve_ids(
         self,
         ctx: RequestContext,
         vector: list[float],
         top_k: int,
         corpus_ids: list[CorpusId],
         filters: dict[str, Any] | None = None,
-    ) -> list[tuple[ChunkId, float]]:
+    ) -> list[ChunkRef]:
         must: list[FieldCondition] = [
             FieldCondition(key=_FIELD_TENANT, match=MatchValue(value=str(ctx.tenant_id)))
         ]
@@ -151,13 +167,23 @@ async def query(
             with_payload=True,
         )
 
-        return [
-            (ChunkId(str(hit.payload[_FIELD_CHUNK])), float(hit.score))
-            for hit in response.points
-            if hit.payload and _FIELD_CHUNK in hit.payload
-        ]
+        tenant = TenantId(ctx.tenant_id)
+        refs: list[ChunkRef] = []
+        for hit in response.points:
+            if not hit.payload or _FIELD_CHUNK not in hit.payload:
+                continue
+            corpus_payload = hit.payload.get(_FIELD_CORPUS) or None
+            refs.append(
+                ChunkRef(
+                    chunk_id=ChunkId(str(hit.payload[_FIELD_CHUNK])),
+                    tenant_id=tenant,
+                    score=float(hit.score),
+                    corpus_id=CorpusId(str(corpus_payload)) if corpus_payload else None,
+                )
+            )
+        return refs
 
-    async def delete(self, ctx: RequestContext, chunk_ids: list[ChunkId]) -> None:
+    async def bulk_delete(self, ctx: RequestContext, chunk_ids: list[ChunkId]) -> None:
         if not chunk_ids:
             return
 

packages/core/src/rag_core/__init__.py

@@ -53,7 +53,11 @@
     Cache,
     Connector,
     Embedder,
+    GraphIndexBackend,
+    GraphRetrievalBackend,
     GraphStore,
+    KeywordIndexBackend,
+    KeywordRetrievalBackend,
     KeywordStore,
     LLMMessage,
     LLMResponse,
@@ -67,6 +71,8 @@
     Secrets,
     Storage,
     Telemetry,
+    VectorIndexBackend,
+    VectorRetrievalBackend,
     VectorStore,
 )
 from rag_core.telemetry import (
@@ -95,6 +101,7 @@
     DocumentStatus,
     Embedding,
     EmbeddingDtype,
+    IndexHint,
     PiiAction,
     PiiPolicy,
     PlanNode,
@@ -112,6 +119,7 @@
     TenantId,
     TraceContext,
     TrustLevel,
+    WriteVolume,
 )
 
 __version__ = "0.5.0"
@@ -165,6 +173,7 @@
     "DocumentStatus",
     "Embedding",
     "EmbeddingDtype",
+    "IndexHint",
     "PiiAction",
     "PiiPolicy",
     "PlanNode",
@@ -182,6 +191,7 @@
     "TenantId",
     "TraceContext",
     "TrustLevel",
+    "WriteVolume",
     # errors
     "ACLDeniedError",
     "AuthError",
@@ -211,7 +221,11 @@
     "Cache",
     "Connector",
     "Embedder",
+    "GraphIndexBackend",
+    "GraphRetrievalBackend",
     "GraphStore",
+    "KeywordIndexBackend",
+    "KeywordRetrievalBackend",
     "KeywordStore",
     "LLM",
     "LLMMessage",
@@ -227,5 +241,7 @@
     "Secrets",
     "Storage",
     "Telemetry",
+    "VectorIndexBackend",
+    "VectorRetrievalBackend",
     "VectorStore",
 ]

packages/core/src/rag_core/gen_schemas.py

@@ -26,6 +26,7 @@
     Cost,
     Document,
     Embedding,
+    IndexHint,
     PiiPolicy,
     PlanNode,
     Principal,
@@ -54,6 +55,7 @@
     Citation,
     ChunkRef,
     Cost,
+    IndexHint,
     PlanNode,
     QueryPlan,
     StageEvent,