import asyncio
import copy
import json
import os
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 Investigator,InvestigationConfig,InvestigationError,InvestigationLimits
from viano_bug_analysis.budget import BudgetConfig,BudgetManager
from viano_bug_analysis.lifecycle import CheckpointStore,Workspace,State
from viano_bug_analysis.lifecycle.async_tests import AsyncTestCoordinator
from viano_bug_analysis.lifecycle.errors import LockBusy
from viano_bug_analysis.lifecycle.locking import FileLock
from viano_bug_analysis.lifecycle.storage import key
from viano_bug_analysis.llm import LLMGateway,ProviderResponse,ProviderUnavailable,GatewayConfig,ProviderConfig,LLMError
from viano_bug_analysis.package import PackageReader,PackageValidator
from test_case_builder import case_setup
from test_investigation_tools import tool_case
from test_case_interpreter import gateway
from test_llm_gateway import http_endpoint

CASE=("case-A","stable-package")


class LoopAdapter:
    def __init__(self,change=None,test=False,unavailable=False):
        self.contexts=[];self.change=change;self.test=test;self.unavailable=unavailable

    async def generate(self,config,request,*,max_response_bytes):
        context=json.loads(request["messages"][1]["content"])["input"]
        self.contexts.append(copy.deepcopy(context))
        if self.unavailable:raise ProviderUnavailable("unavailable")
        previous=context["previous_result"]
        eid=previous["record"]["id"] if previous else None
        result={"assessment":{"semantic_score":50,"reason":"Scoped useful evidence","evidence_refs":[eid]} if previous else None,
                "update":{"confidence":"plausible" if previous else "unassessed","evidence_refs":[eid] if previous else [],
                          "resolve_questions":[],"add_questions":[],"note":"Check the packaged observation before drawing conclusions"},
                "next":{"kind":"finish","reason":"Initial question checked; preserve uncertainty"} if previous else
                       {"kind":"tool","operation":"source.read_context","arguments":{"source_ref":"tool/code.cpp","start_line":1,"end_line":2}}}
        if self.test and previous is None and context["selected_hypothesis"]=="H1":
            result["next"]={"kind":"test","executor":"human","question":"What occurs after interaction?","steps":["Interact once"],
                            "prerequisites":["Use the specified build"],"requested_observations":["Record the observed state"],"environment":"build-A"}
        if self.change:self.change(result,context)
        return ProviderResponse(result,"fixture-investigation",100,50)


def run(loop,steps=20,**kwargs):return asyncio.run(loop.run(*CASE,max_steps=steps,**kwargs))


def state(result):
    with PackageReader(result.package) as reader:return json.loads(reader.read_bytes(result.state_ref))


def test_full_loop_round_robin_assessment_budget_evidence_and_preservation(tool_case):
    adapter=LoopAdapter();loop=Investigator(tool_case[0],gateway(adapter))
    result=run(loop)
    assert result.status=="COMPLETED"
    assert [c["selected_hypothesis"] for c in adapter.contexts]==["H1","H2","H3","H1","H2","H3"]
    snapshot=BudgetManager(tool_case[0]).snapshot(*CASE)
    assert {a["status"] for a in snapshot.state["accounts"].values()}=={"investigation_finished"}
    assert snapshot.state["accounts"]["H1"]["remaining"]==94
    assert snapshot.state["accounts"]["H2"]["remaining"]==90
    assert snapshot.ready_for_evaluation
    assert state(result)["countercheck_complete"]
    assert {h["confidence"] for h in state(result)["hypotheses"].values()}=={"plausible"}
    with PackageReader(tool_case[1]) as old,PackageReader(result.package) as new:
        assert new.manifest["state"]==State.BUILDING_DIAGNOSIS
        assert new.manifest["artifacts"]["diagnosis"] is None
        assert b"Outstanding questions" in new.read_bytes(new.manifest["artifacts"]["investigation_summary"])
        for name in old.files:
            if name!="manifest.json":assert old.read_bytes(name)==new.read_bytes(name)
    assert PackageValidator().validate(result.package).valid
    assert run(loop).package==result.package and len(adapter.contexts)==6


def test_yield_resume_and_portable_state(tool_case,tmp_path):
    adapter=LoopAdapter();first=run(Investigator(tool_case[0],gateway(adapter)),steps=1)
    assert first.status=="YIELDED"
    portable=CheckpointStore(Workspace(tmp_path/"portable"));portable.intake(first.package)
    result=run(Investigator(portable,gateway(adapter)))
    assert result.status=="COMPLETED" and len(adapter.contexts)==6


@pytest.mark.parametrize("change",[
    lambda r,c:r.update(budget=100000),
    lambda r,c:r["next"].update(operation="shell.exec"),
    lambda r,c:r["next"]["arguments"].update(source_ref="../outside"),
    lambda r,c:r["update"].update(confidence="strongly_supported"),
    lambda r,c:r["update"].update(evidence_refs=["E999999"]),
    lambda r,c:r["update"].update(resolve_questions=["Q9999"]),
    lambda r,c:r["update"].update(add_questions=[c["hypotheses"][c["selected_hypothesis"]]["questions"][0]["text"]]),
    lambda r,c:r.update(assessment={"semantic_score":100,"reason":"Invented prior result","evidence_refs":["E000001"]}),
])
def test_invalid_entire_reply_rejected_before_tools_or_spending(tool_case,change):
    adapter=LoopAdapter(change);loop=Investigator(tool_case[0],gateway(adapter))
    result=asyncio.run(loop.step(*CASE))
    assert result.status=="REJECTED"
    snapshot=BudgetManager(tool_case[0]).snapshot(*CASE)
    assert {a["remaining"] for a in snapshot.state["accounts"].values()}=={100}
    assert not any(a["actions"] for a in snapshot.state["accounts"].values())


def test_valid_assessment_with_invalid_next_action_is_not_partially_applied(tool_case):
    def change(reply,context):
        if context["previous_result"]:reply["next"]={"kind":"tool","operation":"source.read_context","arguments":{"source_ref":"missing","start_line":1,"end_line":1}}
    loop=Investigator(tool_case[0],gateway(LoopAdapter(change)))
    run(loop,steps=3)
    before=BudgetManager(tool_case[0]).snapshot(*CASE)
    result=asyncio.run(loop.step(*CASE))
    assert result.status=="REJECTED"
    after=BudgetManager(tool_case[0]).snapshot(*CASE)
    assert after.state==before.state


def test_invalid_response_limit_is_bounded_and_no_free_infinite_loop(tool_case):
    adapter=LoopAdapter(lambda r,c:r.update(forbidden=True))
    result=run(Investigator(tool_case[0],gateway(adapter),InvestigationConfig(max_invalid_responses=2)))
    assert result.status=="COMPLETED" and len(adapter.contexts)==6
    assert {h["disposition"] for h in state(result)["hypotheses"].values()}=={"invalid_responses"}


def test_query_iteration_limit_allows_final_assessment_only(tool_case):
    def change(reply,context):
        if context["previous_result"]:
            reply["next"]={"kind":"tool","operation":"source.read_context","arguments":{"source_ref":"tool/code.cpp","start_line":1,"end_line":1}}
    result=run(Investigator(tool_case[0],gateway(LoopAdapter(change)),InvestigationConfig(max_iterations_per_hypothesis=1,max_invalid_responses=1)))
    assert result.status=="COMPLETED"
    assert all(len(a["actions"])==1 for a in BudgetManager(tool_case[0]).snapshot(*CASE).state["accounts"].values())


def test_strong_support_does_not_skip_alternative_hypotheses(tool_case):
    def change(reply,context):
        if context["previous_result"] and context["selected_hypothesis"]=="H1":reply["update"]["confidence"]="strongly_supported"
    adapter=LoopAdapter(change);result=run(Investigator(tool_case[0],gateway(adapter)))
    assert result.status=="COMPLETED"
    assert state(result)["hypotheses"]["H1"]["confidence"]=="strongly_supported"
    assert {c["selected_hypothesis"] for c in adapter.contexts}=={"H1","H2","H3"}


def test_question_updates_keep_history_and_require_evidence(tool_case):
    def change(reply,context):
        if context["previous_result"]:
            reply["update"]["resolve_questions"]=["Q0001"]
            reply["update"]["add_questions"]=["Which later transition still needs checking?"]
    result=run(Investigator(tool_case[0],gateway(LoopAdapter(change))))
    questions=state(result)["hypotheses"]["H1"]["questions"]
    assert questions[0]["resolved"] and questions[1]["id"]=="Q0002" and not questions[1]["resolved"]


def test_state_pointer_deletion_and_policy_change_do_not_restart(tool_case):
    loop=Investigator(tool_case[0],gateway(LoopAdapter()))
    first=run(loop,steps=1)
    before=BudgetManager(tool_case[0]).snapshot(*CASE).state
    changed=Investigator(tool_case[0],gateway(LoopAdapter()),InvestigationConfig(max_iterations_per_hypothesis=100))
    with pytest.raises(InvestigationError,match="INVESTIGATION_POLICY_CHANGED"):asyncio.run(changed.step(*CASE))
    assert BudgetManager(tool_case[0]).snapshot(*CASE).state==before
    tool_case[0].advance(*CASE,State.INVESTIGATING_HYPOTHESES,step_id="deleted-loop-pointer",artifacts={"investigation_state":None})
    with pytest.raises(InvestigationError,match="INVESTIGATION_POINTER_CHANGED"):asyncio.run(loop.step(*CASE))


def test_async_test_wait_confirmation_and_budget_preservation(tool_case):
    config=BudgetConfig(action_costs={"testing.request_human":40})
    adapter=LoopAdapter(test=True);loop=Investigator(tool_case[0],gateway(adapter),budget_config=config)
    blocked=run(loop)
    assert blocked.status=="BLOCKED" and blocked.code=="AWAITING_TEST_FEEDBACK"
    before=BudgetManager(tool_case[0],config).snapshot(*CASE)
    assert before.state["accounts"]["H1"]["remaining"]==60
    assert run(loop).status=="BLOCKED" and len(adapter.contexts)==1
    with PackageReader(blocked.package) as reader:
        request=json.loads(reader.read_bytes(reader.manifest["artifacts"]["async_test_request"]))
    feedback={f:request[f] for f in ("format_version","case_id","package_id","analysis_revision","request_id","hypothesis_id","action_id","executor")}
    feedback.update(format="viano-async-test-feedback",outcome="completed",observations=["Observed missing transition"],environment="build-A",tester_id="tester",
                    attachments=[],confirmation={"confirmed":True,"confirmed_by":"reviewer","confirmed_at":"2026-10-14T12:00:00Z"})
    AsyncTestCoordinator(tool_case[0]).confirm_feedback(feedback)
    assert BudgetManager(tool_case[0],config).snapshot(*CASE).state==before.state
    result=run(Investigator(tool_case[0],gateway(adapter)))
    assert result.status=="COMPLETED" and state(result)["analysis_revision"]==1
    assert BudgetManager(tool_case[0],config).snapshot(*CASE).state["accounts"]["H1"]["remaining"]==76


def test_provider_retry_defer_exhaustion_and_explicit_retry(tool_case):
    now=[datetime.now(timezone.utc)-timedelta(days=1)]
    adapter=LoopAdapter(unavailable=True)
    base=gateway(adapter);cfg=replace(base.config,max_attempts=2,retry_delay_seconds=1,blocked_retry_seconds=1)
    gw=LLMGateway(cfg,{"fixture":adapter})
    loop=Investigator(tool_case[0],gw,clock=lambda:now[0])
    first=asyncio.run(loop.step(*CASE));assert first.status=="RETRY_PENDING"
    assert asyncio.run(loop.step(*CASE)).status=="DEFERRED"
    now[0]+=timedelta(seconds=2)
    blocked=asyncio.run(loop.step(*CASE));assert blocked.status=="BLOCKED"
    assert {a["remaining"] for a in BudgetManager(tool_case[0]).snapshot(*CASE).state["accounts"].values()}=={100}
    now[0]+=timedelta(seconds=2)
    # CheckpointStore uses wall clock for eligibility; make the retry timestamp due.
    with PackageReader(blocked.package) as reader:due=reader.manifest["routing"]["retry_not_before"]
    assert datetime.fromisoformat(due)<=datetime.now(timezone.utc)
    tool_case[0].advance(*CASE,State.INVESTIGATING_HYPOTHESES,step_id="manual-provider-resume")
    adapter.unavailable=False
    result=asyncio.run(loop.step(*CASE,retry_id="provider-restored"))
    assert result.status=="PROGRESSED"


def test_after_evidence_publication_crash_no_duplicate_calls_costs_or_ids(tool_case,monkeypatch):
    adapter=LoopAdapter();loop=Investigator(tool_case[0],gateway(adapter))
    original=BudgetManager.record_evidence;once=[True]
    def crash(self,*args,**kwargs):
        if once[0]:once[0]=False;raise OSError("Crash after evidence publication")
        return original(self,*args,**kwargs)
    monkeypatch.setattr(BudgetManager,"record_evidence",crash)
    with pytest.raises(InvestigationError):asyncio.run(loop.step(*CASE))
    result=run(loop)
    assert result.status=="COMPLETED" and len(adapter.contexts)==6
    assert all(len(a["actions"])==1 for a in BudgetManager(tool_case[0]).snapshot(*CASE).state["accounts"].values())


def test_same_case_busy_and_configuration_limits(tool_case):
    loop=Investigator(tool_case[0],gateway(LoopAdapter()))
    with FileLock(tool_case[0].workspace.runtime/"locks"/f"investigation-{key(CASE[0])}.lock"):
        with pytest.raises(LockBusy):asyncio.run(loop.step(*CASE))
    with pytest.raises(InvestigationError):InvestigationConfig(max_invalid_responses=True)
    with pytest.raises(InvestigationError):InvestigationLimits(max_context_bytes=0)
    with pytest.raises(InvestigationError):run(loop,steps=0)


def test_validation_cache_uses_archive_bytes_not_timestamps(tool_case):
    from viano_bug_analysis.lifecycle import LifecycleError
    store=tool_case[0]
    store.latest(*CASE)
    old=sorted(tool_case[1].parent.glob("g*.zip"))[0]
    stat=old.stat()
    old.write_bytes(b"x"*stat.st_size)
    os.utime(old,ns=(stat.st_atime_ns,stat.st_mtime_ns))
    with pytest.raises(LifecycleError):store.latest(*CASE)


def test_unchanged_generations_do_not_repeat_zip_validation(tool_case,monkeypatch):
    store=tool_case[0];expected=store.latest(*CASE)
    def unexpected(*args):raise AssertionError("Unchanged archive was unnecessarily revalidated")
    monkeypatch.setattr(PackageReader,"__enter__",unexpected)
    assert store.latest(*CASE)==expected


def test_loop_through_actual_http_adapter(tool_case,http_endpoint):
    url,seen,reply=http_endpoint
    reply["body"]["choices"][0]["message"]["content"]=json.dumps({
        "assessment":None,"update":{"confidence":"unassessed","evidence_refs":[],"resolve_questions":[],"add_questions":[],"note":"Transport regression"},
        "next":{"kind":"tool","operation":"source.read_context","arguments":{"source_ref":"tool/code.cpp","start_line":1,"end_line":2}}})
    cfg=GatewayConfig((ProviderConfig(id="http",model="configured",base_url=url,response_format="json_schema"),))
    result=asyncio.run(Investigator(tool_case[0],LLMGateway(cfg)).step(*CASE))
    assert result.status=="PROGRESSED"
    assert seen[0]["body"]["response_format"]["json_schema"]["schema"]["additionalProperties"] is False
    assert state(result)["hypotheses"]["H1"]["last_record_ref"] is not None


def test_provider_configuration_error_blocks_without_abandoning_hypotheses(tool_case):
    class BrokenProvider:
        async def generate(self,*args,**kwargs):raise LLMError("LLM_CREDENTIALS")
    result=run(Investigator(tool_case[0],gateway(BrokenProvider())))
    assert result.status=="BLOCKED" and result.code=="LLM_CREDENTIALS"
    assert {h["disposition"] for h in state(result)["hypotheses"].values()}=={"active"}
    assert {a["remaining"] for a in BudgetManager(tool_case[0]).snapshot(*CASE).state["accounts"].values()}=={100}


@pytest.mark.parametrize("name",["investigation-output","investigation-state","investigation-context"])
def test_loop_schemas_valid(name):
    Draft202012Validator.check_schema(json.loads(files("viano_bug_analysis.schemas").joinpath(name+'-v1.schema.json').read_text()))