"""Opt-in full-archive regression, without machine-specific paths or binary fixtures.""" import hashlib import json import os from pathlib import Path import pytest from viano_bug_analysis.case import CaseBuilder from viano_bug_analysis.lifecycle import CheckpointStore, Workspace from viano_bug_analysis.package import PackageReader, PackageValidator from viano_bug_analysis.profiles.vcore import ArchiveReader, VcoreImporter from viano_bug_analysis.analysis import CaseInterpreter, HypothesisBuilder from test_case_interpreter import FixtureAdapter, gateway, missing_payload from test_hypothesis_builder import ProposalAdapter import asyncio @pytest.mark.skipif(not os.environ.get("VBA_WOWDIAG_SAMPLE") or not os.environ.get("VBA_WOWSRC_SAMPLE"), reason="Set VBA_WOWDIAG_SAMPLE and VBA_WOWSRC_SAMPLE for the BUG-20 full archive regression") def test_full_bug20_package_pair(tmp_path): diag, src = Path(os.environ["VBA_WOWDIAG_SAMPLE"]), Path(os.environ["VBA_WOWSRC_SAMPLE"]) original_hashes = [hashlib.sha256(path.read_bytes()).hexdigest() for path in (diag, src)] package = tmp_path / "anything.bin" VcoreImporter().create(diag, package, package_id="regression-package", case_id="regression-case", wowsrc=src) store = CheckpointStore(Workspace(tmp_path / "service")) store.intake(package) result = CaseBuilder(store).build("regression-case", "regression-package") assert PackageValidator().validate(result.package).valid with PackageReader(result.package) as reader: case = json.loads(reader.read_bytes(result.case_ref)) assert len(case["human_observations"]["observations"]) == 3 assert case["expected_behavior"]["completeness"] == "missing" assert case["human_observations"]["tester_hypotheses"] == [] assert len(case["database_context"]["refs"]) == 103 assert len(case["source_context"]["refs"]) == 93 for input_path in (diag, src): with ArchiveReader(input_path) as source: for member in source.members: matching = [path for path in reader.files if path.endswith("/members/" + member)] assert len(matching) == 1 assert reader.read_bytes(matching[0]) == source.read_member(member) assert CaseBuilder(store).build("regression-case", "regression-package").reused # Offline phase integration only: this fixture is not a test of model reasoning. response = missing_payload() response["observed_behavior"][0]["text"] = case["human_observations"]["observations"][0]["text"] adapter = FixtureAdapter(response) interpreter = CaseInterpreter(store, gateway(adapter)) interpreted = asyncio.run(interpreter.interpret("regression-case", "regression-package")) assert interpreted.status == "COMPLETED" with PackageReader(result.package) as before, PackageReader(interpreted.package) as after: document = json.loads(after.read_bytes(interpreted.interpretation_ref)) assert document["expected_behavior_completeness"] == "missing" assert document["interpretation"]["expected_behavior"] == [] assert document["interpretation"]["behavior_gap"] is None assert after.manifest["artifacts"]["diagnosis"] is None assert after.manifest["artifacts"]["change_request"] is None for name in before.files: if name != "manifest.json": assert before.read_bytes(name) == after.read_bytes(name) assert PackageValidator().validate(interpreted.package).valid assert asyncio.run(interpreter.interpret("regression-case", "regression-package")).reused assert len(adapter.requests) == 1 hypotheses_adapter = ProposalAdapter() generated = asyncio.run(HypothesisBuilder(store, gateway(hypotheses_adapter)).build("regression-case", "regression-package")) assert generated.status == "COMPLETED" assert PackageValidator().validate(generated.package).valid with PackageReader(interpreted.package) as before, PackageReader(generated.package) as after: hypotheses = json.loads(after.read_bytes(generated.hypotheses_ref)) registry = json.loads(after.read_bytes(generated.registry_ref)) assert len(hypotheses["hypotheses"]) == 3 assert [record["budget_allocation"]["initial"] for record in hypotheses["hypotheses"]] == [100, 100, 100] assert registry["created_count"] == 3 and registry["max_hypotheses"] == 5 assert after.manifest["artifacts"]["diagnosis"] is None assert after.manifest["artifacts"]["change_request"] is None for name in before.files: if name != "manifest.json": assert before.read_bytes(name) == after.read_bytes(name) assert asyncio.run(HypothesisBuilder(store, gateway(hypotheses_adapter)).build("regression-case", "regression-package")).reused assert len(hypotheses_adapter.requests) == 1 from viano_bug_analysis.budget import BudgetManager budget = BudgetManager(store) identity = ("regression-case", "regression-package") budget.initialize(*identity) budget.reserve(*identity, "H1", "offline-query", "source.read_context") source_ref = case["source_context"]["refs"][0]["path"] budget.record_result(*identity, "H1", "offline-query", {"fixture_observation":"Stored context is available"}, source_refs=[source_ref]) accounted = budget.settle(*identity, "H1", "offline-query", semantic_score=50, reason="Simulated assessment, no diagnosis") assert accounted.state["accounts"]["H1"]["remaining"] == 94 assert PackageValidator().validate(accounted.package).valid with PackageReader(generated.package) as before, PackageReader(accounted.package) as after: for name in before.files: if name != "manifest.json": assert before.read_bytes(name) == after.read_bytes(name) assert CaseBuilder(store).build(*identity).reused from viano_bug_analysis.tools import InvestigationToolRouter from viano_bug_analysis.tools.database_query import DatabaseQueryTool from viano_bug_analysis.profiles.vcore.db_tools import normalize_schema_descriptor code_ref = next(r["path"] for r in case["source_context"]["refs"] if r["path"].endswith(".cpp")) schema_ref = next(r["path"] for r in case["database_context"]["refs"] if r["path"].endswith("/database/schemas/quest_template.json")) rows_ref = next(r["path"] for r in case["database_context"]["refs"] if r["path"].endswith("/database/quest_template.json")) request = {"format":"viano-investigation-request","format_version":1,"case_id":identity[0],"package_id":identity[1],"analysis_revision":1, "requests":[{"request_id":"source-query","hypothesis_id":"H1","action_id":"source-query","operation":"source.read_context", "arguments":{"source_ref":code_ref,"start_line":1,"end_line":3}}, {"request_id":"schema-query","hypothesis_id":"H2","action_id":"schema-query","operation":"database.describe", "arguments":{"source_ref":schema_ref}}, {"request_id":"rows-query","hypothesis_id":"H3","action_id":"rows-query","operation":"database.lookup", "arguments":{"source_ref":rows_ref,"filters":{"ID":11117},"columns":["ID"],"offset":0,"limit":10}}]} before_query_hash = hashlib.sha256(accounted.package.read_bytes()).hexdigest() queried = InvestigationToolRouter(database_tool=DatabaseQueryTool(normalize_schema_descriptor)).query(accounted.package,request) assert len(queried["results"][0]["data"]["matches"]) == 3 assert queried["results"][1]["data"]["table"] == "quest_template" assert queried["results"][1]["data"]["coverage"] == "schema_only" assert queried["results"][2]["data"]["matches"][0]["row"] == {"ID":11117} assert queried["results"][2]["data"]["coverage"] == "unknown" assert hashlib.sha256(accounted.package.read_bytes()).hexdigest() == before_query_hash from viano_bug_analysis.evidence import EvidenceStore for asked in request["requests"]: budget.reserve(*identity,asked["hypothesis_id"],asked["action_id"],asked["operation"]) current=budget.snapshot(*identity) queried=InvestigationToolRouter(database_tool=DatabaseQueryTool(normalize_schema_descriptor)).query(current.package,request) evidence=EvidenceStore(store,router=InvestigationToolRouter(database_tool=DatabaseQueryTool(normalize_schema_descriptor))) saved=evidence.store_tool_results(*identity,"real-snapshot-query",request,queried) assert len(saved.record_refs)==3 assert budget.snapshot(*identity).state==current.state assert PackageValidator().validate(saved.package).valid for asked,ref in zip(request["requests"],saved.record_refs): budget.record_evidence(*identity,asked["hypothesis_id"],asked["action_id"],ref) budget.settle(*identity,asked["hypothesis_id"],asked["action_id"],semantic_score=50,reason="Offline provenance regression") assert CaseBuilder(store).build(*identity).reused from viano_bug_analysis.analysis import Investigator,InvestigationConfig,InvestigationLimits from test_investigator import LoopAdapter def choose_packaged_source(reply,context): if context["previous_result"] is None: reply["next"]={"kind":"tool","operation":"source.read_context","arguments":{"source_ref":code_ref,"start_line":1,"end_line":3}} loop_adapter=LoopAdapter(choose_packaged_source) investigated=asyncio.run(Investigator(store,gateway(loop_adapter,max_request_bytes=4*1024*1024), InvestigationConfig(max_iterations_per_hypothesis=1),InvestigationLimits(max_context_bytes=2*1024*1024), router=InvestigationToolRouter(database_tool=DatabaseQueryTool(normalize_schema_descriptor))).run(*identity,max_steps=10)) assert investigated.status=="COMPLETED" and len(loop_adapter.contexts)==6 with PackageReader(investigated.package) as reader: assert reader.manifest["state"]=="BUILDING_DIAGNOSIS" assert reader.manifest["artifacts"]["diagnosis"] is None assert reader.manifest["artifacts"]["change_request"] is None assert [hashlib.sha256(path.read_bytes()).hexdigest() for path in (diag, src)] == original_hashes from viano_bug_analysis.analysis import DiagnosisBuilder from test_diagnosis_builder import FinalAdapter final_adapter=FinalAdapter() diagnosed=asyncio.run(DiagnosisBuilder(store,gateway(final_adapter,max_request_bytes=4*1024*1024), router=InvestigationToolRouter(database_tool=DatabaseQueryTool(normalize_schema_descriptor))).build(*identity)) assert diagnosed.status=='INCONCLUSIVE' and final_adapter.calls==1 with PackageReader(diagnosed.package) as reader: assert reader.manifest['state']=='FAILED_TO_ANALYZE' assert reader.manifest['routing']['destination']=='review' assert reader.manifest['artifacts']['change_request'] is None assert reader.manifest['artifacts']['failure_report']==diagnosed.failure_ref assert PackageValidator().validate(diagnosed.package).valid assert [hashlib.sha256(path.read_bytes()).hexdigest() for path in (diag, src)] == original_hashes