diff --git a/README.md b/README.md index 5c1a258..632d696 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,10 @@ delivery with telemetry: [first real LLM trial](docs/first-llm-trial.md). Use `python -m viano_bug_analysis.trial --help`. Only its `run` command makes LLM requests; `prepare` and `check` do not contact model services. +The trial enables bounded, durable early model-output corrections with explicit +validation feedback: [early correction](docs/early-correction-v1.md), configured +through [correction.example.toml](config/correction.example.toml). + Expected-behavior resolution and the read-only provider contract: [expected-behavior-v1.md](docs/expected-behavior-v1.md). Configurable LLM transport, failover and durable retries: [llm-gateway-v1.md](docs/llm-gateway-v1.md), with [example configuration](config/llm.example.toml). diff --git a/config/correction.example.toml b/config/correction.example.toml new file mode 100644 index 0000000..420469d --- /dev/null +++ b/config/correction.example.toml @@ -0,0 +1,4 @@ +[correction] +# Invalid model responses per phase/revision, including earlier confirmed failures. +# Transport failures/backoff do not consume this count. +max_invalid_responses = 3 diff --git a/docs/early-correction-v1.md b/docs/early-correction-v1.md new file mode 100644 index 0000000..62e5fcf --- /dev/null +++ b/docs/early-correction-v1.md @@ -0,0 +1,95 @@ +# Begrenzte Korrektur früher LLM-Antworten + +Der Testtreiber aktiviert Korrekturen standardmäßig für Interpretation und +Hypothesengenerierung. Bibliotheksaufrufe können `correction_config=CorrectionConfig()` +übergeben. Bestehende direkte Aufrufe ohne diese Option behalten ihr bisheriges +Verhalten; eine bereits gespeicherte Korrektur-Policy wird automatisch übernommen. +Bestätigte alte Ergebnisse bleiben unverändert wiederverwendbar. + +```toml +[correction] +max_invalid_responses = 3 +``` + +Mit `--correction-config config/correction.local.toml` lässt sich die Grenze im +Testtreiber konfigurieren. Sie wird beim ersten aktivierten Dispatch pro Phase +und Analysis Revision eingefroren. Dieselbe Config bei Wiederaufnahme verwenden; +nachträgliche Änderungen werden vor einem weiteren Modellaufruf abgelehnt. +Die Beispielkonfiguration liegt in `config/correction.example.toml`. + +## Ablauf + +Ein Dispatch verarbeitet höchstens eine fällige Provider-Runde. Ungültige +Modellantworten ergeben `REJECTED`. Der Testtreiber setzt mit dem nächsten Dispatch +fort; `--max-dispatches 1` gibt stattdessen `YIELDED` zurück. Die nächste Anfrage +enthält den ursprünglichen Context/Schema, einen festen Korrektur-Systemprompt, +die verworfene JSON-Antwort (falls verfügbar) und deterministisches Feedback. + +Bei Quellenfehlern nennt das Feedback beispielsweise Feld, Source-ID, tatsächlichen +Quellentyp und erlaubte Typen. Bei Hypothesen werden doppelte Titel/Mechanismen, +ungültige Quellen, Mengen- und Grounding-Regeln erläutert. Für andere Fehler steht +der kontrollierte Validierungscode mit allgemeinen Korrekturanforderungen zur +Verfügung. Schema-/Transportfehler, deren verworfener Body nicht gespeichert wird, +erhalten keine erfundene vorherige Antwort. Anbieter-Error-Bodies bleiben ausgeschlossen. + +Die vorherige Antwort ist untrusted candidate data; sie wird niemals zur neuen +Faktengrundlage. Jede neue Antwort durchläuft wieder die vollständige Prüfung. +Es gibt kein automatisches Entfernen falscher Quellen, keine Teilübernahme und +kein Lockern der Quellenregeln. Hypothesen und Budgetkonten entstehen erst nach +Annahme des vollständigen gültigen Hypothesensatzes. + +## Grenzen und Persistenz + +`max_invalid_responses` zählt ungültige Antworten, einschließlich des ersten +Versuchs und früherer bestätigter Modellfehler in derselben Phase/Revision. +Ein neuer Prozess oder eine andere `--attempt-id` setzt diesen Zähler nicht zurück. +Bereits vorhandene fachliche Fehler werden bei Aktivierung anhand ihrer gebundenen +LLM-Aufrufe übernommen; wiederholtes Lesen desselben Fehlers zählt nicht erneut. + +Policy und Dispatch-Beschreibung werden vor der Anfrage als unveränderliche +Artefakte unter `analysis/interpretations/NNNN/corrections/` beziehungsweise +`analysis/hypotheses/NNNN/corrections/` gespeichert. Stabile interne Versuchs-IDs +trennen Modellkorrekturen. Provider-Retries innerhalb eines Versuchs verwenden +dagegen dieselbe eingefrorene Anfrage und Operation, einschließlich ihres +Konfigurationsfingerprints. Die Transportkonfiguration während eines ausstehenden +Retries nicht wechseln. `--attempt-id` ist in diesem Modus kein manueller Reset. + +Provider-Ausfälle/Timeouts/Backoff verbrauchen keinen Modellfehler-Versuch. Sie +behalten die bestehenden Regeln: `RETRY_PENDING`, `DEFERRED`, nach Ausschöpfung +`BLOCKED`. Konfigurations-, Credential-, Größen- und Infrastrukturfehler werden +nicht als fachliche Fehlantwort automatisch wiederholt. Ein Absturz zwischen +Provider-Antwort und Speicherung kann weiterhin den unbestätigten Aufruf wiederholen. +Ein gespeicherter Erfolg oder Fehler wird dagegen ohne erneuten Remote-Aufruf verarbeitet. + +Nach Ausschöpfung entsteht `FAILED_TO_ANALYZE/review` mit +`MODEL_VALIDATION_EXHAUSTED`, Fehlerreferenzen, Context-Hash, offenen Fragen und +empfohlenen menschlichen Aktionen. Das ist ein früher Fehlerbericht, kein erfundener +Diagnosebericht. OutputValidator prüft seine Policy-, Context-, Fehler-/Call- und +Checkpoint-Bindung sowie Input-Frische und veröffentlicht ein validiertes Review-ZIP. +Es entstehen weder Diagnose noch Coding-Auftrag. Bereits vorhandene Budgets werden +nicht geändert; neue Modelldispatches sind keine Investigation-Budgetbuchungen. +LLM-Tokens/Anbieterkosten werden weiterhin separat durch Telemetrie erfasst. + +Der JSON-Vertrag ist `early-correction-v1.schema.json`. Originale, verworfene +Antworten und ältere Generationen bleiben erhalten. Fall-ID, Paket-ID und +Analysis Revision bleiben gleich; bestätigte Artefakte erhöhen package_generation. +Korrekturartefakte sind nach Checkpoint deklarierte Analyseausgaben und werden +nicht erneut als Rohinputs des CaseBuilders interpretiert. Unbekannte Dateien +bleiben Rohinput und unverändert erhalten. + +## Fortsetzung des aktuellen BUG-20-Tests + +Die bisherige Historie enthält zwei fachlich abgelehnte Interpretationen. Mit +der Default-Grenze von drei bleibt genau ein weiterer Modellfehler erlaubt. +Provider-Ausfälle zählen dabei nicht. Der erste echte Korrekturversuch kann so +aus der PowerShell mit API-Key gestartet werden: + +```powershell +.\.venv\Scripts\python.exe -m viano_bug_analysis.trial run --workspace service-data --llm-config config/llm.timeout300.local.toml --case-id BUG-000020 --package-id trial-BUG-000020 --max-dispatches 1 +``` + +Bei akzeptierter Interpretation endet dieser begrenzte Lauf mit `YIELDED` und +vorhandenem Interpretationsartefakt. Bei erneut ungültiger Antwort veröffentlicht +der Treiber Review. Mit derselben Config und mehr Dispatches kann eine erfolgreiche +Interpretation anschließend zur Hypothesengenerierung und Untersuchung fortgesetzt werden. +Es gibt keine automatische neue Revision oder Wiederaufnahme eines Review-Falls. diff --git a/docs/first-llm-trial.md b/docs/first-llm-trial.md index 26ee580..681a667 100644 --- a/docs/first-llm-trial.md +++ b/docs/first-llm-trial.md @@ -73,16 +73,17 @@ Dispatches. `BLOCKED` bleibt blockiert, auch wenn eine Frist vergeht. Externe Testanfragen erhalten ihre Reservierungen; die spätere bestätigte Feedback- Integration ist für deren Wiederaufnahme zuständig. -Interpretation/Hypothesen mit ungültiger Antwort werden kontrolliert gestoppt. -Ein bewusst neuer Versuch kann mit `--attempt-id NEUE-STABILE-ID` erfolgen, -ohne eine gespeicherte alte Antwort unter derselben Operation zu überschreiben. -Dies ist kein automatischer Retry und kann erneut Kosten verursachen. Die -späteren Phasen behalten ihre bestehenden Grenzen für ungültige Antworten. -Keine automatische neue Analysis Revision oder Budgetzurücksetzung findet statt. +Interpretation/Hypothesen mit ungültiger Antwort verwenden die begrenzte +[Korrekturschleife](early-correction-v1.md): standardmäßig drei ungültige Antworten +pro Phase/Revision, einschließlich vorheriger Fehlversuche. Neue Antworten +erhalten Validierungsfeedback. `--attempt-id` setzt diese Grenze nicht zurück. +Nach Ausschöpfung erfolgt früher Review ohne Diagnose/Coding-Auftrag. Die späteren +Phasen behalten ihre bestehenden Grenzen. Keine automatische neue Analysis +Revision oder Budgetzurücksetzung findet statt. Optionale Konfigurationspfade: `--analysis-config`, `--budget-config`, `--investigation-config`, `--diagnosis-config`, `--change-request-config`, -`--telemetry-config`, `--llm-config`. Ohne Pfade gelten die bestehenden Defaults. +`--telemetry-config`, `--llm-config`, `--correction-config`. Ohne Pfade gelten die bestehenden Defaults. Bereits eingefrorene Policies nicht zwischen Resume-Läufen ändern; dieselben Optionen erneut angeben. Der gesamte Gateway-Konfigurationsfingerprint ist Teil der Operation; Änderungen können kontrolliert als Konflikt abgelehnt werden. diff --git a/src/viano_bug_analysis/analysis/__init__.py b/src/viano_bug_analysis/analysis/__init__.py index c6c766d..f3b8d6e 100644 --- a/src/viano_bug_analysis/analysis/__init__.py +++ b/src/viano_bug_analysis/analysis/__init__.py @@ -34,3 +34,5 @@ def __getattr__(name): __all__ += ["DiagnosisBuilder", "DiagnosisResult", "DiagnosisConfig", "DiagnosisLimits", "DiagnosisError"] __all__ += ["ChangeRequestBuilder", "ChangeRequestResult", "ChangeRequestConfig", "ChangeRequestLimits", "ChangeRequestError", "OutputValidator", "OutputResult", "OutputError"] +from .correction import CorrectionConfig +__all__ += ["CorrectionConfig"] diff --git a/src/viano_bug_analysis/analysis/correction.py b/src/viano_bug_analysis/analysis/correction.py new file mode 100644 index 0000000..14c6b4d --- /dev/null +++ b/src/viano_bug_analysis/analysis/correction.py @@ -0,0 +1,318 @@ +"""Optional durable correction dispatches for early semantic model phases.""" +import copy +import contextvars +import hashlib +import re +import tomllib +from dataclasses import dataclass, replace +from pathlib import Path +from importlib.resources import files + +from viano_bug_analysis.case.contracts import read_json, validate, require_file +from viano_bug_analysis.lifecycle import State +from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain +from viano_bug_analysis.lifecycle.storage import json_bytes, key +from viano_bug_analysis.package import PackageReader +from .errors import InterpretationError + +PHASES = { + "interpretation": (State.INTERPRETING_CASE, "analysis/interpretations", "case-interpretation-rejected", "case-interpretation-error", "interpretation_context", "case-interpretation-context"), + "hypotheses": (State.GENERATING_HYPOTHESES, "analysis/hypotheses", "hypotheses-rejected", "hypothesis-error", "hypothesis_context", "hypothesis-context"), +} +MODEL_CODES = frozenset({"LLM_OUTPUT_SCHEMA", "LLM_RESPONSE", "LLM_INCOMPLETE_OUTPUT", + "INTERPRETATION_SCHEMA", "INTERPRETATION_SOURCE_UNKNOWN", "INTERPRETATION_SOURCE_KIND", + "INTERPRETATION_EXPECTATION_UNKNOWN", "INTERPRETATION_CONTRADICTION_SOURCES", + "INTERPRETATION_GAP_SOURCES", "INTERPRETATION_ENTITY", "HYPOTHESIS_SCHEMA", "HYPOTHESIS_COUNT", + "HYPOTHESIS_DUPLICATE", "HYPOTHESIS_SOURCE_UNKNOWN", "HYPOTHESIS_GROUNDING", "HYPOTHESIS_PRIMARY_FOCUS"}) +_attempt_context = contextvars.ContextVar("vba_early_correction", default=None) + + +def current_attempt(owner): + value = _attempt_context.get() + return value[1] if value and value[0] is owner else None + + +@dataclass(frozen=True) +class CorrectionConfig: + max_invalid_responses: int = 3 + + def __post_init__(self): + if type(self.max_invalid_responses) is not int or self.max_invalid_responses <= 0: + raise InterpretationError("CORRECTION_CONFIG") + + @classmethod + def load(cls, path): + try: + with Path(path).open("rb") as stream: + d = tomllib.load(stream) + if set(d) != {"correction"}: + raise ValueError() + return cls(**d["correction"]) + except (OSError, ValueError, TypeError): + raise InterpretationError("CORRECTION_CONFIG") from None + + +def digest(data): + return hashlib.sha256(json_bytes(data)).hexdigest() + + +def base(phase, revision): + return f"{PHASES[phase][1]}/{revision:04d}/corrections" + + +def identity(m, phase, kind): + return {"format": "viano-early-correction", "format_version": 1, "kind": kind, + "phase": phase, **{f: m[f] for f in ("case_id", "package_id", "analysis_revision")}} + + +def artifacts(reader, phase): + result = [] + for _, cp in reversed(checkpoint_chain(reader)): + d = cp["details"] + if d.get("operation") == "early-correction" and d.get("phase") == phase and cp["analysis_revision"] == reader.manifest["analysis_revision"]: + doc = read_json(reader, d["artifact_ref"], "early-correction", 4 * 1024**2) + expected = base(phase, cp["analysis_revision"]) + "/" + ( + f"attempt-{doc.get('attempt', 0):04d}.json" if doc["kind"] == "attempt" else doc["kind"] + ".json") + if (d["artifact_ref"] != expected or digest(doc) != d.get("artifact_sha256") + or any(doc[f] != reader.manifest[f] for f in ("case_id", "package_id", "analysis_revision")) + or doc["phase"] != phase): + raise InterpretationError("CORRECTION_HISTORY") + if doc["kind"] == "policy" and doc["feedback_prompt_version"] != "correction_v1": + raise InterpretationError("CORRECTION_HISTORY") + expected_fields = set(identity(reader.manifest, phase, doc["kind"])) | { + "policy": {"max_invalid_responses", "feedback_prompt_version"}, + "attempt": {"attempt", "attempt_id", "previous_error_ref", "feedback"}, + "failure": {"policy", "error_refs", "reason", "context_ref", "context_sha256", "remaining_questions", "recommended_human_actions"}, + }[doc["kind"]] + if set(doc) != expected_fields: + raise InterpretationError("CORRECTION_HISTORY") + result.append((d["artifact_ref"], doc)) + return result + + +def failures(reader, phase): + result, calls = [], set() + for _, cp in reversed(checkpoint_chain(reader)): + d = cp["details"] + if cp["analysis_revision"] != reader.manifest["analysis_revision"] or d.get("operation") != PHASES[phase][2] or d.get("code") not in MODEL_CODES: + continue + error = read_json(reader, d["error_ref"], PHASES[phase][3], 4 * 1024**2) + call = read_json(reader, d["llm_result_ref"], "llm-call", 4 * 1024**2) + if (error["code"] != d["code"] or error["llm_result_ref"] != d["llm_result_ref"] + or call["phase"] != PHASES[phase][0] or call["status"] not in {"SUCCESS", "ERROR"} + or any(call[f] != reader.manifest[f] or error[f] != reader.manifest[f] for f in ("case_id", "package_id", "analysis_revision"))): + raise InterpretationError("CORRECTION_HISTORY") + context = read_json(reader, error["context_ref"], PHASES[phase][5], 4 * 1024**2) + require_file(reader, call["request_ref"]) + with reader._archive.open(call["request_ref"]) as stream: + from viano_bug_analysis.package.manifest import parse_manifest + request = parse_manifest(reader.validator._read_bounded(stream, min(4 * 1024**2, reader.validator.limits.max_file_bytes))) + from .prompt import InterpretationPromptBuilder + from .hypothesis_prompt import HypothesisPromptBuilder + expected = (InterpretationPromptBuilder() if phase == "interpretation" else HypothesisPromptBuilder()).build(context) + if (error["context_sha256"] != digest(context) or digest(request) != call["input_request_hash"] + or call["result_ref"] != d["llm_result_ref"] or request.get("messages", [])[:2] != expected["messages"] + or request.get("output_schema") != expected["output_schema"]): + raise InterpretationError("CORRECTION_HISTORY") + if call["status"] == "SUCCESS": + from .contracts import validate_interpretation + from .hypothesis_contracts import validate_proposals + from .hypothesis_config import HypothesisError + try: + (validate_interpretation if phase == "interpretation" else validate_proposals)(call["outcome"]["document"], context) + except (InterpretationError, HypothesisError) as invalid: + if invalid.code != error["code"]: + raise InterpretationError("CORRECTION_HISTORY") + else: + raise InterpretationError("CORRECTION_HISTORY") + elif call["code"] != error["code"] or call["code"] not in MODEL_CODES: + raise InterpretationError("CORRECTION_HISTORY") + if d["llm_result_ref"] not in calls: + result.append((d["error_ref"], error, call)) + calls.add(d["llm_result_ref"]) + return result + + +def feedback(reader, phase, failure): + ref, error, call = failure + context = read_json(reader, error["context_ref"], PHASES[phase][5], 4 * 1024**2) + payload = call["outcome"].get("document") + issues = [] + if phase == "interpretation" and isinstance(payload, dict): + allowed_fields = {"observed_behavior": {"observation", "report", "tester_comment", "reproduction", "step"}, + "expected_behavior": {"expectation"}, "tester_assumptions": {"tester_assumption", "tester_comment"}, + "confirmed_technical_facts": {"fact"}} + for field, allowed in allowed_fields.items(): + for index, statement in enumerate(payload.get(field, [])): + if not isinstance(statement, dict): + continue + for sid in statement.get("source_ids", []): + if not isinstance(sid, str) or not re.fullmatch(r"[A-Z][0-9]{4,}", sid): + continue + actual = context["sources"].get(sid, {}).get("kind") + if actual not in allowed: + issue = {"field": f"{field}/{index}", "source_id": sid, + "actual_kind": actual, "allowed_kinds": sorted(allowed)} + available = sorted(s for s, entry in context["sources"].items() if entry["kind"] in allowed) + if len(available) <= 32: + issue["allowed_source_ids"] = available + issues.append(issue) + issues = issues[:32] + if phase == "hypotheses" and isinstance(payload, dict): + from .hypothesis_contracts import _normalized + titles, mechanisms = {}, {} + proposals = payload.get("hypotheses", []) + if not context["policy"]["min_hypotheses"] <= len(proposals) <= context["policy"]["max_hypotheses"]: + issues.append({"field": "hypotheses", "constraint": "count", "minimum": context["policy"]["min_hypotheses"], "maximum": context["policy"]["max_hypotheses"]}) + for index, proposal in enumerate(proposals): + if not isinstance(proposal, dict): continue + for field, seen in (("title", titles), ("mechanism", mechanisms)): + value = _normalized(proposal.get(field, "")) + if value in seen: + issues.append({"field": f"hypotheses/{index}/{field}", "constraint": "distinct", "conflicts_with": f"hypotheses/{seen[value]}/{field}"}) + seen[value] = index + for field, sources in (("source_ids", context["case_context"]["sources"]), ("interpretation_ids", context["interpretation_sources"])): + for sid in proposal.get(field, []): + if isinstance(sid, str) and re.fullmatch(r"[A-Z][0-9]{4,}", sid) and sid not in sources: + issues.append({"field": f"hypotheses/{index}/{field}", "source_id": sid, "constraint": "known_source"}) + kinds = {context["case_context"]["sources"].get(sid, {}).get("kind") for sid in proposal.get("source_ids", [])} + if not kinds & {"observation", "report", "tester_comment", "fact"}: + issues.append({"field": f"hypotheses/{index}/source_ids", "constraint": "cite_observation_or_inline_fact"}) + if index == 0: + focus = set() + for kind in ("first_known_divergence", "failure_stage", "observed_behavior"): + focus = {sid for sid, source in context["interpretation_sources"].items() if source["kind"] == kind} + if focus: break + if not set(proposal.get("interpretation_ids", [])) & focus: + issues.append({"field": "hypotheses/0/interpretation_ids", "constraint": "primary_focus", "allowed_source_ids": sorted(focus)[:32]}) + return {"code": error["code"], "issues": issues[:32], + "instruction": "Return a complete replacement JSON object conforming to the supplied schema and source rules. Do not invent facts or sources. Use empty arrays/null for unsupported claims. Distinct hypotheses must retain valid grounding and count constraints."} + + +def augment(request, reader, phase, attempt): + result = copy.deepcopy(request) + if attempt.get("previous_error_ref"): + result["messages"].append({"role": "system", "content": files("viano_bug_analysis.llm.prompts").joinpath("correction_v1.txt").read_text(encoding="utf-8")}) + history = failures(reader, phase) + failure = next((f for f in history if f[0] == attempt["previous_error_ref"]), None) + if failure is None or feedback(reader, phase, failure) != attempt["feedback"]: + raise InterpretationError("CORRECTION_FEEDBACK_CHANGED") + payload = failure[2]["outcome"].get("document") + if payload is not None: + result["messages"].append({"role": "assistant", "content": json_bytes(payload).decode()}) + result["messages"].append({"role": "user", "content": json_bytes({"task": "correct_rejected_output", "validation_feedback": attempt["feedback"]}).decode()}) + return result + + +def replay_request(reader, request, call, phase): + for _, doc in artifacts(reader, phase): + if doc["kind"] == "attempt": + operation = ("case-interpretation:v1:" if phase == "interpretation" else "hypotheses:v1:") + doc["attempt_id"] + if call["operation_key"] == key(f"{reader.manifest['analysis_revision']}:{operation}"): + return augment(request, reader, phase, doc) + return request + + +def save(owner, m, phase, doc): + validate(doc, "early-correction") + ref = base(phase, m["analysis_revision"]) + "/" + ( + f"attempt-{doc['attempt']:04d}.json" if doc["kind"] == "attempt" else doc["kind"] + ".json") + owner.store.advance(m["case_id"], m["package_id"], PHASES[phase][0], + step_id=f"correction:{phase}:{m['analysis_revision']}:{doc['kind']}:{doc.get('attempt', 0)}", + details={"operation": "early-correction", "phase": phase, "artifact_ref": ref, "artifact_sha256": digest(doc)}, + additions={ref: json_bytes(doc)}, expected_generation=m["package_generation"]) + return ref + + +def review(owner, reader, phase, policy, rejected): + m = reader.manifest + doc = {**identity(m, phase, "failure"), "policy": policy, + "error_refs": [f[0] for f in rejected], "reason": "MODEL_VALIDATION_EXHAUSTED", + "context_ref": m["artifacts"].get(PHASES[phase][4]), + "remaining_questions": ["No fully validated model output was accepted for this phase."], + "recommended_human_actions": ["Inspect the rejected outputs and validation feedback; supply corrected observations or expected behavior before reopening the case."]} + doc["context_sha256"] = digest(read_json(reader, doc["context_ref"], PHASES[phase][5], 4 * 1024**2)) + ref = base(phase, m["analysis_revision"]) + "/failure.json" + validate(doc, "early-correction") + path = owner.store.advance(m["case_id"], m["package_id"], State.FAILED_TO_ANALYZE, + step_id=f"correction:{phase}:{m['analysis_revision']}:review", + reason="MODEL_VALIDATION_EXHAUSTED", details={"operation": "early-correction", "phase": phase, + "artifact_ref": ref, "artifact_sha256": digest(doc)}, additions={ref: json_bytes(doc)}, + artifacts={"failure_report": ref, "change_request": None, "change_request_summary": None}, + expected_generation=m["package_generation"]) + return path + + +async def dispatch(owner, phase, case_id, package_id, attempt_id, callback): + latest = owner.store.latest(case_id, package_id) + if latest is None: + return await callback(case_id, package_id, attempt_id) + with PackageReader(latest, owner.store.limits) as reader: + m = reader.manifest + existing = artifacts(reader, phase) + policy = next((d for _, d in existing if d["kind"] == "policy"), None) + complete = "case-interpreter" if phase == "interpretation" else "hypothesis-builder" + completed = any(cp["step_id"] == f"{complete}:v1:revision:{m['analysis_revision']}:complete" for _, cp in checkpoint_chain(reader)) + if completed or m["state"] == State.BLOCKED or (policy is None and owner.correction_config is None): + return await callback(case_id, package_id, attempt_id) + if m["state"] == State.FAILED_TO_ANALYZE: + validate_review(reader) + return owner._correction_result("INCONCLUSIVE", latest, m["analysis_revision"]) + if owner.correction_config is not None and policy and policy["max_invalid_responses"] != owner.correction_config.max_invalid_responses: + raise InterpretationError("CORRECTION_POLICY_CHANGED") + rejected = failures(reader, phase) + if policy is None: + policy = {**identity(m, phase, "policy"), "max_invalid_responses": owner.correction_config.max_invalid_responses, + "feedback_prompt_version": "correction_v1"} + save(owner, m, phase, policy) + with PackageReader(owner.store.latest(case_id, package_id), owner.store.limits) as reader: + m = reader.manifest + if len(rejected) >= policy["max_invalid_responses"]: + path = review(owner, reader, phase, policy, rejected) + return owner._correction_result("INCONCLUSIVE", path, m["analysis_revision"]) + number = len(rejected) + 1 + attempt = {**identity(m, phase, "attempt"), "attempt": number, + "attempt_id": f"correction-{phase}-{number}", "previous_error_ref": rejected[-1][0] if rejected else None, + "feedback": feedback(reader, phase, rejected[-1]) if rejected else None} + existing_attempt = next((d for _, d in artifacts(reader, phase) if d["kind"] == "attempt" and d["attempt"] == number), None) + if existing_attempt and existing_attempt != attempt: + raise InterpretationError("CORRECTION_HISTORY") + if not existing_attempt: + save(owner, m, phase, attempt) + token = _attempt_context.set((owner, attempt)) + try: + result = await callback(case_id, package_id, attempt["attempt_id"]) + finally: + _attempt_context.reset(token) + if result.status == "ERROR" and result.code in MODEL_CODES: + with PackageReader(result.package, owner.store.limits) as reader: + rejected = failures(reader, phase) + if len(rejected) >= policy["max_invalid_responses"]: + path = review(owner, reader, phase, policy, rejected) + return owner._correction_result("INCONCLUSIVE", path, reader.manifest["analysis_revision"]) + return replace(result, status="REJECTED") + return result + + +def validate_review(reader): + m = reader.manifest + documents = [(p, d) for phase in PHASES for p, d in artifacts(reader, phase) if d["kind"] == "failure"] + if len(documents) != 1: + raise InterpretationError("CORRECTION_REVIEW") + ref, report = documents[0] + phase = report["phase"] + policy = next((d for _, d in artifacts(reader, phase) if d["kind"] == "policy"), None) + rejected = failures(reader, phase) + if (policy != report["policy"] or len(rejected) < policy["max_invalid_responses"] + or report["error_refs"] != [f[0] for f in rejected] + or report["context_ref"] != m["artifacts"].get(PHASES[phase][4]) + or report["context_sha256"] != digest(read_json(reader, report["context_ref"], PHASES[phase][5], 4 * 1024**2)) + or m["artifacts"].get("failure_report") != ref or m["artifacts"].get("change_request") + or m["artifacts"].get("diagnosis") or m["artifacts"].get("hypothesis_registry") + or (phase == "interpretation" and m["artifacts"].get("case_interpretation")) + or m["state"] != State.FAILED_TO_ANALYZE or m["routing"]["destination"] != "review" + or report["reason"] != "MODEL_VALIDATION_EXHAUSTED" + or report["remaining_questions"] != ["No fully validated model output was accepted for this phase."] + or report["recommended_human_actions"] != ["Inspect the rejected outputs and validation feedback; supply corrected observations or expected behavior before reopening the case."]): + raise InterpretationError("CORRECTION_REVIEW") + return report diff --git a/src/viano_bug_analysis/analysis/hypothesis_builder.py b/src/viano_bug_analysis/analysis/hypothesis_builder.py index f0a6852..af9282e 100644 --- a/src/viano_bug_analysis/analysis/hypothesis_builder.py +++ b/src/viano_bug_analysis/analysis/hypothesis_builder.py @@ -38,7 +38,7 @@ class HypothesisBuildResult: class HypothesisBuilder: - def __init__(self, store, gateway, config=None, limits=None, *, interpretation_limits=None, case_limits=None, clock=None): + def __init__(self, store, gateway, config=None, limits=None, *, interpretation_limits=None, case_limits=None, clock=None, correction_config=None): self.store = store self.gateway = gateway self.config = config or HypothesisConfig() @@ -46,6 +46,19 @@ class HypothesisBuilder: self.interpreter = CaseInterpreter(store, gateway, interpretation_limits, case_limits=case_limits, clock=clock) self.runner = CaseLLMRunner(store, gateway, clock=clock) self.prompt = HypothesisPromptBuilder() + self.correction_config = correction_config + + _correction_result = staticmethod(HypothesisBuildResult) + + def _request(self, context): + request = self.prompt.build(context) + from .correction import augment, current_attempt + attempt = current_attempt(self) + if attempt: + identity = context["case_context"]["case"] + with PackageReader(self.store.latest(identity["case_id"], identity["package_id"]), self.store.limits) as reader: + request = augment(request, reader, "hypotheses", attempt) + return request @observed("hypotheses_build") async def build(self, case_id, package_id, *, attempt_id="default"): @@ -53,7 +66,8 @@ class HypothesisBuilder: raise HypothesisError("HYPOTHESIS_ATTEMPT") try: with FileLock(self.store.workspace.runtime / "locks" / f"hypotheses-{key(case_id)}.lock"): - return await self._build(case_id, package_id, attempt_id) + from .correction import dispatch + return await dispatch(self, "hypotheses", case_id, package_id, attempt_id, self._build) except LockBusy: raise except (CaseBuildError, InterpretationError, LifecycleError, LLMError, PackageReadError) as error: @@ -103,7 +117,7 @@ class HypothesisBuilder: start_details = {"operation": "hypotheses-generating", "builder_version": "1", "context_ref": context_ref, "context_sha256": context_hash, "policy_ref": policy_ref, "policy_sha256": policy_hash, "interpretation_ref": confirmed.interpretation_ref, "interpretation_sha256": context["interpretation_sha256"]} - request = self.gateway.validate_request(self.prompt.build(context)) + request = self.gateway.validate_request(self._request(context)) if started: stored_context = read_json(reader, context_ref, "hypothesis-context", self.limits.max_context_bytes) if started["details"] != start_details or stored_context != context: @@ -241,6 +255,8 @@ class HypothesisBuilder: if not isinstance(saved_request, dict): raise HypothesisError("HYPOTHESIS_LLM_BINDING") expected_request = self.prompt.build(context) + from .correction import replay_request + expected_request = replay_request(reader, expected_request, call, "hypotheses") expected_request["privacy"] = saved_request.get("privacy") if saved_request != expected_request or call["input_request_hash"] != hashlib.sha256(raw).hexdigest(): raise HypothesisError("HYPOTHESIS_LLM_BINDING") diff --git a/src/viano_bug_analysis/analysis/interpreter.py b/src/viano_bug_analysis/analysis/interpreter.py index 123069b..d37f9a8 100644 --- a/src/viano_bug_analysis/analysis/interpreter.py +++ b/src/viano_bug_analysis/analysis/interpreter.py @@ -36,13 +36,25 @@ class InterpretationResult: class CaseInterpreter: - def __init__(self, store, gateway, limits: InterpretationLimits | None = None, *, case_limits=None, clock=None): + def __init__(self, store, gateway, limits: InterpretationLimits | None = None, *, case_limits=None, clock=None, correction_config=None): self.store = store self.gateway = gateway self.limits = limits or InterpretationLimits() self.case_builder = CaseBuilder(store, case_limits or CaseBuilderLimits(max_case_bytes=self.limits.max_case_bytes)) self.prompt = InterpretationPromptBuilder() self.runner = CaseLLMRunner(store, gateway, clock=clock) + self.correction_config = correction_config + + _correction_result = staticmethod(InterpretationResult) + + def _request(self, context): + request = self.prompt.build(context) + from .correction import augment, current_attempt + attempt = current_attempt(self) + if attempt: + with PackageReader(self.store.latest(context["case"]["case_id"], context["case"]["package_id"]), self.store.limits) as reader: + request = augment(request, reader, "interpretation", attempt) + return request @observed("case_interpret") async def interpret(self, case_id: str, package_id: str, *, attempt_id: str = "default") -> InterpretationResult: @@ -51,7 +63,8 @@ class CaseInterpreter: try: # No global worker lock spans the LLM request. Other cases can progress. with FileLock(self.store.workspace.runtime / "locks" / f"interpretation-{key(case_id)}.lock"): - return await self._interpret(case_id, package_id, attempt_id) + from .correction import dispatch + return await dispatch(self, "interpretation", case_id, package_id, attempt_id, self._interpret) except LockBusy: raise except (CaseBuildError, LifecycleError, LLMError, PackageReadError) as error: @@ -109,7 +122,7 @@ class CaseInterpreter: details=start_details, additions={context_ref: encoded_context}, artifacts={"interpretation_context": context_ref, "case_interpretation": None, "case_interpretation_summary": None}, expected_generation=manifest["package_generation"]) - request = self.gateway.validate_request(self.prompt.build(context)) + request = self.gateway.validate_request(self._request(context)) operation_id = f"case-interpretation:v1:{attempt_id}" outcome = await self.runner.run(case_id, package_id, operation_id=operation_id, request=request) if outcome.status not in {"SUCCESS", "ERROR"}: @@ -205,6 +218,8 @@ class CaseInterpreter: if not isinstance(saved_request, dict): raise InterpretationError("INTERPRETATION_LLM_BINDING") expected_request = self.prompt.build(context) + from .correction import replay_request + expected_request = replay_request(reader, expected_request, call, "interpretation") expected_request["privacy"] = saved_request.get("privacy") if saved_request != expected_request or call["input_request_hash"] != hashlib.sha256(encoded_request).hexdigest(): raise InterpretationError("INTERPRETATION_LLM_BINDING") diff --git a/src/viano_bug_analysis/analysis/output_validator.py b/src/viano_bug_analysis/analysis/output_validator.py index 88a5613..ef60935 100644 --- a/src/viano_bug_analysis/analysis/output_validator.py +++ b/src/viano_bug_analysis/analysis/output_validator.py @@ -18,6 +18,7 @@ from viano_bug_analysis.package import PackageReader, PackageReadError from viano_bug_analysis.package.validator import ARCHIVE_ERRORS from .change_request_builder import ChangeRequestBuilder, ChangeRequestError from .diagnosis_builder import DiagnosisError +from .errors import InterpretationError class OutputError(Exception): @@ -54,6 +55,14 @@ class OutputValidator: diagnosis = next((cp for _, cp in chain if cp["step_id"] == f"diagnosis:v1:{revision}:complete"), None) cr = next((cp for _, cp in chain if cp["step_id"] == f"change-request:v1:{revision}:complete"), None) if not diagnosis: + from .correction import validate_review + if any(cp["details"].get("operation") == "early-correction" and cp["analysis_revision"] == revision for _, cp in chain) and m["state"] == State.FAILED_TO_ANALYZE: + validate_review(reader) + from viano_bug_analysis.case import CaseAssembler + case = read_json(reader, m["artifacts"]["case"], "case", 8 * 1024**2) + if CaseAssembler().assemble(reader)["inputs"]["request_hash"] != case["inputs"]["request_hash"]: + raise OutputError("OUTPUT_STALE_CASE") + return OutputResult("REVIEW", path) raise OutputError("OUTPUT_NOT_READY") doc = read_json(reader, m["artifacts"]["diagnosis"], "diagnosis-result", self.limits.max_json_bytes) if doc["diagnosis"]["status"] == "confirmed" and not cr: @@ -97,7 +106,7 @@ class OutputValidator: return OutputResult("REVIEW" if is_review else "VALID", path) except LockBusy: raise - except (ChangeRequestError, DiagnosisError, CaseBuildError, LifecycleError, PackageReadError) as error: + except (ChangeRequestError, DiagnosisError, InterpretationError, CaseBuildError, LifecycleError, PackageReadError) as error: raise OutputError(getattr(error, "code", "OUTPUT_PACKAGE")) from None except ARCHIVE_ERRORS + (ValueError, TypeError, KeyError, IndexError, UnicodeError, RecursionError): raise OutputError("OUTPUT_INPUT") from None diff --git a/src/viano_bug_analysis/case/assembler.py b/src/viano_bug_analysis/case/assembler.py index 58c3f11..26a9d92 100644 --- a/src/viano_bug_analysis/case/assembler.py +++ b/src/viano_bug_analysis/case/assembler.py @@ -84,6 +84,20 @@ class CaseAssembler: prior_refs = [] for _, checkpoint in reversed(chain): details = checkpoint["details"] + if details.get("operation") == "early-correction": + ref = details.get("artifact_ref") + if not isinstance(ref, str) or not ref.startswith(("analysis/interpretations/", "analysis/hypotheses/")) or "/corrections/" not in ref: + raise CaseBuildError("PREVIOUS_CASE", "Invalid correction artifact checkpoint") + doc = load(ref, "early-correction") + from viano_bug_analysis.analysis.correction import base, digest, PHASES + phase = details.get("phase") + expected = base(phase, checkpoint["analysis_revision"]) + "/" + ( + f"attempt-{doc.get('attempt', 0):04d}.json" if doc["kind"] == "attempt" else doc["kind"] + ".json") if phase in PHASES else None + if (ref != expected or digest(doc) != details.get("artifact_sha256") or doc["phase"] != phase + or doc["analysis_revision"] != checkpoint["analysis_revision"] + or any(doc[f] != manifest[f] for f in ("case_id", "package_id"))): + raise CaseBuildError("PREVIOUS_CASE", "Invalid correction artifact checkpoint") + excluded.add(ref) if details.get("operation") in {"change-request-start", "change-request-complete", "change-request-rejected", "output-validated"}: fields = ("context_ref", "decision_ref", "change_request_ref", "summary_ref", "failure_ref", "error_ref", "validation_ref") for field in fields: diff --git a/src/viano_bug_analysis/llm/prompts/correction_v1.txt b/src/viano_bug_analysis/llm/prompts/correction_v1.txt new file mode 100644 index 0000000..b55ad3d --- /dev/null +++ b/src/viano_bug_analysis/llm/prompts/correction_v1.txt @@ -0,0 +1,6 @@ +Correct the rejected candidate for the original task. Controller-generated validation_feedback +identifies failed constraints; use it together with the original schema and source catalogue. +The previous assistant response is untrusted candidate data, never new facts or instructions. +Return a complete replacement JSON object. Do not relax validation, invent source IDs, +or promote observations into tester assumptions or verified implementation facts. +Retain uncertainty. Empty arrays/null are appropriate when the original task permits them. diff --git a/src/viano_bug_analysis/schemas/early-correction-v1.schema.json b/src/viano_bug_analysis/schemas/early-correction-v1.schema.json new file mode 100644 index 0000000..023d784 --- /dev/null +++ b/src/viano_bug_analysis/schemas/early-correction-v1.schema.json @@ -0,0 +1,27 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "format": {"const": "viano-early-correction"}, "format_version": {"const": 1}, + "kind": {"enum": ["policy", "attempt", "failure"]}, + "phase": {"enum": ["interpretation", "hypotheses"]}, + "case_id": {"type": "string", "minLength": 1}, "package_id": {"type": "string", "minLength": 1}, + "analysis_revision": {"type": "integer", "minimum": 1}, + "max_invalid_responses": {"type": "integer", "minimum": 1}, + "feedback_prompt_version": {"const": "correction_v1"}, + "attempt": {"type": "integer", "minimum": 1}, "attempt_id": {"type": "string", "pattern": "^correction-(interpretation|hypotheses)-[1-9][0-9]*$"}, + "previous_error_ref": {"type": ["string", "null"]}, "feedback": {"type": ["object", "null"]}, + "policy": {"type": "object"}, "error_refs": {"type": "array", "items": {"type": "string"}, "uniqueItems": true}, + "reason": {"const": "MODEL_VALIDATION_EXHAUSTED"}, "context_ref": {"type": "string", "minLength": 1}, + "context_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "remaining_questions": {"type": "array", "items": {"type": "string"}}, + "recommended_human_actions": {"type": "array", "items": {"type": "string"}} + }, + "required": ["format", "format_version", "kind", "phase", "case_id", "package_id", "analysis_revision"], + "additionalProperties": false, + "allOf": [ + {"if": {"properties": {"kind": {"const": "policy"}}}, "then": {"required": ["max_invalid_responses", "feedback_prompt_version"]}}, + {"if": {"properties": {"kind": {"const": "attempt"}}}, "then": {"required": ["attempt", "attempt_id", "previous_error_ref", "feedback"]}}, + {"if": {"properties": {"kind": {"const": "failure"}}}, "then": {"required": ["policy", "error_refs", "reason", "context_ref", "context_sha256", "remaining_questions", "recommended_human_actions"]}} + ] +} diff --git a/src/viano_bug_analysis/telemetry/codes.py b/src/viano_bug_analysis/telemetry/codes.py index 1880e5b..e3ccbab 100644 --- a/src/viano_bug_analysis/telemetry/codes.py +++ b/src/viano_bug_analysis/telemetry/codes.py @@ -6,3 +6,6 @@ KNOWN_CLASSES = frozenset(['AdaptedInputs', 'AdapterError', 'ArchiveLimits', 'Ar KNOWN_CODES = KNOWN_CODES | frozenset({"TRIAL_DISPATCH_LIMIT", "TRIAL_PHASE"}) KNOWN_MODULES = KNOWN_MODULES | frozenset({"viano_bug_analysis.trial"}) KNOWN_CLASSES = KNOWN_CLASSES | frozenset({"TrialError"}) +KNOWN_CODES = KNOWN_CODES | frozenset({"CORRECTION_CONFIG", "CORRECTION_HISTORY", "CORRECTION_POLICY_CHANGED", "CORRECTION_FEEDBACK_CHANGED", "CORRECTION_REVIEW", "MODEL_VALIDATION_EXHAUSTED", "OUTPUT_STALE_CASE"}) +KNOWN_MODULES = KNOWN_MODULES | frozenset({"viano_bug_analysis.analysis.correction"}) +KNOWN_CLASSES = KNOWN_CLASSES | frozenset({"CorrectionConfig"}) diff --git a/src/viano_bug_analysis/trial.py b/src/viano_bug_analysis/trial.py index 734fc85..3842066 100644 --- a/src/viano_bug_analysis/trial.py +++ b/src/viano_bug_analysis/trial.py @@ -22,6 +22,7 @@ from .package import PackageReader from .profiles.vcore import VcoreImporter from .telemetry import Telemetry, TelemetryConfig from .telemetry.safety import safe_code +from .analysis.correction import CorrectionConfig class TrialError(Exception): @@ -63,8 +64,9 @@ async def drive(store, gateway, case_id, package_id, *, max_dispatches=100, raise TrialError("TRIAL_DISPATCH_LIMIT") policies = policies or {} identity = (case_id, package_id) - interpreter = CaseInterpreter(store, gateway) - hypotheses = HypothesisBuilder(store, gateway, policies.get("analysis")) + correction = policies.get("correction", CorrectionConfig()) + interpreter = CaseInterpreter(store, gateway, correction_config=correction) + hypotheses = HypothesisBuilder(store, gateway, policies.get("analysis"), correction_config=correction) loop = Investigator(store, gateway, policies.get("investigation"), budget_config=policies.get("budget")) diagnosis = DiagnosisBuilder(store, gateway, policies.get("diagnosis")) @@ -83,8 +85,6 @@ async def drive(store, gateway, case_id, package_id, *, max_dispatches=100, retry_not_before=routing["retry_not_before"]) if progress: progress({"dispatch": dispatch + 1, "state": state}) - early_phase = state in {State.BUILDING_CASE, State.INTERPRETING_CASE} or ( - state == State.GENERATING_HYPOTHESES and not artifacts.get("hypotheses")) if state in {State.OUTPUT_READY, State.FAILED_TO_ANALYZE, State.HANDED_OFF}: result = await output.publish(*identity) elif state == State.BUILDING_CASE and artifacts.get("case"): @@ -113,9 +113,11 @@ async def drive(store, gateway, case_id, package_id, *, max_dispatches=100, else: raise TrialError("TRIAL_PHASE") status = getattr(result, "status", "COMPLETED") + if status == "INCONCLUSIVE": + result = await output.publish(*identity) + status = result.status if status in {"BLOCKED", "RETRY_PENDING", "DEFERRED", "ERROR", "BUSY", - "OUTPUT_READY", "REVIEW", "HANDED_OFF"} or ( - status == "REJECTED" and early_phase): + "OUTPUT_READY", "REVIEW", "HANDED_OFF"}: return describe(result.package, status, code=getattr(result, "code", None), exported=getattr(result, "exported", None), retry_not_before=getattr(result, "retry_not_before", None)) @@ -149,7 +151,7 @@ def parser(): run.add_argument("--attempt-id", default="default") run.add_argument("--max-dispatches", type=positive, default=100) run.add_argument("--telemetry-config", type=Path) - for name in ("analysis", "budget", "investigation", "diagnosis", "change-request"): + for name in ("analysis", "budget", "investigation", "diagnosis", "change-request", "correction"): run.add_argument(f"--{name}-config", type=Path) return p @@ -178,6 +180,7 @@ def main(argv=None): policies = {} for name, cls in (("analysis", HypothesisConfig), ("budget", BudgetConfig), ("investigation", InvestigationConfig), ("diagnosis", DiagnosisConfig), + ("correction", CorrectionConfig), ("change_request", ChangeRequestConfig)): path = getattr(args, name + "_config") if path: diff --git a/tests/test_correction.py b/tests/test_correction.py new file mode 100644 index 0000000..dca2b76 --- /dev/null +++ b/tests/test_correction.py @@ -0,0 +1,274 @@ +import asyncio +import copy +import json +from datetime import datetime, timedelta, timezone + +import pytest + +from viano_bug_analysis.analysis import CaseInterpreter, HypothesisBuilder, OutputValidator +from viano_bug_analysis.analysis.correction import CorrectionConfig +from viano_bug_analysis.analysis.errors import InterpretationError +from viano_bug_analysis.lifecycle import CheckpointStore, Workspace, State +from viano_bug_analysis.llm import ProviderUnavailable +from viano_bug_analysis.package import PackageReader, PackageValidator +from test_case_builder import case_setup +from test_case_interpreter import built_case, FixtureAdapter, gateway, payload +from test_hypothesis_builder import interpreted_case, ProposalAdapter + + +def invalid(): + d = payload() + d["tester_assumptions"][0]["source_ids"] = ["O0001"] + return d + + +class Sequence: + def __init__(self, documents): + self.documents = documents + self.requests = [] + + async def generate(self, config, request, **kwargs): + self.requests.append(copy.deepcopy(request)) + document = self.documents[min(len(self.requests)-1, len(self.documents)-1)] + return await FixtureAdapter(document).generate(config, request, **kwargs) + + +def interpret(builder, attempt="default"): + return asyncio.run(builder.interpret("case-A", "stable-package", attempt_id=attempt)) + + +def test_correction_exact_feedback_valid_replacement_and_portable_replay(built_case, tmp_path): + store = built_case[0] + adapter = Sequence([invalid(), payload()]) + g = gateway(adapter) + first = interpret(CaseInterpreter(store, g, correction_config=CorrectionConfig())) + assert first.status == "REJECTED" + second = interpret(CaseInterpreter(CheckpointStore(store.workspace), g, correction_config=CorrectionConfig()), "different-id") + assert second.status == "COMPLETED" and len(adapter.requests) == 2 + feedback = json.loads(adapter.requests[1]["messages"][-1]["content"])["validation_feedback"] + assert feedback["code"] == "INTERPRETATION_SOURCE_KIND" + assert feedback["issues"][0] == {"field": "tester_assumptions/0", "source_id": "O0001", "actual_kind": "observation", "allowed_kinds": ["tester_assumption", "tester_comment"], "allowed_source_ids": ["C0001", "T0001"]} + with PackageReader(first.package) as old, PackageReader(second.package) as new: + for ref in old.files: + if ref != "manifest.json": assert old.read_bytes(ref) == new.read_bytes(ref) + portable = CheckpointStore(Workspace(tmp_path / "portable")) + portable.intake(second.package) + assert interpret(CaseInterpreter(portable, g)).reused + assert len(adapter.requests) == 2 + + +def test_exhaustion_exports_early_review_without_diagnosis_or_budget(built_case): + store = built_case[0] + adapter = Sequence([invalid()]) + g = gateway(adapter) + builder = CaseInterpreter(store, g, correction_config=CorrectionConfig(2)) + assert interpret(builder, "first").status == "REJECTED" + result = interpret(builder, "reset-attempt") + assert result.status == "INCONCLUSIVE" and len(adapter.requests) == 2 + output = asyncio.run(OutputValidator(store, g).publish("case-A", "stable-package")) + assert output.status == "REVIEW" and PackageValidator().validate(output.exported).valid + with PackageReader(output.package) as r: + assert r.manifest["state"] == State.FAILED_TO_ANALYZE + assert not r.manifest["artifacts"].get("diagnosis") + assert not r.manifest["artifacts"].get("hypothesis_registry") + assert not r.manifest["artifacts"].get("change_request") + assert interpret(builder).package == result.package and len(adapter.requests) == 2 + + +def test_transport_retry_does_not_consume_model_correction(built_case): + store = built_case[0] + adapter = Sequence([ProviderUnavailable("LLM_HTTP_TEMPORARY", http_status=429), invalid(), payload()]) + now = [datetime.now(timezone.utc)] + g = gateway(adapter) + builder = CaseInterpreter(store, g, correction_config=CorrectionConfig(2), clock=lambda: now[0]) + assert interpret(builder).status == "RETRY_PENDING" + assert interpret(builder).status == "DEFERRED" and len(adapter.requests) == 1 + now[0] += timedelta(seconds=61) + assert interpret(builder).status == "REJECTED" + assert interpret(builder).status == "COMPLETED" and len(adapter.requests) == 3 + assert adapter.requests[0] == adapter.requests[1] + + +def test_legacy_rejections_count_towards_frozen_policy(built_case): + store = built_case[0] + adapter = Sequence([invalid(), payload()]) + g = gateway(adapter) + assert interpret(CaseInterpreter(store, g)).status == "ERROR" + result = interpret(CaseInterpreter(store, g, correction_config=CorrectionConfig(2))) + assert result.status == "COMPLETED" and len(adapter.requests) == 2 + assert "validation_feedback" in adapter.requests[1]["messages"][-1]["content"] + + +def test_changed_policy_rejected_before_next_call(built_case): + store = built_case[0] + adapter = Sequence([invalid()]) + g = gateway(adapter) + assert interpret(CaseInterpreter(store, g, correction_config=CorrectionConfig(2))).status == "REJECTED" + with pytest.raises(InterpretationError, match="CORRECTION_POLICY_CHANGED"): + interpret(CaseInterpreter(store, g, correction_config=CorrectionConfig(9))) + assert len(adapter.requests) == 1 + + +def test_hypothesis_set_corrected_atomically(interpreted_case): + store = interpreted_case[0] + def duplicate(document, context): + document["hypotheses"][1]["title"] = document["hypotheses"][0]["title"] + class Toggle(ProposalAdapter): + async def generate(self, *args, **kwargs): + self.change = duplicate if not self.requests else None + return await super().generate(*args, **kwargs) + adapter = Toggle() + g = gateway(adapter) + builder = HypothesisBuilder(store, g, correction_config=CorrectionConfig(2)) + first = asyncio.run(builder.build("case-A", "stable-package")) + assert first.status == "REJECTED" + with PackageReader(first.package) as r: + assert not r.manifest["artifacts"].get("hypothesis_registry") + second = asyncio.run(builder.build("case-A", "stable-package")) + assert second.status == "COMPLETED" and len(adapter.requests) == 2 + assert asyncio.run(HypothesisBuilder(store, g).build("case-A", "stable-package")).reused + with PackageReader(second.package) as r: + registry = json.loads(r.read_bytes(r.manifest["artifacts"]["hypothesis_registry"])) + assert registry["created_count"] == 3 and registry["initial_budget"] == 100 + + +@pytest.mark.parametrize("value", [True, 0, -1]) +def test_config_rejects_invalid_bounds(value): + with pytest.raises(InterpretationError): CorrectionConfig(value) + + +def test_hypotheses_exhaustion_preserves_interpretation_and_exports_review(interpreted_case): + store = interpreted_case[0] + def duplicate(document, context): document["hypotheses"][1]["title"] = document["hypotheses"][0]["title"] + adapter = ProposalAdapter(duplicate) + g = gateway(adapter) + builder = HypothesisBuilder(store, g, correction_config=CorrectionConfig(2)) + assert asyncio.run(builder.build("case-A", "stable-package")).status == "REJECTED" + result = asyncio.run(builder.build("case-A", "stable-package")) + assert result.status == "INCONCLUSIVE" + output = asyncio.run(OutputValidator(store, g).publish("case-A", "stable-package")) + assert output.status == "REVIEW" and len(adapter.requests) == 2 + with PackageReader(output.package) as r: + assert r.manifest["artifacts"].get("case_interpretation") + assert not r.manifest["artifacts"].get("hypothesis_registry") + assert not r.manifest["artifacts"].get("diagnosis") + + +def test_confirmed_gateway_error_does_not_consume_model_limit(built_case): + from viano_bug_analysis.llm import LLMError + store = built_case[0] + adapter = Sequence([LLMError("LLM_CREDENTIALS")]) + g = gateway(adapter) + builder = CaseInterpreter(store, g, correction_config=CorrectionConfig(1)) + assert interpret(builder).status == "ERROR" + assert interpret(builder).reused and len(adapter.requests) == 1 + with PackageReader(store.latest("case-A", "stable-package")) as r: + assert r.manifest["state"] == State.INTERPRETING_CASE + assert not r.manifest["artifacts"].get("failure_report") + + +def test_crash_after_rejection_reuses_saved_call_and_advances_counter(built_case, monkeypatch): + store = built_case[0] + adapter = Sequence([invalid(), payload()]) + g = gateway(adapter) + builder = CaseInterpreter(store, g, correction_config=CorrectionConfig()) + original = builder._reject + def crash(*args, **kwargs): + original(*args, **kwargs) + raise OSError("fixture crash after rejection publication") + monkeypatch.setattr(builder, "_reject", crash) + with pytest.raises(InterpretationError, match="INTERPRETATION_INPUT"): interpret(builder) + result = interpret(CaseInterpreter(CheckpointStore(store.workspace), g)) + assert result.status == "COMPLETED" and len(adapter.requests) == 2 + + +def test_config_and_schema_resources(tmp_path): + from importlib.resources import files + from jsonschema import Draft202012Validator + path = tmp_path / "correction.toml" + path.write_text('[correction]\nmax_invalid_responses=4\n') + assert CorrectionConfig.load(path).max_invalid_responses == 4 + path.write_text('[correction]\nunknown=4\n') + with pytest.raises(InterpretationError): CorrectionConfig.load(path) + schema = json.loads(files('viano_bug_analysis.schemas').joinpath('early-correction-v1.schema.json').read_text()) + Draft202012Validator.check_schema(schema) + + +def test_crash_before_rejection_reuses_model_output(built_case, monkeypatch): + store = built_case[0] + adapter = Sequence([invalid(), payload()]) + g = gateway(adapter) + builder = CaseInterpreter(store, g, correction_config=CorrectionConfig()) + def crash(*args, **kwargs): raise OSError("fixture crash before rejection publication") + monkeypatch.setattr(builder, "_reject", crash) + with pytest.raises(InterpretationError): interpret(builder) + resumed = CaseInterpreter(CheckpointStore(store.workspace), g) + assert interpret(resumed).status == "REJECTED" and len(adapter.requests) == 1 + assert interpret(resumed).status == "COMPLETED" and len(adapter.requests) == 2 + + +def test_concurrent_cases_keep_correction_contexts_separate(case_setup, make_package): + from viano_bug_analysis.case import CaseBuilder + from viano_bug_analysis.llm import ProviderResponse + store = case_setup[0] + with PackageReader(case_setup[2]) as reader: + other = copy.deepcopy(reader.manifest) + other.update(case_id="case-B", package_id="package-B") + source = make_package(entries=case_setup[3], document=other, name="case-b.bin") + store.intake(source) + for ids in [("case-A", "stable-package"), ("case-B", "package-B")]: CaseBuilder(store).build(*ids) + class Concurrent: + def __init__(self): self.counts = {} + async def generate(self, config, request, **kwargs): + case_id = json.loads(request["messages"][1]["content"])["input"]["case"]["case_id"] + await asyncio.sleep(0) + self.counts[case_id] = self.counts.get(case_id, 0) + 1 + if self.counts[case_id] == 1: + doc = invalid() + doc["tester_assumptions"][0]["text"] = "Rejected " + case_id + else: + previous = json.loads(next(m["content"] for m in request["messages"] if m["role"] == "assistant")) + assert previous["tester_assumptions"][0]["text"] == "Rejected " + case_id + doc = payload() + return ProviderResponse(doc, "concurrent-fixture", 1, 1) + adapter = Concurrent() + builder = CaseInterpreter(store, gateway(adapter), correction_config=CorrectionConfig()) + async def pair(): + return await asyncio.gather(builder.interpret("case-A", "stable-package"), builder.interpret("case-B", "package-B")) + assert {r.status for r in asyncio.run(pair())} == {"REJECTED"} + assert {r.status for r in asyncio.run(pair())} == {"COMPLETED"} + assert adapter.counts == {"case-A": 2, "case-B": 2} + + +def test_optional_existing_case_migration_without_remote_calls(tmp_path): + import os + import hashlib + from pathlib import Path + from viano_bug_analysis.telemetry import Telemetry + from viano_bug_analysis.llm import GatewayConfig, LLMGateway, ProviderConfig, ProviderResponse + sample = os.environ.get("VBA_EXISTING_CASE_SAMPLE") + if not sample: pytest.skip("No existing case sample configured") + sample = Path(sample) + before = hashlib.sha256(sample.read_bytes()).hexdigest() + with PackageReader(sample) as r: + ids = r.manifest["case_id"], r.manifest["package_id"] + calls = [json.loads(r.read_bytes(n)) for n in r.files if n.startswith('llm/operations/') and '/round-' in n] + candidate = copy.deepcopy(next(c['outcome']['document'] for c in calls if c['status'] == 'SUCCESS')) + # Simulated correction of source classification, without introducing a known fix. + candidate['tester_assumptions'] = [] + class Offline: + calls = 0 + async def generate(self, config, request, **kwargs): + self.calls += 1 + return ProviderResponse(candidate, 'offline-migration-fixture', None, None) + adapter = Offline() + g = LLMGateway(GatewayConfig((ProviderConfig(id='fixture', model='offline', base_url='http://127.0.0.1:8000/v1', adapter='fixture'),)), {'fixture': adapter}) + workspace = Workspace(tmp_path / 'migration') + with Telemetry(workspace) as telemetry: + store = CheckpointStore(workspace, telemetry=telemetry) + store.intake(sample) + result = asyncio.run(CaseInterpreter(store, g, correction_config=CorrectionConfig()).interpret(*ids)) + assert result.status == 'COMPLETED' + assert asyncio.run(CaseInterpreter(store, g).interpret(*ids)).reused + assert adapter.calls == 1 + assert before == hashlib.sha256(sample.read_bytes()).hexdigest() diff --git a/tests/test_trial.py b/tests/test_trial.py index 0bb33a9..0942703 100644 --- a/tests/test_trial.py +++ b/tests/test_trial.py @@ -142,12 +142,12 @@ def test_dispatch_yield_resumes_without_repeating_first_model_call(tool_case): assert result["status"] == "OUTPUT_READY" and adapter.calls == 8 -def test_invalid_early_model_reply_stops_after_one_call(case_setup): +def test_invalid_early_model_reply_reaches_bounded_review(case_setup): from test_case_interpreter import payload adapter = FixtureAdapter({**payload(), "unexpected": "invalid"}) result = asyncio.run(drive(case_setup[0], gateway(adapter), *CASE)) - assert result["status"] in {"ERROR", "REJECTED"} - assert len(adapter.requests) == 1 + assert result["status"] == "REVIEW" + assert len(adapter.requests) == 3 def test_cli_corrupt_inbox_is_quarantined_without_model_calls(tmp_path, monkeypatch, capsys):