diff --git a/TRACKER.md b/TRACKER.md
index aa989d1..e2ab717 100644
--- a/TRACKER.md
+++ b/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 |
 
 ---
 
diff --git a/docs/README.md b/docs/README.md
index 3e5df2b..2c74c33 100644
--- a/docs/README.md
+++ b/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/
diff --git a/docs/architecture/caching.md b/docs/architecture/caching.md
index daa0a9b..8d725f9 100644
--- a/docs/architecture/caching.md
+++ b/docs/architecture/caching.md
@@ -22,6 +22,10 @@ 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
@@ -29,10 +33,21 @@ 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
 
diff --git a/docs/architecture/performance.md b/docs/architecture/performance.md
index f3d0aa8..8f9af7a 100644
--- a/docs/architecture/performance.md
+++ b/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
 
diff --git a/docs/reference/rag-core.md b/docs/reference/rag-core.md
index 45ded06..829c482 100644
--- a/docs/reference/rag-core.md
+++ b/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.
diff --git a/docs/reference/rag-observability.md b/docs/reference/rag-observability.md
new file mode 100644
index 0000000..2c062f9
--- /dev/null
+++ b/docs/reference/rag-observability.md
@@ -0,0 +1,99 @@
+# Reference — `rag-observability`
+
+`rag-observability` owns the structured-logging configuration, the event
+registry, the `set_log_context()` ContextVar manager, and (Step 1.1e)
+the `AsyncTelemetrySink` — a bounded, drop-on-overflow buffer for
+telemetry export so that exporter back-pressure never reaches the
+request path.
+
+---
+
+## Public surface
+
+| Name | Purpose |
+|---|---|
+| `configure_logging(...)` | Configure the 7-field JSON logger; idempotent. Call once at startup. |
+| `get_logger(name)` | Get a logger under the `rag.*` namespace (RAG001-compliant). |
+| `set_log_context(...)` | ContextVar-scoped trace/tenant/span enrichment. |
+| `reset_logging()` | Test-only reset of logger state. |
+| `AsyncTelemetrySink(export_fn, *, max_buffer, drain_timeout_s)` | Bounded buffer + background drainer for any async exporter. |
+| `SinkStats.from_sink(sink)` | Snapshot of buffer size + per-kind drop and exporter-error counters. |
+| Event constants (`EVT_*`) + Pydantic event types | The typed event registry consumed by structured log emission. |
+
+---
+
+## `AsyncTelemetrySink`
+
+### Usage
+
+```python
+from rag_observability import AsyncTelemetrySink
+
+async def export(record):
+    await otel_exporter.send(record)
+
+sink = AsyncTelemetrySink(export, max_buffer=10_000)
+sink.start()
+
+# In the request path (hot, must not block):
+sink.submit({"span": "vector.retrieve", "ms": 12}, kind="spans")
+
+# At shutdown:
+await sink.stop()
+```
+
+`submit()` returns `True` on enqueue, `False` if the queue was full and
+the record was dropped. The drop counter for that `kind` is incremented;
+read it via `sink.dropped_total("spans")` or all-kinds with
+`sink.dropped_total()`.
+
+### Semantics
+
+- **Non-blocking submit.** Microsecond latency regardless of exporter
+  state. Stalls in the exporter are absorbed by the queue, then by
+  drops once the queue is full.
+- **Single drainer task.** Started by `start()`, cancelled by `stop()`.
+  Records are exported FIFO.
+- **Exporter exceptions swallowed and counted.** A failing `export_fn`
+  never crashes the drainer or surfaces to the caller; the failure is
+  counted via `sink.export_errors_total(kind)`.
+- **Cooperative shutdown.** `stop()` awaits `queue.join()` up to
+  `drain_timeout_s` (default 5 s) before cancelling the drainer.
+  Idempotent.
+- **Default capacity.** 10,000 records, matching the value published in
+  [docs/architecture/performance.md](../architecture/performance.md).
+
+### When to wire one in
+
+Any code path that fires telemetry from inside the request path:
+
+- OTel exporter (logs / traces / metrics)
+- `StageEvent` stream to the Eval Recorder
+- Audit log shipper if the underlying `AuditStore` does network I/O
+- Any future log-to-Loki / event-to-Kafka pipeline
+
+### Counters and alerting
+
+Surface the per-kind counters in your Telemetry SPI of choice:
+
+```python
+metrics.gauge("telemetry.dropped_total", sink.dropped_total("spans"), labels={"kind": "spans"})
+metrics.gauge("telemetry.export_errors_total", sink.export_errors_total("spans"), labels={"kind": "spans"})
+```
+
+Alert on either being non-zero. A drop count > 0 means the request path
+saved itself by sacrificing telemetry — investigate the exporter, not
+the request path.
+
+---
+
+## Related
+
+- [docs/architecture/performance.md](../architecture/performance.md) —
+  hot-path discipline + reviewer checklist; defines the published default
+  capacity and alerting expectation.
+- [docs/guides/logging-policy.md](../guides/logging-policy.md) — RAG001
+  policy (no raw `logging.getLogger`) enforced for everything that emits
+  log records.
+- TRACKER.md Step 1.1e (this work), Step 0.7b (structured logging
+  foundation), Step 0.7 (OTel tracing + SPI metrics).
diff --git a/packages/core/pyproject.toml b/packages/core/pyproject.toml
index 38eb17c..a9f3c8f 100644
--- a/packages/core/pyproject.toml
+++ b/packages/core/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "rag-core"
-version = "0.6.0"
+version = "0.7.0"
 description = "AgentContextOS — core domain types, error hierarchy, and plugin SPIs"
 readme = "README.md"
 requires-python = ">=3.12"
diff --git a/packages/core/src/rag_core/__init__.py b/packages/core/src/rag_core/__init__.py
index 0aafb3c..df13f51 100644
--- a/packages/core/src/rag_core/__init__.py
+++ b/packages/core/src/rag_core/__init__.py
@@ -51,10 +51,12 @@
 from rag_core.spi import (
     LLM,
     OCR,
+    AnswerCache,
     Auth,
     Cache,
     Connector,
     Embedder,
+    EmbeddingCache,
     GraphIndexBackend,
     GraphRetrievalBackend,
     GraphStore,
@@ -70,6 +72,7 @@
     Queue,
     QueueMessage,
     Reranker,
+    RetrievalCache,
     Secrets,
     Storage,
     Telemetry,
@@ -124,7 +127,7 @@
     WriteVolume,
 )
 
-__version__ = "0.6.0"
+__version__ = "0.7.0"
 
 __all__ = [
     # eval
@@ -223,10 +226,12 @@
     "span_from_trace_context",
     "trace_context_from_span",
     # SPIs
+    "AnswerCache",
     "Auth",
     "Cache",
     "Connector",
     "Embedder",
+    "EmbeddingCache",
     "GraphIndexBackend",
     "GraphRetrievalBackend",
     "GraphStore",
@@ -244,6 +249,7 @@
     "Queue",
     "QueueMessage",
     "Reranker",
+    "RetrievalCache",
     "Secrets",
     "Storage",
     "Telemetry",
diff --git a/packages/core/src/rag_core/spi/__init__.py b/packages/core/src/rag_core/spi/__init__.py
index fcad201..97f3675 100644
--- a/packages/core/src/rag_core/spi/__init__.py
+++ b/packages/core/src/rag_core/spi/__init__.py
@@ -10,11 +10,13 @@
 serve both sides.
 """
 
+from rag_core.spi.answer_cache import AnswerCache
 from rag_core.spi.audit_store import AuditStore
 from rag_core.spi.auth import Auth
 from rag_core.spi.cache import Cache
 from rag_core.spi.connector import Connector
 from rag_core.spi.embedder import Embedder
+from rag_core.spi.embedding_cache import EmbeddingCache
 from rag_core.spi.graph_store import (
     GraphIndexBackend,
     GraphRetrievalBackend,
@@ -31,6 +33,7 @@
 from rag_core.spi.pii_detector import PIIDetector, PIISpan
 from rag_core.spi.queue import Queue, QueueMessage
 from rag_core.spi.reranker import Reranker
+from rag_core.spi.retrieval_cache import RetrievalCache
 from rag_core.spi.secrets import Secrets
 from rag_core.spi.storage import Storage
 from rag_core.spi.telemetry import Telemetry
@@ -41,11 +44,13 @@
 )
 
 __all__ = [
+    "AnswerCache",
     "AuditStore",
     "Auth",
     "Cache",
     "Connector",
     "Embedder",
+    "EmbeddingCache",
     "GraphIndexBackend",
     "GraphRetrievalBackend",
     "GraphStore",
@@ -63,6 +68,7 @@
     "Queue",
     "QueueMessage",
     "Reranker",
+    "RetrievalCache",
     "Secrets",
     "Storage",
     "Telemetry",
diff --git a/packages/core/src/rag_core/spi/answer_cache.py b/packages/core/src/rag_core/spi/answer_cache.py
new file mode 100644
index 0000000..75eb882
--- /dev/null
+++ b/packages/core/src/rag_core/spi/answer_cache.py
@@ -0,0 +1,75 @@
+"""AnswerCache SPI — keyed by ``(plan_hash, corpus_version, policy_version)``.
+
+The strictest invalidation contract of the three cache SPIs: an answer is
+only valid as long as the plan, the corpus version, *and* the active
+policy version are unchanged. Rotating a tenant's PII or ACL policy must
+not return previously-redacted answers — those entries are dropped on the
+next lookup by virtue of the key miss.
+
+Values are opaque bytes because the answer envelope (text + citations +
+trace metadata) is still firming up in Phase 3; callers handle their own
+serialization. The SPI does not constrain it.
+
+See ``docs/architecture/caching.md`` for the full three-cache split and
+``docs/architecture/performance.md`` for the per-call p99 budget (≤ 2 ms).
+"""
+
+from __future__ import annotations
+
+import abc
+
+from rag_core.spi._base import HealthCheckMixin
+from rag_core.types import RequestContext
+
+
+class AnswerCache(HealthCheckMixin, abc.ABC):
+    """Cache of serialized answer envelopes.
+
+    ``policy_version`` is a hash of the active ``PolicyEngine``
+    configuration for the tenant — exposed by the control plane and bumped
+    on any policy change. Use ``policy_version=ctx.policy_version`` once
+    that field lands on ``RequestContext``; for now callers pass it
+    explicitly.
+
+    Hit rate target: ≥ 15% on production query mix.
+    """
+
+    @abc.abstractmethod
+    async def get(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+        policy_version: str,
+    ) -> bytes | None:
+        """Return the cached serialized answer or ``None`` on a miss."""
+
+    @abc.abstractmethod
+    async def put(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+        policy_version: str,
+        value: bytes,
+        ttl_seconds: int | None = None,
+    ) -> None:
+        """Store ``value`` under the composite key.
+
+        ``ttl_seconds`` defaults to 900s in the reference Redis backend.
+        """
+
+    @abc.abstractmethod
+    async def invalidate_policy(
+        self,
+        ctx: RequestContext,
+        *,
+        policy_version: str,
+    ) -> int:
+        """Drop every entry tagged with ``policy_version`` for this tenant.
+
+        Called by the control plane when a policy rotates. Returns the
+        count removed (best-effort).
+        """
diff --git a/packages/core/src/rag_core/spi/embedding_cache.py b/packages/core/src/rag_core/spi/embedding_cache.py
new file mode 100644
index 0000000..51b7c09
--- /dev/null
+++ b/packages/core/src/rag_core/spi/embedding_cache.py
@@ -0,0 +1,73 @@
+"""EmbeddingCache SPI — keyed by ``(model_id, model_version, text_hash)``.
+
+Distinct from the other two cache SPIs (`RetrievalCache`, `AnswerCache`)
+because its invalidation rule is different: an entry stays valid until the
+embedder model or model version changes, regardless of corpus mutation or
+policy change. Re-chunking a corpus does not invalidate cached embeddings
+(the text hash is unchanged), which keeps re-ingest cheap.
+
+See ``docs/architecture/caching.md`` for the full three-cache split and
+``docs/architecture/performance.md`` for the per-call p99 budget (≤ 2 ms).
+"""
+
+from __future__ import annotations
+
+import abc
+
+from rag_core.spi._base import HealthCheckMixin
+from rag_core.types import Embedding, RequestContext
+
+
+class EmbeddingCache(HealthCheckMixin, abc.ABC):
+    """Cache of ``Embedding`` values keyed by model + model version + text hash.
+
+    Implementations are expected to be tenant-aware via ``ctx``: a key from
+    tenant A must never resolve to tenant B's value. The reference Redis
+    backend bakes ``ctx.tenant_id`` into the storage key; alternative
+    backends must do the same.
+
+    Hit rate target: ≥ 99% on re-ingest of identical corpora.
+    """
+
+    @abc.abstractmethod
+    async def get(
+        self,
+        ctx: RequestContext,
+        *,
+        model_id: str,
+        model_version: str,
+        text_hash: str,
+    ) -> Embedding | None:
+        """Return cached embedding or ``None`` on a miss."""
+
+    @abc.abstractmethod
+    async def put(
+        self,
+        ctx: RequestContext,
+        *,
+        model_id: str,
+        model_version: str,
+        text_hash: str,
+        value: Embedding,
+        ttl_seconds: int | None = None,
+    ) -> None:
+        """Store ``value`` under the composite key.
+
+        ``ttl_seconds=None`` means never expire — embeddings are invalidated
+        by model / version change, not by time.
+        """
+
+    @abc.abstractmethod
+    async def invalidate_model(
+        self,
+        ctx: RequestContext,
+        *,
+        model_id: str,
+        model_version: str,
+    ) -> int:
+        """Drop every entry for ``(model_id, model_version)`` for this tenant.
+
+        Called by the control plane when an embedder model is retired or
+        upgraded. Returns the number of entries removed (best-effort —
+        backends with no scan support may return 0).
+        """
diff --git a/packages/core/src/rag_core/spi/noop/__init__.py b/packages/core/src/rag_core/spi/noop/__init__.py
index 2987fb3..03b7d5a 100644
--- a/packages/core/src/rag_core/spi/noop/__init__.py
+++ b/packages/core/src/rag_core/spi/noop/__init__.py
@@ -1,10 +1,12 @@
 """Noop (in-memory) SPI implementations — suitable for tests and local dev."""
 
+from rag_core.spi.noop.answer_cache import NoopAnswerCache
 from rag_core.spi.noop.audit_store import NoopAuditStore
 from rag_core.spi.noop.auth import NoopAuth
 from rag_core.spi.noop.cache import NoopCache
 from rag_core.spi.noop.connector import NoopConnector
 from rag_core.spi.noop.embedder import NoopEmbedder
+from rag_core.spi.noop.embedding_cache import NoopEmbeddingCache
 from rag_core.spi.noop.graph_store import NoopGraphStore
 from rag_core.spi.noop.keyword_store import NoopKeywordStore
 from rag_core.spi.noop.llm import NoopLLM
@@ -13,17 +15,20 @@
 from rag_core.spi.noop.pii_detector import NoopPIIDetector
 from rag_core.spi.noop.queue import NoopQueue
 from rag_core.spi.noop.reranker import NoopReranker
+from rag_core.spi.noop.retrieval_cache import NoopRetrievalCache
 from rag_core.spi.noop.secrets import NoopSecrets
 from rag_core.spi.noop.storage import NoopStorage
 from rag_core.spi.noop.telemetry import NoopTelemetry
 from rag_core.spi.noop.vector_store import NoopVectorStore
 
 __all__ = [
+    "NoopAnswerCache",
     "NoopAuditStore",
     "NoopAuth",
     "NoopCache",
     "NoopConnector",
     "NoopEmbedder",
+    "NoopEmbeddingCache",
     "NoopGraphStore",
     "NoopKeywordStore",
     "NoopLLM",
@@ -32,6 +37,7 @@
     "NoopPIIDetector",
     "NoopQueue",
     "NoopReranker",
+    "NoopRetrievalCache",
     "NoopSecrets",
     "NoopStorage",
     "NoopTelemetry",
diff --git a/packages/core/src/rag_core/spi/noop/answer_cache.py b/packages/core/src/rag_core/spi/noop/answer_cache.py
new file mode 100644
index 0000000..8b4daf8
--- /dev/null
+++ b/packages/core/src/rag_core/spi/noop/answer_cache.py
@@ -0,0 +1,60 @@
+"""In-memory noop AnswerCache."""
+
+from __future__ import annotations
+
+from rag_core.spi.answer_cache import AnswerCache
+from rag_core.types import RequestContext
+
+
+class NoopAnswerCache(AnswerCache):
+    """Per-process dict-backed answer cache for tests and local dev."""
+
+    def __init__(self) -> None:
+        # key: (tenant_id, plan_hash, corpus_version, policy_version) -> bytes
+        self._store: dict[tuple[str, str, int, str], bytes] = {}
+
+    @staticmethod
+    def _key(
+        ctx: RequestContext,
+        plan_hash: str,
+        corpus_version: int,
+        policy_version: str,
+    ) -> tuple[str, str, int, str]:
+        return (str(ctx.tenant_id), plan_hash, corpus_version, policy_version)
+
+    async def get(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+        policy_version: str,
+    ) -> bytes | None:
+        return self._store.get(self._key(ctx, plan_hash, corpus_version, policy_version))
+
+    async def put(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+        policy_version: str,
+        value: bytes,
+        ttl_seconds: int | None = None,
+    ) -> None:
+        self._store[self._key(ctx, plan_hash, corpus_version, policy_version)] = value
+
+    async def invalidate_policy(
+        self,
+        ctx: RequestContext,
+        *,
+        policy_version: str,
+    ) -> int:
+        tid = str(ctx.tenant_id)
+        keys = [k for k in self._store if k[0] == tid and k[3] == policy_version]
+        for k in keys:
+            del self._store[k]
+        return len(keys)
+
+    async def health(self) -> bool:
+        return True
diff --git a/packages/core/src/rag_core/spi/noop/embedding_cache.py b/packages/core/src/rag_core/spi/noop/embedding_cache.py
new file mode 100644
index 0000000..6c43f8f
--- /dev/null
+++ b/packages/core/src/rag_core/spi/noop/embedding_cache.py
@@ -0,0 +1,60 @@
+"""In-memory noop EmbeddingCache."""
+
+from __future__ import annotations
+
+from rag_core.spi.embedding_cache import EmbeddingCache
+from rag_core.types import Embedding, RequestContext
+
+
+class NoopEmbeddingCache(EmbeddingCache):
+    """Per-process dict-backed embedding cache for tests and local dev."""
+
+    def __init__(self) -> None:
+        # key: (tenant_id, model_id, model_version, text_hash) -> Embedding
+        self._store: dict[tuple[str, str, str, str], Embedding] = {}
+
+    @staticmethod
+    def _key(
+        ctx: RequestContext, model_id: str, model_version: str, text_hash: str
+    ) -> tuple[str, str, str, str]:
+        return (str(ctx.tenant_id), model_id, model_version, text_hash)
+
+    async def get(
+        self,
+        ctx: RequestContext,
+        *,
+        model_id: str,
+        model_version: str,
+        text_hash: str,
+    ) -> Embedding | None:
+        return self._store.get(self._key(ctx, model_id, model_version, text_hash))
+
+    async def put(
+        self,
+        ctx: RequestContext,
+        *,
+        model_id: str,
+        model_version: str,
+        text_hash: str,
+        value: Embedding,
+        ttl_seconds: int | None = None,
+    ) -> None:
+        # TTL ignored in noop — embeddings are model-invalidated.
+        self._store[self._key(ctx, model_id, model_version, text_hash)] = value
+
+    async def invalidate_model(
+        self,
+        ctx: RequestContext,
+        *,
+        model_id: str,
+        model_version: str,
+    ) -> int:
+        tid = str(ctx.tenant_id)
+        prefix = (tid, model_id, model_version)
+        keys = [k for k in self._store if k[:3] == prefix]
+        for k in keys:
+            del self._store[k]
+        return len(keys)
+
+    async def health(self) -> bool:
+        return True
diff --git a/packages/core/src/rag_core/spi/noop/retrieval_cache.py b/packages/core/src/rag_core/spi/noop/retrieval_cache.py
new file mode 100644
index 0000000..b6a2d18
--- /dev/null
+++ b/packages/core/src/rag_core/spi/noop/retrieval_cache.py
@@ -0,0 +1,55 @@
+"""In-memory noop RetrievalCache."""
+
+from __future__ import annotations
+
+from rag_core.spi.retrieval_cache import RetrievalCache
+from rag_core.types import ChunkRef, RequestContext
+
+
+class NoopRetrievalCache(RetrievalCache):
+    """Per-process dict-backed retrieval cache for tests and local dev.
+
+    The noop has no scan support, so ``invalidate_corpus`` is a no-op
+    (returns 0) — corpus-version bumping in the key already invalidates
+    affected entries on the next read.
+    """
+
+    def __init__(self) -> None:
+        # key: (tenant_id, plan_hash, corpus_version) -> list[ChunkRef]
+        self._store: dict[tuple[str, str, int], list[ChunkRef]] = {}
+
+    @staticmethod
+    def _key(ctx: RequestContext, plan_hash: str, corpus_version: int) -> tuple[str, str, int]:
+        return (str(ctx.tenant_id), plan_hash, corpus_version)
+
+    async def get(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+    ) -> list[ChunkRef] | None:
+        return self._store.get(self._key(ctx, plan_hash, corpus_version))
+
+    async def put(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+        value: list[ChunkRef],
+        ttl_seconds: int | None = None,
+    ) -> None:
+        self._store[self._key(ctx, plan_hash, corpus_version)] = list(value)
+
+    async def invalidate_corpus(
+        self,
+        ctx: RequestContext,
+        *,
+        corpus_id: str,
+    ) -> int:
+        # Noop has no plan→corpus mapping; rely on corpus_version key bump.
+        return 0
+
+    async def health(self) -> bool:
+        return True
diff --git a/packages/core/src/rag_core/spi/retrieval_cache.py b/packages/core/src/rag_core/spi/retrieval_cache.py
new file mode 100644
index 0000000..407514b
--- /dev/null
+++ b/packages/core/src/rag_core/spi/retrieval_cache.py
@@ -0,0 +1,70 @@
+"""RetrievalCache SPI — keyed by ``(plan_hash, corpus_version)``.
+
+Distinct from `EmbeddingCache` (which survives plan changes) and
+`AnswerCache` (which also invalidates on policy change). A successful
+hit returns the same ordered ``list[ChunkRef]`` the retrieval pipeline
+would have produced — callers still rerank and hydrate from there.
+
+See ``docs/architecture/caching.md`` for the full three-cache split and
+``docs/architecture/performance.md`` for the per-call p99 budget (≤ 2 ms).
+"""
+
+from __future__ import annotations
+
+import abc
+
+from rag_core.spi._base import HealthCheckMixin
+from rag_core.types import ChunkRef, RequestContext
+
+
+class RetrievalCache(HealthCheckMixin, abc.ABC):
+    """Cache of ranked ``ChunkRef`` lists keyed by ``plan_hash`` + corpus version.
+
+    ``plan_hash`` carries every input that determines retrieval: tenant,
+    query text, top-k, filter expression, backend selection, ACL labels.
+    ``corpus_version`` bumps on any successful upsert/delete to the queried
+    corpus, so adding one document invalidates exactly the affected
+    corpus's cache entries — not the entire keyspace.
+
+    Hit rate target: ≥ 30% on production query mix (Pareto-distributed).
+    """
+
+    @abc.abstractmethod
+    async def get(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+    ) -> list[ChunkRef] | None:
+        """Return the cached ChunkRef list or ``None`` on a miss."""
+
+    @abc.abstractmethod
+    async def put(
+        self,
+        ctx: RequestContext,
+        *,
+        plan_hash: str,
+        corpus_version: int,
+        value: list[ChunkRef],
+        ttl_seconds: int | None = None,
+    ) -> None:
+        """Store ``value`` under ``(plan_hash, corpus_version)``.
+
+        ``ttl_seconds`` defaults to a belt-and-braces 3600s in the reference
+        Redis backend; pass explicitly to override.
+        """
+
+    @abc.abstractmethod
+    async def invalidate_corpus(
+        self,
+        ctx: RequestContext,
+        *,
+        corpus_id: str,
+    ) -> int:
+        """Drop every entry whose plan touched ``corpus_id`` for this tenant.
+
+        Called by the write path on a successful ingest. Returns the count
+        removed (best-effort — backends with no tag-based scan may return 0
+        and rely on ``corpus_version`` bumping the key instead).
+        """
diff --git a/packages/observability/src/rag_observability/__init__.py b/packages/observability/src/rag_observability/__init__.py
index 06a6860..8821d0f 100644
--- a/packages/observability/src/rag_observability/__init__.py
+++ b/packages/observability/src/rag_observability/__init__.py
@@ -1 +1,5 @@
-"""AgentContextOS observability — structured logging, events, and context propagation."""
+"""AgentContextOS observability — logging, events, context, async telemetry sink."""
+
+from rag_observability.async_sink import AsyncTelemetrySink, SinkStats
+
+__all__ = ["AsyncTelemetrySink", "SinkStats"]
diff --git a/packages/observability/src/rag_observability/async_sink.py b/packages/observability/src/rag_observability/async_sink.py
new file mode 100644
index 0000000..b0687a0
--- /dev/null
+++ b/packages/observability/src/rag_observability/async_sink.py
@@ -0,0 +1,175 @@
+"""AsyncTelemetrySink — bounded, drop-on-overflow buffer for telemetry export.
+
+OTel exporters, log shippers, and event emitters cannot be allowed to back-
+pressure the request path. A spike of records, a slow downstream collector,
+or a transient network blip would otherwise stall every coroutine that
+tries to emit telemetry.
+
+This sink wraps an inner async ``export`` callable in a bounded
+``asyncio.Queue`` with a background drainer task. ``submit(record)`` is
+non-blocking: if the queue is full, the record is dropped and a per-kind
+counter is incremented. The counter is exposed via :meth:`dropped_total`
+and surfaced as the ``telemetry.dropped_total{kind}`` metric (alertable)
+when wired into the Telemetry SPI.
+
+Design notes
+------------
+
+* **No blocking on submit.** Even at 100k QPS, ``submit`` returns in
+  microseconds. The exporter's tail latency is decoupled from the request
+  path entirely.
+* **Drop policy is explicit.** Dropping is the intentional safety
+  behaviour. Alerts on ``telemetry.dropped_total`` tell operators to
+  scale the exporter, not the request path.
+* **Per-record ``kind`` label.** Lets callers (logs, spans, metrics,
+  StageEvents) share a single sink while attributing drops to a specific
+  stream.
+* **Cooperative shutdown.** :meth:`stop` waits up to ``drain_timeout_s``
+  for the queue to flush, then cancels the drainer.
+
+Default capacity is 10,000 records — the doc's published default.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections import defaultdict
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+ExportFn = Callable[[Any], Awaitable[None]]
+"""Async callable invoked by the drainer for each record."""
+
+
+@dataclass
+class _Entry:
+    kind: str
+    record: Any
+
+
+class AsyncTelemetrySink:
+    """Bounded buffer + background drainer for an async export callable.
+
+    Parameters
+    ----------
+    export_fn:
+        Async callable invoked with each record. Exceptions raised by the
+        exporter are caught and counted; they never propagate to the
+        request path or stop the drainer.
+    max_buffer:
+        Maximum number of records held in the queue. Defaults to 10,000.
+    drain_timeout_s:
+        Seconds :meth:`stop` waits for the queue to flush before cancelling.
+    """
+
+    def __init__(
+        self,
+        export_fn: ExportFn,
+        *,
+        max_buffer: int = 10_000,
+        drain_timeout_s: float = 5.0,
+    ) -> None:
+        if max_buffer < 1:
+            raise ValueError("max_buffer must be >= 1")
+        if drain_timeout_s < 0:
+            raise ValueError("drain_timeout_s must be >= 0")
+
+        self._export_fn = export_fn
+        self._max_buffer = max_buffer
+        self._drain_timeout_s = drain_timeout_s
+
+        self._queue: asyncio.Queue[_Entry] = asyncio.Queue(maxsize=max_buffer)
+        self._dropped: dict[str, int] = defaultdict(int)
+        self._export_errors: dict[str, int] = defaultdict(int)
+        self._drainer: asyncio.Task[None] | None = None
+        self._started = False
+
+    def start(self) -> None:
+        """Spawn the background drainer task. Idempotent."""
+        if self._started:
+            return
+        self._drainer = asyncio.create_task(self._drain(), name="telemetry.drainer")
+        self._started = True
+
+    def submit(self, record: Any, *, kind: str = "default") -> bool:
+        """Enqueue ``record`` for export. Non-blocking.
+
+        Returns ``True`` if enqueued, ``False`` if dropped due to a full
+        buffer. The drop counter for ``kind`` is incremented on a drop.
+        """
+        try:
+            self._queue.put_nowait(_Entry(kind=kind, record=record))
+            return True
+        except asyncio.QueueFull:
+            self._dropped[kind] += 1
+            return False
+
+    def dropped_total(self, kind: str | None = None) -> int:
+        """Return the total drops for ``kind`` (or across all kinds)."""
+        if kind is None:
+            return sum(self._dropped.values())
+        return self._dropped.get(kind, 0)
+
+    def export_errors_total(self, kind: str | None = None) -> int:
+        """Return the total exporter errors for ``kind`` (or across all kinds)."""
+        if kind is None:
+            return sum(self._export_errors.values())
+        return self._export_errors.get(kind, 0)
+
+    def dropped_snapshot(self) -> dict[str, int]:
+        """Return a copy of the per-kind drop counters."""
+        return dict(self._dropped)
+
+    @property
+    def buffer_size(self) -> int:
+        """Number of records currently buffered (not yet exported)."""
+        return self._queue.qsize()
+
+    async def stop(self) -> None:
+        """Flush pending records (up to ``drain_timeout_s``) and cancel the drainer."""
+        if not self._started:
+            return
+        try:
+            await asyncio.wait_for(self._queue.join(), timeout=self._drain_timeout_s)
+        except TimeoutError:
+            pass
+
+        drainer = self._drainer
+        self._drainer = None
+        self._started = False
+        if drainer is not None and not drainer.done():
+            drainer.cancel()
+            try:
+                await drainer
+            except (asyncio.CancelledError, Exception):  # noqa: S110, BLE001
+                # Drainer cancellation is expected during shutdown; any
+                # remaining exporter error would already have been counted.
+                pass
+
+    async def _drain(self) -> None:
+        while True:
+            entry = await self._queue.get()
+            try:
+                await self._export_fn(entry.record)
+            except Exception:  # noqa: BLE001 — exporter failures must never crash drainer
+                self._export_errors[entry.kind] += 1
+            finally:
+                self._queue.task_done()
+
+
+@dataclass
+class SinkStats:
+    """Lightweight snapshot of an :class:`AsyncTelemetrySink`'s counters."""
+
+    buffer_size: int
+    dropped: dict[str, int] = field(default_factory=dict)
+    export_errors: dict[str, int] = field(default_factory=dict)
+
+    @classmethod
+    def from_sink(cls, sink: AsyncTelemetrySink) -> SinkStats:
+        return cls(
+            buffer_size=sink.buffer_size,
+            dropped=dict(sink.dropped_snapshot()),
+            export_errors=dict(sink._export_errors),  # noqa: SLF001
+        )
diff --git a/tests/contract/conftest.py b/tests/contract/conftest.py
index baced0d..e139eb5 100644
--- a/tests/contract/conftest.py
+++ b/tests/contract/conftest.py
@@ -4,11 +4,13 @@
 
 import pytest
 from rag_core.spi.noop import (
+    NoopAnswerCache,
     NoopAuditStore,
     NoopAuth,
     NoopCache,
     NoopConnector,
     NoopEmbedder,
+    NoopEmbeddingCache,
     NoopGraphStore,
     NoopKeywordStore,
     NoopLLM,
@@ -17,6 +19,7 @@
     NoopPIIDetector,
     NoopQueue,
     NoopReranker,
+    NoopRetrievalCache,
     NoopSecrets,
     NoopStorage,
     NoopTelemetry,
@@ -124,6 +127,21 @@ def cache() -> NoopCache:
     return NoopCache()
 
 
+@pytest.fixture()
+def embedding_cache() -> NoopEmbeddingCache:
+    return NoopEmbeddingCache()
+
+
+@pytest.fixture()
+def retrieval_cache() -> NoopRetrievalCache:
+    return NoopRetrievalCache()
+
+
+@pytest.fixture()
+def answer_cache() -> NoopAnswerCache:
+    return NoopAnswerCache()
+
+
 @pytest.fixture()
 def queue() -> NoopQueue:
     return NoopQueue()
diff --git a/tests/contract/spi_signature.py b/tests/contract/spi_signature.py
index 04ed826..8861f43 100644
--- a/tests/contract/spi_signature.py
+++ b/tests/contract/spi_signature.py
@@ -33,9 +33,11 @@
 from rag_core.spi import (
     LLM,
     OCR,
+    AnswerCache,
     Cache,
     Connector,
     Embedder,
+    EmbeddingCache,
     GraphIndexBackend,
     GraphRetrievalBackend,
     GraphStore,
@@ -46,6 +48,7 @@
     PIIDetector,
     Queue,
     Reranker,
+    RetrievalCache,
     Secrets,
     Storage,
     VectorIndexBackend,
@@ -60,9 +63,11 @@
 # Per-class set of methods that legitimately do *not* take ctx as first arg.
 _PER_CLASS_EXEMPT: dict[type, set[str]] = {
     # health() lives on HealthCheckMixin and is universally pre-ctx.
+    AnswerCache: {"health"},
     Cache: {"health"},
     Connector: {"health"},
     Embedder: {"health", "model", "dimension"},
+    EmbeddingCache: {"health"},
     GraphRetrievalBackend: {"health"},
     GraphIndexBackend: {"health"},
     GraphStore: {"health"},
@@ -75,6 +80,7 @@
     PIIDetector: {"health"},
     Queue: {"health"},
     Reranker: {"health"},
+    RetrievalCache: {"health"},
     Secrets: {"health"},
     Storage: {"health"},
     VectorRetrievalBackend: {"health"},
@@ -85,9 +91,11 @@
 
 def _ctx_threaded_classes() -> list[type]:
     return [
+        AnswerCache,
         Cache,
         Connector,
         Embedder,
+        EmbeddingCache,
         GraphRetrievalBackend,
         GraphIndexBackend,
         GraphStore,
@@ -100,6 +108,7 @@ def _ctx_threaded_classes() -> list[type]:
         PIIDetector,
         Queue,
         Reranker,
+        RetrievalCache,
         Secrets,
         Storage,
         VectorRetrievalBackend,
diff --git a/tests/contract/test_answer_cache.py b/tests/contract/test_answer_cache.py
new file mode 100644
index 0000000..9e858d1
--- /dev/null
+++ b/tests/contract/test_answer_cache.py
@@ -0,0 +1,70 @@
+"""Conformance tests for AnswerCache SPI."""
+
+from __future__ import annotations
+
+import pytest
+from rag_core.spi.noop import NoopAnswerCache
+from rag_core.types import RequestContext
+
+pytestmark = pytest.mark.contract
+
+
+async def test_health(answer_cache: NoopAnswerCache) -> None:
+    assert await answer_cache.health() is True
+
+
+async def test_put_and_get(answer_cache: NoopAnswerCache, ctx: RequestContext) -> None:
+    await answer_cache.put(
+        ctx, plan_hash="p1", corpus_version=1, policy_version="v1", value=b"answer"
+    )
+    got = await answer_cache.get(ctx, plan_hash="p1", corpus_version=1, policy_version="v1")
+    assert got == b"answer"
+
+
+async def test_get_miss_returns_none(answer_cache: NoopAnswerCache, ctx: RequestContext) -> None:
+    got = await answer_cache.get(ctx, plan_hash="p1", corpus_version=1, policy_version="v1")
+    assert got is None
+
+
+async def test_policy_version_misses_on_rotation(
+    answer_cache: NoopAnswerCache, ctx: RequestContext
+) -> None:
+    await answer_cache.put(ctx, plan_hash="p1", corpus_version=1, policy_version="v1", value=b"a1")
+    # New policy version → miss
+    assert (
+        await answer_cache.get(ctx, plan_hash="p1", corpus_version=1, policy_version="v2") is None
+    )
+    # Original policy version still hits
+    assert (
+        await answer_cache.get(ctx, plan_hash="p1", corpus_version=1, policy_version="v1") == b"a1"
+    )
+
+
+async def test_invalidate_policy_drops_matching_entries(
+    answer_cache: NoopAnswerCache, ctx: RequestContext
+) -> None:
+    await answer_cache.put(ctx, plan_hash="p1", corpus_version=1, policy_version="v1", value=b"a")
+    await answer_cache.put(ctx, plan_hash="p2", corpus_version=1, policy_version="v1", value=b"b")
+    await answer_cache.put(ctx, plan_hash="p1", corpus_version=1, policy_version="v2", value=b"c")
+
+    removed = await answer_cache.invalidate_policy(ctx, policy_version="v1")
+    assert removed == 2
+    assert (
+        await answer_cache.get(ctx, plan_hash="p1", corpus_version=1, policy_version="v1") is None
+    )
+    # v2 survives
+    assert (
+        await answer_cache.get(ctx, plan_hash="p1", corpus_version=1, policy_version="v2") == b"c"
+    )
+
+
+async def test_tenant_isolation(
+    answer_cache: NoopAnswerCache,
+    ctx: RequestContext,
+    other_ctx: RequestContext,
+) -> None:
+    await answer_cache.put(
+        ctx, plan_hash="p1", corpus_version=1, policy_version="v1", value=b"secret"
+    )
+    got = await answer_cache.get(other_ctx, plan_hash="p1", corpus_version=1, policy_version="v1")
+    assert got is None
diff --git a/tests/contract/test_embedding_cache.py b/tests/contract/test_embedding_cache.py
new file mode 100644
index 0000000..b584204
--- /dev/null
+++ b/tests/contract/test_embedding_cache.py
@@ -0,0 +1,78 @@
+"""Conformance tests for EmbeddingCache SPI."""
+
+from __future__ import annotations
+
+import pytest
+from rag_core.spi.noop import NoopEmbeddingCache
+from rag_core.types import ChunkId, Embedding, RequestContext, TenantId
+
+pytestmark = pytest.mark.contract
+
+
+def _emb(chunk_id: str = "c1", tenant_id: str = "tenant-test") -> Embedding:
+    return Embedding(
+        chunk_id=ChunkId(chunk_id),
+        tenant_id=TenantId(tenant_id),
+        model="bge-large-en-v1.5",
+        vector=[0.1, 0.2, 0.3, 0.4],
+        dimension=4,
+    )
+
+
+async def test_health(embedding_cache: NoopEmbeddingCache) -> None:
+    assert await embedding_cache.health() is True
+
+
+async def test_put_and_get(embedding_cache: NoopEmbeddingCache, ctx: RequestContext) -> None:
+    e = _emb()
+    await embedding_cache.put(ctx, model_id="bge", model_version="1.5", text_hash="h1", value=e)
+    got = await embedding_cache.get(ctx, model_id="bge", model_version="1.5", text_hash="h1")
+    assert got == e
+
+
+async def test_get_miss_returns_none(
+    embedding_cache: NoopEmbeddingCache, ctx: RequestContext
+) -> None:
+    got = await embedding_cache.get(ctx, model_id="bge", model_version="1.5", text_hash="missing")
+    assert got is None
+
+
+async def test_model_version_partitions_keys(
+    embedding_cache: NoopEmbeddingCache, ctx: RequestContext
+) -> None:
+    a = _emb()
+    b = _emb(chunk_id="c2")
+    await embedding_cache.put(ctx, model_id="bge", model_version="1.5", text_hash="h1", value=a)
+    await embedding_cache.put(ctx, model_id="bge", model_version="2.0", text_hash="h1", value=b)
+
+    assert await embedding_cache.get(ctx, model_id="bge", model_version="1.5", text_hash="h1") == a
+    assert await embedding_cache.get(ctx, model_id="bge", model_version="2.0", text_hash="h1") == b
+
+
+async def test_invalidate_model_drops_only_matching_version(
+    embedding_cache: NoopEmbeddingCache, ctx: RequestContext
+) -> None:
+    e = _emb()
+    await embedding_cache.put(ctx, model_id="bge", model_version="1.5", text_hash="h1", value=e)
+    await embedding_cache.put(ctx, model_id="bge", model_version="1.5", text_hash="h2", value=e)
+    await embedding_cache.put(ctx, model_id="bge", model_version="2.0", text_hash="h1", value=e)
+
+    removed = await embedding_cache.invalidate_model(ctx, model_id="bge", model_version="1.5")
+    assert removed == 2
+    assert (
+        await embedding_cache.get(ctx, model_id="bge", model_version="1.5", text_hash="h1") is None
+    )
+    # Different version survives
+    assert await embedding_cache.get(ctx, model_id="bge", model_version="2.0", text_hash="h1") == e
+
+
+async def test_tenant_isolation(
+    embedding_cache: NoopEmbeddingCache,
+    ctx: RequestContext,
+    other_ctx: RequestContext,
+) -> None:
+    e = _emb(tenant_id=str(ctx.tenant_id))
+    await embedding_cache.put(ctx, model_id="bge", model_version="1.5", text_hash="h1", value=e)
+    # Other tenant must not see it
+    got = await embedding_cache.get(other_ctx, model_id="bge", model_version="1.5", text_hash="h1")
+    assert got is None
diff --git a/tests/contract/test_retrieval_cache.py b/tests/contract/test_retrieval_cache.py
new file mode 100644
index 0000000..441c4d7
--- /dev/null
+++ b/tests/contract/test_retrieval_cache.py
@@ -0,0 +1,65 @@
+"""Conformance tests for RetrievalCache SPI."""
+
+from __future__ import annotations
+
+import pytest
+from rag_core.spi.noop import NoopRetrievalCache
+from rag_core.types import ChunkId, ChunkRef, RequestContext, TenantId
+
+pytestmark = pytest.mark.contract
+
+
+def _refs(tenant_id: str = "tenant-test") -> list[ChunkRef]:
+    return [
+        ChunkRef(chunk_id=ChunkId("c1"), tenant_id=TenantId(tenant_id), score=0.9),
+        ChunkRef(chunk_id=ChunkId("c2"), tenant_id=TenantId(tenant_id), score=0.8),
+    ]
+
+
+async def test_health(retrieval_cache: NoopRetrievalCache) -> None:
+    assert await retrieval_cache.health() is True
+
+
+async def test_put_and_get(retrieval_cache: NoopRetrievalCache, ctx: RequestContext) -> None:
+    refs = _refs()
+    await retrieval_cache.put(ctx, plan_hash="p1", corpus_version=1, value=refs)
+    got = await retrieval_cache.get(ctx, plan_hash="p1", corpus_version=1)
+    assert got == refs
+
+
+async def test_get_miss_returns_none(
+    retrieval_cache: NoopRetrievalCache, ctx: RequestContext
+) -> None:
+    got = await retrieval_cache.get(ctx, plan_hash="missing", corpus_version=1)
+    assert got is None
+
+
+async def test_corpus_version_bump_misses(
+    retrieval_cache: NoopRetrievalCache, ctx: RequestContext
+) -> None:
+    refs = _refs()
+    await retrieval_cache.put(ctx, plan_hash="p1", corpus_version=1, value=refs)
+    # Same plan, new corpus version → miss
+    assert await retrieval_cache.get(ctx, plan_hash="p1", corpus_version=2) is None
+    # Original version still hits
+    assert await retrieval_cache.get(ctx, plan_hash="p1", corpus_version=1) == refs
+
+
+async def test_invalidate_corpus_is_noop_for_noop_backend(
+    retrieval_cache: NoopRetrievalCache, ctx: RequestContext
+) -> None:
+    # The noop relies on key-bumping rather than tag scans; it must still
+    # return without error.
+    removed = await retrieval_cache.invalidate_corpus(ctx, corpus_id="corpus-a")
+    assert removed == 0
+
+
+async def test_tenant_isolation(
+    retrieval_cache: NoopRetrievalCache,
+    ctx: RequestContext,
+    other_ctx: RequestContext,
+) -> None:
+    refs = _refs(tenant_id=str(ctx.tenant_id))
+    await retrieval_cache.put(ctx, plan_hash="p1", corpus_version=1, value=refs)
+    got = await retrieval_cache.get(other_ctx, plan_hash="p1", corpus_version=1)
+    assert got is None
diff --git a/tests/observability/__init__.py b/tests/observability/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/observability/test_async_sink.py b/tests/observability/test_async_sink.py
new file mode 100644
index 0000000..9de34de
--- /dev/null
+++ b/tests/observability/test_async_sink.py
@@ -0,0 +1,167 @@
+"""Unit tests for ``rag_observability.async_sink.AsyncTelemetrySink``."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from rag_observability.async_sink import AsyncTelemetrySink, SinkStats
+
+
+async def test_submit_and_drain_in_order() -> None:
+    exported: list[int] = []
+
+    async def export(record: int) -> None:
+        exported.append(record)
+
+    sink = AsyncTelemetrySink(export, max_buffer=100)
+    sink.start()
+
+    for i in range(5):
+        assert sink.submit(i) is True
+
+    await sink.stop()
+
+    assert exported == [0, 1, 2, 3, 4]
+    assert sink.dropped_total() == 0
+    assert sink.export_errors_total() == 0
+
+
+async def test_drops_on_full_buffer_and_counts_them() -> None:
+    block = asyncio.Event()
+    exported: list[int] = []
+
+    async def slow_export(record: int) -> None:
+        await block.wait()
+        exported.append(record)
+
+    sink = AsyncTelemetrySink(slow_export, max_buffer=2)
+    sink.start()
+
+    # The drainer will pull one record off the queue and then block in
+    # slow_export. So after the drainer wakes, at most max_buffer + 1
+    # records can be in flight without dropping (1 in the exporter, 2 in
+    # the queue). Anything beyond that drops.
+    accepted = 0
+    dropped = 0
+    for i in range(20):
+        if sink.submit(i, kind="ingest"):
+            accepted += 1
+        else:
+            dropped += 1
+        # Give the drainer a tick after the first submit so it pulls one.
+        if i == 0:
+            await asyncio.sleep(0)
+
+    assert dropped > 0, "expected at least one drop"
+    assert sink.dropped_total() == dropped
+    assert sink.dropped_total("ingest") == dropped
+    assert sink.dropped_total("other") == 0
+
+    # Unblock and drain
+    block.set()
+    await sink.stop()
+
+    assert len(exported) == accepted
+
+
+async def test_exporter_exception_is_swallowed_and_counted() -> None:
+    exported: list[int] = []
+
+    async def flaky_export(record: int) -> None:
+        if record == 2:
+            raise RuntimeError("boom")
+        exported.append(record)
+
+    sink = AsyncTelemetrySink(flaky_export, max_buffer=100)
+    sink.start()
+
+    for i in range(5):
+        sink.submit(i, kind="span")
+
+    await sink.stop()
+
+    assert exported == [0, 1, 3, 4]  # record 2 raised, others fine
+    assert sink.export_errors_total() == 1
+    assert sink.export_errors_total("span") == 1
+
+
+async def test_stop_is_idempotent() -> None:
+    async def export(record: int) -> None:
+        pass
+
+    sink = AsyncTelemetrySink(export, max_buffer=10)
+    sink.start()
+    await sink.stop()
+    await sink.stop()  # must not raise
+
+
+async def test_start_is_idempotent() -> None:
+    async def export(record: int) -> None:
+        pass
+
+    sink = AsyncTelemetrySink(export, max_buffer=10)
+    sink.start()
+    sink.start()  # must not raise or double-spawn
+    await sink.stop()
+
+
+async def test_submit_before_start_buffers() -> None:
+    exported: list[int] = []
+
+    async def export(record: int) -> None:
+        exported.append(record)
+
+    sink = AsyncTelemetrySink(export, max_buffer=10)
+
+    # submitted before start — still queued
+    assert sink.submit(1) is True
+    assert sink.submit(2) is True
+    assert sink.buffer_size == 2
+
+    sink.start()
+    await sink.stop()
+
+    assert exported == [1, 2]
+
+
+async def test_sink_stats_snapshot() -> None:
+    async def export(record: int) -> None:
+        pass
+
+    sink = AsyncTelemetrySink(export, max_buffer=2)
+    sink.start()
+    # Force a drop by overrunning the buffer with the drainer paused.
+    # Easiest: stop the drainer, then submit past the limit.
+    await sink.stop()
+    sink.submit(1, kind="k1")
+    sink.submit(2, kind="k1")
+    sink.submit(3, kind="k1")  # dropped
+    sink.submit(4, kind="k2")  # dropped
+
+    stats = SinkStats.from_sink(sink)
+    assert stats.buffer_size == 2
+    assert stats.dropped == {"k1": 1, "k2": 1}
+
+
+async def test_invalid_construction() -> None:
+    async def export(record: int) -> None:
+        pass
+
+    with pytest.raises(ValueError):
+        AsyncTelemetrySink(export, max_buffer=0)
+    with pytest.raises(ValueError):
+        AsyncTelemetrySink(export, drain_timeout_s=-1)
+
+
+async def test_stop_drainer_under_timeout_does_not_hang() -> None:
+    blocking = asyncio.Event()
+
+    async def really_slow(record: int) -> None:
+        await blocking.wait()  # never set
+
+    sink = AsyncTelemetrySink(really_slow, max_buffer=10, drain_timeout_s=0.05)
+    sink.start()
+    sink.submit(1)
+    # stop should give up after drain_timeout_s and cancel the drainer.
+    await sink.stop()
diff --git a/uv.lock b/uv.lock
index b35d026..f1034ba 100644
--- a/uv.lock
+++ b/uv.lock
@@ -3247,7 +3247,7 @@ provides-extras = ["eval"]
 
 [[package]]
 name = "rag-core"
-version = "0.6.0"
+version = "0.7.0"
 source = { editable = "packages/core" }
 dependencies = [
     { name = "opentelemetry-api" },