import asyncio
import copy
import json

import pytest
from jsonschema import Draft202012Validator

from test_case_builder import case_setup
from test_investigation_tools import tool_case
from test_case_interpreter import gateway
from test_compact_investigation import context
from test_lean_investigation_decision import reply
from test_investigator import LoopAdapter,CASE,state
from test_project_research import rows
from test_response_recovery import loop,RepairAdapter
from viano_bug_analysis.analysis.investigation_prompt import output_schema
from viano_bug_analysis.budget import BudgetManager
from viano_bug_analysis.llm import ProviderResponse
from viano_bug_analysis.llm.response_recovery import (make_repair,repair_request,apply_repair,
    filter_contract_feedback,sha,admission_reason,REPAIR_PROMPT)
from viano_bug_analysis.package import PackageReader,PackageValidator


def setup_case(filters):
    c=context();r=reply()
    r['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=filters)}
    c['previous_rejection']='INVESTIGATION_OUTPUT'
    c['previous_rejection_details']={'validation_path':'/next/arguments/filters/0',
        'validation_keyword':'type','expected_types':['object'],'actual_type':'string'}
    return c,{'outcome':{'document':r}}


def test_flat_filter_is_repaired_as_whole_list_not_first_scalar():
    c,call=setup_case(['id','eq',1])
    plan,binding=make_repair(c,call,'source-call.json')
    assert plan['path']==['next','arguments','filters'] and plan['schema']['type']=='array'
    c.update(response_repair=plan,repair_binding=binding)
    fixed=apply_repair(repair_request(c),{'replacement':[{'column':'id','operator':'eq','value':1}]})
    assert fixed['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
    assert fixed['update']==call['outcome']['document']['update']
    Draft202012Validator(output_schema(c,strict=True)).validate(fixed)


@pytest.mark.parametrize('filters',[
    ['LogTitle','like','%Wolpertinger%'],
    [['LogTitle','like','%Wolpertinger%']],
    [{'column':'LogTitle','operator':'like','value':'%Wolpertinger%'}],
    ['column','LogTitle','operator','like','value','%Wolpertinger%'],
])
def test_unknown_operator_has_no_format_repair_or_translation(filters):
    c,call=setup_case(filters);before=copy.deepcopy(call)
    feedback=filter_contract_feedback(call['outcome']['document'])
    assert feedback['validation_fields']==['operator']
    assert 'Allowed: eq,ne,lt,le,gt,ge,contains,in' in feedback['requirement']
    assert make_repair(c,call,'source-call.json') is None
    assert call==before


@pytest.mark.parametrize('filters',[['id'],['id','eq'],[{'column':'id','operator':'eq'}],{'0':{'column':'id'}}])
def test_incomplete_filter_cannot_invent_missing_values(filters):
    c,call=setup_case(filters)
    assert make_repair(c,call,'source-call.json') is None


def test_direct_scalar_target_is_rejected_before_api():
    doc={'next':{'arguments':{'filters':['LogTitle','like','%Wolpertinger%']}}}
    assert admission_reason(doc,['next','arguments','filters',0],{'type':'object'})=='scalar_target'


def test_operator_contract_feedback_reaches_next_regular_call(tool_case):
    def change(r,c):
        filters=['id','like','%1%'] if not c.get('previous_rejection') else [{'column':'id','operator':'eq','value':1}]
        r['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=filters)}
    adapter=LoopAdapter(change);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 'response_repair' not in c and c['previous_rejection_details']['validation_fields']==['operator']
    second=asyncio.run(worker.step(*CASE));assert second.status=='PROGRESSED'
    assert len(adapter.contexts)==2 and PackageValidator().validate(second.package).valid


def test_rejected_patch_is_saved_but_not_logged(tool_case,caplog):
    adapter=RepairAdapter(tamper=True);worker=loop(tool_case,adapter)
    assert asyncio.run(worker.step(*CASE)).status=='REJECTED'
    second=asyncio.run(worker.step(*CASE));assert second.code=='LLM_REPAIR_CONTENT_CHANGED'
    with PackageReader(second.package) as r:
        call=json.loads(r.read_bytes(state(second)['hypotheses']['H1']['notes'][-1]['llm_ref']))
    saved=call['outcome']['rejected_repair']
    assert saved['patch']=={'replacement':{'column':'id','operator':'eq','value':999}}
    assert saved['sha256']==sha(saved['patch']) and saved['reason']==second.code
    assert 'replacement' not in caplog.text
    assert PackageValidator().validate(second.package).valid
    assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']


def test_schema_rejected_patch_is_also_retained(tool_case):
    class Adapter(RepairAdapter):
        async def generate(self,config,request,**kwargs):
            if request['prompt_version']==REPAIR_PROMPT:
                return ProviderResponse({'replacement':{'column':'id','operator':'unknown','value':1}},'fixture',100,20)
            return await super().generate(config,request,**kwargs)
    worker=loop(tool_case,Adapter());assert asyncio.run(worker.step(*CASE)).status=='REJECTED'
    second=asyncio.run(worker.step(*CASE));assert second.code=='LLM_OUTPUT_SCHEMA'
    with PackageReader(second.package) as r:
        call=json.loads(r.read_bytes(state(second)['hypotheses']['H1']['notes'][-1]['llm_ref']))
    assert call['outcome']['rejected_repair']['patch']['replacement']['operator']=='unknown'
    assert call['outcome']['diagnostics']['validation_path']=='/replacement/operator'


def test_malformed_repair_json_is_retained_without_log_echo(tool_case,caplog):
    from viano_bug_analysis.llm.response_recovery import parse_text
    raw='{"replacement":"PRIVATE_REPAIR_TEXT"'
    class Adapter(RepairAdapter):
        async def generate(self,config,request,**kwargs):
            if request['prompt_version']==REPAIR_PROMPT:parse_text(raw,request)
            return await super().generate(config,request,**kwargs)
    worker=loop(tool_case,Adapter());assert asyncio.run(worker.step(*CASE)).status=='REJECTED'
    second=asyncio.run(worker.step(*CASE));assert second.code=='LLM_RESPONSE'
    with PackageReader(second.package) as r:
        call=json.loads(r.read_bytes(state(second)['hypotheses']['H1']['notes'][-1]['llm_ref']))
    saved=call['outcome']['rejected_repair']
    assert saved['original_text']==raw and saved['sha256']==sha(raw)
    assert 'PRIVATE_REPAIR_TEXT' not in caplog.text
    assert PackageValidator().validate(second.package).valid


def test_another_invalid_field_prevents_format_repair():
    c,call=setup_case(['id','eq',1]);call['outcome']['document']['next']['kind']='invented_action'
    assert make_repair(c,call,'source-call.json') is None


def test_ineligible_saved_plan_blocks_without_another_provider_call(tool_case):
    adapter=RepairAdapter();worker=loop(tool_case,adapter)
    first=asyncio.run(worker.step(*CASE));assert first.status=='REJECTED'
    s=state(first)
    with PackageReader(first.package) as r:c=json.loads(r.read_bytes(s['active']['context_ref']))
    plan=c['response_repair'];plan['document']['next']['arguments']['filters']=['id','eq',1]
    plan['document_sha256']=sha(plan['document'])
    assert plan['path']==['next','arguments','filters',0] # now points to a scalar
    s['active']['context_ref']=s['active']['context_ref'].replace('.json','-ineligible.json')
    worker._save(s,context=c)
    blocked=asyncio.run(worker.step(*CASE))
    assert blocked.status=='BLOCKED' and blocked.code=='LLM_REPAIR_INELIGIBLE'
    assert len(adapter.requests)==1
    assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']


def test_oversized_rejected_patch_is_not_retained(tool_case):
    class Adapter(RepairAdapter):
        async def generate(self,config,request,**kwargs):
            if request['prompt_version']==REPAIR_PROMPT:
                return ProviderResponse({'replacement':{'column':'id','operator':'eq','value':'x'*70000}},'fixture',100,20)
            return await super().generate(config,request,**kwargs)
    worker=loop(tool_case,Adapter());assert asyncio.run(worker.step(*CASE)).status=='REJECTED'
    second=asyncio.run(worker.step(*CASE));assert second.status=='REJECTED'
    with PackageReader(second.package) as r:
        call=json.loads(r.read_bytes(state(second)['hypotheses']['H1']['notes'][-1]['llm_ref']))
    assert 'rejected_repair' not in call['outcome']
    assert PackageValidator().validate(second.package).valid