280 lines
15 KiB
Python
280 lines
15 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import built_case, gateway, payload, statement
|
|
from viano_bug_analysis.analysis import CaseInterpreter, InterpretationError
|
|
from viano_bug_analysis.analysis.context import build_context, InterpretationLimits
|
|
from viano_bug_analysis.analysis.entity_resolution import policy, validate_plan
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.lifecycle import CheckpointStore
|
|
from viano_bug_analysis.llm import ProviderResponse, ProviderUnavailable
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.tools import ToolError
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
|
|
|
|
SETTINGS = {"enabled": True, "max_searches": 4, "max_candidates": 8, "max_invalid_responses": 3,
|
|
"catalogs": [{"id": "components", "entity_type": "component", "table": "components",
|
|
"id_column": "id", "name_column": "name", "columns": ["id", "name", "script"]}]}
|
|
PLAN = {"searches": [{"catalog": "components", "term": "switch", "role": "affected_subject", "source_ids": ["O0001"]}]}
|
|
|
|
|
|
class Database:
|
|
def __init__(self, rows=None, truncated=False):
|
|
self.rows = rows if rows is not None else [{"id": 17, "name": "Wall switch", "script": "switch_control"},
|
|
{"id": 42, "name": "Switch operator", "script": "operator_dialog"}]
|
|
self.truncated = truncated
|
|
self.calls = []
|
|
self.unavailable = False
|
|
self.snapshot = "fixture-1"
|
|
|
|
def binding(self):
|
|
return {"kind": "fixture", "snapshot_id": self.snapshot}
|
|
|
|
def validate(self, query):
|
|
assert query["table"] == "components" and query["kind"] == "database.rows"
|
|
if self.unavailable:
|
|
raise ToolError("PROJECT_DATABASE_UNAVAILABLE")
|
|
|
|
def query(self, query):
|
|
self.calls.append(copy.deepcopy(query))
|
|
return {"items": [{"table": "components", "row": r} for r in self.rows], "coverage": "complete_filtered_rows",
|
|
"truncated": self.truncated, "next_offset": 8 if self.truncated else None}
|
|
|
|
|
|
class Model:
|
|
def __init__(self, plan=None, status="selected", candidate="K0001"):
|
|
self.plan = plan or PLAN
|
|
self.status, self.candidate = status, candidate
|
|
self.requests = []
|
|
self.invalid_plan = 0
|
|
self.invalid_assessment = 0
|
|
self.unavailable_stage = None
|
|
self.omit_resolved = 0
|
|
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.requests.append(copy.deepcopy(request))
|
|
version = request["prompt_version"]
|
|
if version == self.unavailable_stage:
|
|
raise ProviderUnavailable("LLM_UNAVAILABLE")
|
|
if version.startswith('entity_resolution_') and version.endswith('_v2'):
|
|
# This is the channel json_object providers actually receive.
|
|
assert json.loads(request['messages'][1]['content'])['output_schema'] == request['output_schema']
|
|
version = version.replace('_v1', '_v2') if version.startswith('entity_resolution_') else version
|
|
if version == "entity_resolution_plan_v2":
|
|
document = copy.deepcopy(self.plan)
|
|
if self.invalid_plan:
|
|
self.invalid_plan -= 1
|
|
document["searches"][0]["term"] = "invented device"
|
|
elif version == "entity_resolution_assessment_v2":
|
|
document = {"resolutions": [{"search_index": i, "status": self.status,
|
|
"candidate_id": self.candidate if self.status == "selected" else None,
|
|
"reason": "The affected subject is the wall switch; the operator is an interaction participant.",
|
|
"source_ids": ["O0001", f"V{i+1:04d}"]} for i,_ in enumerate(self.plan["searches"])]}
|
|
if self.invalid_assessment:
|
|
self.invalid_assessment -= 1
|
|
document["resolutions"][0]["candidate_id"] = "K9999"
|
|
else:
|
|
document = payload()
|
|
data = json.loads(request["messages"][1]["content"])["input"]
|
|
selected = [(k,v) for k,v in data["sources"].items() if v["kind"] == "entity" and v["case_pointer"].startswith("/entity_resolution/")]
|
|
for k,v in selected:
|
|
document["relevant_entities"].append({"entity_id": k, "reason": "Contextually matched affected switch", "source_ids": [k,"O0001"]})
|
|
if self.omit_resolved:
|
|
self.omit_resolved -= 1
|
|
document["relevant_entities"] = payload()["relevant_entities"]
|
|
unresolved = [k for k,v in data["sources"].items() if v["kind"] == "data_quality" and v["case_pointer"].startswith("/entity_resolution/")]
|
|
for k in unresolved:
|
|
document["unknowns"].append(statement("Identity is unresolved, not proof of a missing runtime entity",k))
|
|
return ProviderResponse(document, "fixture-model", 100, 20)
|
|
|
|
|
|
def service_for(store, database=None):
|
|
with PackageReader(store.latest("case-A", "stable-package")) as reader:
|
|
project = reader.manifest["project"]
|
|
service = ProjectResearch(project, database=database or Database())
|
|
service.entity_resolution = copy.deepcopy(SETTINGS)
|
|
return service
|
|
|
|
|
|
def run(store, model, service=None, **kwargs):
|
|
return asyncio.run(CaseInterpreter(store, gateway(model), project_research=service, **kwargs).interpret("case-A", "stable-package"))
|
|
|
|
|
|
def context_of(store):
|
|
with PackageReader(store.latest("case-A", "stable-package")) as reader:
|
|
ref = reader.manifest["artifacts"]["case"]
|
|
return build_context(json.loads(reader.read_bytes(ref)),ref,InterpretationLimits())
|
|
|
|
|
|
def test_resolves_unrecorded_subject_preserves_case_and_portable_replay(built_case, tmp_path):
|
|
store, _, built, _, _ = built_case
|
|
with PackageReader(built.package) as reader:
|
|
old = {n:reader.read_bytes(n) for n in reader.files if n != "manifest.json"}
|
|
model = Model(); service = service_for(store)
|
|
result = run(store, model, service)
|
|
assert result.status == "COMPLETED" and PackageValidator().validate(result.package).valid
|
|
assert [r["prompt_version"] for r in model.requests] == ["entity_resolution_plan_v2","entity_resolution_assessment_v2","case_interpretation_resolved_v1"]
|
|
assert len(service.database.calls) == 1
|
|
assert service.database.calls[0]["filters"] == [{"column":"name","operator":"contains","value":"switch"}]
|
|
with PackageReader(result.package) as reader:
|
|
for n,data in old.items():
|
|
assert reader.read_bytes(n) == data
|
|
doc = json.loads(reader.read_bytes(result.interpretation_ref))
|
|
context = json.loads(reader.read_bytes(doc["context_ref"]))
|
|
subject = context["sources"]["N0002"]["data"]
|
|
assert subject["id"] == "17" and subject["roles"] == ["affected_subject"]
|
|
assert subject["association"] == "model_contextual_selection_not_causation"
|
|
assert context["sources"]["N0001"]["data"]["id"] == "42"
|
|
assert "captures/" in subject["provenance"]["source_ref"]
|
|
assert not reader.manifest["artifacts"].get("hypotheses")
|
|
assert CaseBuilder(store).build("case-A","stable-package").reused
|
|
# No live DB binding required when all records are complete.
|
|
receiving = CheckpointStore(type(store.workspace)(tmp_path / "portable"))
|
|
receiving.intake(result.package)
|
|
again = run(receiving, model)
|
|
assert again.reused and len(model.requests) == 3
|
|
|
|
|
|
@pytest.mark.parametrize("status,rows,truncated", [("ambiguous",None,False),("no_match",[],False),("ambiguous",None,True)])
|
|
def test_unresolved_candidates_remain_explicit_unknowns(built_case,status,rows,truncated):
|
|
store=built_case[0]; model=Model(status=status); service=service_for(store,Database(rows,truncated))
|
|
result=run(store,model,service)
|
|
with PackageReader(result.package) as r:
|
|
doc=json.loads(r.read_bytes(result.interpretation_ref)); context=json.loads(r.read_bytes(doc['context_ref']))
|
|
assert 'N0002' not in context['sources']
|
|
assert any(v['data'].get('code')=='ENTITY_RESOLUTION_UNRESOLVED' for v in context['sources'].values() if isinstance(v['data'],dict))
|
|
assert any('Identity is unresolved' in s['text'] for s in doc['interpretation']['unknowns'])
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["invalid_plan","invalid_assessment"])
|
|
def test_invalid_outputs_corrected_without_repeating_queries(built_case,field):
|
|
store=built_case[0]; model=Model(); setattr(model,field,1); service=service_for(store)
|
|
assert run(store,model,service).status=='REJECTED'
|
|
assert run(store,model,service).status=='COMPLETED'
|
|
assert len(service.database.calls)==1 and len(model.requests)==4
|
|
assert any('validation_feedback' in r['messages'][1]['content'] for r in model.requests)
|
|
assert run(store,model).reused and len(model.requests)==4
|
|
|
|
|
|
def test_invalid_exhaustion_blocks_without_hypotheses_or_budget_mutation(built_case):
|
|
store=built_case[0]; model=Model();model.invalid_plan=3;service=service_for(store)
|
|
for _ in range(3):
|
|
assert run(store,model,service).status=='REJECTED'
|
|
result=run(store,model,service)
|
|
assert result.status=='BLOCKED' and result.code=='ENTITY_RESOLUTION_INVALID_RESPONSES'
|
|
assert len(model.requests)==3 and not service.database.calls
|
|
assert run(store,model,service).status=='BLOCKED'
|
|
|
|
|
|
def test_database_outage_preserves_accepted_plan(built_case,monkeypatch):
|
|
store=built_case[0];model=Model(); service=service_for(store);service.database.unavailable=True
|
|
result=run(store,model,service)
|
|
assert result.status=='BLOCKED' and result.code=='RESEARCH_UNAVAILABLE'
|
|
with PackageReader(result.package) as r:
|
|
assert 'analysis/interpretations/0001/entity-resolution/plan.json' in r.files
|
|
assert r.manifest['artifacts'].get('case_interpretation') is None
|
|
service.database.unavailable=False
|
|
monkeypatch.setattr(store,'retry_due',lambda m: True)
|
|
store.advance('case-A','stable-package','INTERPRETING_CASE',step_id='manual-resume',details={'operation':'manual-resume'})
|
|
assert run(store,model,service).status=='COMPLETED'
|
|
assert len(model.requests)==3
|
|
|
|
|
|
def test_provider_retry_resume_uses_frozen_input(built_case):
|
|
store=built_case[0];model=Model(); model.unavailable_stage='entity_resolution_plan_v2'; service=service_for(store)
|
|
# Tiny positive retry delay permits resumption in this offline fixture.
|
|
g=gateway(model,retry_delay_seconds=0.001)
|
|
owner=CaseInterpreter(store,g,project_research=service)
|
|
first=asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert first.status=='RETRY_PENDING'
|
|
model.unavailable_stage=None
|
|
assert asyncio.run(owner.interpret('case-A','stable-package')).status=='COMPLETED'
|
|
|
|
|
|
def test_changed_snapshot_rejected_before_additional_calls(built_case):
|
|
store=built_case[0];model=Model();model.invalid_plan=1;service=service_for(store)
|
|
assert run(store,model,service).status=='REJECTED'
|
|
service.database.snapshot='different'
|
|
with pytest.raises(InterpretationError,match='ENTITY_RESOLUTION_BINDING_CHANGED'):
|
|
run(store,model,service)
|
|
assert len(model.requests)==1
|
|
|
|
|
|
@pytest.mark.parametrize("mutation,code", [
|
|
(lambda p:p['searches'][0].update(term='Unknown NPC'),'ENTITY_RESOLUTION_UNGROUNDED_TERM'),
|
|
(lambda p:p['searches'][0].update(source_ids=['X0001']),'ENTITY_RESOLUTION_SOURCE'),
|
|
(lambda p:p['searches'][0].update(catalog='account'),'ENTITY_RESOLUTION_CATALOG'),
|
|
(lambda p:p['searches'].append(copy.deepcopy(p['searches'][0])),'ENTITY_RESOLUTION_DUPLICATE_SEARCH'),
|
|
])
|
|
def test_plan_validation_rejects_invented_terms_unread_refs_and_unknown_catalogs(built_case,mutation,code):
|
|
context=context_of(built_case[0]); p=copy.deepcopy(PLAN);mutation(p)
|
|
with pytest.raises(InterpretationError,match=code):validate_plan(p,context,policy(SETTINGS))
|
|
|
|
|
|
def test_truncated_search_cannot_be_selected(built_case):
|
|
store=built_case[0]; model=Model();service=service_for(store,Database(truncated=True))
|
|
result=run(store,model,service)
|
|
assert result.status=='REJECTED' and result.code=='ENTITY_RESOLUTION_SELECTION'
|
|
|
|
|
|
def test_search_cannot_be_grounded_in_a_capture_filename(built_case):
|
|
context=context_of(built_case[0]);p=copy.deepcopy(PLAN)
|
|
context['sources']['F0001']['data']['value']={'origin':{'source_ref':'raw/invented device.json'},'row':{'name':'switch'}}
|
|
p['searches'][0].update(term='invented device',source_ids=['F0001'])
|
|
with pytest.raises(InterpretationError,match='ENTITY_RESOLUTION_UNGROUNDED_TERM'):
|
|
validate_plan(p,context,policy(SETTINGS))
|
|
|
|
|
|
def test_empty_plan_keeps_old_interpretation_contract(built_case):
|
|
store=built_case[0];model=Model(plan={'searches':[]});service=service_for(store)
|
|
assert run(store,model,service).status=='COMPLETED'
|
|
assert model.requests[-1]['prompt_version']=='case_interpretation_v1' and not service.database.calls
|
|
|
|
|
|
def test_trial_phase_stop_includes_resolution_and_is_idempotent(built_case):
|
|
from viano_bug_analysis.trial import drive
|
|
store=built_case[0];model=Model();service=service_for(store)
|
|
async def trial():
|
|
return await drive(store,gateway(model),'case-A','stable-package',max_dispatches=6,
|
|
policies={'project_research':service},stop_after='interpretation')
|
|
result=asyncio.run(trial())
|
|
assert result['status']=='PHASE_COMPLETE' and result['state']=='INTERPRETING_CASE'
|
|
assert asyncio.run(trial())['status']=='PHASE_COMPLETE' and len(model.requests)==3
|
|
|
|
|
|
def test_final_interpretation_cannot_discard_resolved_identity(built_case):
|
|
from viano_bug_analysis.analysis.correction import CorrectionConfig
|
|
store=built_case[0];model=Model();model.omit_resolved=1;service=service_for(store)
|
|
result=run(store,model,service,correction_config=CorrectionConfig())
|
|
assert result.status=='REJECTED' and result.code=='INTERPRETATION_RESOLVED_ENTITY_OMITTED'
|
|
result=run(store,model,service,correction_config=CorrectionConfig())
|
|
assert result.status=='COMPLETED' and len(model.requests)==4
|
|
assert len(service.database.calls)==1
|
|
assert run(store,model).reused and len(model.requests)==4
|
|
|
|
|
|
@pytest.mark.parametrize('settings', [{'enabled':True,'catalogs':[]},{**SETTINGS,'max_searches':True},{**SETTINGS,'unexpected':5}])
|
|
def test_invalid_operator_configuration(settings):
|
|
with pytest.raises(InterpretationError,match='ENTITY_RESOLUTION_CONFIG'):policy(settings)
|
|
|
|
|
|
def test_legacy_resolution_requests_and_completed_packages_remain_replayable(built_case):
|
|
from viano_bug_analysis.analysis.entity_resolution import EntityResolver, request_for
|
|
store=built_case[0];service=service_for(store);model=Model()
|
|
owner=CaseInterpreter(store,gateway(model),project_research=service)
|
|
context=context_of(store)
|
|
# A v1 artifact has no protocol_version; preserve its original request bytes.
|
|
EntityResolver(owner).save(context,'input',{'context':context,'policy':policy(SETTINGS),'binding':service.database.binding()})
|
|
old=request_for('plan',context,policy(SETTINGS))
|
|
assert old['prompt_version']=='entity_resolution_plan_v1'
|
|
assert 'output_schema' not in json.loads(old['messages'][1]['content'])
|
|
result=asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status=='COMPLETED'
|
|
assert model.requests[0]['prompt_version']=='entity_resolution_plan_v1'
|
|
assert run(store,model).reused and len(model.requests)==3
|