278 lines
14 KiB
Python
278 lines
14 KiB
Python
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"]
|
|
document["hypotheses"][1]["mechanism"] = document["hypotheses"][0]["mechanism"]
|
|
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"]
|
|
document["hypotheses"][1]["mechanism"] = document["hypotheses"][0]["mechanism"]
|
|
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()
|