Add optional Gemini context caching and cache usage accounting

This commit is contained in:
Nico committed 2026-10-16 14:57:33 +02:00
1 parent 61138955ec
commit 4142e2dc2b
16 files changed
+965 -20

No files matched your search

+4
View File
@@ -42,6 +42,10 @@ Expected-behavior resolution and the read-only provider contract: [expected-beha
Configurable LLM transport, failover and durable retries: [llm-gateway-v1.md](docs/llm-gateway-v1.md), with [example configuration](config/llm.example.toml).
Optional Gemini instruction/schema caching, implicit cache measurements and
separate token/storage cost accounting: [context caching](docs/llm-context-caching.md),
with [native Gemini example](config/llm.gemini-cache.example.toml).
The first domain LLM phase and its source-reference contracts: [case-interpretation-v1.md](docs/case-interpretation-v1.md).
Initial hypotheses, revision-wide creation limits and equal budget allocations: [hypothesis-builder-v1.md](docs/hypothesis-builder-v1.md), with [analysis configuration](config/analysis.example.toml).
+35
View File
@@ -0,0 +1,35 @@
# Optional separate configuration. Copy to an ignored *.local.toml file.
# This does not replace the existing OpenAI-compatible configuration.
[llm]
privacy_profile = "cloud_allowed"
max_attempts = 10
retry_delay_seconds = 60
blocked_retry_seconds = 3600
[[llm.providers]]
id = "gemini-native"
adapter = "gemini"
privacy = "cloud"
base_url = "https://generativelanguage.googleapis.com/v1beta"
model = "gemini-3.8-flash"
api_key_env = "GEMINI_API_KEY"
timeout_seconds = 300
response_format = "json_object"
max_output_tokens = 16384
reasoning_effort = "medium"
entity_resolution_reasoning_effort = "low"
# Opt-in. Instruction/schema prefix only; live evidence is always complete.
context_cache = true
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
# Optional estimates in USD per million tokens; verify your model/tariff/date.
# These illustrative standard-tier rates are from 2026-10-16, valid through
# 2026-12-31 for the model above, not universal/default prices.
input_cost_per_million = 0.75
cached_input_cost_per_million = 0.075
output_cost_per_million = 3.75
cache_storage_cost_per_million_hour = 0.50
+131
View File
@@ -0,0 +1,131 @@
# Optional context caching and usage accounting
The package remains the complete, authoritative record. Saved requests, evidence,
responses, checkpoints and provider configuration references remain portable.
No cache handle, credential or provider conversation ID becomes required case
state. `docs/architecture.md` and the investigation full-history policy are
unchanged. All evidence remains in every investigation input.
## First implementation
The existing `openai_compatible` adapter now reads
`usage.prompt_tokens_details.cached_tokens` and
`usage.completion_tokens_details.reasoning_tokens`. This exposes implicit cache
hits when the provider actually reports them, without changing existing requests
or enabling explicit caching. An absent count means unknown, not zero.
The optional `gemini` adapter uses the native GenerateContent API. It moves the
unchanged system instructions and authoritative output schema to the beginning
of the wire request. Only an exact top-level schema duplicate is removed from
the live user JSON; no case facts, identifiers, evidence or assessments are
removed. The archived request is not changed. This is deterministic transport
preparation, not another LLM summary.
With `context_cache = true`, this instruction/schema prefix can be stored as a
Gemini `cachedContents` resource. Case-specific input and all research history
are still sent in full. The prefix is independent of the currently selected
hypothesis. Reuse requires an exact match of endpoint, provider ID, model,
credential fingerprint, prefix and TTL. The adapter retains at most eight
entries (including short-lived failed-creation entries), in process memory only.
Different phases have different instructions/schema and therefore different
cache entries. Final diagnosis remains an independent request, without a prior
investigation conversation.
This intentionally does **not** implement a provider conversation/session or
cache the growing evidence history. It establishes measured caching without
changing what the model knows. A later session adapter must preserve full replay,
provider failover and an independent final evaluation; its quality and total
cost need a separate comparison. Caching itself promises no reasoning-quality
improvement.
## Configuration and fallback
See `config/llm.gemini-cache.example.toml`. Existing local configuration files
are not edited. The native base URL ends in `/v1beta`, **not** `/v1beta/openai`.
`context_cache` defaults to false and is allowed only for the native adapter.
`cached_input_cost_per_million` can also be set for OpenAI-compatible providers
that report cache usage. Prices remain optional, model-specific configuration.
New default fields are omitted from configuration snapshots to preserve frozen
operation hashes and replay of old cases.
`cache_min_prefix_chars` is an eligibility heuristic, not a token count or API
minimum. A short prefix may therefore obtain no explicit cache saving. The
provider remains authoritative about minimum tokens and model support. There
is no padding to meet a minimum. TTL defaults to five minutes; expiry is not
extended automatically. Restarting a worker rebuilds a cache from complete
requests rather than depending on a remote handle.
Cache creation has a small bounded part of the overall request deadline (at
most five seconds and one quarter of the provider timeout). Rejection, timeout,
unsupported caching or an invalid cache-create response falls back to the full
request. Failed creation is suppressed for sixty seconds. A cached generation
rejected with HTTP 400, 403 or 404 invalidates the handle and retries once with
the full request, within the same overall deadline. Other failures use normal
provider failover/retry. No unbounded transport retry loop is introduced.
Successful allocations expire remotely by TTL even if the worker exits. Keys
are sent in headers; redirects and ambient proxies are disabled. Native
`reasoning_effort` uses Gemini `thinkingLevel` (model support must be checked);
`none` is rejected rather than silently mapped to another setting.
## Reading the measurements
Each round's metadata and safe `llm_round` telemetry can include:
| Field | Meaning |
| --- | --- |
| `input_tokens` | Total effective input, including cached input |
| `cached_input_tokens` | Provider-reported cached part |
| `uncached_input_tokens` | Total minus cached part, only if both are known and consistent |
| `output_tokens` | Billed generated tokens; native Gemini candidates plus thoughts |
| `reasoning_tokens` | Reported thinking subset, not added to output cost a second time |
| `estimated_input_cost` | Uncached input estimate |
| `estimated_cached_input_cost` | Cached input estimate |
| `estimated_output_cost` | Output estimate, including thinking |
| `estimated_cost` | Generation estimate; excludes separately reported cache storage |
| `cache_status` | `disabled`, `below_minimum`, `created`, `reused`, or `fallback` |
| `cache_created_tokens` | Newly allocated tokens, recorded once per allocation |
| `estimated_cache_storage_cost` | Newly allocated tokens × configured TTL hours × storage rate / 1M |
Token counters are available as `vba_llm_tokens_total{direction="..."}`; the
directions overlap (total input includes cached input, output includes thinking).
Do not sum all directions. Cost components use
`vba_llm_cost_component_total{component="...",currency="..."}`. The storage
component is distinct from generation costs. Unknown values remain null and
usage-unknown counters identify incomplete accounting.
Storage is an estimate for the **whole reserved TTL**, counted on creation only,
not on every hit. It is not a provider invoice. A cache created before a
temporary generation failure is recorded in the failed provider attempt's
`cache_usage`; ordinary unsuccessful generation attempts can still have unknown
usage. A crash between remote allocation and local persistence can also leave
unrecorded charges. Never treat these estimates as a precise account balance.
If cache creation loses its response, newly allocated tokens remain unknown;
they are not reported as zero.
For old configurations with no cache price and no reported cached count, the
legacy full-input-price estimate remains available. With a known positive cache
count and no cache rate, or with a configured cache rate and missing cache usage,
the exact generation estimate is unknown. No fake discount is inferred.
For an economic comparison, use measured cached/uncached input and output
estimates plus newly allocated storage, latency and outcome quality. There is
no program-side Euro abort cap. The existing local configuration remains usable
for collecting implicit-hit measurements before choosing the native adapter.
No paid call is needed to run the offline protocol/accounting regression tests.
An offline inspection of the archived BUG-000017 selected-context trial found an
instruction/schema prefix of 18,703 characters for `investigation_project_v9`,
with 59,900 characters remaining live in the inspected request. That prefix
passes the default character heuristic; only a real provider response can
confirm token eligibility, hits and savings. The smaller entity-resolution
prefixes (3,730 / 4,340 characters) do not pass it. No live API call was made
for this inspection; it is not a new quality or cost benchmark.
## API sources checked 2026-10-16
- [Gemini caching guide](https://ai.google.dev/gemini-api/docs/generate-content/caching)
- [CachedContents reference](https://ai.google.dev/api/caching)
- [GenerateContent usage metadata](https://ai.google.dev/api/generate-content#UsageMetadata)
- [Thinking configuration](https://ai.google.dev/gemini-api/docs/generate-content/thinking)
- [Model-specific pricing](https://ai.google.dev/gemini-api/docs/pricing)
- [Interactions session API, for a separate future adapter](https://ai.google.dev/gemini-api/docs/interactions-overview)
+22 -2
View File
@@ -34,6 +34,11 @@ class ProviderConfig:
cost_currency: str = "USD"
reasoning_effort: str | None = None
entity_resolution_reasoning_effort: str | None = None
cached_input_cost_per_million: float | None = None
cache_storage_cost_per_million_hour: float | None = None
context_cache: bool = False
cache_ttl_seconds: int = 300
cache_min_prefix_chars: int = 16000
def __post_init__(self):
if not isinstance(self.id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", self.id):
@@ -63,7 +68,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):
for value in (self.enabled, self.allow_insecure_http, self.context_cache):
if not isinstance(value, bool):
raise LLMError("LLM_CONFIG")
for effort in (self.reasoning_effort, self.entity_resolution_reasoning_effort):
@@ -71,9 +76,17 @@ class ProviderConfig:
raise LLMError("LLM_CONFIG")
positive(self.timeout_seconds)
positive(self.max_output_tokens, integer=True)
positive(self.cache_ttl_seconds, integer=True)
positive(self.cache_min_prefix_chars, integer=True)
if self.context_cache and self.adapter != 'gemini':
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')
if not isinstance(self.cost_currency, str) or not re.fullmatch(r"[A-Z]{3}", self.cost_currency):
raise LLMError("LLM_CONFIG")
for cost in (self.input_cost_per_million, self.output_cost_per_million):
for cost in (self.input_cost_per_million, self.output_cost_per_million,
self.cached_input_cost_per_million, self.cache_storage_cost_per_million_hour):
if cost is not None and (isinstance(cost, bool) or not isinstance(cost, (int, float)) or not math.isfinite(cost) or cost < 0):
raise LLMError("LLM_CONFIG")
@@ -117,6 +130,13 @@ class GatewayConfig:
# Only configuration references; never resolve credentials for persistence.
result = asdict(self)
for provider in result["providers"]:
# Omit new defaults so frozen operations keep their configuration hash.
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_min_prefix_chars': 16000}.items():
if provider[name] == default:
del provider[name]
for name in ("reasoning_effort", "entity_resolution_reasoning_effort"):
if provider[name] is None:
del provider[name]
+7 -1
View File
@@ -1,6 +1,7 @@
"""Allowlisted failure details: never response text, headers or provider errors."""
REASONS = {'stop','length','content_filter','tool_calls','function_call','MAX_TOKENS','SAFETY','STOP','OTHER','unknown','missing'}
COUNTS = {'input_tokens','output_tokens','reasoning_tokens','response_bytes','output_characters','max_output_tokens'}
COUNTS = {'input_tokens','output_tokens','reasoning_tokens','cached_input_tokens',
'cache_created_tokens','cache_ttl_seconds','response_bytes','output_characters','max_output_tokens'}
KEYWORDS = {'type','required','additionalProperties','enum','const','pattern','minItems','maxItems','minimum','maximum','anyOf','oneOf'}
@@ -12,6 +13,8 @@ def sanitize(data):
if not isinstance(data,dict):
return {}
result = {k:count(v) for k,v in data.items() if k in COUNTS}
if isinstance(data.get('cache_status'), str) and data['cache_status'] in {'disabled','below_minimum','created','reused','fallback'}:
result['cache_status'] = data['cache_status']
if 'finish_reason' in data:
value = data['finish_reason']
result['finish_reason'] = value if isinstance(value,str) and value in REASONS else ('missing' if value is None else 'unknown')
@@ -29,6 +32,9 @@ def envelope_details(envelope, max_output_tokens, response_bytes):
usage = envelope.get('usage')
if isinstance(usage,dict):
result.update(input_tokens=usage.get('prompt_tokens'),output_tokens=usage.get('completion_tokens'))
prompt_details = usage.get('prompt_tokens_details')
if isinstance(prompt_details,dict):
result['cached_input_tokens'] = prompt_details.get('cached_tokens')
details = usage.get('completion_tokens_details')
if isinstance(details,dict):
result['reasoning_tokens'] = details.get('reasoning_tokens')
+21 -14
View File
@@ -2,7 +2,6 @@ import asyncio
import copy
import logging
import time
import math
import re
from dataclasses import replace
@@ -16,6 +15,8 @@ from .errors import GatewayUnavailable, LLMError, ProviderUnavailable
from .openai_compatible import OpenAICompatibleProvider
from .provider_base import ProviderResponse
from .diagnostics import sanitize, count
from .gemini import GeminiProvider
from .usage import usage_metadata
LOGGER = logging.getLogger("viano_bug_analysis.llm")
PRIVACY = {"local_only": {"local"}, "self_hosted": {"local", "self_hosted"}, "cloud_allowed": {"local", "self_hosted", "cloud"}}
@@ -36,7 +37,7 @@ def _schema_check(schema):
class LLMGateway:
def __init__(self, config, adapters=None):
self.config = config
self.adapters = {"openai_compatible": OpenAICompatibleProvider()} | (adapters or {})
self.adapters = {"openai_compatible": OpenAICompatibleProvider(), "gemini": GeminiProvider()} | (adapters or {})
if any(p.enabled and p.adapter not in self.adapters for p in config.providers):
raise LLMError("LLM_ADAPTER")
@@ -84,8 +85,8 @@ class LLMGateway:
response = await asyncio.wait_for(self.adapters[provider.adapter].generate(provider, copy.deepcopy(request), max_response_bytes=self.config.max_response_bytes), timeout=provider.timeout_seconds)
if not isinstance(response.model, str) or not response.model.strip() or len(response.model) > 256 or any(ord(c) < 32 for c in response.model):
raise LLMError("LLM_RESPONSE")
for count in (response.input_tokens, response.output_tokens):
if count is not None and (isinstance(count, bool) or not isinstance(count, int) or not 0 <= count <= 10**12):
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")
encoded = json_bytes(response.document)
if len(encoded) > self.config.max_response_bytes:
@@ -93,13 +94,16 @@ class LLMGateway:
if not isinstance(response.document, dict):
raise LLMError("LLM_OUTPUT_SCHEMA")
Draft202012Validator(request["output_schema"], format_checker=FORMAT_CHECKER).validate(response.document)
cost = None
if all(value is not None for value in (provider.input_cost_per_million, provider.output_cost_per_million, response.input_tokens, response.output_tokens)):
cost = (response.input_tokens * provider.input_cost_per_million + response.output_tokens * provider.output_cost_per_million) / 1_000_000
usage = usage_metadata(provider, response)
cost = usage['estimated_cost']
metadata = {"provider": provider.id, "model": response.model, "prompt_version": request["prompt_version"],
"input_tokens": response.input_tokens, "output_tokens": response.output_tokens,
"estimated_cost": cost, "elapsed_seconds": round(time.monotonic() - started, 6), "attempts": attempts}
metadata["cost_currency"] = provider.cost_currency if cost is not None else None
metadata.update(usage)
if any(usage[k] is not None for k in ('estimated_input_cost', 'estimated_cached_input_cost',
'estimated_output_cost', 'estimated_cache_storage_cost')):
metadata['cost_currency'] = provider.cost_currency
LOGGER.info("LLM request succeeded provider=%s", provider.id)
return {"document": parse_manifest(encoded), "metadata": metadata}
except (ProviderUnavailable, TimeoutError) as error:
@@ -108,6 +112,11 @@ class LLMGateway:
http_status = getattr(error, "http_status", None)
if type(http_status) is int and 100 <= http_status <= 599:
attempts[-1]["http_status"] = http_status
cache_details = getattr(error, 'diagnostics', {})
if 'cache_created_tokens' in cache_details and cache_details['cache_created_tokens'] != 0:
# Allocation can have succeeded even if generation was unavailable.
attempts[-1]['cache_usage'] = usage_metadata(provider, diagnostics=error.diagnostics)
attempts[-1]['cache_usage']['cost_currency'] = provider.cost_currency
LOGGER.warning("LLM endpoint unavailable provider=%s code=%s", provider.id, code)
except ValidationError as invalid:
error = LLMError("LLM_OUTPUT_SCHEMA",diagnostics={'validation_keyword':invalid.validator})
@@ -131,12 +140,10 @@ class LLMGateway:
details['max_output_tokens'] = count(provider.max_output_tokens)
error.diagnostics = details
input_tokens, output_tokens = details.get('input_tokens'),details.get('output_tokens')
cost = None
if all(v is not None for v in (input_tokens,output_tokens,provider.input_cost_per_million,provider.output_cost_per_million)):
value = (input_tokens*provider.input_cost_per_million+output_tokens*provider.output_cost_per_million)/1_000_000
if math.isfinite(value):
cost = value
error.metadata = {'provider':provider.id,'model':provider.model,'prompt_version':request['prompt_version'],
'input_tokens':input_tokens,'output_tokens':output_tokens,'estimated_cost':cost,
'cost_currency':provider.cost_currency if cost is not None else None,
'input_tokens':input_tokens,'output_tokens':output_tokens,'cost_currency':None,
'elapsed_seconds':round(time.monotonic()-started,6),'attempts':copy.deepcopy(attempts)}
error.metadata.update(usage_metadata(provider, response, details))
if any(error.metadata[k] is not None for k in ('estimated_cost','estimated_input_cost',
'estimated_cached_input_cost','estimated_output_cost','estimated_cache_storage_cost')):
error.metadata['cost_currency'] = provider.cost_currency
+179
View File
@@ -0,0 +1,179 @@
"""Optional native Gemini transport with disposable instruction/schema caching.
No conversation state or research evidence is stored exclusively at the provider.
Cache handles and credential fingerprints never enter the portable case.
"""
import asyncio
import hashlib
import http.client
import os
import re
import threading
import time
import urllib.error
import urllib.request
from viano_bug_analysis.lifecycle.storage import json_bytes
from viano_bug_analysis.package.manifest import parse_manifest
from .errors import LLMError, ProviderUnavailable
from .openai_compatible import _NoRedirect
from .provider_base import ProviderResponse
from .diagnostics import count, sanitize
class GeminiProvider:
def __init__(self):
self._cache = {}
self._lock = threading.Lock()
async def generate(self, config, request, *, max_response_bytes):
return await asyncio.to_thread(self._generate, config, request, max_response_bytes)
@staticmethod
def _post(url, payload, headers, limit, deadline):
remaining = deadline - time.monotonic()
if remaining <= 0:
raise ProviderUnavailable('LLM_TIMEOUT')
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect())
try:
req = urllib.request.Request(url, data=json_bytes(payload), headers=headers)
with opener.open(req, timeout=remaining) as response:
chunks, size = [], 0
while True:
if time.monotonic() >= deadline:
raise ProviderUnavailable('LLM_TIMEOUT')
chunk = response.read(min(65536, limit + 1 - size))
if not chunk:
break
size += len(chunk)
if size > limit:
raise LLMError('LLM_RESPONSE_LIMIT')
chunks.append(chunk)
return parse_manifest(b''.join(chunks))
except urllib.error.HTTPError as error:
error.close()
if error.code in {408, 429} or 500 <= error.code < 600:
raise ProviderUnavailable('LLM_HTTP_TEMPORARY', http_status=error.code) from None
raise LLMError('LLM_HTTP_CONFIGURATION', http_status=error.code) from None
except (urllib.error.URLError, TimeoutError, OSError, http.client.HTTPException):
raise ProviderUnavailable('LLM_CONNECTION') from None
except (ValueError, TypeError, RecursionError):
raise LLMError('LLM_RESPONSE') from None
@staticmethod
def _messages(request):
system, contents = [], []
for message in request['messages']:
if message['role'] == 'system':
system.append({'text': message['content']})
continue
text = message['content']
# Move only an exact duplicate of the authoritative schema. Everything
# else, including every research result, stays in the live input.
try:
value = parse_manifest(text.encode('utf-8'))
if isinstance(value, dict) and value.get('output_schema') == request['output_schema']:
value.pop('output_schema')
text = json_bytes(value).decode('utf-8')
except (ValueError, TypeError, RecursionError):
pass
contents.append({'role': 'model' if message['role'] == 'assistant' else 'user',
'parts': [{'text': text}]})
prefix = {'role': 'user', 'parts': [{'text': json_bytes({
'output_schema': request['output_schema']}).decode('utf-8')}]}
return {'parts': system}, [prefix], contents
def _generate(self, config, request, limit):
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')
headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
if secret:
headers['x-goog-api-key'] = secret
base = config.base_url.rstrip('/')
deadline = time.monotonic() + config.timeout_seconds
system, prefix, contents = self._messages(request)
generation = {'maxOutputTokens': config.max_output_tokens}
if config.reasoning_effort is not None:
generation['thinkingConfig'] = {'thinkingLevel': config.reasoning_effort}
if config.response_format != 'none':
generation['responseMimeType'] = 'application/json'
if config.response_format == 'json_schema':
generation['responseJsonSchema'] = request['output_schema']
full = {'contents': prefix + contents, 'generationConfig': generation}
if system['parts']:
full['systemInstruction'] = system
status, created, name = 'disabled', 0, None
fingerprint = hashlib.sha256(json_bytes([base, config.id, config.model, secret,
system, prefix, config.cache_ttl_seconds])).hexdigest()
if config.context_cache:
status = 'below_minimum'
if len(json_bytes([system, prefix]).decode('utf-8')) >= config.cache_min_prefix_chars:
with self._lock:
now = time.monotonic()
self._cache = {k: v for k, v in self._cache.items() if v[1] > now}
entry = self._cache.get(fingerprint)
if entry:
name, _ = entry
status = 'reused' if name else 'fallback'
elif len(self._cache) < 8:
status = 'fallback'
try:
cache_payload = {'model': 'models/' + config.model, 'contents': prefix,
'ttl': str(config.cache_ttl_seconds) + 's'}
if system['parts']:
cache_payload['systemInstruction'] = system
# A lost creation response may still have allocated
# 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)))
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):
name = None
# Negative caching prevents repeated failed creation requests.
self._cache[fingerprint] = (name, now + (config.cache_ttl_seconds if name else 60))
else:
status = 'fallback'
payload = {'contents': contents, 'generationConfig': generation, 'cachedContent': name} if name else full
diagnostics = {'cache_status': status, 'cache_created_tokens': created,
'cache_ttl_seconds': config.cache_ttl_seconds, 'max_output_tokens': config.max_output_tokens}
try:
try:
envelope = self._post(base + '/models/' + config.model + ':generateContent', payload, headers, limit, deadline)
except LLMError as error:
if not name or error.http_status not in {400, 403, 404}:
raise
with self._lock:
self._cache[fingerprint] = (None, time.monotonic() + 60)
diagnostics['cache_status'] = 'fallback'
envelope = self._post(base + '/models/' + config.model + ':generateContent', full, headers, limit, deadline)
usage = envelope.get('usageMetadata') or {}
prompt = count(usage.get('promptTokenCount'))
cached = count(usage.get('cachedContentTokenCount'))
thoughts = count(usage.get('thoughtsTokenCount'))
candidates = count(usage.get('candidatesTokenCount'))
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)
candidate = envelope['candidates'][0]
diagnostics['finish_reason'] = candidate.get('finishReason')
if candidate.get('finishReason') != 'STOP':
raise LLMError('LLM_INCOMPLETE_OUTPUT')
parts = candidate['content']['parts']
if any('functionCall' in part for part in parts):
raise LLMError('LLM_INCOMPLETE_OUTPUT')
text = ''.join(part['text'] for part in parts if 'text' in part and not part.get('thought'))
document = parse_manifest(text.encode('utf-8'))
return ProviderResponse(document, envelope.get('modelVersion', config.model), prompt, output,
sanitize(diagnostics), cached, thoughts)
except LLMError as error:
error.diagnostics = sanitize(diagnostics) | sanitize(error.diagnostics)
raise
except (ValueError, TypeError, KeyError, IndexError, AttributeError, RecursionError):
raise LLMError('LLM_RESPONSE', diagnostics=sanitize(diagnostics)) from None
@@ -85,6 +85,7 @@ class OpenAICompatibleProvider:
for count in (input_tokens, output_tokens):
if count is not None and (isinstance(count, bool) or not isinstance(count, int) or count < 0):
raise LLMError("LLM_RESPONSE",diagnostics=diagnostics)
return ProviderResponse(document, model, input_tokens, output_tokens, diagnostics)
return ProviderResponse(document, model, input_tokens, output_tokens, diagnostics,
diagnostics.get('cached_input_tokens'), diagnostics.get('reasoning_tokens'))
except (ValueError, TypeError, KeyError, IndexError, AttributeError, RecursionError):
raise LLMError("LLM_RESPONSE",diagnostics=diagnostics) from None
+6 -1
View File
@@ -11,9 +11,14 @@ class ProviderResponse:
input_tokens: int | None = None
output_tokens: int | None = None
diagnostics: dict | None = None
cached_input_tokens: int | None = None
reasoning_tokens: int | None = None
class LLMProvider(Protocol):
async def generate(self, config: ProviderConfig, request: dict, *, max_response_bytes: int) -> ProviderResponse:
"""One request only; adapters do not retry or change lifecycle state."""
"""One generation attempt; optional cache setup/fallback stays transport-only.
Adapters never retry model answers or change case lifecycle state.
"""
...
+49
View File
@@ -0,0 +1,49 @@
"""Provider-neutral usage accounting; missing cache counts are never zero hits."""
import math
from .diagnostics import count, sanitize
def usage_metadata(provider, response=None, diagnostics=None):
details = sanitize(diagnostics)
if response is not None:
details = sanitize(response.diagnostics) | details
details.update(input_tokens=count(response.input_tokens), output_tokens=count(response.output_tokens),
cached_input_tokens=count(response.cached_input_tokens),
reasoning_tokens=count(response.reasoning_tokens))
total, output = details.get('input_tokens'), details.get('output_tokens')
cached = details.get('cached_input_tokens')
# Inconsistent optional provider accounting must not poison the case.
if cached is not None and (total is None or cached > total):
cached = None
uncached = total - cached if total is not None and cached is not None else None
def charge(tokens, rate):
if tokens == 0:
return 0.0
if tokens is None or rate is None:
return None
value = tokens * rate / 1_000_000
return value if math.isfinite(value) else None
input_cost = charge(uncached, provider.input_cost_per_million)
cached_cost = charge(cached, provider.cached_input_cost_per_million)
output_cost = charge(output, provider.output_cost_per_million)
estimated = None
if cached is None and provider.cached_input_cost_per_million is None:
# Legacy estimate at full input price; not a measured cache saving.
legacy = charge(total, provider.input_cost_per_million)
if legacy is not None and output_cost is not None:
estimated = legacy + output_cost
elif all(v is not None for v in (input_cost, cached_cost, output_cost)):
estimated = input_cost + cached_cost + output_cost
created = details.get('cache_created_tokens')
storage = charge(created, provider.cache_storage_cost_per_million_hour)
if storage is not None:
ttl = details.get('cache_ttl_seconds')
storage = storage * ttl / 3600 if ttl is not None else None
return {'cached_input_tokens': cached, 'uncached_input_tokens': uncached,
'reasoning_tokens': details.get('reasoning_tokens'),
'estimated_input_cost': input_cost, 'estimated_cached_input_cost': cached_cost,
'estimated_output_cost': output_cost, 'estimated_cost': estimated,
# Charged once per allocation, separately from generation estimates.
'cache_created_tokens': created, 'estimated_cache_storage_cost': storage,
'cache_status': details.get('cache_status')}
@@ -160,6 +160,72 @@
},
"additionalProperties": false
}
},
"cached_input_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"uncached_input_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"reasoning_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"cache_created_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"estimated_input_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"estimated_cached_input_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"estimated_output_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"estimated_cache_storage_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"cache_status": {
"enum": [
null,
"disabled",
"below_minimum",
"created",
"reused",
"fallback"
]
}
},
"required": [
@@ -193,6 +193,72 @@
},
"additionalProperties": false
}
},
"cached_input_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"uncached_input_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"reasoning_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"cache_created_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0
},
"estimated_input_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"estimated_cached_input_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"estimated_output_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"estimated_cache_storage_cost": {
"type": [
"number",
"null"
],
"minimum": 0
},
"cache_status": {
"enum": [
null,
"disabled",
"below_minimum",
"created",
"reused",
"fallback"
]
}
},
"required": [
@@ -194,6 +194,39 @@
"required",
"type"
]
},
"cached_input_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0,
"maximum": 1000000000000
},
"cache_created_tokens": {
"type": [
"integer",
"null"
],
"minimum": 0,
"maximum": 1000000000000
},
"cache_ttl_seconds": {
"type": [
"integer",
"null"
],
"minimum": 0,
"maximum": 1000000000000
},
"cache_status": {
"enum": [
"disabled",
"below_minimum",
"created",
"reused",
"fallback"
]
}
},
"additionalProperties": false
+19 -1
View File
@@ -350,7 +350,7 @@ class Telemetry:
phase, status = record["phase"], record["status"]
self._metric("vba_llm_rounds_total", phase=phase, status=status)
metadata = record["outcome"].get("metadata", {})
for field in ("input_tokens", "output_tokens"):
for field in ("input_tokens", "output_tokens", "cached_input_tokens", "uncached_input_tokens", "reasoning_tokens"):
value = metadata.get(field)
if type(value) is int and value >= 0:
self._metric("vba_llm_tokens_total", value, direction=field)
@@ -362,7 +362,21 @@ class Telemetry:
self._metric("vba_llm_estimated_cost_total", cost, currency=currency)
else:
self._metric("vba_llm_usage_unknown_total", field="estimated_cost")
for field in ('estimated_input_cost', 'estimated_cached_input_cost', 'estimated_output_cost',
'estimated_cache_storage_cost'):
value = metadata.get(field)
if type(value) in {int, float} and math.isfinite(value) and value >= 0 and currency_valid:
self._metric('vba_llm_cost_component_total', value, component=field, currency=currency)
attempts = metadata.get("attempts", record["outcome"].get("attempts", []))
failed_cache_storage = []
for attempt in attempts:
allocation = attempt.get('cache_usage', {})
value, unit = allocation.get('estimated_cache_storage_cost'), allocation.get('cost_currency')
if (type(value) in {int, float} and math.isfinite(value) and value >= 0
and isinstance(unit, str) and re.fullmatch('[A-Z]{3}', unit)):
self._metric('vba_llm_cost_component_total', value,
component='estimated_cache_storage_cost', currency=unit)
failed_cache_storage.append({'estimated_cost': value, 'currency': unit})
self._metric("vba_llm_provider_failures_total", len(attempts), phase=phase)
self._metric("vba_llm_failed_attempt_usage_unknown_total", len(attempts))
self.emit("llm_round", level="INFO" if status == "SUCCESS" else "WARNING", status=status, code=record.get("code"),
@@ -371,8 +385,12 @@ class Telemetry:
"diagnostics": record["outcome"].get("diagnostics"),
"elapsed_seconds": metadata.get("elapsed_seconds"),
"input_tokens": metadata.get("input_tokens"), "output_tokens": metadata.get("output_tokens"),
**{field: metadata.get(field) for field in ('cached_input_tokens','uncached_input_tokens',
'reasoning_tokens','cache_status','cache_created_tokens','estimated_input_cost',
'estimated_cached_input_cost','estimated_output_cost','estimated_cache_storage_cost')},
"estimated_cost": cost if currency_valid else None,
"currency": currency if currency_valid else None,
'failed_cache_storage': failed_cache_storage,
"failed_providers": [{"provider_key": identity(a.get("provider")), "code": safe_code(a.get("code")), "http_status": a.get("http_status")} for a in attempts]})
def refresh_inventory(self, *, force=False):
@@ -104,6 +104,16 @@ def detail(event, value):
"currency": value.get("currency") if isinstance(value.get("currency"), str) and re.fullmatch("[A-Z]{3}", value["currency"]) else None,
"failed_providers": [{"provider_key": hashed(a.get("provider_key")), "code": safe_code(a.get("code")), "http_status": http_status(a.get("http_status"))} for a in value.get("failed_providers", [])[:64] if isinstance(a, dict)]}
from viano_bug_analysis.llm.diagnostics import sanitize
result['failed_cache_storage'] = [
{'estimated_cost': number(a.get('estimated_cost')), 'currency': a['currency']}
for a in value.get('failed_cache_storage', [])[:64]
if isinstance(a, dict) and isinstance(a.get('currency'), str) and re.fullmatch('[A-Z]{3}', a['currency'])]
for field in ('cached_input_tokens','uncached_input_tokens','reasoning_tokens','cache_created_tokens',
'estimated_input_cost','estimated_cached_input_cost','estimated_output_cost','estimated_cache_storage_cost'):
if field in value:
result[field] = number(value[field])
if isinstance(value.get('cache_status'), str) and value['cache_status'] in {'disabled','below_minimum','created','reused','fallback'}:
result['cache_status'] = value['cache_status']
diagnostics = sanitize(value.get('diagnostics'))
if diagnostics:
result['diagnostics'] = diagnostics
+315
View File
@@ -0,0 +1,315 @@
"""Offline cache protocol, accounting, privacy and portable-request regressions."""
import asyncio
import copy
import json
from dataclasses import replace
from importlib.resources import files
import pytest
from jsonschema import Draft202012Validator
from viano_bug_analysis.llm import GatewayConfig, LLMError, LLMGateway, ProviderConfig, ProviderResponse, ProviderUnavailable
from viano_bug_analysis.llm.gemini import GeminiProvider
from viano_bug_analysis.llm.usage import usage_metadata
from viano_bug_analysis.llm.diagnostics import sanitize
from test_llm_gateway import Adapter, request, http_endpoint
def config(**kw):
return ProviderConfig(id='gemini-test', model='gemini-test', base_url='http://localhost/v1beta',
adapter='gemini', context_cache=True, cache_min_prefix_chars=1, **kw)
def native(cached=80, finish='STOP'):
return {'modelVersion': 'gemini-test', 'candidates': [{'finishReason': finish,
'content': {'parts': [{'text': '{"answer":"ok"}'}]}}],
'usageMetadata': {'promptTokenCount': 100, 'cachedContentTokenCount': cached,
'candidatesTokenCount': 10, 'thoughtsTokenCount': 5}}
def scripted(adapter, monkeypatch, actions):
seen = []
def post(url, payload, headers, limit, deadline):
seen.append((url, copy.deepcopy(payload), headers.copy()))
result = actions.pop(0)
if isinstance(result, Exception):
raise result
return result
monkeypatch.setattr(adapter, '_post', post)
return seen
CACHE = {'name': 'cachedContents/test-cache', 'usageMetadata': {'totalTokenCount': 80}}
def test_cache_reuses_only_instructions_schema_keeps_full_live_research(monkeypatch):
adapter, cfg = GeminiProvider(), config(api_key_env='CACHE_TEST_KEY')
monkeypatch.setenv('CACHE_TEST_KEY', 'never-persist-this')
seen = scripted(adapter, monkeypatch, [CACHE, native(), native()])
req = request()
value = json.loads(req['messages'][1]['content'])
value['input'] = {'evidence': {'E000001': {'text': 'complete first finding'}}}
req['messages'][1]['content'] = json.dumps(value)
original = copy.deepcopy(req)
first = asyncio.run(adapter.generate(cfg, req, max_response_bytes=10000))
value['input']['evidence']['E000002'] = {'text': 'complete second finding'}
second_req = copy.deepcopy(req)
second_req['messages'][1]['content'] = json.dumps(value)
second = asyncio.run(adapter.generate(cfg, second_req, max_response_bytes=10000))
assert req == original
assert len(seen) == 3 and seen[0][0].endswith('/cachedContents')
assert 'complete first finding' not in json.dumps(seen[0][1])
assert 'complete first finding' in json.dumps(seen[2][1])
assert 'complete second finding' in json.dumps(seen[2][1])
assert seen[2][1]['cachedContent'] == CACHE['name']
assert 'systemInstruction' not in seen[2][1]
assert first.diagnostics['cache_status'] == 'created'
assert second.diagnostics['cache_status'] == 'reused'
assert first.output_tokens == 15 and first.reasoning_tokens == 5
assert second.diagnostics['cache_created_tokens'] == 0
assert 'never-persist-this' not in repr(first) + repr(adapter._cache)
@pytest.mark.parametrize('change', ['model','system','schema','credential','url'])
def test_changed_prefix_or_provider_never_reuses_cache(monkeypatch, change):
adapter, cfg, req = GeminiProvider(), config(api_key_env='CACHE_TEST_KEY'), request()
monkeypatch.setenv('CACHE_TEST_KEY', 'first-key')
seen = scripted(adapter, monkeypatch, [CACHE, native(), CACHE, native()])
asyncio.run(adapter.generate(cfg, req, max_response_bytes=10000))
if change == 'model': cfg = replace(cfg, model='other')
elif change == 'system': req['messages'][0]['content'] += ' changed'
elif change == 'schema': req['output_schema']['description'] = 'changed'
elif change == 'credential': monkeypatch.setenv('CACHE_TEST_KEY', 'second-key')
else: cfg = replace(cfg, base_url='http://localhost/other')
asyncio.run(adapter.generate(cfg, req, max_response_bytes=10000))
assert sum(url.endswith('/cachedContents') for url, _, _ in seen) == 2
def test_expired_cache_recreated(monkeypatch):
adapter, cfg = GeminiProvider(), config()
seen = scripted(adapter, monkeypatch, [CACHE, native(), CACHE, native()])
asyncio.run(adapter.generate(cfg, request(), max_response_bytes=10000))
adapter._cache = {key: (name, 0) for key, (name, _) in adapter._cache.items()}
asyncio.run(adapter.generate(cfg, request(), max_response_bytes=10000))
assert len(seen) == 4
@pytest.mark.parametrize('code', [400,403,404,429,503])
def test_cache_creation_rejected_falls_back_and_is_not_retried_every_round(monkeypatch, code):
adapter, cfg = GeminiProvider(), config()
seen = scripted(adapter, monkeypatch, [LLMError('LLM_HTTP_CONFIGURATION', http_status=code), native(0), native(0)])
for _ in range(2):
result = asyncio.run(adapter.generate(cfg, request(), max_response_bytes=10000))
assert result.diagnostics['cache_status'] == 'fallback'
assert len(seen) == 3
assert all('cachedContent' not in payload and 'systemInstruction' in payload for _, payload, _ in seen[1:])
def test_rejected_cached_generation_retries_full_input_once(monkeypatch):
adapter, cfg = GeminiProvider(), config()
seen = scripted(adapter, monkeypatch, [CACHE, LLMError('LLM_HTTP_CONFIGURATION', http_status=404), native(0)])
result = asyncio.run(adapter.generate(cfg, request(), max_response_bytes=10000))
assert result.diagnostics['cache_status'] == 'fallback'
assert 'cachedContent' in seen[1][1] and 'cachedContent' not in seen[2][1]
assert 'systemInstruction' in seen[2][1]
def test_cache_loss_and_new_adapter_need_no_provider_state(monkeypatch):
cfg, req = config(), request()
for adapter in (GeminiProvider(), GeminiProvider()):
seen = scripted(adapter, monkeypatch, [CACHE, native()])
result = asyncio.run(adapter.generate(cfg, req, max_response_bytes=10000))
assert result.document == {'answer':'ok'} and len(seen) == 2
@pytest.mark.parametrize('cfg', [replace(config(), context_cache=False), replace(config(), cache_min_prefix_chars=1000000)])
def test_cache_optional_no_create_when_disabled_or_small(monkeypatch, cfg):
adapter = GeminiProvider()
seen = scripted(adapter, monkeypatch, [native(0)])
asyncio.run(adapter.generate(cfg, request(), max_response_bytes=10000))
assert len(seen) == 1 and 'cachedContent' not in seen[0][1]
def test_usage_counts_and_cost_do_not_double_charge_thinking_or_cached_input():
cfg = config(input_cost_per_million=2, cached_input_cost_per_million=.2,
output_cost_per_million=4, cache_storage_cost_per_million_hour=.5)
response = ProviderResponse({}, 'm', 100, 15, {'cache_created_tokens':80,'cache_ttl_seconds':300}, 80, 5)
usage = usage_metadata(cfg, response)
assert usage['uncached_input_tokens'] == 20
assert usage['estimated_cost'] == pytest.approx((20*2 + 80*.2 + 15*4)/1e6)
assert usage['estimated_cache_storage_cost'] == pytest.approx(80*.5/1e6/12)
assert usage['reasoning_tokens'] == 5
def test_unknown_cache_usage_remains_unknown():
usage = usage_metadata(config(cached_input_cost_per_million=.1), ProviderResponse({}, 'm', 100, 10))
assert usage['cached_input_tokens'] is None and usage['uncached_input_tokens'] is None
assert usage['estimated_cost'] is None
assert usage_metadata(config(), ProviderResponse({}, 'm', 10, 1, cached_input_tokens=11))['cached_input_tokens'] is None
def test_config_defaults_keep_frozen_snapshot_compatible():
snapshot = GatewayConfig((ProviderConfig('x','m','http://localhost'),)).snapshot()['providers'][0]
for key in ('context_cache','cache_ttl_seconds','cache_min_prefix_chars','cached_input_cost_per_million',
'cache_storage_cost_per_million_hour'):
assert key not in snapshot
assert GatewayConfig((config(),)).snapshot()['providers'][0]['context_cache'] is True
@pytest.mark.parametrize('changes', [{'adapter':'openai_compatible'}, {'cache_ttl_seconds':True},
{'cache_min_prefix_chars':0}, {'cached_input_cost_per_million':-1}, {'model':'../other'}, {'reasoning_effort':'none'}])
def test_bad_cache_configuration_rejected(changes):
with pytest.raises(LLMError): replace(config(), **changes)
def test_openai_compatible_reports_implicit_cache_hits(http_endpoint):
url, _, reply = http_endpoint
reply['body']['usage'].update(prompt_tokens=100, prompt_tokens_details={'cached_tokens':80},
completion_tokens_details={'reasoning_tokens':2})
cfg = ProviderConfig('test','m',url, input_cost_per_million=2,
cached_input_cost_per_million=.2, output_cost_per_million=4)
metadata = asyncio.run(LLMGateway(GatewayConfig((cfg,))).complete_round(request()))['metadata']
assert metadata['cached_input_tokens'] == 80 and metadata['uncached_input_tokens'] == 20
assert metadata['reasoning_tokens'] == 2
assert metadata['estimated_cost'] == pytest.approx((20*2+80*.2+5*4)/1e6)
def test_native_real_http_transport(http_endpoint):
url, seen, reply = http_endpoint
reply['body'] = native()
cfg = replace(config(), base_url=url, context_cache=False, response_format='json_schema', reasoning_effort='low')
result = asyncio.run(LLMGateway(GatewayConfig((cfg,))).complete_round(request()))
assert result['document'] == {'answer':'ok'}
assert seen[0]['path'].endswith('/models/gemini-test:generateContent')
generation = seen[0]['body']['generationConfig']
assert generation['thinkingConfig'] == {'thinkingLevel':'low'}
assert generation['responseJsonSchema'] == request()['output_schema']
def test_invalid_model_output_keeps_usage_and_storage_diagnostics(monkeypatch):
adapter, cfg = GeminiProvider(), config()
scripted(adapter, monkeypatch, [CACHE, native(finish='MAX_TOKENS')])
gw = LLMGateway(GatewayConfig((cfg,)), {'gemini':adapter})
with pytest.raises(LLMError) as caught:
asyncio.run(gw.complete_round(request()))
assert caught.value.code == 'LLM_INCOMPLETE_OUTPUT'
assert caught.value.metadata['cached_input_tokens'] == 80
assert caught.value.metadata['cache_created_tokens'] == 80
schema = json.loads(files('viano_bug_analysis.schemas').joinpath('llm-call-v1.schema.json').read_text())
Draft202012Validator(schema['properties']['outcome']['properties']['diagnostics']).validate(caught.value.diagnostics)
def test_strict_phase_metadata_schemas_accept_optional_usage_fields():
cfg = replace(config(), adapter='fixture', context_cache=False)
response = ProviderResponse({'answer':'ok'}, 'm', 100, 15, cached_input_tokens=80, reasoning_tokens=5)
adapter = Adapter({cfg.id:response})
metadata = asyncio.run(LLMGateway(GatewayConfig((cfg,)), {'fixture':adapter}).complete_round(request()))['metadata']
for name, version in [('case-interpretation-v1.schema.json','case_interpretation_resolved_v8'),
('hypotheses-v1.schema.json','hypotheses_project_v4')]:
schema = json.loads(files('viano_bug_analysis.schemas').joinpath(name).read_text())
# Use an allowed phase-specific prompt; usage contract stays identical.
target = schema['properties']['llm_metadata']
phase_metadata = metadata | {'prompt_version':target['properties']['prompt_version']['enum'][-1]}
Draft202012Validator(target).validate(phase_metadata)
legacy = {k:v for k,v in phase_metadata.items() if k in target['required']}
Draft202012Validator(target).validate(legacy)
def test_diagnostics_never_include_provider_cache_handles():
assert sanitize({'cache_status':{}}) == {}
assert sanitize({'cache_status':'cachedContents/secret', 'cache_name':'secret',
'cached_input_tokens':80}) == {'cached_input_tokens':80}
def test_allocation_is_preserved_through_provider_failover(monkeypatch):
adapter, cfg = GeminiProvider(), config(cache_storage_cost_per_million_hour=.5)
scripted(adapter, monkeypatch, [CACHE, ProviderUnavailable('LLM_CONNECTION')])
fallback = ProviderConfig('fallback','m','http://localhost',adapter='fixture')
gw = LLMGateway(GatewayConfig((cfg, fallback)), {'gemini':adapter,
'fixture':Adapter({'fallback':ProviderResponse({'answer':'ok'},'m')})})
result = asyncio.run(gw.complete_round(request()))
usage = result['metadata']['attempts'][0]['cache_usage']
assert usage['cache_created_tokens'] == 80
assert usage['estimated_cache_storage_cost'] == pytest.approx(80*.5/1e6/12)
assert 'cachedContents/' not in json.dumps(result)
@pytest.mark.parametrize('status', [401,302,503])
def test_native_http_failures_are_controlled(http_endpoint, status):
url, seen, reply = http_endpoint
reply.update(status=status, body=b'private provider body', headers={'Location':url+'/secret'})
cfg = replace(config(), base_url=url, context_cache=False)
with pytest.raises(LLMError) as caught:
asyncio.run(LLMGateway(GatewayConfig((cfg,))).complete_round(request()))
assert len(seen) == 1
assert 'private' not in str(caught.value)
def test_native_response_limit(http_endpoint):
url, _, reply = http_endpoint
reply['body'] = b'x'*1000
cfg = replace(config(), base_url=url, context_cache=False)
with pytest.raises(LLMError) as caught:
asyncio.run(LLMGateway(GatewayConfig((cfg,),max_response_bytes=100)).complete_round(request()))
assert caught.value.code == 'LLM_RESPONSE_LIMIT'
def test_native_credentials_checked_before_any_request(monkeypatch):
adapter = GeminiProvider()
monkeypatch.delenv('CACHE_TEST_MISSING', raising=False)
seen = scripted(adapter, monkeypatch, [])
with pytest.raises(LLMError) as caught:
asyncio.run(adapter.generate(config(api_key_env='CACHE_TEST_MISSING'),request(),max_response_bytes=1000))
assert caught.value.code == 'LLM_CREDENTIALS' and not seen
def test_cache_usage_visible_in_telemetry(tmp_path):
from viano_bug_analysis.lifecycle import Workspace
from viano_bug_analysis.telemetry import Telemetry
from viano_bug_analysis.telemetry.__main__ import read_events
cfg = config(input_cost_per_million=2, cached_input_cost_per_million=.2,
output_cost_per_million=4, cache_storage_cost_per_million_hour=.5)
response = ProviderResponse({},'m',100,15,{'cache_status':'created','cache_created_tokens':80,'cache_ttl_seconds':300},80,5)
metadata = usage_metadata(cfg,response) | {'input_tokens':100,'output_tokens':15,'cost_currency':'USD'}
failed = {'cache_usage':{'estimated_cache_storage_cost':.01,'cost_currency':'USD'}}
metadata['attempts'] = [failed]
with Telemetry(Workspace(tmp_path/'workspace')) as telemetry:
telemetry.llm_round({'phase':'INVESTIGATING_HYPOTHESES','status':'SUCCESS','outcome':{'metadata':metadata}})
metrics = telemetry.prometheus()
assert 'vba_llm_tokens_total{direction="cached_input_tokens"} 80' in metrics
assert 'vba_llm_tokens_total{direction="uncached_input_tokens"} 20' in metrics
assert 'vba_llm_tokens_total{direction="reasoning_tokens"} 5' in metrics
assert 'component="estimated_cache_storage_cost"' in metrics
event = next(e for e in read_events(telemetry.workspace.root) if e['event']=='llm_round')['detail']
assert event['cached_input_tokens'] == 80 and event['cache_status'] == 'created'
assert event['failed_cache_storage'] == [{'estimated_cost':.01,'currency':'USD'}]
def test_example_config_loads_without_credentials_or_network():
from pathlib import Path
cfg = GatewayConfig.load(Path(__file__).parents[1]/'config/llm.gemini-cache.example.toml')
assert cfg.providers[0].adapter == 'gemini' and cfg.providers[0].context_cache
LLMGateway(cfg)
def test_cache_memory_and_remote_allocations_are_bounded(monkeypatch):
adapter, cfg = GeminiProvider(), config()
actions = []
for _ in range(8): actions.extend([CACHE, native()])
actions.append(native(0))
seen = scripted(adapter, monkeypatch, actions)
for index in range(9):
req = request()
req['messages'][0]['content'] += str(index)
asyncio.run(adapter.generate(cfg,req,max_response_bytes=10000))
assert len(adapter._cache) == 8
assert sum(url.endswith('/cachedContents') for url,_,_ in seen) == 8
@pytest.mark.parametrize('invalid', [{}, {'name':'https://evil.example/secret'}, {'name':'cachedContents/../secret'}])
def test_invalid_cache_handles_never_used(monkeypatch, invalid):
adapter = GeminiProvider()
seen = scripted(adapter,monkeypatch,[invalid,native(0)])
result = asyncio.run(adapter.generate(config(),request(),max_response_bytes=10000))
assert result.diagnostics['cache_status'] == 'fallback'
assert 'cachedContent' not in seen[-1][1]