Prepare four isolated bilingual minimal prompt benchmarks

docs/minimal-prompt-benchmark.md

@@ -0,0 +1,46 @@
+# Minimaler Investigation-Prompt: vier Vergleichsanfragen
+
+Vorbereitung ohne Netzwerk:
+
+```powershell
+.\.venv\Scripts\python.exe -m viano_bug_analysis.minimal_prompt_test
+```
+
+Ausführung durch den Nutzer (vier potenziell kostenpflichtige API-Aufrufe):
+
+```powershell
+.\.venv\Scripts\python.exe -m viano_bug_analysis.minimal_prompt_test --run
+```
+
+Standardkonfiguration: config/llm.working-brief.local.toml. Standardausgabe:
+service-data/benchmarks/minimal-prompts-20261017/plan.json und results.json.
+Der vorbereitete Plan muss unverändert zur Konfiguration passen. Ein vorhandenes
+results.json verhindert eine unbeabsichtigte Wiederholung. Für einen bewusst
+neuen Durchlauf eine neue --output-Angabe verwenden.
+
+Vier Varianten: de_plain, de_json, en_plain, en_json. Deutsch 429 Zeichen,
+Englisch 375 Zeichen, inhaltlich gleichwertiger Bug/Hypothese/Arbeitsauftrag.
+Innerhalb einer Sprache ist der Text identisch. Nur JSON-Varianten verwenden
+responseMimeType und das native Schema für step und purpose. Kein Systemprompt,
+kein Investigation-Kontext, keine Werkzeuge, kein Fallpaket, keine Fallakte,
+keine Schemaeinbettung als zusätzliche Nachricht und kein expliziter Cache.
+Der isolierte native Gemini-Aufruf vermeidet die zusätzlichen Hüllen des normalen
+Gateways. Das ist ausschließlich eine Messhilfe; Produktionspfade bleiben gleich.
+
+Modell, Denkstufe, Outputlimit und Timeout werden aus derselben Providerkonfiguration
+übernommen. Aktuell: gemini-3.8-flash, medium, 16384 Outputtokens als Obergrenze.
+Das hohe Limit ist keine Aufforderung, es auszuschöpfen. Kurzer Input garantiert
+keine kurze Denkphase. Keine automatischen Wiederholungen und kein Failover;
+anderenfalls wäre die Zahl und Vergleichbarkeit der vier Aufrufe unklar.
+
+Erfasst werden tatsächliche Modellversion, Antwort, Abschlussgrund, Laufzeit,
+Inputtokens, sichtbare Kandidatentokens, Denktokens, Cachetokens und Gesamttokens.
+Fehlende API-Zähler bleiben null; Denk- und Ausgabetokens werden getrennt berichtet,
+nicht doppelt addiert. Implizites Caching bleibt anbieterseitig möglich und wird
+anhand der Antwortmetriken sichtbar. Ein Einzelwert je Variante ist eine
+Untergrenzen-Stichprobe, kein stabiler Leistungs- oder Sprachvergleich.
+
+Es wird keine SQL-Abfrage ausgeführt. Generierte Abfragevorschläge sind nur Text.
+Die Antworten bewerten wir nicht als vollständige Buganalyse. Vorbereitung und
+Offline-Tests verursachen keine API-Kosten. Speicherverbrauch der vorbereiteten
+Messung liegt nur im Kilobytebereich; Testbestand bei Vorbereitung ca. 18,85 GB.

src/viano_bug_analysis/minimal_prompt_test.py

@@ -0,0 +1,106 @@
+"""Four isolated Gemini measurements; default prepares only, --run makes four calls."""
+import argparse
+import json
+import os
+import time
+from pathlib import Path
+
+from jsonschema import Draft202012Validator
+from .llm import GatewayConfig, LLMError
+from .llm.gateway import PRIVACY
+from .llm.gemini import GeminiProvider
+from .llm.diagnostics import count
+from .telemetry.safety import safe_code
+
+SCHEMA = {"type":"object","properties":{"step":{"type":"string"},"purpose":{"type":"string"}},
+          "required":["step","purpose"],"additionalProperties":False}
+PROMPTS = {
+    "de": "Bug: Beim Braufest erscheinen vor Eisenschmiede keine Wolpertinger, obwohl der Spieler die erforderliche Brille trägt. In Orgrimmar funktioniert die entsprechende Quest. Hypothese: In der Datenbank fehlen Wolpertinger-Spawns für Dun Morogh. Welchen einzelnen Untersuchungsschritt würdest du als Nächstes durchführen? Antworte knapp auf Deutsch mit der benötigten Abfrage und ihrem Zweck. Du kannst hier keine Werkzeuge ausführen.",
+    "en": "Bug: During Brewfest, no Wolpertingers appear outside Ironforge although the player wears the required goggles. The corresponding quest works in Orgrimmar. Hypothesis: The database lacks Wolpertinger spawns for Dun Morogh. Which single investigation step would you perform next? Reply briefly in English with the required query and its purpose. You cannot execute tools here.",
+}
+
+
+def make_plan(provider):
+    cases=[]
+    for language,prompt in PROMPTS.items():
+        for structured in (False,True):
+            generation={"maxOutputTokens":provider.max_output_tokens}
+            if provider.reasoning_effort is not None:
+                generation["thinkingConfig"]={"thinkingLevel":provider.reasoning_effort}
+            if structured:
+                generation.update(responseMimeType="application/json",responseJsonSchema=SCHEMA)
+            cases.append({"id":language+("_json" if structured else "_plain"),
+                "prompt_characters":len(prompt),"payload":{"contents":[{"role":"user","parts":[{"text":prompt}]}],
+                "generationConfig":generation}})
+    return {"format":"viano-minimal-prompt-test-v1","provider":provider.id,"model":provider.model,
+            "explicit_cache":False,"automatic_cache":"provider controlled; usage reported",
+            "calls":cases,"notes":"No system prompt, history, case data, tools, retries, failover or investigation wrapper. JSON variants differ only in native generation format/schema."}
+
+
+def measure(provider,case,limit,post=None):
+    secret=os.environ.get(provider.api_key_env) if provider.api_key_env else None
+    if provider.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
+    start=time.monotonic()
+    result={"id":case["id"],"prompt_characters":case["prompt_characters"]}
+    try:
+        envelope=(post or GeminiProvider._post)(provider.base_url.rstrip('/')+'/models/'+provider.model+':generateContent',
+            case['payload'],headers,limit,start+provider.timeout_seconds)
+        usage=envelope.get('usageMetadata') or {}
+        result['usage']={name:count(usage.get(key)) for name,key in {
+            'input_tokens':'promptTokenCount','candidate_output_tokens':'candidatesTokenCount',
+            'reasoning_tokens':'thoughtsTokenCount','cached_input_tokens':'cachedContentTokenCount',
+            'total_tokens':'totalTokenCount'}.items()}
+        candidate=envelope['candidates'][0]
+        text=''.join(part['text'] for part in candidate.get('content',{}).get('parts',[]) if 'text' in part and not part.get('thought'))
+        result.update(answer=text,finish_reason=candidate.get('finishReason'),model=envelope.get('modelVersion',provider.model))
+        result['status']='SUCCESS' if candidate.get('finishReason')=='STOP' and text else 'INCOMPLETE'
+        if case['id'].endswith('_json') and result['status']=='SUCCESS':
+            try: valid=Draft202012Validator(SCHEMA).is_valid(json.loads(text))
+            except ValueError: valid=False
+            result['schema_valid']=valid
+            if not valid: result['status']='INVALID_JSON'
+    except LLMError as error:
+        result.update(status='ERROR',code=safe_code(error.code),http_status=error.http_status)
+    except (ValueError,TypeError,KeyError,IndexError,AttributeError):
+        result.update(status='ERROR',code='LLM_RESPONSE')
+    result['elapsed_seconds']=round(time.monotonic()-start,3)
+    return result
+
+
+def main(argv=None):
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--llm-config',default='config/llm.working-brief.local.toml')
+    parser.add_argument('--provider',help='Optional exact Gemini provider ID; otherwise first eligible provider')
+    parser.add_argument('--output',type=Path,default=Path('service-data/benchmarks/minimal-prompts-20261017'))
+    parser.add_argument('--run',action='store_true',help='Make four potentially billable calls; default prepares only')
+    args=parser.parse_args(argv)
+    try:
+        config=GatewayConfig.load(args.llm_config)
+        eligible=[p for p in config.providers if p.enabled and p.privacy in PRIVACY[config.privacy_profile]]
+        provider=next((p for p in eligible if p.id==args.provider),None) if args.provider else next(iter(eligible),None)
+        if provider is None or provider.adapter!='gemini':raise LLMError('MINIMAL_TEST_PROVIDER')
+        plan=make_plan(provider);args.output.mkdir(parents=True,exist_ok=True)
+        plan_path=args.output/'plan.json';result_path=args.output/'results.json'
+        if plan_path.exists() and json.loads(plan_path.read_text(encoding='utf-8'))!=plan:
+            raise LLMError('MINIMAL_TEST_PLAN_CHANGED')
+        if not plan_path.exists():plan_path.write_text(json.dumps(plan,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
+        if not args.run:
+            print(json.dumps({'status':'PREPARED','plan':str(plan_path),'calls':4,'network_used':False}));return 0
+        # Reserve result file before the first call; never silently repeat a paid run.
+        with result_path.open('x',encoding='utf-8') as f:json.dump({'status':'RUNNING','results':[]},f)
+        results=[]
+        for case in plan['calls']:
+            result=measure(provider,case,config.max_response_bytes);results.append(result)
+            result_path.write_text(json.dumps({'status':'COMPLETE' if len(results)==4 else 'RUNNING','results':results},ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
+            print(json.dumps(result,ensure_ascii=True),flush=True)
+        return 0 if all(r['status']=='SUCCESS' for r in results) else 2
+    except KeyboardInterrupt:
+        print(json.dumps({'status':'INTERRUPTED'}));return 130
+    except (LLMError,OSError,ValueError) as error:
+        print(json.dumps({'status':'ERROR','code':safe_code(getattr(error,'code',None))}));return 2
+
+
+if __name__=='__main__':raise SystemExit(main())

tests/test_minimal_prompt_test.py

@@ -0,0 +1,55 @@
+import copy
+import pytest
+from viano_bug_analysis.llm.config import ProviderConfig
+from viano_bug_analysis.llm.errors import ProviderUnavailable
+from viano_bug_analysis.minimal_prompt_test import make_plan,measure
+
+
+def provider():
+    return ProviderConfig(id='fixture',model='fixture-model',base_url='https://example.invalid/v1beta',
+        adapter='gemini',privacy='cloud',reasoning_effort='low',max_output_tokens=8192)
+
+
+def test_four_requests_differ_only_in_language_and_native_format():
+    p=provider();plan=make_plan(p)
+    assert [c['id'] for c in plan['calls']]==['de_plain','de_json','en_plain','en_json']
+    for plain,structured in (plan['calls'][:2],plan['calls'][2:]):
+        assert plain['payload']['contents']==structured['payload']['contents']
+        assert plain['prompt_characters']<500
+        assert set(plain['payload'])=={'contents','generationConfig'}
+        generation=copy.deepcopy(structured['payload']['generationConfig'])
+        generation.pop('responseMimeType');generation.pop('responseJsonSchema')
+        assert generation==plain['payload']['generationConfig']
+        assert generation=={'maxOutputTokens':8192,'thinkingConfig':{'thinkingLevel':'low'}}
+
+
+@pytest.mark.parametrize('index',[0,1,2,3])
+def test_usage_reports_candidates_and_thoughts_separately(index):
+    case=make_plan(provider())['calls'][index]
+    def post(url,payload,headers,limit,deadline):
+        assert payload==case['payload']
+        return {'modelVersion':'fixture','candidates':[{'finishReason':'STOP','content':{'parts':[
+            {'text':'hidden','thought':True},{'text':'{"step":"lookup","purpose":"test"}' if index%2 else 'Look up spawns.'}]}}],
+            'usageMetadata':{'promptTokenCount':123,'candidatesTokenCount':12,'thoughtsTokenCount':30,'totalTokenCount':165}}
+    result=measure(provider(),case,10000,post)
+    assert result['status']=='SUCCESS'
+    assert result['usage']['input_tokens']==123
+    assert result['usage']['candidate_output_tokens']==12
+    assert result['usage']['reasoning_tokens']==30
+    assert result['usage']['cached_input_tokens'] is None
+    assert 'hidden' not in result['answer']
+
+
+def test_no_retries_and_no_provider_body_leak():
+    calls=[]
+    def post(*args):
+        calls.append(1);raise ProviderUnavailable('LLM_HTTP_TEMPORARY',http_status=503)
+    result=measure(provider(),make_plan(provider())['calls'][0],1000,post)
+    assert len(calls)==1 and result['status']=='ERROR' and result['http_status']==503
+
+
+def test_incomplete_retains_usage():
+    def post(*args):return {'candidates':[{'finishReason':'MAX_TOKENS','content':{'parts':[]}}],
+        'usageMetadata':{'promptTokenCount':99,'thoughtsTokenCount':100}}
+    result=measure(provider(),make_plan(provider())['calls'][0],1000,post)
+    assert result['status']=='INCOMPLETE' and result['usage']['input_tokens']==99