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 = 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_v2"
    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")]
    adapter = ProposalAdapter(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) == 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=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