Prepare bounded casebuilding trial with explicit phase stop

docs/casebuilding-trial.md

@@ -0,0 +1,50 @@
+# Begrenzter Casebuilding-Test
+
+Der Test vom 14.10.2026 verwendet eine frische Fallakte aus den ursprünglichen
+wowdiag/wowsrc-Paketen, die lokale World-Testdatenbank und den festgelegten Git-Commit.
+Die vorherige gestoppte Analyse wird nicht fortgesetzt.
+
+Workspace: `service-data/trials/casebuilding-20261014`
+Case-ID: `BUG-000020`
+Package-ID: `trial-casebuilding-BUG-000020-20261014`
+
+Der deterministische CaseBuilder wurde bereits ohne LLM ausgeführt. Das gültige
+Paket enthält vier belegte Projektabfragen: die Git-Schemafundstellen für Creature
+und Quest sowie DB-Namen/Details für die gemeldete Creature und Quest. Insbesondere
+ist `23486 = Goldark Snipehunter` vor der Interpretation sichtbar.
+Die Lösung des bekannten Bugs wurde nicht vorgegeben. Snapshots für die Prüfung
+liegen als `case.before-llm.json`, `case.before-llm.md` und `preparation.json` im Workspace.
+
+In der PowerShell mit gesetztem API-Key aus dem Repository-Verzeichnis starten:
+
+```powershell
+.\.venv\Scripts\python.exe -m viano_bug_analysis.trial run --workspace service-data/trials/casebuilding-20261014 --llm-config config/llm.timeout300.local.toml --project-config config/project.local.toml --case-id BUG-000020 --package-id trial-casebuilding-BUG-000020-20261014 --stop-after interpretation --max-dispatches 6
+```
+
+Dieser Befehl führt kostenpflichtige LLM-Anfragen aus. `--stop-after interpretation`
+beendet den Lauf bei der bestätigten Interpretation mit `PHASE_COMPLETE`, auch bei
+einer erneuten Ausführung desselben Befehls. Er startet keine Hypothesengenerierung
+oder Investigation. `--max-dispatches 6` begrenzt zusätzlich die einzelnen Schritte;
+die bestehende Korrekturpolicy begrenzt ungültige Antworten separat. Provider-Retries
+folgen der LLM-Konfiguration; der Dienst wartet nicht aktiv auf einen Retry-Zeitpunkt.
+
+Bei `RETRY_PENDING`/`DEFERRED` nach dem angegebenen Zeitpunkt denselben Befehl
+erneut ausführen. `YIELDED` bedeutet, dass das Schrittlimit vor einem bestätigten
+Ergebnis erreicht wurde; derselbe Befehl setzt fort. Bei `ERROR`, `BLOCKED` oder
+`REVIEW` zuerst die Ausgabe auswerten.
+
+Prüfkriterien nach `PHASE_COMPLETE`:
+
+- Goldark wird korrekt von den betroffenen Wolpertingern unterschieden.
+- Testerbeobachtung, erwartetes Verhalten und Annahmen bleiben getrennt.
+- Neue technische Aussagen zitieren die DB-/Git-Fakten aus dem Case.
+- Nicht gelesene Spawns werden nicht als bekannte Daten ausgegeben.
+- DB-Snapshot und Sourcecode-Schemata werden nicht mit Live-Serverbeobachtungen verwechselt.
+- Verbleibende Wissenslücken sind sichtbar; die Interpretation behauptet noch keine Root Cause.
+- Modellaufrufe, Input-/Output-Tokens, Korrekturversuche und Laufzeit werden aus der
+  gespeicherten LLM-Historie ausgewertet. Ohne konfigurierte Preise ist `estimated_cost`
+  gegebenenfalls null; tatsächliche Kosten sind zusätzlich im Provider-Dashboard sichtbar.
+
+Der JSON-Abschluss zeigt den Pfad der neuen Paketgeneration. Die vollständige
+Interpretation, JSON/Markdown-Zusammenfassung, Prompt und Tokenmetadaten liegen
+in diesem Paket. Der vorbereitete Case bleibt append-only erhalten.

src/viano_bug_analysis/trial.py

@@ -59,9 +59,11 @@ def describe(path, status, *, code=None, exported=None, retry_not_before=None):
 
 
 async def drive(store, gateway, case_id, package_id, *, max_dispatches=100,
-                attempt_id="default", policies=None, progress=None):
+                attempt_id="default", policies=None, progress=None, stop_after=None):
     if type(max_dispatches) is not int or max_dispatches <= 0:
         raise TrialError("TRIAL_DISPATCH_LIMIT")
+    if stop_after not in {None, "case", "interpretation"}:
+        raise TrialError("TRIAL_STOP_PHASE")
     policies = policies or {}
     identity = (case_id, package_id)
     correction = policies.get("correction", CorrectionConfig())
@@ -86,6 +88,8 @@ async def drive(store, gateway, case_id, package_id, *, max_dispatches=100,
         if state == State.BLOCKED:
             return describe(latest, "BLOCKED", code=routing["reason"],
                             retry_not_before=routing["retry_not_before"])
+        if phase_complete(latest, stop_after):
+            return describe(latest, "PHASE_COMPLETE")
         if progress:
             progress({"dispatch": dispatch + 1, "state": state})
         if state in {State.OUTPUT_READY, State.FAILED_TO_ANALYZE, State.HANDED_OFF}:
@@ -124,7 +128,27 @@ async def drive(store, gateway, case_id, package_id, *, max_dispatches=100,
             return describe(result.package, status, code=getattr(result, "code", None),
                             exported=getattr(result, "exported", None),
                             retry_not_before=getattr(result, "retry_not_before", None))
-    return describe(store.latest(*identity), "YIELDED")
+    latest = store.latest(*identity)
+    return describe(latest, "PHASE_COMPLETE" if phase_complete(latest, stop_after) else "YIELDED")
+
+
+def phase_complete(path, stop_after):
+    if stop_after is None:
+        return False
+    from .case.contracts import read_json
+    with PackageReader(path) as reader:
+        manifest = reader.manifest
+        if stop_after == "case":
+            ref = manifest["artifacts"].get("case")
+            if manifest["state"] != State.BUILDING_CASE or not ref:
+                return False
+            doc = read_json(reader, ref, "case", 8 * 1024 * 1024)["case"]
+        else:
+            ref = manifest["artifacts"].get("case_interpretation")
+            if manifest["state"] != State.INTERPRETING_CASE or not ref:
+                return False
+            doc = read_json(reader, ref, "case-interpretation", 2 * 1024 * 1024)
+        return all(doc[field] == manifest[field] for field in ("case_id", "package_id", "analysis_revision"))
 
 
 def positive(text):
@@ -153,6 +177,7 @@ def parser():
     run.add_argument("--package-id", help="Stable managed package identity")
     run.add_argument("--attempt-id", default="default")
     run.add_argument("--max-dispatches", type=positive, default=100)
+    run.add_argument("--stop-after", choices=["case", "interpretation"], help="Stop at the confirmed phase artifact, including on resume")
     run.add_argument("--telemetry-config", type=Path)
     run.add_argument("--project-config", type=Path, help="Read-only source/DB research providers")
     for name in ("analysis", "budget", "investigation", "diagnosis", "change-request", "correction"):
@@ -210,12 +235,13 @@ def main(argv=None):
                     with FileLock(workspace.runtime / "locks" / ("trial-" + key(case_id) + ".lock")):
                         result = asyncio.run(drive(queue.store, gateway, case_id, package_id,
                             max_dispatches=args.max_dispatches, attempt_id=args.attempt_id, policies=policies,
+                            stop_after=args.stop_after,
                             progress=lambda event: print(json.dumps(event), file=sys.stderr, flush=True)))
                 except Exception as error:
                     telemetry.guard(telemetry.exception, error)
                     raise
         print(json.dumps(result, ensure_ascii=True))
-        return 0 if result["status"] in {"CONFIG_READY", "PREPARED", "OUTPUT_READY", "REVIEW", "HANDED_OFF"} else 3
+        return 0 if result["status"] in {"CONFIG_READY", "PREPARED", "PHASE_COMPLETE", "OUTPUT_READY", "REVIEW", "HANDED_OFF"} else 3
     except KeyboardInterrupt:
         print(json.dumps({"status": "INTERRUPTED"}), file=sys.stderr)
         return 130

tests/test_trial.py

@@ -65,6 +65,18 @@ def test_initial_builder_interpretation_and_hypotheses(case_setup):
     assert adapter.calls == 2
 
 
+@pytest.mark.parametrize("phase,dispatches,calls",[("case",1,0),("interpretation",2,1),("interpretation",10,1)])
+def test_stop_after_confirmed_phase_and_safe_repeated_command(case_setup,phase,dispatches,calls):
+    store=case_setup[0]
+    adapter=PipelineAdapter();g=gateway(adapter)
+    result=asyncio.run(drive(store,g,*CASE,max_dispatches=dispatches,stop_after=phase))
+    assert result["status"]=="PHASE_COMPLETE" and adapter.calls==calls
+    with PackageReader(result["package"]) as reader:
+        assert not reader.manifest["artifacts"].get("hypotheses")
+    replay=asyncio.run(drive(store,g,*CASE,max_dispatches=10,stop_after=phase))
+    assert replay["package"]==result["package"] and adapter.calls==calls
+
+
 def test_retry_returns_without_sleep_and_blocked_budget_kept(tool_case):
     store = tool_case[0]
     adapter = LoopAdapter(unavailable=True)
@@ -105,6 +117,17 @@ def test_cli_rejects_ambiguous_identity():
         main(["run", "--package", "one.zip", "--case-id", "a", "--package-id", "b"])
 
 
+def test_cli_interpretation_stop_exits_successfully(case_setup,monkeypatch,capsys):
+    import viano_bug_analysis.trial as trial
+    store=case_setup[0];adapter=PipelineAdapter()
+    monkeypatch.setattr(trial,"preflight",lambda path:gateway(adapter))
+    command=["run","--workspace",str(store.workspace.root),"--case-id",CASE[0],"--package-id",CASE[1],
+             "--stop-after","interpretation","--max-dispatches","6"]
+    assert main(command)==0
+    assert json.loads(capsys.readouterr().out)["status"]=="PHASE_COMPLETE" and adapter.calls==1
+    assert main(command)==0 and adapter.calls==1
+
+
 def test_cli_run_connects_telemetry_and_publishes(tool_case, monkeypatch, capsys):
     import viano_bug_analysis.trial as trial
     store = tool_case[0]