diff --git a/README.md b/README.md index 4b61a35..f2e52a6 100644 --- a/README.md +++ b/README.md @@ -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). diff --git a/config/llm.gemini-cache.example.toml b/config/llm.gemini-cache.example.toml new file mode 100644 index 0000000..8d79d96 --- /dev/null +++ b/config/llm.gemini-cache.example.toml @@ -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 diff --git a/docs/llm-context-caching.md b/docs/llm-context-caching.md new file mode 100644 index 0000000..554e87e --- /dev/null +++ b/docs/llm-context-caching.md @@ -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) diff --git a/src/viano_bug_analysis/llm/config.py b/src/viano_bug_analysis/llm/config.py index 6eb8c73..d89cf93 100644 --- a/src/viano_bug_analysis/llm/config.py +++ b/src/viano_bug_analysis/llm/config.py @@ -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] diff --git a/src/viano_bug_analysis/llm/diagnostics.py b/src/viano_bug_analysis/llm/diagnostics.py index d445654..aeb2294 100644 --- a/src/viano_bug_analysis/llm/diagnostics.py +++ b/src/viano_bug_analysis/llm/diagnostics.py @@ -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') diff --git a/src/viano_bug_analysis/llm/gateway.py b/src/viano_bug_analysis/llm/gateway.py index bb47fa1..9ed3c56 100644 --- a/src/viano_bug_analysis/llm/gateway.py +++ b/src/viano_bug_analysis/llm/gateway.py @@ -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 diff --git a/src/viano_bug_analysis/llm/gemini.py b/src/viano_bug_analysis/llm/gemini.py new file mode 100644 index 0000000..ec61de3 --- /dev/null +++ b/src/viano_bug_analysis/llm/gemini.py @@ -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 diff --git a/src/viano_bug_analysis/llm/openai_compatible.py b/src/viano_bug_analysis/llm/openai_compatible.py index 87aa297..a4e0fa1 100644 --- a/src/viano_bug_analysis/llm/openai_compatible.py +++ b/src/viano_bug_analysis/llm/openai_compatible.py @@ -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 diff --git a/src/viano_bug_analysis/llm/provider_base.py b/src/viano_bug_analysis/llm/provider_base.py index 735c3ce..f5e0165 100644 --- a/src/viano_bug_analysis/llm/provider_base.py +++ b/src/viano_bug_analysis/llm/provider_base.py @@ -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. + """ ... diff --git a/src/viano_bug_analysis/llm/usage.py b/src/viano_bug_analysis/llm/usage.py new file mode 100644 index 0000000..a57b0b5 --- /dev/null +++ b/src/viano_bug_analysis/llm/usage.py @@ -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')} diff --git a/src/viano_bug_analysis/schemas/case-interpretation-v1.schema.json b/src/viano_bug_analysis/schemas/case-interpretation-v1.schema.json index 9738812..2db1510 100644 --- a/src/viano_bug_analysis/schemas/case-interpretation-v1.schema.json +++ b/src/viano_bug_analysis/schemas/case-interpretation-v1.schema.json @@ -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": [ diff --git a/src/viano_bug_analysis/schemas/hypotheses-v1.schema.json b/src/viano_bug_analysis/schemas/hypotheses-v1.schema.json index 1829b39..c3822d6 100644 --- a/src/viano_bug_analysis/schemas/hypotheses-v1.schema.json +++ b/src/viano_bug_analysis/schemas/hypotheses-v1.schema.json @@ -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": [ diff --git a/src/viano_bug_analysis/schemas/llm-call-v1.schema.json b/src/viano_bug_analysis/schemas/llm-call-v1.schema.json index f639d43..be39795 100644 --- a/src/viano_bug_analysis/schemas/llm-call-v1.schema.json +++ b/src/viano_bug_analysis/schemas/llm-call-v1.schema.json @@ -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 diff --git a/src/viano_bug_analysis/telemetry/core.py b/src/viano_bug_analysis/telemetry/core.py index a6181da..3c4bc43 100644 --- a/src/viano_bug_analysis/telemetry/core.py +++ b/src/viano_bug_analysis/telemetry/core.py @@ -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): diff --git a/src/viano_bug_analysis/telemetry/safety.py b/src/viano_bug_analysis/telemetry/safety.py index edcb205..68d9a14 100644 --- a/src/viano_bug_analysis/telemetry/safety.py +++ b/src/viano_bug_analysis/telemetry/safety.py @@ -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 diff --git a/tests/test_llm_caching.py b/tests/test_llm_caching.py new file mode 100644 index 0000000..03297b4 --- /dev/null +++ b/tests/test_llm_caching.py @@ -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]