feat(adapters): caching interceptor

[Glorf]

Adapter interceptor follow-up on top of feat/adapter-base (the adapter middleware framework). Independent of the observability and request-rewriting follow-ups.

Adds the caching interceptor + its sqlite disk-cache module:

• caching (request → response): SHA-256-keyed disk cache. On hit, short-circuits the chain (RequestToResponseInterceptor); on miss, writes the upstream response back via the matching ResponseInterceptor. Keys are session-scoped when ctx.extra["session_id"] is set.
• nemo_gym/adapters/cache/disk_cache.py: sqlite-backed cache with canonical-body hashing.

Tests: cache-key goldens, sqlite round-trip, interceptor behavior, parity replay — 16 added (green locally).

Supersedes Glorf#2 (fork discontinued; rebased clean onto upstream feat/adapter-base).

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[github-actions]

🌿 Preview your docs: https://nvidia-preview-feat-adapter-caching.docs.buildwithfern.com/nemo/gym

Here are the markdown pages you've updated:

• nemo/gym/model-server/adapters-caching

[ffrujeri]

Thanks for this — the caching interceptor is clean, well-tested (16 tests, green locally), and the docs + sqlite-via-to_thread design are nicely done. A 6-agent review surfaced one correctness question worth resolving before merge (cache-key field coverage) plus a handful of smaller suggestions and test fill-ins; all are left as inline comments, phrased as questions where intent wasn't clear. Nothing blocking on merge mechanics (clean, no conflicts).

Generated by Claude Code

[ffrujeri]

disk_cache.py:30

Trying to understand the intended scope of the cache key. _RELEVANT_KEYS covers model/messages/tools/temperature/max_tokens/top_p/seed (+ extra_body), but several other output-affecting params aren't included — so two requests differing only in one of them would hash to the same key and the second would get the first's cached response. A few cases I wasn't sure how you'd want handled:

• response_format — would a {"type": "json_object"} call collide with a prior free-text call on the same prompt, and hand the verifier back un-structured output?
• n — does an n: 1 entry get served to a later n: 8 (pass@k) request, collapsing it to a single choice?
• stop — would an agent's stop: ["\n\nObservation:"] call collide with a no-stop call, so the cached body runs past the boundary the harness expects?
• logprobs / tool_choice — similar story for GenRM/logprob scoring and forced-tool calls?

Is the intent that callers always route these through extra_body (which is keyed), or should the allowlist be expanded — or the key derived from the full canonical body minus a small denylist of volatile fields? Mostly want to make sure the cache can't silently return a wrong-shaped response in the eval/RL paths.

[ffrujeri]

caching.py:38

How does this interceptor expect to interact with streaming requests? I see stream_safe = False declared, but I couldn't find where the pipeline reads that flag (it looks inert for now), and stream isn't part of the cache key. So it seems like a {"stream": true} request could take a cache hit and be returned a JSON body instead of an SSE stream — would that break the client's stream parser? Would it make sense to short-circuit early here (e.g. if req.body.get("stream"): return req) until the pipeline honors stream_safe, and flip the cache_key(stream=True) == cache_key(stream=False) assertion accordingly?

[ffrujeri]

adapters-caching.mdx:22

This line notes keys are session-scoped "when present (via ctx.extra["session_id"])" — I wanted to check how that plays out in proxy mode. It looks like session_id is only set from the /s/<hex>/ path in middleware mode, and the proxy builds a bare context, so in the bring-your-own-inference path the prefix would always be empty. If that's right, then with temperature > 0 / no seed, repeated samples of the same prompt would share one cache entry and collapse sampling diversity. Could we add a short caveat that proxy mode currently has no session isolation, and that caching is best used with deterministic decoding (or a distinct session_id per sample) to avoid silently de-duplicating independent samples?

[ffrujeri]

test_adapter_parity_replay_caching.py:46

It looks like the fixture-loading machinery here (_load_fixtures, _FIXTURES, _normalise_headers, _VOLATILE_HEADERS, FIXTURE_DIR) isn't consumed by any test — adapter_fixtures/ doesn't exist yet, so only test_caching_round_trip_returns_same_response runs. Is the fixture replay intended to land in a follow-up? If so it might be worth a quick # TODO; if not, maybe drop the unused helpers and trim the module docstring so it matches what actually runs.

[ffrujeri]

test_adapter_interceptors_caching.py:125

Two response-side behaviors look correct but aren't directly asserted: error responses not being cached (right now key is None short-circuits before not resp.ok, so the not-ok path is never exercised), and the non-dict-body skip. These are also the uncovered branches. Would it be worth adding (both pass locally)? The suggestion appends two methods to the TestCachingInterceptor class:

Suggested change

	assert isinstance(result_b, AdapterRequest), "must be a cache miss, not a hit"
	assert isinstance(result_b, AdapterRequest), "must be a cache miss, not a hit"
	async def test_error_response_not_cached(self, tmp_path):
	from nemo_gym.adapters.interceptors.caching import Interceptor
	i = Interceptor(cache_dir=str(tmp_path))
	ctx = InterceptorContext()
	body = {"messages": [{"role": "user", "content": "err"}]}
	req = AdapterRequest(method="POST", path="/chat/completions", headers={}, body=body, ctx=ctx)
	await i.intercept_request(req)
	await i.intercept_response(AdapterResponse(status_code=500, headers={}, body={"error": "boom"}, ctx=ctx))
	req2 = AdapterRequest(method="POST", path="/chat/completions", headers={}, body=body, ctx=InterceptorContext())
	assert isinstance(await i.intercept_request(req2), AdapterRequest), "error response must not be cached"
	async def test_non_dict_body_not_cached(self, tmp_path):
	from nemo_gym.adapters.interceptors.caching import Interceptor
	i = Interceptor(cache_dir=str(tmp_path))
	ctx = InterceptorContext()
	body = {"messages": [{"role": "user", "content": "bytes"}]}
	req = AdapterRequest(method="POST", path="/chat/completions", headers={}, body=body, ctx=ctx)
	await i.intercept_request(req)
	out = await i.intercept_response(AdapterResponse(status_code=200, headers={}, body=b"raw-bytes", ctx=ctx))
	assert out.ctx.extra.get("cache_key") is None
	req2 = AdapterRequest(method="POST", path="/chat/completions", headers={}, body=body, ctx=InterceptorContext())
	assert isinstance(await i.intercept_request(req2), AdapterRequest), "non-dict body must not be cached"

[ffrujeri]

test_adapter_cache_keys.py:68

The extra_body branch of cache_key (dict → folded into the key, non-dict → ignored) doesn't look covered yet. Since extra_body is the one "extra" field that does affect the key, pinning it with a test also documents that contract. Both pass locally — the suggestion appends two module-level tests:

Suggested change

	assert k_none == DiskCache.cache_key(body, session_prefix="")
	assert k_none == DiskCache.cache_key(body, session_prefix="")
	def test_key_includes_extra_body():
	base = {"model": "gpt-4", "messages": [{"role": "user", "content": "hi"}]}
	with_extra = {**base, "extra_body": {"guided_json": {"type": "object"}}}
	assert _compute_key(base) != _compute_key(with_extra)
	def test_key_ignores_non_dict_extra_body():
	base = {"model": "gpt-4", "messages": [{"role": "user", "content": "hi"}]}
	with_bad = {**base, "extra_body": "not-a-dict"}
	assert _compute_key(base) == _compute_key(with_bad)

[ffrujeri]

disk_cache.py:53

Small one: busy_timeout is set in _init_db, but since it's per-connection (WAL persists on the file, busy_timeout doesn't), the fresh connections in _get_sync/_set_sync would use the default of 0. In the shared-cache_dir scenario the docs mention, that could turn a transient lock into a silently-dropped write. Would it be worth re-applying PRAGMA busy_timeout=5000 on each connection (or factoring a small _connect() helper) so the 5s wait applies to reads/writes too?

[ffrujeri]

test_adapter_interceptors_caching.py:64

test_bypass_mode confirms the request passes through, but not that bypass actually skips the cache read/write — it'd still pass if bypass did a lookup. Might be worth strengthening it to populate an entry first, then assert bypass still returns an AdapterRequest (no hit) and doesn't set cache_key in the ctx, so the test pins the "bypass means no cache interaction" contract.

[ffrujeri]

caching.py:31

Optional nit: since the hit-short-circuit / miss-write-back contract is a little subtle, a short docstring on the public surface might help future readers:

Suggested change

	class Interceptor(RequestToResponseInterceptor, ResponseInterceptor):
	stream_safe = False
	class Interceptor(RequestToResponseInterceptor, ResponseInterceptor):
	"""SHA-256-keyed disk cache. On hit, short-circuits the chain with the
	stored response; on miss, records the cache key so the matching response
	phase writes the upstream body back. Keys are session-scoped when
	``ctx.extra["session_id"]`` is set. ``stream_safe = False`` is declared
	but not yet enforced by the pipeline (see streaming note)."""
	stream_safe = False

(Same idea for DiskCache / cache_key / get / set in disk_cache.py if you agree it's worth it.)