516 lines
29 KiB
Python
516 lines
29 KiB
Python
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
import asyncio
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from dataclasses import replace
|
|
from datetime import datetime, timedelta, timezone
|
|
from importlib.resources import files
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from viano_bug_analysis.analysis import (CaseInterpreter, HypothesisBuilder, HypothesisConfig, HypothesisError,
|
|
HypothesisLimits, render_hypotheses)
|
|
from viano_bug_analysis.analysis.hypothesis_contracts import validate_generation
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, State, Workspace
|
|
from viano_bug_analysis.lifecycle.errors import LockBusy
|
|
from viano_bug_analysis.llm import GatewayConfig, LLMGateway, ProviderConfig, ProviderResponse, ProviderUnavailable
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator, PackageWriter
|
|
from test_case_interpreter import built_case, FixtureAdapter, gateway, http_endpoint
|
|
from test_case_builder import case_setup
|
|
|
|
|
|
def proposals(context, count=3):
|
|
sources = context["interpretation_sources"]
|
|
primary = next((key for kind in ("first_known_divergence", "failure_stage", "observed_behavior")
|
|
for key, entry in sources.items() if entry["kind"] == kind), next(iter(sources)))
|
|
any_ref = next(iter(sources))
|
|
mechanisms = ["The interaction event may not reach the state handler", "The stored state may be overwritten after interaction",
|
|
"The client display may not reflect the updated state", "The interaction may target another instance", "The operation may finish after the observation window"]
|
|
return {"hypotheses": [{"title":f"Alternative {index + 1}", "mechanism":mechanisms[index % len(mechanisms)] + (f" variant {index}" if index >= 5 else ""),
|
|
"rationale":"The reported observation is consistent with this untested possibility",
|
|
"distinction":f"Candidate mechanism {index + 1} concerns a different stage",
|
|
"source_ids":["O0001"], "interpretation_ids":[primary if index == 0 else any_ref],
|
|
"expected_evidence":[f"A trace would expose candidate mechanism {index + 1}"],
|
|
"falsifying_evidence":[f"An intact stage {index + 1} would disconfirm this candidate"],
|
|
"open_questions":[f"Does candidate stage {index + 1} operate as expected?"]} for index in range(count)]}
|
|
|
|
|
|
class ProposalAdapter:
|
|
def __init__(self, change=None, count=3, response=None):
|
|
self.change, self.count, self.response = change, count, response
|
|
self.requests = []
|
|
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.requests.append(copy.deepcopy(request))
|
|
if self.response:
|
|
raise self.response
|
|
context = restore(json.loads(request["messages"][1]["content"])["input"])
|
|
result = proposals(context,self.count)
|
|
if self.change: self.change(result,context)
|
|
return ProviderResponse(result,"fixture-hypothesis-model",12345,678)
|
|
|
|
|
|
@pytest.fixture
|
|
def interpreted_case(built_case):
|
|
store, builder, built, original, entries = built_case
|
|
interpreted = asyncio.run(CaseInterpreter(store,gateway(FixtureAdapter())).interpret("case-A","stable-package"))
|
|
return store, builder, interpreted, original, entries
|
|
|
|
|
|
def run(builder, **kw):
|
|
return asyncio.run(builder.build("case-A","stable-package",**kw))
|
|
|
|
|
|
def test_generation_budget_registry_and_preservation(interpreted_case):
|
|
store, case_builder, interpreted, original, entries = interpreted_case
|
|
original_bytes = original.read_bytes()
|
|
adapter = ProposalAdapter()
|
|
result = run(HypothesisBuilder(store,gateway(adapter)))
|
|
assert result.status == "COMPLETED" and not result.reused
|
|
assert PackageValidator().validate(result.package).valid
|
|
with PackageReader(interpreted.package) as before, PackageReader(result.package) as after:
|
|
document = json.loads(after.read_bytes(result.hypotheses_ref))
|
|
registry = json.loads(after.read_bytes(result.registry_ref))
|
|
assert [h["id"] for h in document["hypotheses"]] == ["H1","H2","H3"]
|
|
assert {h["status"] for h in document["hypotheses"]} == {"new"}
|
|
assert {h["confidence"] for h in document["hypotheses"]} == {"unassessed"}
|
|
assert {h["budget_allocation"]["initial"] for h in document["hypotheses"]} == {100}
|
|
assert document["hypotheses"][0]["focus"] == "tester_primary"
|
|
assert document["budget_accounting"] == "not_started"
|
|
assert registry["created_count"] == 3 and registry["remaining_slots"] == 2 and registry["next_id"] == 4
|
|
assert [c["allocated_budget"] for c in registry["creations"]] == [100,100,100]
|
|
assert registry["hypotheses_sha256"] == hashlib.sha256(after.read_bytes(result.hypotheses_ref)).hexdigest()
|
|
assert document["llm_metadata"]["input_tokens"] == 12345
|
|
assert after.manifest["state"] == State.GENERATING_HYPOTHESES
|
|
assert after.manifest["package_generation"] == 9 and after.manifest["analysis_revision"] == 1
|
|
assert after.manifest["package_id"] == "stable-package"
|
|
assert after.manifest["artifacts"]["diagnosis"] is None
|
|
assert after.manifest["artifacts"]["change_request"] is None
|
|
assert after.read_bytes(result.summary_ref).decode() == render_hypotheses(document)
|
|
for name in before.files:
|
|
if name != "manifest.json": assert before.read_bytes(name) == after.read_bytes(name)
|
|
assert original.read_bytes() == original_bytes
|
|
assert len(adapter.requests) == 1
|
|
assert adapter.requests[0]["prompt_version"] == "hypotheses_v4"
|
|
assert case_builder.build("case-A","stable-package").reused
|
|
assert asyncio.run(CaseInterpreter(store,gateway(adapter)).interpret("case-A","stable-package")).reused
|
|
|
|
|
|
def test_configuration_defaults_and_invalid_values(tmp_path):
|
|
assert HypothesisConfig.load("config/analysis.example.toml") == HypothesisConfig()
|
|
assert HypothesisConfig(max_hypotheses=7,initial_budget=23).initial_budget == 23
|
|
for values in ({"min_hypotheses":0},{"max_hypotheses":True},{"initial_budget":1.5},{"initial_budget":-1},{"min_hypotheses":6}):
|
|
with pytest.raises(HypothesisError): HypothesisConfig(**values)
|
|
invalid = tmp_path / "invalid.toml"
|
|
invalid.write_text('[hypotheses]\nunknown=1')
|
|
with pytest.raises(HypothesisError): HypothesisConfig.load(invalid)
|
|
|
|
|
|
def test_custom_equal_budget_and_count(interpreted_case):
|
|
cfg = HypothesisConfig(min_hypotheses=4,max_hypotheses=4,initial_budget=23)
|
|
result = run(HypothesisBuilder(interpreted_case[0],gateway(ProposalAdapter(count=4)),cfg))
|
|
assert result.status == "COMPLETED"
|
|
with PackageReader(result.package) as reader:
|
|
document = json.loads(reader.read_bytes(result.hypotheses_ref))
|
|
assert len(document["hypotheses"]) == 4
|
|
assert [h["budget_allocation"]["initial"] for h in document["hypotheses"]] == [23]*4
|
|
assert json.loads(reader.read_bytes(result.registry_ref))["remaining_slots"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["budget","id","status","confidence","priority","tool_requests","patch","root_cause"])
|
|
def test_model_cannot_assign_budget_ids_state_or_actions(interpreted_case,field):
|
|
def change(payload,context): payload["hypotheses"][0][field] = 1000000
|
|
result = run(HypothesisBuilder(interpreted_case[0],gateway(ProposalAdapter(change))))
|
|
assert result.status == "ERROR" and result.code == "LLM_OUTPUT_SCHEMA"
|
|
with PackageReader(result.package) as reader:
|
|
assert reader.manifest["artifacts"]["hypothesis_registry"] is None
|
|
assert reader.manifest["artifacts"]["hypotheses"] is None
|
|
assert not any(name.endswith("/hypotheses.json") for name in reader.files)
|
|
|
|
|
|
@pytest.mark.parametrize("count", [0,1,2,6,7])
|
|
def test_out_of_policy_count_is_rejected_before_creation(interpreted_case,count):
|
|
result = run(HypothesisBuilder(interpreted_case[0],gateway(ProposalAdapter(count=count))))
|
|
assert result.status == "ERROR" and result.code == "LLM_OUTPUT_SCHEMA"
|
|
with PackageReader(result.package) as reader: assert reader.manifest["artifacts"]["hypothesis_registry"] is None
|
|
|
|
|
|
@pytest.mark.parametrize("change,code", [("duplicate_mechanism","HYPOTHESIS_DUPLICATE"),("duplicate_title","HYPOTHESIS_DUPLICATE"),
|
|
("unknown_source","HYPOTHESIS_SOURCE_UNKNOWN"),("unknown_interpretation","HYPOTHESIS_SOURCE_UNKNOWN"),
|
|
("ungrounded","HYPOTHESIS_GROUNDING"),("primary_not_human","HYPOTHESIS_PRIMARY_FOCUS"),("primary_wrong_focus","HYPOTHESIS_PRIMARY_FOCUS")])
|
|
def test_invalid_set_rejected_atomically_and_replayed(interpreted_case,change,code):
|
|
def mutate(payload,context):
|
|
first, second = payload["hypotheses"][:2]
|
|
if change == "duplicate_mechanism": second["mechanism"] = " " + first["mechanism"].upper() + "!!!"
|
|
if change == "duplicate_title": second["title"] = first["title"].upper()
|
|
if change == "unknown_source": second["source_ids"] = ["O9999"]
|
|
if change == "unknown_interpretation": second["interpretation_ids"] = ["I9999"]
|
|
if change == "ungrounded": second["source_ids"] = ["T0001","E0001"]
|
|
if change == "primary_not_human": first["source_ids"] = ["F0001"]
|
|
if change == "primary_wrong_focus": first["interpretation_ids"] = [next(key for key,row in context["interpretation_sources"].items() if row["kind"] == "unknowns")]
|
|
class InvalidPatch(ProposalAdapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
if request['prompt_version']=='response_field_repair_v1':
|
|
self.requests.append(copy.deepcopy(request))
|
|
data=json.loads(request['messages'][1]['content'])['repair']
|
|
return ProviderResponse({'replacement':data['subject']['title']},'fixture',1,1)
|
|
return await super().generate(config,request,**kwargs)
|
|
adapter = InvalidPatch(mutate)
|
|
builder = HypothesisBuilder(interpreted_case[0],gateway(adapter))
|
|
result = run(builder)
|
|
assert result.status == "ERROR" and result.code == code
|
|
with PackageReader(result.package) as reader:
|
|
assert reader.manifest["artifacts"]["hypothesis_registry"] is None
|
|
assert reader.manifest["artifacts"]["hypotheses"] is None
|
|
again = run(builder)
|
|
assert again.reused and again.package == result.package
|
|
assert len(adapter.requests) == (2 if change == "duplicate_title" else 1)
|
|
|
|
|
|
def test_completed_replay_never_refills_or_regenerates_on_config_change(interpreted_case):
|
|
adapter = ProposalAdapter()
|
|
store = interpreted_case[0]
|
|
first = run(HypothesisBuilder(store,gateway(adapter)))
|
|
changed = HypothesisConfig(min_hypotheses=1,max_hypotheses=99,initial_budget=1000000)
|
|
again = run(HypothesisBuilder(CheckpointStore(store.workspace),gateway(adapter),changed),attempt_id="another")
|
|
assert again.reused and again.package == first.package
|
|
assert len(adapter.requests) == 1
|
|
with PackageReader(again.package) as reader:
|
|
registry = json.loads(reader.read_bytes(again.registry_ref))
|
|
assert registry["created_count"] == 3 and registry["max_hypotheses"] == 5
|
|
assert registry["initial_budget"] == 100
|
|
|
|
|
|
def test_policy_is_pinned_across_pending_retries(interpreted_case):
|
|
store = interpreted_case[0]
|
|
adapter = ProposalAdapter(response=ProviderUnavailable("LLM_CONNECTION"))
|
|
first = run(HypothesisBuilder(store,gateway(adapter)))
|
|
assert first.status == "RETRY_PENDING"
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_POLICY_CHANGED"):
|
|
run(HypothesisBuilder(store,gateway(adapter),HypothesisConfig(initial_budget=200)),attempt_id="new")
|
|
assert len(adapter.requests) == 1
|
|
|
|
|
|
def test_requires_confirmed_interpretation_without_calling_earlier_phase(built_case):
|
|
adapter = ProposalAdapter()
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_INTERPRETATION_NOT_READY"):
|
|
run(HypothesisBuilder(built_case[0],gateway(adapter)))
|
|
assert not adapter.requests
|
|
|
|
|
|
def test_portable_completed_generation(interpreted_case,tmp_path):
|
|
adapter = ProposalAdapter()
|
|
first = run(HypothesisBuilder(interpreted_case[0],gateway(adapter)))
|
|
receiving = CheckpointStore(Workspace(tmp_path / "other"))
|
|
receiving.intake(first.package)
|
|
result = run(HypothesisBuilder(receiving,gateway(adapter)))
|
|
assert result.reused and len(adapter.requests) == 1
|
|
assert result.package == receiving.latest("case-A","stable-package")
|
|
|
|
|
|
def test_explicit_recovery_after_rejection_does_not_create_ghost_ids(interpreted_case):
|
|
adapter = ProposalAdapter(count=6)
|
|
builder = HypothesisBuilder(interpreted_case[0],gateway(adapter))
|
|
assert run(builder).status == "ERROR"
|
|
adapter.count = 3
|
|
recovered = run(builder,attempt_id="corrected")
|
|
assert recovered.status == "COMPLETED"
|
|
with PackageReader(recovered.package) as reader:
|
|
registry = json.loads(reader.read_bytes(recovered.registry_ref))
|
|
assert registry["created_count"] == 3 and registry["next_id"] == 4
|
|
assert [r["id"] for r in registry["creations"]] == ["H1","H2","H3"]
|
|
assert len([name for name in reader.files if "/hypotheses/0001/errors/" in name]) == 1
|
|
|
|
|
|
def test_registry_and_allocation_tampering_is_rejected(interpreted_case):
|
|
result = run(HypothesisBuilder(interpreted_case[0],gateway(ProposalAdapter())))
|
|
with PackageReader(result.package) as reader:
|
|
document = json.loads(reader.read_bytes(result.hypotheses_ref))
|
|
registry = json.loads(reader.read_bytes(result.registry_ref))
|
|
context = json.loads(reader.read_bytes(document["context_ref"]))
|
|
changed = copy.deepcopy(registry)
|
|
changed["created_count"] = 0
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_REGISTRY_CHANGED"):
|
|
validate_generation(document,changed,context,result.hypotheses_ref)
|
|
changed = copy.deepcopy(document)
|
|
changed["hypotheses"][1]["budget_allocation"]["initial"] = 1000
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_ALLOCATION_CHANGED"):
|
|
validate_generation(changed,registry,context,result.hypotheses_ref)
|
|
|
|
|
|
def test_retry_deferral_exhaustion_and_phase_logging(interpreted_case,caplog):
|
|
store = interpreted_case[0]
|
|
now = [datetime(2026,1,1,tzinfo=timezone.utc)]
|
|
adapter = ProposalAdapter(response=ProviderUnavailable("LLM_CONNECTION"))
|
|
cfg = gateway(adapter,max_attempts=2,retry_delay_seconds=1)
|
|
builder = HypothesisBuilder(store,cfg,clock=lambda:now[0])
|
|
first = run(builder)
|
|
assert first.status == "RETRY_PENDING"
|
|
deferred = run(HypothesisBuilder(CheckpointStore(store.workspace),cfg,clock=lambda:now[0]))
|
|
assert deferred.status == "DEFERRED" and deferred.package == first.package
|
|
assert len(adapter.requests) == 1
|
|
now[0] += timedelta(seconds=1)
|
|
blocked = run(builder)
|
|
assert blocked.status == "BLOCKED" and blocked.code == "LLM_UNAVAILABLE"
|
|
with PackageReader(blocked.package) as reader:
|
|
assert reader.manifest["state"] == State.BLOCKED
|
|
assert reader.manifest["artifacts"]["hypothesis_registry"] is None
|
|
assert "phase GENERATING_HYPOTHESES failed" in caplog.text
|
|
|
|
|
|
@pytest.mark.parametrize("field,value,code", [("max_context_bytes",100,"HYPOTHESIS_CONTEXT_LIMIT"),
|
|
("max_output_bytes",100,"HYPOTHESIS_OUTPUT_LIMIT"),("max_summary_bytes",100,"HYPOTHESIS_SUMMARY_LIMIT")])
|
|
def test_operational_limits(interpreted_case,field,value,code):
|
|
adapter = ProposalAdapter()
|
|
builder = HypothesisBuilder(interpreted_case[0],gateway(adapter),limits=HypothesisLimits(**{field:value}))
|
|
if field == "max_context_bytes":
|
|
with pytest.raises(HypothesisError,match=code): run(builder)
|
|
assert not adapter.requests
|
|
else:
|
|
result = run(builder)
|
|
assert result.status == "ERROR" and result.code == code
|
|
|
|
|
|
@pytest.mark.parametrize("contract", ["hypotheses-output","hypotheses","hypothesis-policy","hypothesis-context","hypothesis-registry","hypothesis-error"])
|
|
def test_hypothesis_schemas(contract):
|
|
Draft202012Validator.check_schema(json.loads(files("viano_bug_analysis.schemas").joinpath(contract+'-v1.schema.json').read_text()))
|
|
|
|
|
|
def test_real_http_adapter(interpreted_case,http_endpoint):
|
|
url, seen, reply = http_endpoint
|
|
original_body = reply["body"]
|
|
# Obtain the exact I catalogue from the persisted phase context through an offline preparation.
|
|
adapter = ProposalAdapter(response=ProviderUnavailable("LLM_CONNECTION"))
|
|
store = interpreted_case[0]
|
|
pending = run(HypothesisBuilder(store,gateway(adapter)))
|
|
with PackageReader(pending.package) as reader:
|
|
context = json.loads(reader.read_bytes(reader.manifest["artifacts"]["hypothesis_context"]))
|
|
original_body["choices"][0]["message"]["content"] = json.dumps(proposals(context))
|
|
cfg = GatewayConfig((ProviderConfig(id="http",model="model",base_url=url,response_format="json_schema"),))
|
|
# New LLM operation, same pinned generation policy, after the existing retry's due time.
|
|
result = run(HypothesisBuilder(store,LLMGateway(cfg)),attempt_id="http")
|
|
assert result.status == "COMPLETED"
|
|
assert seen[0]["body"]["response_format"]["json_schema"]["schema"]["properties"]["hypotheses"]["maxItems"] == 5
|
|
|
|
|
|
def test_summary_escapes_untrusted_text(interpreted_case):
|
|
def change(payload,context): payload["hypotheses"][0]["title"] = '<script>x</script> [link](javascript:x)\n# injected'
|
|
result = run(HypothesisBuilder(interpreted_case[0],gateway(ProposalAdapter(change))))
|
|
with PackageReader(result.package) as reader:
|
|
summary = reader.read_bytes(result.summary_ref).decode()
|
|
assert '<script>' not in summary and '[link](javascript:' not in summary and '\n# injected' not in summary
|
|
|
|
|
|
def test_crash_after_final_publication_recovers_without_new_allocation(interpreted_case,monkeypatch):
|
|
store = interpreted_case[0]
|
|
index = store._index
|
|
def crash(path,manifest):
|
|
if manifest["artifacts"].get("hypotheses"): raise RuntimeError("simulated")
|
|
return index(path,manifest)
|
|
monkeypatch.setattr(store,"_index",crash)
|
|
adapter = ProposalAdapter()
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_INPUT"):
|
|
run(HypothesisBuilder(store,gateway(adapter)))
|
|
recovered = run(HypothesisBuilder(CheckpointStore(store.workspace),gateway(adapter)))
|
|
assert recovered.reused and recovered.status == "COMPLETED"
|
|
assert len(adapter.requests) == 1
|
|
assert json.loads((store.workspace.runtime / "current_job.json").read_bytes()) is None
|
|
|
|
|
|
def test_duplicate_dispatch_reports_busy(interpreted_case):
|
|
store = interpreted_case[0]
|
|
async def dispatch():
|
|
entered, release = asyncio.Event(), asyncio.Event()
|
|
class Waiting(ProposalAdapter):
|
|
async def generate(self,*args,**kw):
|
|
entered.set()
|
|
await release.wait()
|
|
return await super().generate(*args,**kw)
|
|
builder = HypothesisBuilder(store,gateway(Waiting()))
|
|
first = asyncio.create_task(builder.build("case-A","stable-package"))
|
|
await entered.wait()
|
|
with pytest.raises(LockBusy): await builder.build("case-A","stable-package")
|
|
release.set()
|
|
assert (await first).status == "COMPLETED"
|
|
asyncio.run(dispatch())
|
|
|
|
|
|
def test_concurrent_change_rejects_stale_generation(interpreted_case):
|
|
store = interpreted_case[0]
|
|
class Concurrent(ProposalAdapter):
|
|
async def generate(self,*args,**kwargs):
|
|
store.advance("case-A","stable-package",State.GENERATING_HYPOTHESES,step_id="concurrent",additions={"new-evidence":b"data"})
|
|
return await super().generate(*args,**kwargs)
|
|
with pytest.raises(HypothesisError,match="GENERATION_CHANGED"):
|
|
run(HypothesisBuilder(store,gateway(Concurrent())))
|
|
|
|
|
|
def test_reopen_preserves_previous_hypotheses_and_allocates_only_new_revision(interpreted_case):
|
|
store, case_builder = interpreted_case[:2]
|
|
adapter = ProposalAdapter()
|
|
first = run(HypothesisBuilder(store,gateway(adapter)))
|
|
with PackageReader(first.package) as reader: previous = reader.read_bytes(first.hypotheses_ref)
|
|
for state in (State.INVESTIGATING_HYPOTHESES,State.BUILDING_DIAGNOSIS,State.OUTPUT_READY,State.HANDED_OFF,State.REOPENED):
|
|
store.advance("case-A","stable-package",state,step_id="test-state:"+state)
|
|
case_builder.build("case-A","stable-package")
|
|
asyncio.run(CaseInterpreter(store,gateway(FixtureAdapter())).interpret("case-A","stable-package"))
|
|
second = run(HypothesisBuilder(store,gateway(adapter),HypothesisConfig(initial_budget=23)))
|
|
assert second.analysis_revision == 2
|
|
with PackageReader(second.package) as reader:
|
|
assert reader.read_bytes(first.hypotheses_ref) == previous
|
|
document = json.loads(reader.read_bytes(second.hypotheses_ref))
|
|
context = json.loads(reader.read_bytes(document["context_ref"]))
|
|
assert document["hypotheses"][0]["id"] == "H1" and document["hypotheses"][0]["budget_allocation"]["initial"] == 23
|
|
assert context["prior_generations"][0]["analysis_revision"] == 1
|
|
assert len(context["prior_generations"][0]["initial_proposals"]) == 3
|
|
|
|
|
|
def test_stale_case_rejected_before_generation(interpreted_case,tmp_path):
|
|
store = interpreted_case[0]
|
|
source = store.latest("case-A","stable-package")
|
|
changed = tmp_path / "changed.zip"
|
|
PackageWriter().write(source,changed,additions={"new-evidence":b"data"})
|
|
store.intake(changed)
|
|
adapter = ProposalAdapter()
|
|
with pytest.raises(HypothesisError,match="STALE_CASE"):
|
|
run(HypothesisBuilder(store,gateway(adapter)))
|
|
assert not adapter.requests
|
|
|
|
|
|
def test_terminal_status_markers_do_not_reopen_generation_capacity(interpreted_case):
|
|
store = interpreted_case[0]
|
|
adapter = ProposalAdapter()
|
|
builder = HypothesisBuilder(store,gateway(adapter))
|
|
first = run(builder)
|
|
# Lifecycle setup only; hypothesis state management is a later milestone.
|
|
marked = store.advance("case-A","stable-package",State.INVESTIGATING_HYPOTHESES,step_id="test-terminal-markers",
|
|
details={"test_hypothesis_statuses":{"H1":"rejected","H2":"rejected","H3":"budget_exhausted"}})
|
|
repeated = run(builder,attempt_id="new-candidates")
|
|
assert repeated.reused and repeated.package == marked
|
|
assert len(adapter.requests) == 1
|
|
with PackageReader(repeated.package) as reader:
|
|
registry = json.loads(reader.read_bytes(repeated.registry_ref))
|
|
assert registry["created_count"] == 3 and registry["remaining_slots"] == 2
|
|
|
|
|
|
def test_removing_index_pointers_cannot_reset_the_revision(interpreted_case,tmp_path):
|
|
store = interpreted_case[0]
|
|
adapter = ProposalAdapter()
|
|
builder = HypothesisBuilder(store,gateway(adapter))
|
|
first = run(builder)
|
|
changed = tmp_path / "removed-pointers.zip"
|
|
PackageWriter().write(first.package,changed,manifest_updates={"artifacts":{"hypotheses":None,"hypothesis_registry":None}})
|
|
store.intake(changed)
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_OUTPUT_CHANGED"):
|
|
run(builder,attempt_id="reset")
|
|
assert len(adapter.requests) == 1
|
|
|
|
|
|
def test_no_observations_prevents_unsupported_generation(make_package,tmp_path):
|
|
from test_case_interpreter import statement
|
|
store = CheckpointStore(Workspace(tmp_path / "empty"))
|
|
store.intake(make_package())
|
|
CaseBuilder(store).build("case-A","stable-package")
|
|
class Empty:
|
|
async def generate(self,*args,**kwargs):
|
|
return ProviderResponse({"observed_behavior":[],"expected_behavior":[],"tester_assumptions":[],"confirmed_technical_facts":[],
|
|
"behavior_gap":None,"first_known_divergence":None,"failure_stage":None,"contradictions":[],
|
|
"unknowns":[statement("No expected behavior","Q0001")],"relevant_entities":[],"investigation_direction":[],
|
|
"interpretation_summary":statement("No usable inputs","Q0001")},"fixture")
|
|
asyncio.run(CaseInterpreter(store,gateway(Empty())).interpret("case-A","stable-package"))
|
|
adapter = ProposalAdapter()
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_NO_OBSERVATIONS"):
|
|
run(HypothesisBuilder(store,gateway(adapter)))
|
|
assert not adapter.requests
|
|
|
|
|
|
def test_context_references_match_the_accepted_interpretation(interpreted_case):
|
|
result = run(HypothesisBuilder(interpreted_case[0],gateway(ProposalAdapter())))
|
|
with PackageReader(result.package) as reader:
|
|
document = json.loads(reader.read_bytes(result.hypotheses_ref))
|
|
context = json.loads(reader.read_bytes(document["context_ref"]))
|
|
interpretation = json.loads(reader.read_bytes(document["interpretation_ref"]))
|
|
for row in context["interpretation_sources"].values():
|
|
value = interpretation
|
|
for part in row["interpretation_pointer"].split("/")[1:]:
|
|
value = value[int(part)] if isinstance(value,list) else value[part]
|
|
assert value == row["data"]
|
|
|
|
|
|
def test_imported_hypotheses_without_creation_history_cannot_be_forgotten(interpreted_case,make_package,tmp_path):
|
|
store, _, _, original, entries = interpreted_case
|
|
first = run(HypothesisBuilder(store,gateway(ProposalAdapter())))
|
|
with PackageReader(first.package) as reader:
|
|
declarations = reader.read_bytes(first.hypotheses_ref)
|
|
with PackageReader(original) as reader:
|
|
manifest = reader.manifest
|
|
manifest["artifacts"]["hypotheses"] = "imported/declarations.json"
|
|
package = make_package(document=manifest,entries={**entries,"imported/declarations.json":declarations},name="foreign.zip")
|
|
foreign = CheckpointStore(Workspace(tmp_path / "foreign"))
|
|
foreign.intake(package)
|
|
CaseBuilder(foreign).build("case-A","stable-package")
|
|
asyncio.run(CaseInterpreter(foreign,gateway(FixtureAdapter())).interpret("case-A","stable-package"))
|
|
adapter = ProposalAdapter()
|
|
with pytest.raises(HypothesisError,match="HYPOTHESIS_EXISTING_REGISTRY"):
|
|
run(HypothesisBuilder(foreign,gateway(adapter)))
|
|
assert not adapter.requests
|
|
|
|
|
|
def test_historical_v1_start_resumes_and_completed_generation_replays(interpreted_case):
|
|
from viano_bug_analysis.analysis.hypothesis_contracts import build_context
|
|
from viano_bug_analysis.lifecycle.storage import json_bytes
|
|
store,_,interpreted,_,_=interpreted_case
|
|
with PackageReader(interpreted.package) as reader:
|
|
interpretation=json.loads(reader.read_bytes(interpreted.interpretation_ref))
|
|
case_context=json.loads(reader.read_bytes(interpretation['context_ref']))
|
|
policy=HypothesisConfig().policy()
|
|
context=build_context(case_context,interpretation,interpreted.interpretation_ref,policy,[],HypothesisLimits())
|
|
context_ref='analysis/hypotheses/0001/context.json'
|
|
policy_ref='analysis/hypotheses/0001/policy.json'
|
|
details={'operation':'hypotheses-generating','builder_version':'1','context_ref':context_ref,
|
|
'context_sha256':hashlib.sha256(json_bytes(context)).hexdigest(),'policy_ref':policy_ref,
|
|
'policy_sha256':hashlib.sha256(json_bytes(policy)).hexdigest(),
|
|
'interpretation_ref':interpreted.interpretation_ref,'interpretation_sha256':context['interpretation_sha256']}
|
|
generation=reader.manifest['package_generation']
|
|
store.advance('case-A','stable-package',State.GENERATING_HYPOTHESES,
|
|
step_id='hypothesis-builder:v1:revision:1:start',details=details,
|
|
additions={context_ref:json_bytes(context),policy_ref:json_bytes(policy)},
|
|
artifacts={'hypothesis_context':context_ref,'hypothesis_policy':policy_ref},expected_generation=generation)
|
|
model=ProposalAdapter()
|
|
builder=HypothesisBuilder(store,gateway(model))
|
|
result=run(builder)
|
|
assert result.status=='COMPLETED'
|
|
assert model.requests[0]['prompt_version']=='hypotheses_v1'
|
|
assert run(builder).reused and len(model.requests)==1
|
|
|
|
|
|
def test_compact_hypothesis_prompt_preserves_claims_sources_and_budget(interpreted_case):
|
|
from viano_bug_analysis.analysis.hypothesis_prompt import HypothesisPromptBuilder
|
|
from viano_bug_analysis.analysis.hypothesis_contracts import build_context
|
|
store,_,interpreted,_,_=interpreted_case
|
|
with PackageReader(interpreted.package) as reader:
|
|
interpretation=json.loads(reader.read_bytes(interpreted.interpretation_ref))
|
|
case_context=json.loads(reader.read_bytes(interpretation['context_ref']))
|
|
context=build_context(case_context,interpretation,interpreted.interpretation_ref,HypothesisConfig().policy(),[],HypothesisLimits())
|
|
context['case_context']['sources']['X9999']={'kind':'unread_reference','case_pointer':'/research/refs/0',
|
|
'data':{'path':'original/'+'a'*64+'/evidence.json','sha256':'b'*64,'size':100}}
|
|
prompt=HypothesisPromptBuilder()
|
|
old=prompt.build(context)
|
|
before=copy.deepcopy(context)
|
|
context['prompt_version']='hypotheses_v2'
|
|
new=prompt.build(context)
|
|
data=restore(json.loads(new['messages'][1]['content'])['input'])
|
|
assert data['policy']==context['policy']
|
|
assert data['interpretation_sources']==context['interpretation_sources']
|
|
assert set(data['case_context']['sources'])==set(case_context['sources'])|{'X9999'}
|
|
assert data['case_context']['sources']['O0001']['data']['text']==case_context['sources']['O0001']['data']['text']
|
|
assert len(new['messages'][1]['content'])<len(old['messages'][1]['content'])
|
|
assert prompt.build(before)==old
|
|
rules=new['messages'][0]['content']
|
|
assert 'conditional suspected' in rules and 'symbol-to-number mappings from memory' in rules
|
|
# Neither prompt supplies the known test-case cause or specific entity IDs.
|
|
assert '23487' not in rules and 'Dun Morogh' not in rules
|