feat(core,observability): Cache SPI split + async telemetry (Step 1.1e)#51
Merged
officialCodeWork merged 1 commit intoMay 24, 2026
Conversation
…p 1.1e)
Adds the three distinct cache SPIs the platform needs (each with a
different key and invalidation contract) plus the bounded, drop-on-
overflow telemetry path that decouples the request path from any
exporter back-pressure.
Cache SPIs (rag_core.spi)
-------------------------
- ``EmbeddingCache`` — key ``(tenant_id, model_id, model_version,
text_hash)``; ``invalidate_model(ctx, model_id, model_version)`` for
model rotation. Survives re-chunking; target hit rate ≥ 99% on
identical re-ingest.
- ``RetrievalCache`` — key ``(tenant_id, plan_hash, corpus_version)``;
``invalidate_corpus(ctx, corpus_id)`` called by the write path on
any successful upsert/delete. Stores ``list[ChunkRef]``; target hit
rate ≥ 30%.
- ``AnswerCache`` — key ``(tenant_id, plan_hash, corpus_version,
policy_version)``; ``invalidate_policy(ctx, policy_version)`` called
by the control plane on policy rotation. Stores opaque bytes; target
hit rate ≥ 15%.
All three are ``ctx``-first; ``tests/contract/spi_signature.py``
extended to cover them. Noop in-memory reference impls live in
``rag_core.spi.noop``. The base ``Cache`` SPI is retained for generic
byte-level caching needs.
Async telemetry sink (rag_observability)
----------------------------------------
- ``AsyncTelemetrySink(export_fn, *, max_buffer=10_000,
drain_timeout_s=5.0)`` — bounded ``asyncio.Queue`` fronting any async
exporter. ``submit(record, kind=...)`` is non-blocking and returns
``False`` on a full queue. Per-kind drop and exporter-error counters
are surfaceable as ``telemetry.dropped_total{kind}`` and
``telemetry.export_errors_total{kind}``.
- Idempotent ``start()`` / ``stop()``; cooperative drain with timeout.
- Records submitted before ``start()`` are queued; the drainer flushes
them once started.
Tests
-----
- 18 new conformance tests across the three caches (hit/miss, version
partition, invalidate semantics, tenant isolation).
- 9 new sink tests covering drain ordering, drops under simulated
back-pressure, exporter-exception swallowing, idempotent lifecycle,
drain-timeout cancellation.
- Full suite: **576 passed**, 1 skipped (was 549 before this step).
- ``mypy --strict`` clean across the workspace (67 source files).
Documentation
-------------
- Updated [caching.md](../architecture/caching.md): SPI signatures now
show ``ctx`` as first arg; extension-point block lists the canonical
shapes including ``invalidate_*`` methods.
- Updated [performance.md](../architecture/performance.md): hot-path
discipline section gains a concrete hot/cold call-site table; async
telemetry section rewritten around ``AsyncTelemetrySink`` with
counter / alerting guidance.
- Updated [rag-core.md](../reference/rag-core.md): new Cache SPIs
section.
- New [rag-observability.md](../reference/rag-observability.md):
package reference covering logger, event registry, and the new
``AsyncTelemetrySink``.
``rag-core`` 0.6.0 → 0.7.0.
Step 1.1e of the Phase 1 refactor window. Last remaining step in the
window is 1.1f (ADRs 0005–0009 finalized + reviewer checklist).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
4 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
rag_core.spi.{embedding,retrieval,answer}_cache), each with a different cache key and a different invalidation contract per caching.md:EmbeddingCache—(tenant, model_id, model_version, text_hash);invalidate_model(...).RetrievalCache—(tenant, plan_hash, corpus_version);invalidate_corpus(...).AnswerCache—(tenant, plan_hash, corpus_version, policy_version);invalidate_policy(...).AsyncTelemetrySinkinrag-observability— bounded buffer (default 10 000), non-blockingsubmit(), per-kind drop + exporter-error counters, idempotent start/stop, drain-timeout cancel.tests/contract/spi_signature.pyto enforcectx-first on the three new caches.AsyncTelemetrySink(counters, alerting).rag-core0.6.0 → 0.7.0.Step 1.1e of the Phase 1 refactor window (1.1a–1.1f). Last remaining step in the window is 1.1f (ADRs 0005–0009 finalized + reviewer checklist).
Tests
AsyncTelemetrySinktests (drain ordering, drops under simulated back-pressure, exporter-exception swallowing, idempotent lifecycle, drain-timeout cancellation).mypy --strictclean across the workspace (67 source files).ruff check .clean.Documentation
Per the per-PR docs requirement (CLAUDE.md):
invalidate_*).AsyncTelemetrySinksection).AsyncTelemetrySinkusage and alerting guidance.Tracker
Test plan
uv run pytest tests/contract/test_embedding_cache.py tests/contract/test_retrieval_cache.py tests/contract/test_answer_cache.py -xuv run pytest tests/observability/test_async_sink.py -xuv run pytest tests/ packages/ -k "not integration"— 576 passeduv run ruff check .cleanuv run mypy packages/core/ packages/observability/ apps/gateway/cleanPYTHONPATH=... uv run python -m rag_core.gen_schemas dist/schemas/— no drift🤖 Generated with Claude Code