228 lines
12 KiB
Python
228 lines
12 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.analysis import CaseInterpreter, HypothesisBuilder, HypothesisConfig
|
|
from viano_bug_analysis.analysis.context import build_context, InterpretationLimits
|
|
from viano_bug_analysis.analysis.hypothesis_contracts import build_context as hypothesis_context
|
|
from viano_bug_analysis.analysis.hypothesis_config import HypothesisLimits
|
|
from viano_bug_analysis.analysis.correction import CorrectionConfig
|
|
from viano_bug_analysis.analysis.joint_planning import VERSION, REPAIR
|
|
from viano_bug_analysis.lifecycle.storage import json_bytes
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import built_case, payload, gateway
|
|
from test_hypothesis_builder import proposals
|
|
|
|
|
|
class JointAdapter:
|
|
def __init__(self, change=None):
|
|
self.requests = []
|
|
self.change = change
|
|
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.requests.append(copy.deepcopy(request))
|
|
if request["prompt_version"] in {VERSION,"case_plan_v2"}:
|
|
interpretation = json.loads(json_bytes(payload()))
|
|
# Match the deterministic index created after canonical serialization.
|
|
sources = {}
|
|
for field,value in interpretation.items():
|
|
for index,item in enumerate(value if isinstance(value,list) else [value] if value is not None else []):
|
|
sources[f"I{len(sources)+1:04d}"] = {"kind":field, "interpretation_pointer":f"/interpretation/{field}"+(f"/{index}" if isinstance(value,list) else "")}
|
|
hp = proposals({"interpretation_sources": sources})
|
|
for p in hp["hypotheses"]:
|
|
p["interpretation_paths"] = [sources[k]["interpretation_pointer"] for k in p.pop("interpretation_ids")]
|
|
if request["prompt_version"]=="case_plan_v2":
|
|
for p in hp["hypotheses"]:
|
|
p["premises"]=[{"kind":"grounded","text":"Reported symptom", "source_ids":p["source_ids"],"interpretation_paths":p["interpretation_paths"]}]
|
|
result = {"interpretation":interpretation, "hypotheses":hp}
|
|
if self.change: self.change(result)
|
|
elif request["prompt_version"] == REPAIR:
|
|
result = payload()
|
|
else:
|
|
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
# Plain fixture contexts do not need shared decoding beyond consolidation.
|
|
context = restore(json.loads(request["messages"][1]["content"])["input"])
|
|
result = proposals(context)
|
|
return ProviderResponse(result,"fixture",100,50)
|
|
|
|
|
|
def interpret(store, gw):
|
|
return asyncio.run(CaseInterpreter(store,gw,joint_config=HypothesisConfig(),joint_version=1,
|
|
correction_config=CorrectionConfig()).interpret("case-A","stable-package"))
|
|
|
|
|
|
def hypotheses(store, gw):
|
|
return asyncio.run(HypothesisBuilder(store,gw,correction_config=CorrectionConfig()).build("case-A","stable-package"))
|
|
|
|
|
|
def test_joint_call_separate_artifacts_and_replay(built_case):
|
|
store = built_case[0]
|
|
adapter=JointAdapter();gw=gateway(adapter)
|
|
first=interpret(store,gw)
|
|
assert first.status == "COMPLETED"
|
|
with PackageReader(first.package) as reader:
|
|
before=reader.read_bytes(first.interpretation_ref)
|
|
assert not reader.manifest["artifacts"].get("hypotheses")
|
|
result=hypotheses(store,gw)
|
|
assert result.status == "COMPLETED"
|
|
assert PackageValidator().validate(result.package).valid
|
|
with PackageReader(result.package) as reader:
|
|
assert reader.read_bytes(first.interpretation_ref)==before
|
|
h=json.loads(reader.read_bytes(result.hypotheses_ref))
|
|
i=json.loads(before)
|
|
assert h["llm_result_ref"]==i["llm_result_ref"]
|
|
assert len(h["hypotheses"])==3
|
|
assert h["hypotheses"][0]["proposal"]["interpretation_ids"]
|
|
assert hypotheses(store,gw).reused
|
|
assert asyncio.run(CaseInterpreter(store,gw).interpret("case-A","stable-package")).reused
|
|
assert len(adapter.requests)==1
|
|
request=adapter.requests[0]
|
|
assert request["prompt_version"]==VERSION
|
|
assert "Generate distinct, untested" not in request["messages"][0]["content"]
|
|
|
|
|
|
@pytest.mark.parametrize("bad",["missing","schema","unknown_pointer","duplicate"])
|
|
def test_valid_interpretation_survives_invalid_hypotheses(built_case,bad):
|
|
def change(d):
|
|
if bad=="missing":d.pop("hypotheses")
|
|
elif bad=="schema":d["hypotheses"]="broken"
|
|
elif bad=="unknown_pointer":d["hypotheses"]["hypotheses"][0]["interpretation_paths"]=["/interpretation/unknowns/999"]
|
|
else:d["hypotheses"]["hypotheses"][1]["mechanism"]=d["hypotheses"]["hypotheses"][0]["mechanism"]
|
|
store=built_case[0];adapter=JointAdapter(change);gw=gateway(adapter)
|
|
first=interpret(store,gw)
|
|
assert first.status=="COMPLETED"
|
|
with PackageReader(first.package) as reader:original=reader.read_bytes(first.interpretation_ref)
|
|
rejected=hypotheses(store,gw)
|
|
assert rejected.status=="REJECTED"
|
|
with PackageReader(rejected.package) as reader:
|
|
assert reader.read_bytes(first.interpretation_ref)==original
|
|
assert not reader.manifest["artifacts"].get("hypotheses")
|
|
assert len(adapter.requests)==1
|
|
accepted=hypotheses(store,gw)
|
|
assert accepted.status=="COMPLETED"
|
|
assert len(adapter.requests)==2
|
|
assert adapter.requests[-1]["prompt_version"]=="hypotheses_v4"
|
|
assert hypotheses(store,gw).reused
|
|
|
|
|
|
def test_invalid_interpretation_repairs_only_that_block_and_invalidates_candidates(built_case):
|
|
def change(d):d["interpretation"]["observed_behavior"][0]["source_ids"]=["F0001"]
|
|
store=built_case[0];adapter=JointAdapter(change);gw=gateway(adapter)
|
|
rejected=interpret(store,gw)
|
|
assert rejected.status=="REJECTED"
|
|
with PackageReader(rejected.package) as reader:
|
|
assert not reader.manifest["artifacts"].get("case_interpretation")
|
|
assert not reader.manifest["artifacts"].get("hypotheses")
|
|
fixed=interpret(store,gw)
|
|
assert fixed.status=="COMPLETED"
|
|
assert [r["prompt_version"] for r in adapter.requests]==[VERSION,REPAIR]
|
|
assert "block_contracts" not in adapter.requests[-1]["messages"][1]["content"]
|
|
assert hypotheses(store,gw).status=="COMPLETED"
|
|
assert len(adapter.requests)==3
|
|
assert hypotheses(store,gw).reused
|
|
|
|
|
|
def test_multiple_invalid_interpretation_repairs_exhaust_cleanly(built_case):
|
|
class AlwaysInvalid(JointAdapter):
|
|
async def generate(self,config,request,**kw):
|
|
response=await super().generate(config,request,**kw)
|
|
if request["prompt_version"]==REPAIR:
|
|
response.document["observed_behavior"][0]["source_ids"]=["F0001"]
|
|
return response
|
|
store=built_case[0]
|
|
adapter=AlwaysInvalid(lambda d:d["interpretation"]["observed_behavior"][0].update(source_ids=["F0001"]))
|
|
gw=gateway(adapter)
|
|
assert interpret(store,gw).status=="REJECTED"
|
|
assert interpret(store,gw).status=="REJECTED"
|
|
assert interpret(store,gw).status=="INCONCLUSIVE"
|
|
assert len(adapter.requests)==3
|
|
|
|
|
|
def test_joint_mode_does_not_change_already_frozen_separate_context(built_case):
|
|
from test_case_interpreter import FixtureAdapter
|
|
store=built_case[0];adapter=FixtureAdapter();gw=gateway(adapter)
|
|
first=asyncio.run(CaseInterpreter(store,gw).interpret("case-A","stable-package"))
|
|
assert first.status=="COMPLETED"
|
|
assert interpret(store,gw).reused
|
|
assert len(adapter.requests)==1
|
|
|
|
|
|
def test_schema_normalization_of_each_block_is_independent(built_case):
|
|
def change(d):
|
|
d["hypotheses"]["hypotheses"][0]["expected_evidence"]={"not":"a string list"}
|
|
d["interpretation"]=json.dumps(d["interpretation"])
|
|
store=built_case[0];adapter=JointAdapter(change);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=="COMPLETED"
|
|
assert hypotheses(store,gw).status=="REJECTED"
|
|
assert len(adapter.requests)==1
|
|
|
|
|
|
def test_joint_trial_cli_selects_new_mode_and_preserves_phase_boundaries(built_case,monkeypatch,capsys):
|
|
from viano_bug_analysis import trial
|
|
store=built_case[0];adapter=JointAdapter();gw=gateway(adapter)
|
|
monkeypatch.setattr(trial,"preflight",lambda _:gw)
|
|
args=["run","--workspace",str(store.workspace.root),"--case-id","case-A", "--package-id","stable-package",
|
|
"--planning-mode","joint","--max-dispatches","1"]
|
|
assert trial.main(args)==3
|
|
assert len(adapter.requests)==1
|
|
with PackageReader(store.latest("case-A","stable-package")) as reader:
|
|
assert reader.manifest["artifacts"]["case_interpretation"]
|
|
assert not reader.manifest["artifacts"].get("hypotheses")
|
|
# Omit the mode at resume: the frozen case context still selects the persisted plan.
|
|
args=args[:-4]+["--max-dispatches","1"]
|
|
assert trial.main(args)==3
|
|
assert len(adapter.requests)==1
|
|
with PackageReader(store.latest("case-A","stable-package")) as reader:
|
|
assert reader.manifest["artifacts"]["hypotheses"]
|
|
|
|
|
|
def test_joint_candidate_with_modified_pointer_is_rejected(built_case):
|
|
store=built_case[0];adapter=JointAdapter();gw=gateway(adapter)
|
|
first=interpret(store,gw)
|
|
from viano_bug_analysis.analysis.hypothesis_contracts import build_context
|
|
from viano_bug_analysis.analysis.hypothesis_config import HypothesisLimits, HypothesisError
|
|
from viano_bug_analysis.analysis.joint_planning import joint_call,hypothesis_payload
|
|
with PackageReader(first.package) as reader:
|
|
i=json.loads(reader.read_bytes(first.interpretation_ref))
|
|
c=json.loads(reader.read_bytes(i["context_ref"]))
|
|
hc=build_context(c,i,first.interpretation_ref,HypothesisConfig().policy(),[],HypothesisLimits())
|
|
call=joint_call(reader,hc)
|
|
call["outcome"]["document"]["interpretation"]["observed_behavior"][0]["text"]="Tampered"
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_LLM_BINDING"):
|
|
hypothesis_payload(reader,call,hc)
|
|
|
|
|
|
def test_joint_pipeline_reaches_output_and_reimports_without_another_model_call(case_setup,tmp_path):
|
|
from test_trial import PipelineAdapter
|
|
from viano_bug_analysis.trial import drive
|
|
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore
|
|
from viano_bug_analysis.package import PackageWriter
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
store=case_setup[0]
|
|
incoming=tmp_path/"joint-input.zip"
|
|
index=copy.deepcopy(case_setup[-1])
|
|
index["context"]["source"].append("tool/code.cpp")
|
|
PackageWriter().write(store.latest("case-A","stable-package"),incoming,
|
|
additions={"tool/code.cpp":b"void activate() {\n activate();\n}\n", "tool/index.json":json_bytes(index)},
|
|
manifest_updates={"artifacts":{"case_inputs":"tool/index.json"}})
|
|
store.intake(incoming)
|
|
CaseBuilder(store).build("case-A","stable-package")
|
|
joint=JointAdapter();rest=PipelineAdapter()
|
|
class Adapter:
|
|
async def generate(self,config,request,**kw):
|
|
return await (joint if request["prompt_version"] in {VERSION,"case_plan_v2"} else rest).generate(config,request,**kw)
|
|
gw=gateway(Adapter())
|
|
result=asyncio.run(drive(store,gw,"case-A","stable-package",policies={"planning_mode":"joint"}))
|
|
assert result["status"]=="OUTPUT_READY"
|
|
assert len(joint.requests)==1
|
|
assert all(not a.requests for k,a in rest.adapters.items() if k in {"case","hypotheses"})
|
|
imported=CheckpointStore(Workspace(tmp_path/"recipient"))
|
|
imported.intake(result["exported"])
|
|
assert asyncio.run(CaseInterpreter(imported,gw).interpret("case-A","stable-package")).reused
|
|
assert hypotheses(imported,gw).reused
|
|
assert len(joint.requests)==1
|