feat(core,observability): Cache SPI split + async telemetry (Step 1.1e)

TRACKER.md

@@ -10,7 +10,7 @@
 
 **Last updated:** 2026-05-24
 **Current phase:** Phase 1 — Ingestion + Knowledge Store
-**Next action:** Phase 1 Step 1.1e — Cache SPI split (`EmbeddingCache` / `RetrievalCache` / `AnswerCache`) + hot-path discipline + async telemetry path with drop-on-overflow counter
+**Next action:** Phase 1 Step 1.1f — ADRs 0005–0009 (already drafted) finalized + reviewer checklist landed in `docs/architecture/performance.md` and cross-referenced from every relevant module
 
 > **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 | **4** | 12 |
+| 1 | Ingestion + Knowledge Store | 16 | **5** | 11 |
 | 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** | **17** | **67** |
+| **Total** | | **84** | **18** | **66** |
 
 ---
 
@@ -71,8 +71,8 @@
 | 1.1a | Core type & SPI refactor | ✅ | `build/phase-1/step-1.1a-core-type-spi-refactor` | [#44](https://github.com/officialCodeWork/AgentContextOS/pull/44) | `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 | ✅ | `build/phase-1/step-1.1b-spi-split-retrieval-index` | [#45](https://github.com/officialCodeWork/AgentContextOS/pull/45) | 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 | ✅ | `build/phase-1/step-1.1c-policy-engine-package` | [#46](https://github.com/officialCodeWork/AgentContextOS/pull/46) | New `packages/policy/` (`rag-policy` v0.1.0): `PolicyEngine` SPI + `NoopPolicyEngine` (always-ALLOW with tenant-scoped `filter_pushdown`); `PolicyDecision` enum (read_chunk / ingest_doc / egress_text / quota_check / rate_limit / execute_plan); `PolicyResult` (allow/deny/transform) with predicate helpers; `QuotaSubject` / `RateLimitSubject`; `FilterExpr` mini-language (Eq / AnyIn / And / Or / Not / TrueExpr) returned by `filter_pushdown`; `PolicyWriter` facade mirroring `AuditWriter` and emitting `policy.decision` structured logs via `rag-observability`. Coverage linter `tests/policy/coverage.py` greps for governance-relevant SPI calls without adjacent `PolicyEngine`/`PolicyWriter` consultation, with file allowlist that consumers shrink as they wire the PDP in. Workspace + pytest pythonpath updated. 20 conformance tests added. ADR-0005 finalized. |
-| 1.1d | Pipeline + Batcher primitives | 🚧 | `build/phase-1/step-1.1d-pipeline-batcher-primitives` | — | `Pipeline` primitive in `rag-core` (`rag_core.pipeline`): async DAG with bounded queues, per-stage worker counts, backpressure, joiner-based shutdown handling per-stage worker-count differences, `on_error="fail"`/`"skip"` policies, optional `error_handler`. `Batcher[Req, Resp]` middleware (`rag_core.batcher`, DataLoader pattern) coalescing concurrent `load(req)` calls into batched `batch_fn(reqs)` invocations; size + time triggers (`max_batch_size`, `max_wait_ms`); per-request `asyncio.Future` for isolated error propagation; `flush()` for deterministic shutdown. PEP-695-typed generics. 22 unit tests (8 Batcher + 14 Pipeline). `docs/architecture/pipeline-batcher.md` + `rag-core` reference + `performance.md` backlinks. `rag-core` 0.5.0 → 0.6.0. |
-| 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. |
+| 1.1d | Pipeline + Batcher primitives | ✅ | `build/phase-1/step-1.1d-pipeline-batcher-primitives` | [#50](https://github.com/officialCodeWork/AgentContextOS/pull/50) | `Pipeline` primitive in `rag-core` (`rag_core.pipeline`): async DAG with bounded queues, per-stage worker counts, backpressure, joiner-based shutdown handling per-stage worker-count differences, `on_error="fail"`/`"skip"` policies, optional `error_handler`. `Batcher[Req, Resp]` middleware (`rag_core.batcher`, DataLoader pattern) coalescing concurrent `load(req)` calls into batched `batch_fn(reqs)` invocations; size + time triggers (`max_batch_size`, `max_wait_ms`); per-request `asyncio.Future` for isolated error propagation; `flush()` for deterministic shutdown. PEP-695-typed generics. 22 unit tests (8 Batcher + 14 Pipeline). `docs/architecture/pipeline-batcher.md` + `rag-core` reference + `performance.md` backlinks. `rag-core` 0.5.0 → 0.6.0. |
+| 1.1e | Cache SPI split + perf discipline + async telemetry | 🚧 | `build/phase-1/step-1.1e-cache-split-perf-telemetry` | — | `EmbeddingCache` (`rag_core.spi.embedding_cache`, key tenant+model_id+model_version+text_hash, `invalidate_model`), `RetrievalCache` (`rag_core.spi.retrieval_cache`, key tenant+plan_hash+corpus_version, `invalidate_corpus`), `AnswerCache` (`rag_core.spi.answer_cache`, key tenant+plan_hash+corpus_version+policy_version, `invalidate_policy`). All ctx-first; signature linter extended. Noop in-memory impls + 18 conformance tests covering hit/miss/version-partition/invalidate/tenant-isolation. `AsyncTelemetrySink` in `rag-observability` — bounded buffer (default 10_000), non-blocking `submit()`, per-kind drop + exporter-error counters, idempotent start/stop, drain-timeout-cancel; 9 unit tests. Hot-path discipline doc in `performance.md` expanded with concrete hot/cold call-site table. New `docs/reference/rag-observability.md`. `caching.md` SPI shapes updated with ctx-first signatures. `rag-core` 0.6.0 → 0.7.0. |
 | 1.1f | ADRs 0005–0009 + reviewer checklist | ⏳ | 1.1a–1.1e | — | ADR-0005 (PolicyEngine PDP), ADR-0006 (two-stage reranker default), ADR-0007 (tiered storage with BlobRef), ADR-0008 (cost-aware planner replacing reactive fallback), ADR-0009 (vector index strategy + quantization). Reviewer checklist in `docs/architecture/performance.md`. |
 | 1.2 | Connectors framework | ⏳ | 1.1a–1.1f | — | `Connector` SPI implementation (now receives `RequestContext` and `ConnectorState` watermark); built-in: filesystem, S3, GCS; crawler base class |
 | 1.3 | Document parsers | ⏳ | — | — | PDF, DOCX, PPTX, XLSX, HTML, Markdown, plain text, JSON, CSV, YAML parsers; MIME detection |
@@ -218,6 +218,7 @@
 | [#47](https://github.com/officialCodeWork/AgentContextOS/pull/47) | chore(schemas): track generated JSON Schemas + add drift CI gate | `chore/track-generated-schemas-drift-gate` | ✅ Merged | 2026-05-24 |
 | [#48](https://github.com/officialCodeWork/AgentContextOS/pull/48) | chore(ci): lint scripts/ + ignore S603/S607 in scripts/** | `chore/lint-ignore-subprocess-in-scripts` | ✅ Merged | 2026-05-24 |
 | [#49](https://github.com/officialCodeWork/AgentContextOS/pull/49) | chore(tracker): sync PR links for steps 1.1a/1.1c + log #46–#48 | `chore/tracker-sync-pr44-48` | ✅ Merged | 2026-05-24 |
+| [#50](https://github.com/officialCodeWork/AgentContextOS/pull/50) | feat(core): Pipeline + Batcher primitives (Step 1.1d) | `build/phase-1/step-1.1d-pipeline-batcher-primitives` | ✅ Merged | 2026-05-24 |
 
 ---
 

docs/README.md

@@ -21,7 +21,8 @@
 |------|-------------|
 | [ragctl.md](reference/ragctl.md) | Full `ragctl` command reference — public usage, internals, extension points |
 | [backends.md](reference/backends.md) | `rag-backends` reference — PgVectorStore, QdrantVectorStore, RedisCache, S3Storage, LocalFileStorage |
-| [rag-core.md](reference/rag-core.md) | `rag-core` type surface — `RequestContext`, `Budget`, `BlobRef`, `QueryPlan`, `ChunkRef`, `StageEvent`, `Pipeline`, `Batcher` |
+| [rag-core.md](reference/rag-core.md) | `rag-core` type surface — `RequestContext`, `Budget`, `BlobRef`, `QueryPlan`, `ChunkRef`, `StageEvent`, `Pipeline`, `Batcher`, cache SPIs |
+| [rag-observability.md](reference/rag-observability.md) | `rag-observability` — structured logger, event registry, `AsyncTelemetrySink` (bounded, drop-on-overflow) |
 | [rag-policy.md](reference/rag-policy.md) | `rag-policy` reference — `PolicyEngine`, `PolicyWriter`, `PolicyDecision`, `PolicyResult`, `FilterExpr` |
 
 ## guides/

docs/architecture/caching.md

@@ -22,17 +22,32 @@ The V1.0 plan described a single "L0/L1/L2 semantic cache" in Step 4.1. Three ob
 
 ## Usage
 
+All three SPIs take `ctx: RequestContext` as their first positional
+argument (enforced by `tests/contract/spi_signature.py`). The remaining
+parameters are keyword-only to keep call sites readable.
+
 ### EmbeddingCache
 
 ```python
 from rag_core.spi import EmbeddingCache
 
 cache: EmbeddingCache = ...
 
-cached = await cache.get(model_id="bge-large-en-v1.5", model_version="1.5.0", text_hash=h)
+cached = await cache.get(
+    ctx,
+    model_id="bge-large-en-v1.5",
+    model_version="1.5.0",
+    text_hash=h,
+)
 if cached is None:
-    emb = await embedder.embed(ctx, text)
-    await cache.put(model_id="bge-large-en-v1.5", model_version="1.5.0", text_hash=h, value=emb)
+    emb = await embedder.embed(ctx, text, chunk_id)
+    await cache.put(
+        ctx,
+        model_id="bge-large-en-v1.5",
+        model_version="1.5.0",
+        text_hash=h,
+        value=emb,
+    )
 ```
 
 Hit rate target: ≥ 99% on re-ingest of identical corpora.
@@ -44,12 +59,19 @@ from rag_core.spi import RetrievalCache
 
 cache: RetrievalCache = ...
 
-cached = await cache.get(plan_hash=plan.hash(), corpus_version=corpus.version)
+cached = await cache.get(ctx, plan_hash=plan.hash(), corpus_version=corpus.version)
 if cached is None:
     refs = await retrieve_pipeline(ctx, plan)
-    await cache.put(plan_hash=plan.hash(), corpus_version=corpus.version, value=refs)
+    await cache.put(
+        ctx, plan_hash=plan.hash(), corpus_version=corpus.version, value=refs
+    )
 ```
 
+`RetrievalCache.invalidate_corpus(ctx, corpus_id=...)` is wired into the
+write path: every successful `bulk_index` / `bulk_delete` on a corpus
+calls it after bumping `corpus_version`. Backends with no scan support
+return 0 from `invalidate_corpus` and rely on the version bump alone.
+
 Hit rate target: ≥ 30% on production query mix (Pareto-distributed).
 
 ### AnswerCache
@@ -60,15 +82,29 @@ from rag_core.spi import AnswerCache
 cache: AnswerCache = ...
 
 cached = await cache.get(
+    ctx,
     plan_hash=plan.hash(),
     corpus_version=corpus.version,
-    policy_version=ctx.principal.policy_version,
+    policy_version=current_policy_version(ctx),
 )
 if cached is None:
-    answer = await llm_pipeline(ctx, plan)
-    await cache.put(..., value=answer)
+    answer_bytes = serialize(await llm_pipeline(ctx, plan))
+    await cache.put(
+        ctx,
+        plan_hash=plan.hash(),
+        corpus_version=corpus.version,
+        policy_version=current_policy_version(ctx),
+        value=answer_bytes,
+    )
 ```
 
+`AnswerCache` stores opaque `bytes` — the answer envelope (text +
+citations + trace metadata) is still firming up in Phase 3, so the SPI
+doesn't constrain it. Callers handle their own serialization.
+
+`AnswerCache.invalidate_policy(ctx, policy_version=...)` is called by the
+control plane on any policy rotation.
+
 Hit rate target: ≥ 15% on production query mix.
 
 ## Internals
@@ -115,17 +151,28 @@ cache:
 
 ## Extension points
 
-Implement any of:
+The canonical SPI shapes:
 
 ```python
 class EmbeddingCache(ABC):
-    async def get(self, *, model_id: str, model_version: str, text_hash: str) -> Embedding | None: ...
-    async def put(self, *, model_id: str, model_version: str, text_hash: str, value: Embedding) -> None: ...
+    async def get(self, ctx, *, model_id: str, model_version: str, text_hash: str) -> Embedding | None: ...
+    async def put(self, ctx, *, model_id: str, model_version: str, text_hash: str, value: Embedding, ttl_seconds: int | None = None) -> None: ...
+    async def invalidate_model(self, ctx, *, model_id: str, model_version: str) -> int: ...
+
+class RetrievalCache(ABC):
+    async def get(self, ctx, *, plan_hash: str, corpus_version: int) -> list[ChunkRef] | None: ...
+    async def put(self, ctx, *, plan_hash: str, corpus_version: int, value: list[ChunkRef], ttl_seconds: int | None = None) -> None: ...
+    async def invalidate_corpus(self, ctx, *, corpus_id: str) -> int: ...
+
+class AnswerCache(ABC):
+    async def get(self, ctx, *, plan_hash: str, corpus_version: int, policy_version: str) -> bytes | None: ...
+    async def put(self, ctx, *, plan_hash: str, corpus_version: int, policy_version: str, value: bytes, ttl_seconds: int | None = None) -> None: ...
+    async def invalidate_policy(self, ctx, *, policy_version: str) -> int: ...
 ```
 
-(Similar shapes for `RetrievalCache`, `AnswerCache`.)
-
-Plug in alternative backends (Valkey, Memcached, in-memory LRU) without changing consumers.
+Plug in alternative backends (Valkey, Memcached, in-memory LRU) without
+changing consumers. Reference noop implementations live in
+`rag_core.spi.noop` and are exercised by `tests/contract/test_*_cache.py`.
 
 ## Related
 

docs/architecture/performance.md

@@ -21,6 +21,16 @@ Pydantic v2 validation is fast (~µs per model) but it runs every time. In a hot
 
 When trusted internal construction is needed for a Pydantic model, use `Model.model_construct(**fields)` — this skips validation entirely. Reserve for code paths where inputs are known-good (e.g., reconstructing a `RequestContext` from a cache).
 
+**Concrete examples** of "hot loop" in this codebase:
+
+| Site | Hot or cold | Rule |
+|---|---|---|
+| Gateway request handler (per request) | Cold | Validate `RequestContext` once. |
+| `VectorRetrievalBackend.retrieve_ids` returning 200 `ChunkRef`s | Hot | Construct `ChunkRef` with `ChunkRef.model_construct(...)` inside the comprehension. |
+| `Embedder.bulk_embed` mapping provider response → 100 `Embedding`s | Hot | `Embedding.model_construct(...)` per item; the provider response shape is already validated by the SDK. |
+| `Pipeline` stage function called once per item | Caller-determined | If the item comes from another SPI it's trusted — skip validation; if from `connector.read` (raw bytes) validate once at the boundary. |
+| `AuditWriter.write` per event | Hot under spikes | The `AuditEvent` is built by trusted code — `model_construct` is fine. |
+
 ### Avoid per-call `getLogger`
 
 `logging.getLogger(__name__)` is fast on modern Python but still has overhead in tight loops. Always store at module level:
@@ -78,13 +88,31 @@ When a budget is missed, the conformance suite fails. The fix is either (a) tune
 
 ## Async telemetry path
 
-OTel exporter back-pressure cannot block the request path. The shipped `rag_observability` configuration uses:
+OTel exporter back-pressure cannot block the request path. Shipped as `rag_observability.AsyncTelemetrySink` (Step 1.1e):
+
+- **Bounded buffer:** default 10,000 records. Configurable via `max_buffer=`.
+- **Non-blocking `submit()`:** drops on overflow, never raises and never `await`s the queue.
+- **Per-kind drop counter:** `sink.dropped_total("logs" | "spans" | "stage_events" | ...)`. Surface as the `telemetry.dropped_total{kind}` metric for alerting.
+- **Per-kind exporter-error counter:** `sink.export_errors_total("...")`. Exporter exceptions are swallowed (the drainer stays alive) but counted; surface as `telemetry.export_errors_total{kind}`.
+- **Cooperative shutdown:** `await sink.stop()` flushes pending records up to `drain_timeout_s` (default 5 s) before cancelling the drainer.
+
+```python
+from rag_observability import AsyncTelemetrySink
+
+async def export(rec):  # OTel exporter, log shipper, etc.
+    await collector.send(rec)
+
+sink = AsyncTelemetrySink(export, max_buffer=10_000)
+sink.start()
 
-- **Bounded buffer:** default 10,000 records.
-- **Non-blocking exporter:** drops on overflow.
-- **Drop counter:** `telemetry.dropped_total{kind}` metric, alertable.
+sink.submit({"span": "vector.retrieve", "ms": 12}, kind="spans")   # non-blocking
+sink.submit({"event": "ingest.completed"}, kind="stage_events")
+
+# at shutdown
+await sink.stop()
+```
 
-This is a Step 1.1e deliverable, tested by simulating exporter back-pressure for 60s and asserting the request path's p99 is unchanged.
+Tested via `tests/observability/test_async_sink.py` (9 tests covering drain ordering, drop accounting under simulated back-pressure, exporter-error swallowing, idempotent start/stop, and drain-timeout cancellation).
 
 ## Reviewer checklist
 

docs/reference/rag-core.md

@@ -237,6 +237,22 @@ The reviewer checklist in
 [docs/architecture/performance.md](../architecture/performance.md) calls
 out where each must be used.
 
+## Cache SPIs (Step 1.1e)
+
+Three distinct cache SPIs replace the single semantic-cache plan — each
+keyed differently, each with a different invalidation contract:
+
+| SPI | Module | Key | Invalidation |
+|---|---|---|---|
+| `EmbeddingCache` | `rag_core.spi.embedding_cache` | `(tenant, model_id, model_version, text_hash)` | Model / version change |
+| `RetrievalCache` | `rag_core.spi.retrieval_cache` | `(tenant, plan_hash, corpus_version)` | Corpus mutation (version bump) |
+| `AnswerCache` | `rag_core.spi.answer_cache` | `(tenant, plan_hash, corpus_version, policy_version)` | Any of the above + policy rotation |
+
+Each SPI takes `ctx: RequestContext` as first arg (linter-enforced) and
+ships with a noop in-memory reference impl in `rag_core.spi.noop`. See
+[docs/architecture/caching.md](../architecture/caching.md) for the full
+contract, hit-rate targets, and reference Redis-backed deployment shape.
+
 ## Related
 
 - [docs/architecture/request-context.md](../architecture/request-context.md) — design rationale.