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 == "entity_resolution_plan_v1": document = copy.deepcopy(self.plan) if self.invalid_plan: self.invalid_plan -= 1 document["searches"][0]["term"] = "invented device" elif version == "entity_resolution_assessment_v1": 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_v1","entity_resolution_assessment_v1","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_v1'; 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)