Require confirmed cache usage in strict investigation tests

config/llm.gemini-cache.example.toml

@@ -25,6 +25,10 @@ cache_ttl_seconds = 300
 # A character heuristic, not a tokenizer or a format limit. Provider minimums
 # still apply; rejected/too-small cache requests fall back to the full input.
 cache_min_prefix_chars = 16000
+# For an explicit cache test, require confirmed hits in investigation calls:
+# cache_required = true
+# cache_required_prompt_prefix = "investigation_project_"
+# Omit the prefix to require caching in ALL phases (small prefixes may fail).
 
 # Optional estimates in USD per million tokens; verify your model/tariff/date.
 # These illustrative standard-tier rates are from 2026-10-16, valid through

docs/llm-context-caching.md

@@ -69,6 +69,22 @@ are sent in headers; redirects and ambient proxies are disabled. Native
 
 ## Reading the measurements
 
+For a strict offline-prepared/live-user-run cache test, set `cache_required = true`.
+Optionally restrict it with `cache_required_prompt_prefix = "investigation_project_"`;
+this requires cache usage throughout investigation while leaving small earlier
+phases optional. Without a prefix, the requirement covers all calls from that
+provider. It requires `context_cache = true` and the native Gemini adapter.
+The worker refuses to generate without a handle (`LLM_CACHE_REQUIRED`), refuses
+uncached retry/failover on a required call, and only accepts a response with a
+positive, consistent provider-reported cached token count
+(`LLM_CACHE_USAGE_UNCONFIRMED` otherwise). This is a test assertion, not a
+promise that Gemini will accept a prefix. A rejected/too-small cache ends the
+test explicitly. A response without confirmed usage may already have incurred
+API costs. Default behaviour remains optional; old operation hashes stay valid.
+Strict cache creation may use half of the configured request timeout, instead
+of the five-second optional-optimization allowance. The overall deadline still
+applies; a creation HTTP status is retained for diagnosing a rejected cache.
+
 Each round's metadata and safe `llm_round` telemetry can include:
 
 | Field | Meaning |

src/viano_bug_analysis/llm/config.py

@@ -39,6 +39,8 @@ class ProviderConfig:
     context_cache: bool = False
     cache_ttl_seconds: int = 300
     cache_min_prefix_chars: int = 16000
+    cache_required: bool = False
+    cache_required_prompt_prefix: str | None = None
 
     def __post_init__(self):
         if not isinstance(self.id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", self.id):
@@ -68,7 +70,7 @@ class ProviderConfig:
             raise LLMError("LLM_CONFIG")
         if not isinstance(self.adapter, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", self.adapter):
             raise LLMError("LLM_CONFIG")
-        for value in (self.enabled, self.allow_insecure_http, self.context_cache):
+        for value in (self.enabled, self.allow_insecure_http, self.context_cache, self.cache_required):
             if not isinstance(value, bool):
                 raise LLMError("LLM_CONFIG")
         for effort in (self.reasoning_effort, self.entity_resolution_reasoning_effort):
@@ -80,6 +82,12 @@ class ProviderConfig:
         positive(self.cache_min_prefix_chars, integer=True)
         if self.context_cache and self.adapter != 'gemini':
             raise LLMError('LLM_CONFIG')
+        if self.cache_required and not self.context_cache:
+            raise LLMError('LLM_CONFIG')
+        if self.cache_required_prompt_prefix is not None and (not self.cache_required
+                or not isinstance(self.cache_required_prompt_prefix, str)
+                or not re.fullmatch(r'[A-Za-z0-9_-]{1,128}', self.cache_required_prompt_prefix)):
+            raise LLMError('LLM_CONFIG')
         if self.adapter == 'gemini' and (not re.fullmatch(r'[A-Za-z0-9._-]+', self.model)
                 or self.reasoning_effort == 'none' or self.entity_resolution_reasoning_effort == 'none'):
             raise LLMError('LLM_CONFIG')
@@ -134,6 +142,7 @@ class GatewayConfig:
             for name, default in {'cached_input_cost_per_million': None,
                                   'cache_storage_cost_per_million_hour': None,
                                   'context_cache': False, 'cache_ttl_seconds': 300,
+                                  'cache_required': False, 'cache_required_prompt_prefix': None,
                                   'cache_min_prefix_chars': 16000}.items():
                 if provider[name] == default:
                     del provider[name]

src/viano_bug_analysis/llm/gateway.py

@@ -88,6 +88,11 @@ class LLMGateway:
                 for tokens in (response.input_tokens, response.output_tokens, response.cached_input_tokens, response.reasoning_tokens):
                     if tokens is not None and (isinstance(tokens, bool) or not isinstance(tokens, int) or not 0 <= tokens <= 10**12):
                         raise LLMError("LLM_RESPONSE")
+                if provider.cache_required and (provider.cache_required_prompt_prefix is None
+                        or request['prompt_version'].startswith(provider.cache_required_prompt_prefix)):
+                    if (response.cached_input_tokens is None or response.cached_input_tokens <= 0
+                            or response.input_tokens is None or response.cached_input_tokens > response.input_tokens):
+                        raise LLMError('LLM_CACHE_USAGE_UNCONFIRMED')
                 encoded = json_bytes(response.document)
                 if len(encoded) > self.config.max_response_bytes:
                     raise LLMError("LLM_RESPONSE_LIMIT")
@@ -107,6 +112,12 @@ class LLMGateway:
                 LOGGER.info("LLM request succeeded provider=%s", provider.id)
                 return {"document": parse_manifest(encoded), "metadata": metadata}
             except (ProviderUnavailable, TimeoutError) as error:
+                if provider.cache_required and (provider.cache_required_prompt_prefix is None
+                        or request['prompt_version'].startswith(provider.cache_required_prompt_prefix)):
+                    required_error = LLMError('LLM_CACHE_REQUIRED',
+                        http_status=getattr(error, 'http_status', None), diagnostics=getattr(error, 'diagnostics', None))
+                    self._failure_metadata(required_error, provider, request, started, response, attempts)
+                    raise required_error from None
                 code = error.code if isinstance(error, ProviderUnavailable) else "LLM_TIMEOUT"
                 attempts.append({"provider": provider.id, "code": code})
                 http_status = getattr(error, "http_status", None)

src/viano_bug_analysis/llm/gemini.py

@@ -84,6 +84,8 @@ class GeminiProvider:
         return {'parts': system}, [prefix], contents
 
     def _generate(self, config, request, limit):
+        required = config.cache_required and (config.cache_required_prompt_prefix is None
+                    or request['prompt_version'].startswith(config.cache_required_prompt_prefix))
         secret = os.environ.get(config.api_key_env) if config.api_key_env else None
         if config.api_key_env and (not secret or any(ord(c) < 32 or ord(c) == 127 for c in secret)):
             raise LLMError('LLM_CREDENTIALS')
@@ -104,6 +106,7 @@ class GeminiProvider:
         if system['parts']:
             full['systemInstruction'] = system
         status, created, name = 'disabled', 0, None
+        cache_http_status = None
         fingerprint = hashlib.sha256(json_bytes([base, config.id, config.model, secret,
                                                 system, prefix, config.cache_ttl_seconds])).hexdigest()
         if config.context_cache:
@@ -127,15 +130,17 @@ class GeminiProvider:
                             # remote storage; do not report a fabricated zero.
                             created = None
                             cached = self._post(base + '/cachedContents', cache_payload, headers, limit,
-                                                min(deadline, now + min(5, config.timeout_seconds / 4)))
+                                                min(deadline, now + (config.timeout_seconds / 2 if required
+                                                    else min(5, config.timeout_seconds / 4))))
                             if not isinstance(cached, dict) or not isinstance(cached.get('name'), str) or not re.fullmatch(
                                     r'cachedContents/[A-Za-z0-9_-]{1,256}', cached['name']):
                                 raise LLMError('LLM_RESPONSE')
                             name = cached['name']
                             created = count((cached.get('usageMetadata') or {}).get('totalTokenCount'))
                             status = 'created'
-                        except (LLMError, TypeError, AttributeError):
+                        except (LLMError, TypeError, AttributeError) as error:
                             name = None
+                            cache_http_status = getattr(error, 'http_status', None)
                         # Negative caching prevents repeated failed creation requests.
                         self._cache[fingerprint] = (name, now + (config.cache_ttl_seconds if name else 60))
                     else:
@@ -144,9 +149,13 @@ class GeminiProvider:
         diagnostics = {'cache_status': status, 'cache_created_tokens': created,
                        'cache_ttl_seconds': config.cache_ttl_seconds, 'max_output_tokens': config.max_output_tokens}
         try:
+            if required and not name:
+                raise LLMError('LLM_CACHE_REQUIRED', http_status=cache_http_status)
             try:
                 envelope = self._post(base + '/models/' + config.model + ':generateContent', payload, headers, limit, deadline)
             except LLMError as error:
+                if required:
+                    raise LLMError('LLM_CACHE_REQUIRED', http_status=error.http_status) from None
                 if not name or error.http_status not in {400, 403, 404}:
                     raise
                 with self._lock:
@@ -161,6 +170,8 @@ class GeminiProvider:
             output = candidates + (thoughts or 0) if candidates is not None else None
             diagnostics.update(input_tokens=prompt, cached_input_tokens=cached,
                                output_tokens=output, reasoning_tokens=thoughts)
+            if required and (cached is None or cached <= 0 or prompt is None or cached > prompt):
+                raise LLMError('LLM_CACHE_USAGE_UNCONFIRMED')
             candidate = envelope['candidates'][0]
             diagnostics['finish_reason'] = candidate.get('finishReason')
             if candidate.get('finishReason') != 'STOP':

tests/test_llm_caching.py

@@ -313,3 +313,68 @@ def test_invalid_cache_handles_never_used(monkeypatch, invalid):
     result = asyncio.run(adapter.generate(config(),request(),max_response_bytes=10000))
     assert result.diagnostics['cache_status'] == 'fallback'
     assert 'cachedContent' not in seen[-1][1]
+
+
+def test_required_cache_accepts_confirmed_creation_and_reuse(monkeypatch):
+    adapter, cfg = GeminiProvider(), replace(config(), cache_required=True)
+    seen = scripted(adapter,monkeypatch,[CACHE,native(),native()])
+    for _ in range(2):
+        result=asyncio.run(adapter.generate(cfg,request(),max_response_bytes=10000))
+        assert result.cached_input_tokens == 80
+    assert len(seen)==3 and all('cachedContent' in payload for _,payload,_ in seen[1:])
+
+
+@pytest.mark.parametrize('mode', ['small','rejected','stale','missing_usage','zero_usage','inconsistent_usage'])
+def test_required_cache_never_silently_falls_back(monkeypatch,mode):
+    adapter, cfg = GeminiProvider(), replace(config(),cache_required=True)
+    actions=[CACHE,native()]
+    expected='LLM_CACHE_REQUIRED'
+    if mode=='small':
+        cfg=replace(cfg,cache_min_prefix_chars=1000000)
+        actions=[]
+    elif mode=='rejected': actions=[LLMError('LLM_HTTP_CONFIGURATION',http_status=400)]
+    elif mode=='stale': actions=[CACHE,LLMError('LLM_HTTP_CONFIGURATION',http_status=404)]
+    else:
+        expected='LLM_CACHE_USAGE_UNCONFIRMED'
+        actions[1]=native(None if mode=='missing_usage' else 0 if mode=='zero_usage' else 101)
+    seen=scripted(adapter,monkeypatch,actions)
+    with pytest.raises(LLMError) as caught:
+        asyncio.run(adapter.generate(cfg,request(),max_response_bytes=10000))
+    assert caught.value.code==expected
+    assert all('cachedContent' in payload for url,payload,_ in seen if url.endswith(':generateContent'))
+
+
+def test_required_prefix_keeps_other_phases_optional(monkeypatch):
+    adapter=GeminiProvider()
+    cfg=replace(config(),cache_required=True,cache_required_prompt_prefix='investigation_project_',cache_min_prefix_chars=1000000)
+    seen=scripted(adapter,monkeypatch,[native(0)])
+    assert asyncio.run(adapter.generate(cfg,request(),max_response_bytes=10000)).document=={'answer':'ok'}
+    req=request(); req['prompt_version']='investigation_project_v10'
+    with pytest.raises(LLMError,match='LLM_CACHE_REQUIRED'):
+        asyncio.run(adapter.generate(cfg,req,max_response_bytes=10000))
+    assert len(seen)==1
+
+
+@pytest.mark.parametrize('failure',[ProviderUnavailable('LLM_CONNECTION'),TimeoutError()])
+def test_required_cache_cannot_fail_over_to_uncached_provider(failure):
+    cfg=replace(config(),cache_required=True)
+    fallback=ProviderConfig('fallback','m','http://localhost',adapter='fixture')
+    adapter=Adapter({cfg.id:failure,'fallback':ProviderResponse({'answer':'ok'},'m')})
+    gateway=LLMGateway(GatewayConfig((cfg,fallback)),{'gemini':adapter,'fixture':adapter})
+    with pytest.raises(LLMError,match='LLM_CACHE_REQUIRED'):
+        asyncio.run(gateway.complete_round(request()))
+    assert adapter.calls==[cfg.id]
+
+
+@pytest.mark.parametrize('changes',[{'context_cache':False},{'cache_required':'true'},
+    {'cache_required_prompt_prefix':''},{'cache_required':False,'cache_required_prompt_prefix':'investigation_'}])
+def test_required_cache_config_is_strict(changes):
+    with pytest.raises(LLMError):
+        replace(config(),**({'cache_required':True}|changes))
+
+
+def test_gateway_itself_requires_confirmed_cached_usage():
+    cfg=replace(config(),cache_required=True)
+    adapter=Adapter({cfg.id:ProviderResponse({'answer':'ok'},'m',100,10)})
+    with pytest.raises(LLMError,match='LLM_CACHE_USAGE_UNCONFIRMED'):
+        asyncio.run(LLMGateway(GatewayConfig((cfg,)),{'gemini':adapter}).complete_round(request()))