542 lines
31 KiB
Python
542 lines
31 KiB
Python
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
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
|
|
self.linked_capture = False
|
|
|
|
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','_v3','_v4')):
|
|
# 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', '_v4').replace('_v2', '_v4').replace('_v3', '_v4') if version.startswith('entity_resolution_') else version
|
|
if version == "entity_resolution_plan_v4":
|
|
document = copy.deepcopy(self.plan)
|
|
if self.invalid_plan:
|
|
self.invalid_plan -= 1
|
|
document["searches"][0]["term"] = "invented device"
|
|
elif version == "entity_resolution_assessment_v4":
|
|
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()
|
|
if version in {"case_interpretation_resolved_v3", "case_interpretation_resolved_v4", "case_interpretation_resolved_v5", "case_interpretation_resolved_v6", "case_interpretation_resolved_v7","case_interpretation_resolved_v8"}:
|
|
document.update(described_behavior=[],provisional_behavior_gap=None)
|
|
data = restore(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:
|
|
evidence = k
|
|
if self.linked_capture:
|
|
evidence = next(f for f,s in data['sources'].items() if s['kind']=='fact'
|
|
and s['data'].get('name')=='entity_name_lookup'
|
|
and s['data']['provenance']['source_ref']==v['data']['provenance']['source_ref'])
|
|
document["relevant_entities"].append({"entity_id": k, "reason": "Contextually matched affected switch", "source_ids": [evidence,"O0001"]})
|
|
if self.omit_resolved:
|
|
self.omit_resolved -= 1
|
|
document["relevant_entities"] = payload()["relevant_entities"]
|
|
if version in {'case_interpretation_resolved_v4','case_interpretation_resolved_v5','case_interpretation_resolved_v6','case_interpretation_resolved_v7','case_interpretation_resolved_v8'}:
|
|
from viano_bug_analysis.analysis.quality import unnamed_entity_reasons
|
|
reasons = unnamed_entity_reasons(data)
|
|
for entity in document['relevant_entities']:
|
|
if entity['entity_id'] in reasons:
|
|
entity['reason'] = reasons[entity['entity_id']]
|
|
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())
|
|
|
|
|
|
@pytest.mark.parametrize('qualified', [False, True])
|
|
@pytest.mark.parametrize('wrong_origin', [False, True])
|
|
def test_multidatabase_capture_origin_and_portable_replay(built_case, tmp_path, qualified, wrong_origin):
|
|
class MultiDatabase(Database):
|
|
def binding(self):
|
|
return {'kind':'mariadb', 'database':'world', 'snapshot_id':self.snapshot,
|
|
'tables':['components'], 'additional_databases':{'characters':['components']}}
|
|
|
|
def binding_for(self, query):
|
|
database, table = query['table'].split('.') if '.' in query['table'] else ('world', query['table'])
|
|
return {**self.binding(), 'selected_database':'wrong' if wrong_origin else database,
|
|
'selected_table':table}
|
|
|
|
def validate(self, query):
|
|
assert query['table'] in {'components','characters.components'}
|
|
|
|
store=built_case[0]
|
|
model=Model(); service=service_for(store,MultiDatabase())
|
|
if qualified:
|
|
service.entity_resolution['catalogs'][0]['table']='characters.components'
|
|
if wrong_origin:
|
|
with pytest.raises(InterpretationError, match='ENTITY_RESOLUTION_CAPTURE_CHANGED'):
|
|
run(store,model,service)
|
|
return
|
|
result=run(store,model,service)
|
|
assert result.status=='COMPLETED'
|
|
assert PackageValidator().validate(result.package).valid
|
|
receiving=CheckpointStore(type(store.workspace)(tmp_path/'portable-multidb'))
|
|
receiving.intake(result.package)
|
|
again=run(receiving,model)
|
|
assert again.reused and len(model.requests)==3
|
|
|
|
|
|
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_v4","entity_resolution_assessment_v4","case_interpretation_resolved_v8"]
|
|
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_v4'; 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_still_supports_descriptive_quality_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_resolved_v8' 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
|
|
feedback=json.loads(model.requests[-1]['messages'][-1]['content'])['validation_feedback']
|
|
assert any(i.get('constraint')=='retain_resolved_entity' and i['entity_id']=='N0002' for i in feedback['issues'])
|
|
assert len(service.database.calls)==1
|
|
assert run(store,model).reused and len(model.requests)==4
|
|
|
|
|
|
def test_matching_database_capture_is_valid_without_redundant_entity_citation(built_case):
|
|
store=built_case[0];model=Model();model.linked_capture=True;service=service_for(store)
|
|
result=run(store,model,service)
|
|
assert result.status=='COMPLETED' and len(model.requests)==3
|
|
assert run(store,model).reused
|
|
with PackageReader(result.package) as r:
|
|
doc=json.loads(r.read_bytes(result.interpretation_ref))
|
|
assert 'N0002' not in doc['interpretation']['relevant_entities'][-1]['source_ids']
|
|
|
|
|
|
def test_provenance_feedback_identifies_missing_mention_and_wrong_capture(built_case):
|
|
from viano_bug_analysis.analysis.contracts import entity_citation_issues,validate_interpretation
|
|
store=built_case[0];model=Model();service=service_for(store);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']))
|
|
item=doc['interpretation']['relevant_entities'][-1]
|
|
item['source_ids']=['F0001']
|
|
issues=entity_citation_issues(doc['interpretation'],context)
|
|
assert {i['constraint'] for i in issues}=={'cite_original_mention','cite_entity_or_matching_capture'}
|
|
assert next(i for i in issues if i['constraint']=='cite_original_mention')['allowed_source_ids']==['O0001']
|
|
assert 'N0002' in next(i for i in issues if i['constraint']=='cite_entity_or_matching_capture')['allowed_source_ids']
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_ENTITY_PROVENANCE'):
|
|
validate_interpretation(doc['interpretation'],context)
|
|
# Even a lookup-shaped fact cannot substitute an unrelated capture.
|
|
context['sources']['F0001']['data'].update(name='entity_name_lookup',provenance={'source_ref':'different/capture.json'})
|
|
item['source_ids']=['O0001','F0001']
|
|
assert [i['constraint'] for i in entity_citation_issues(doc['interpretation'],context)]==['cite_entity_or_matching_capture']
|
|
|
|
|
|
def test_real_gemini_reply_is_accepted_with_matching_capture_and_legacy_rule_is_preserved():
|
|
from pathlib import Path
|
|
from viano_bug_analysis.analysis.contracts import validate_interpretation
|
|
fixture=json.loads((Path(__file__).parent/'fixtures/interpretation/real-capture-citations.json').read_text(encoding='utf-8'))
|
|
validate_interpretation(fixture['response'],fixture['context'])
|
|
entity=next(e for e in fixture['response']['relevant_entities'] if e['entity_id']=='N0005')
|
|
assert 'N0005' not in entity['source_ids'] and 'F0027' in entity['source_ids']
|
|
assert fixture['context']['sources']['N0005']['data']['id']=='23487'
|
|
old=copy.deepcopy(fixture['context'])
|
|
del old['sources']['N0005']['data']['citation_policy']
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_RESOLVED_ENTITY_OMITTED'):
|
|
validate_interpretation(fixture['response'],old)
|
|
|
|
|
|
@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)
|
|
|
|
|
|
@pytest.mark.parametrize("legacy_version", [1, 2])
|
|
def test_legacy_resolution_requests_and_completed_packages_remain_replayable(built_case, legacy_version):
|
|
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(),
|
|
**({'protocol_version':2} if legacy_version == 2 else {})})
|
|
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']==f'entity_resolution_plan_v{legacy_version}'
|
|
assert run(store,model).reused and len(model.requests)==3
|
|
|
|
|
|
def test_missing_capture_feedback_resumes_and_replays_without_repeating_research(built_case):
|
|
class MissingCaptureModel(Model):
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
response = await super().generate(config, request, max_response_bytes=max_response_bytes)
|
|
if request['prompt_version'] in {'entity_resolution_assessment_v3','entity_resolution_assessment_v4'}:
|
|
data = json.loads(request['messages'][1]['content'])
|
|
if 'validation_feedback' not in data:
|
|
for resolution in response.document['resolutions']:
|
|
resolution['source_ids'] = ['O0001']
|
|
resolution['reason'] = 'The right capture is V0001 but only mentioned here.'
|
|
else:
|
|
fixes = data['validation_feedback']['citation_corrections']
|
|
assert fixes['version']==2
|
|
assert fixes['issues'][0]['search_index']==0
|
|
assert fixes['issues'][0]['capture_id']=='V0001'
|
|
assert fixes['issues'][0]['code']=='ENTITY_RESOLUTION_CAPTURE_SOURCE'
|
|
return response
|
|
store = built_case[0]
|
|
model = MissingCaptureModel()
|
|
service = service_for(store)
|
|
from viano_bug_analysis.analysis.entity_resolution import EntityResolver
|
|
owner=CaseInterpreter(store,gateway(model),project_research=service)
|
|
context=context_of(store)
|
|
EntityResolver(owner).save(context,'input',{'context':context,'policy':policy(SETTINGS),
|
|
'binding':service.database.binding(),'protocol_version':3,'interpretation_protocol_version':6})
|
|
rejected = run(store, model, service)
|
|
assert rejected.status == 'REJECTED'
|
|
assert rejected.code == 'ENTITY_RESOLUTION_CAPTURE_SOURCE'
|
|
assert len(service.database.calls) == 1
|
|
result = run(store, model, service)
|
|
assert result.status == 'COMPLETED'
|
|
assert len(service.database.calls) == 1
|
|
assert len(model.requests) == 4
|
|
with PackageReader(result.package) as reader:
|
|
feedback_ref = 'analysis/interpretations/0001/entity-resolution/feedback-assessment-0002.json'
|
|
assert feedback_ref in reader.files
|
|
assert PackageValidator().validate(result.package).valid
|
|
assert run(store, model).reused
|
|
assert len(model.requests) == 4
|
|
|
|
|
|
def test_third_assessment_can_use_lower_reasoning_without_resetting_history(built_case):
|
|
from dataclasses import replace
|
|
from viano_bug_analysis.llm import LLMError, LLMGateway
|
|
class TruncatedModel(Model):
|
|
failures = 2
|
|
efforts = []
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.efforts.append((request['prompt_version'], config.reasoning_effort))
|
|
if request['prompt_version']=='entity_resolution_assessment_v4' and self.failures:
|
|
self.failures -= 1
|
|
raise LLMError('LLM_INCOMPLETE_OUTPUT')
|
|
return await super().generate(config, request, max_response_bytes=max_response_bytes)
|
|
store = built_case[0]
|
|
model = TruncatedModel()
|
|
service = service_for(store)
|
|
old = gateway(model)
|
|
old.config = replace(old.config, providers=tuple(replace(p, reasoning_effort='medium') for p in old.config.providers))
|
|
owner = CaseInterpreter(store, old, project_research=service)
|
|
for _ in range(2):
|
|
result = asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status == 'REJECTED'
|
|
adjusted = replace(old.config, providers=tuple(replace(p, entity_resolution_reasoning_effort='low') for p in old.config.providers))
|
|
owner = CaseInterpreter(store, LLMGateway(adjusted, {'fixture': model}), project_research=service)
|
|
result = asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status == 'COMPLETED'
|
|
assert len(service.database.calls) == 1
|
|
assert model.efforts == [('entity_resolution_plan_v4','medium'),
|
|
('entity_resolution_assessment_v4','medium'),('entity_resolution_assessment_v4','medium'),
|
|
('entity_resolution_assessment_v4','low'),('case_interpretation_resolved_v8','medium')]
|
|
with PackageReader(result.package) as reader:
|
|
saved = json.loads(reader.read_bytes('analysis/interpretations/0001/entity-resolution/assessment.json'))
|
|
assert saved['payload']['attempt'] == 3
|
|
assert asyncio.run(owner.interpret('case-A','stable-package')).reused
|
|
assert len(model.efforts) == 5
|
|
|
|
|
|
@pytest.mark.parametrize("version", [3,4,5])
|
|
def test_completed_interpretation_keeps_its_historical_contract(built_case, version):
|
|
from viano_bug_analysis.analysis.entity_resolution import EntityResolver
|
|
store=built_case[0]
|
|
service=service_for(store)
|
|
model=Model()
|
|
owner=CaseInterpreter(store,gateway(model),project_research=service)
|
|
context=context_of(store)
|
|
EntityResolver(owner).save(context,'input',{'context':context,'policy':policy(SETTINGS),
|
|
'binding':service.database.binding(),'protocol_version':3,'interpretation_protocol_version':version})
|
|
result=asyncio.run(owner.interpret('case-A','stable-package'))
|
|
assert result.status=='COMPLETED'
|
|
assert model.requests[-1]['prompt_version']==f'case_interpretation_resolved_v{version}'
|
|
assert run(store,model).reused
|
|
assert len(model.requests)==3
|
|
|
|
|
|
def test_v4_assessment_schema_requires_capture_and_original_mention_per_index():
|
|
from jsonschema import Draft202012Validator
|
|
from viano_bug_analysis.analysis.entity_resolution import request_for
|
|
context={'sources':{'O0001':{'kind':'observation','data':{'text':'Missing switch'}}}}
|
|
plan={'searches':[PLAN['searches'][0],{**PLAN['searches'][0],'term':'Missing switch'}]}
|
|
request=request_for('assessment',context,policy(SETTINGS),plan=plan,captures=[{},{}],candidates=[],protocol_version=4)
|
|
validator=Draft202012Validator(request['output_schema'])
|
|
valid={'resolutions':[{'search_index':i,'status':'no_match','candidate_id':None,
|
|
'reason':'No rows','source_ids':['O0001',f'V{i+1:04d}']} for i in range(2)]}
|
|
validator.validate(valid)
|
|
invalid=copy.deepcopy(valid)
|
|
invalid['resolutions'][1]['source_ids']=['O0001','V0001']
|
|
assert list(validator.iter_errors(invalid))
|
|
invalid=copy.deepcopy(valid)
|
|
invalid['resolutions'][0]['source_ids']=['V0001']
|
|
assert list(validator.iter_errors(invalid))
|
|
|
|
|
|
def test_description_matches_prioritize_specific_names_without_automatic_selection():
|
|
from viano_bug_analysis.analysis.entity_resolution import resolution_guidance
|
|
context={'sources':{
|
|
'F0001':{'kind':'fact','data':{'name':'project_description','value':{'texts':[
|
|
{'column':'title','text':'Find the Wild Sample Creature!'}]}}},
|
|
'F0002':{'kind':'fact','data':{'name':'project_entity_name','value':{
|
|
'type':'creature','id':'42','name':'Known quest giver'}}}}}
|
|
candidates=[{'candidate_id':'K0001','type':'creature','id':'17','name':'Wild Sample Creature'},
|
|
{'candidate_id':'K0002','type':'creature','id':'18','name':'Sample Creature'}]
|
|
guidance=resolution_guidance(context,candidates)
|
|
assert guidance['known_name_bindings'][0]['id']=='42'
|
|
matches=guidance['candidate_description_matches']
|
|
assert next(m for m in matches if m['candidate_id']=='K0002')['more_specific_matches']==['K0001']
|
|
assert next(m for m in matches if m['candidate_id']=='K0001')['more_specific_matches']==[]
|
|
assert 'selected' not in str(guidance)
|
|
|
|
|
|
def test_resolved_selected_interpretation_transitions_to_hypotheses_and_replays(built_case):
|
|
from viano_bug_analysis.analysis import HypothesisBuilder
|
|
from test_hypothesis_builder import ProposalAdapter
|
|
store=built_case[0]
|
|
model=Model();service=service_for(store)
|
|
interpreter=CaseInterpreter(store,gateway(model),project_research=service)
|
|
interpreted=asyncio.run(interpreter.interpret('case-A','stable-package'))
|
|
assert interpreted.status=='COMPLETED'
|
|
with PackageReader(interpreted.package) as reader:
|
|
doc=json.loads(reader.read_bytes(interpreted.interpretation_ref))
|
|
context=json.loads(reader.read_bytes(doc['context_ref']))
|
|
assert context['model_encoding']=='shared_json_v1'
|
|
assert context['context_selection']=='phase_selection_v1'
|
|
frozen={n:reader.read_bytes(n) for n in reader.files if n!='manifest.json'}
|
|
proposal=ProposalAdapter()
|
|
builder=HypothesisBuilder(store,gateway(proposal))
|
|
builder.interpreter=interpreter
|
|
result=asyncio.run(builder.build('case-A','stable-package'))
|
|
assert result.status=='COMPLETED' and PackageValidator().validate(result.package).valid
|
|
assert proposal.requests[0]['prompt_version']=='hypotheses_v4'
|
|
assert len(model.requests)==3 and len(service.database.calls)==1
|
|
assert asyncio.run(builder.build('case-A','stable-package')).reused
|
|
assert len(proposal.requests)==1 and len(model.requests)==3
|
|
with PackageReader(result.package) as reader:
|
|
for name,raw in frozen.items():assert reader.read_bytes(name)==raw
|
|
|
|
|
|
def test_embedded_interpretation_context_contract_stays_in_sync():
|
|
from importlib.resources import files
|
|
schemas=files('viano_bug_analysis.schemas')
|
|
canonical=json.loads(schemas.joinpath('case-interpretation-context-v1.schema.json').read_text())
|
|
hypothesis=json.loads(schemas.joinpath('hypothesis-context-v1.schema.json').read_text())
|
|
embedded=hypothesis['properties']['case_context']
|
|
assert embedded['properties']==canonical['properties']
|
|
assert embedded['required']==canonical['required']
|
|
assert embedded['additionalProperties']==canonical['additionalProperties']
|