import asyncio
import copy
import json

import pytest

from test_case_builder import case_setup
from test_investigation_tools import tool_case
from test_case_interpreter import gateway
from test_investigator import LoopAdapter,CASE,state
from test_working_brief import service_for
from test_project_research import rows
from viano_bug_analysis.analysis import Investigator,InvestigationConfig
from viano_bug_analysis.budget import BudgetManager
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore
from viano_bug_analysis.llm import ProviderResponse,LLMError
from viano_bug_analysis.llm.response_recovery import normalize,parse_text,MARKER,REPAIR_PROMPT,apply_repair,sha
from viano_bug_analysis.package import PackageReader,PackageValidator
from viano_bug_analysis.tools import InvestigationToolRouter


def loop(tool_case,adapter):
    return Investigator(tool_case[0],gateway(adapter),InvestigationConfig(response_recovery_version=1),
                        router=InvestigationToolRouter(project_research=service_for(tool_case)))


def test_local_normalization_is_bounded_explicit_and_preserves_values():
    raw={'update':json.dumps({'confidence':'plausible'}),'next':{'kind':'tool','operation':'project.query',
         'arguments':rows(filters=['id','eq',1])}}
    before=copy.deepcopy(raw);fixed,audit=normalize(raw)
    assert raw==before and audit['original_document']==raw
    assert fixed['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
    assert fixed['update']=={'confidence':'plausible'}
    assert len(audit['rules'])==2
    assert normalize(fixed)==(fixed,None)


@pytest.mark.parametrize('filters',[[],['id','eq'],['id','unknown',1],['id','eq',1,'extra']])
def test_ambiguous_filters_are_not_guessed(filters):
    raw={'next':{'kind':'tool','operation':'project.query','arguments':rows(filters=filters)}}
    assert normalize(raw)==(raw,None)


def test_fence_is_opt_in_and_duplicate_keys_remain_invalid():
    req={'prompt_version':'investigation_bootstrap_v22','output_schema':{MARKER:1}}
    original='```json\n{"next":{}}\n```'
    doc,audit=parse_text(original,req)
    assert doc=={'next':{}} and audit['original_text']==original
    with pytest.raises(LLMError):parse_text(original,{'output_schema':{}})
    with pytest.raises(LLMError):parse_text('```json\n{"a":1,"a":2}\n```',req)
    doc,audit=parse_text(json.dumps('{"next":{}}'),req)
    assert doc=={'next':{}} and audit['rules']==['decode_root_object:/']


def test_flat_key_value_filter_pairs_are_unambiguous():
    raw={'next':{'kind':'tool','operation':'project.query',
                 'arguments':rows(filters=['column','id','operator','eq','value',1])}}
    fixed,audit=normalize(raw)
    assert fixed['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
    assert audit['rules']==['flat_filter_pairs:/next/arguments/filters']


def test_flat_filter_executes_once_with_audit_and_deferred_notice(tool_case):
    def change(reply,c):reply['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=['id','eq',1])}
    adapter=LoopAdapter(change);worker=loop(tool_case,adapter)
    result=asyncio.run(worker.step(*CASE));assert result.status=='PROGRESSED'
    assert len(adapter.contexts)==1
    with PackageReader(result.package) as r:
        s=state(result);call=json.loads(r.read_bytes(s['hypotheses']['H1']['notes'][-1]['llm_ref']))
    assert call['outcome']['normalization']['original_document']['next']['arguments']['filters']==['id','eq',1]
    assert call['outcome']['document']['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
    assert len(BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions'])==1
    second=asyncio.run(worker.step(*CASE));assert second.status=='PROGRESSED'
    with PackageReader(second.package) as r:
        requests=[json.loads(r.read_bytes(n)) for n in r.files if n.startswith('llm/') and n.endswith('/request.json')]
    assert any('format_notice' in req['messages'][1]['content'] for req in requests)
    assert PackageValidator().validate(second.package).valid


class RepairAdapter(LoopAdapter):
    def __init__(self,tamper=False):super().__init__();self.requests=[];self.tamper=tamper
    async def generate(self,config,request,**kwargs):
        self.requests.append(copy.deepcopy(request))
        if request['prompt_version']==REPAIR_PROMPT:
            replacement={'column':'id','operator':'eq','value':999 if self.tamper else 1}
            return ProviderResponse({'replacement':replacement},'fixture',100,20)
        response=await super().generate(config,request,**kwargs)
        response.document['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=[['id','eq',1]])}
        return response


def test_small_repair_resumes_portably_and_preserves_whole_response(tool_case,tmp_path):
    adapter=RepairAdapter();worker=loop(tool_case,adapter)
    first=asyncio.run(worker.step(*CASE));assert first.status=='REJECTED'
    assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']
    with PackageReader(first.package) as r:
        c=json.loads(r.read_bytes(state(first)['active']['context_ref']))
    assert c['response_repair']['path']==['next','arguments','filters',0]
    portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
    from viano_bug_analysis.tools import InvestigationToolRouter
    resumed=Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service_for(tool_case)))
    second=asyncio.run(resumed.step(*CASE));assert second.status=='PROGRESSED'
    assert adapter.requests[-1]['prompt_version']==REPAIR_PROMPT
    sizes=[sum(len(m['content']) for m in req['messages']) for req in adapter.requests]
    assert sizes[-1]<sizes[0]/2
    assert 'case_interpretation' not in adapter.requests[-1]['messages'][1]['content']
    assert len(BudgetManager(portable).snapshot(*CASE).state['accounts']['H1']['actions'])==1
    with PackageReader(second.package) as r:
        s=state(second);call=json.loads(r.read_bytes(s['hypotheses']['H1']['notes'][-1]['llm_ref']))
    before=c['response_repair']['document'];after=call['outcome']['document']
    assert before['update']==after['update'] and before['assessment']==after['assessment']
    assert after['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
    assert call['outcome']['repair']['patch']=={'replacement':{'column':'id','operator':'eq','value':1}}
    assert PackageValidator().validate(second.package).valid
    # Re-reading/replaying the completed tool action must not execute it twice.
    assert second.package==portable.latest(*CASE)


def test_repair_cannot_change_a_query_value(tool_case):
    adapter=RepairAdapter(tamper=True);worker=loop(tool_case,adapter)
    first=asyncio.run(worker.step(*CASE));assert first.status=='REJECTED'
    second=asyncio.run(worker.step(*CASE));assert second.status=='REJECTED'
    assert second.code=='LLM_REPAIR_CONTENT_CHANGED'
    assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']


@pytest.mark.parametrize('field,value',[('kind','invented_action'),('evidence_refs',['E999999'])])
def test_unknown_action_or_invented_evidence_is_not_normalized(field,value,tool_case):
    def change(r,c):
        if field=='kind':r['next']['kind']=value
        else:r['update']['working_brief']={'format_version':2,'findings':[{
            'id':'B0001','direction':'support','text':'A claim.','scope':'Snapshot.',
            'evidence_refs':value}],'uncertainties':[],'next_question':'Next?','retired':[]}
    worker=loop(tool_case,LoopAdapter(change));result=asyncio.run(worker.step(*CASE))
    assert result.status=='REJECTED'
    assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']
    with PackageReader(result.package) as r:
        c=json.loads(r.read_bytes(state(result)['active']['context_ref']))
    assert 'response_repair' not in c


def test_repaired_lean_turn_remains_bound_after_portable_resume(tool_case,tmp_path):
    from test_investigation_hybrid import HybridAdapter
    class Adapter(HybridAdapter):
        def __init__(self):super().__init__(continuous=True);self.bad_sent=False;self.requests=[]
        async def generate(self,config,request,**kwargs):
            self.requests.append(copy.deepcopy(request))
            if request['prompt_version']==REPAIR_PROMPT:
                return ProviderResponse({'replacement':{'column':'id','operator':'eq','value':1}},'fixture',100,20)
            response=await super().generate(config,request,**kwargs)
            if request['prompt_version']=='investigation_project_v20' and not self.bad_sent:
                self.bad_sent=True
                response.document['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=[['id','eq',1]])}
            return response
    adapter=Adapter();router=InvestigationToolRouter(project_research=service_for(tool_case))
    worker=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,
                        lean_checkpoint_steps=0,response_recovery_version=1),router=router)
    for _ in range(3):assert asyncio.run(worker.step(*CASE)).status=='PROGRESSED'
    failed=asyncio.run(worker.step(*CASE));assert failed.status=='REJECTED'
    portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(failed.package)
    resumed=Investigator(portable,gateway(adapter),router=router)
    fixed=asyncio.run(resumed.step(*CASE));assert fixed.status=='PROGRESSED'
    assert adapter.requests[-1]['prompt_version']==REPAIR_PROMPT
    # Loading the next step verifies the repair call's previous-result/read-ID binding.
    following=asyncio.run(resumed.step(*CASE));assert following.status=='PROGRESSED'
    assert PackageValidator().validate(following.package).valid


def test_patch_never_changes_scalar_bindings_or_source_hash():
    original={'next':{'arguments':{'filters':[['id','eq',1]]}}}
    data={'document':original,'document_sha256':sha(original),'path':['next','arguments','filters',0]}
    request={'messages':[{}, {'content':json.dumps({'repair':data})}]}
    assert apply_repair(request,{'replacement':{'column':'id','operator':'eq','value':1}})['next']['arguments']['filters'][0]['value']==1
    with pytest.raises(LLMError,match='CONTENT_CHANGED'):
        apply_repair(request,{'replacement':{'column':'id','operator':'eq','value':2}})
    data['document_sha256']='0'*64;request['messages'][1]['content']=json.dumps({'repair':data})
    with pytest.raises(LLMError,match='BINDING'):apply_repair(request,{'replacement':{}})
    original={'next':{'arguments':{'filters':{'column':'id','operator':'eq','value':1}}}}
    data.update(document=original,document_sha256=sha(original),path=['next','arguments','filters'])
    request['messages'][1]['content']=json.dumps({'repair':data})
    with pytest.raises(LLMError,match='CONTENT_CHANGED'):
        apply_repair(request,{'replacement':[{'column':'eq','operator':'id','value':1}]})
    original={'next':{'arguments':{'filters':[['id','eq','name']]}}}
    data.update(document=original,document_sha256=sha(original),path=['next','arguments','filters',0])
    request['messages'][1]['content']=json.dumps({'repair':data})
    with pytest.raises(LLMError,match='CONTENT_CHANGED'):
        apply_repair(request,{'replacement':{'column':'name','operator':'eq','value':'id'}})


def test_native_adapter_retains_fenced_original_only_when_enabled(monkeypatch):
    from dataclasses import replace
    from test_parameter_and_bootstrap_guard import bootstrap
    from test_lean_investigation_decision import reply
    from test_llm_caching import config,native,scripted
    from viano_bug_analysis.llm.gemini import GeminiProvider
    request=bootstrap();request['output_schema'][MARKER]=1
    document=reply();document['update']['next_step']=None
    text='```json\n'+json.dumps(document)+'\n```'
    envelope=native(cached=0);envelope['candidates'][0]['content']['parts'][0]['text']=text
    adapter=GeminiProvider();scripted(adapter,monkeypatch,[envelope])
    result=adapter._generate(replace(config(),context_cache=False,response_format='json_object'),request,1000000)
    assert result.document==document and result.normalization['original_text']==text


def test_crash_after_normalized_plan_publication_does_not_repeat_llm_or_charge(tool_case,tmp_path,monkeypatch):
    def change(reply,c):reply['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=['id','eq',1])}
    adapter=LoopAdapter(change);worker=loop(tool_case,adapter)
    class SimulatedCrash(Exception):pass
    def crash(*args):raise SimulatedCrash('simulated interruption')
    monkeypatch.setattr(worker,'_execute_plan',crash)
    with pytest.raises(SimulatedCrash):asyncio.run(worker.step(*CASE))
    assert len(adapter.contexts)==1
    assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']
    portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(tool_case[0].latest(*CASE))
    resumed=Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service_for(tool_case)))
    result=asyncio.run(resumed.step(*CASE))
    assert result.status=='PROGRESSED' and len(adapter.contexts)==1
    assert len(BudgetManager(portable).snapshot(*CASE).state['accounts']['H1']['actions'])==1


def test_nonexistent_or_semantically_incomplete_repair_path_is_not_used():
    from viano_bug_analysis.llm.response_recovery import make_repair
    context={'previous_rejection_details':{'validation_keyword':'type','validation_path':'/update/working_brief/findings/0'}}
    assert make_repair(context,{'outcome':{'document':{'other':'data'}}},'unused') is None


def test_oversized_document_is_not_duplicated_for_normalization():
    raw={'text':'x'*65536,'next':{'kind':'tool','operation':'project.query','arguments':rows(filters=['id','eq',1])}}
    assert normalize(raw)==(raw,None)