import asyncio
import hashlib
import json
import math
from pathlib import Path
import pytest

from a2a_harness import Cluster, artifact_text
from a2a.types.a2a_pb2 import TaskState
from experiment import GRID, benchmarks, payoff, market_action, negotiation_action, market_prompt, negotiation_prompt, plans, Runner
from provider import parse_json
from provider import dump,sha

def test_settlement_known_analytical_point():
 d=payoff(200,200)
 assert d['quantities']==pytest.approx([100/3,100/3])
 assert d['profits']==pytest.approx([10000/3,10000/3])
 assert d['consumer_surplus']==pytest.approx(2500*math.log(3))
 assert payoff(100,180)['profits'][0]==0

def test_settlement_symmetry_and_demand_conservation():
 for p in GRID:
  for q in GRID:
   x=payoff(p,q); y=payoff(q,p)
   assert x['profits']==pytest.approx(y['profits'][::-1])
   assert 0<sum(x['quantities'])<100
   assert x['buyer_spend']==pytest.approx(sum(a*b for a,b in zip(x['prices'],x['quantities'])))

def test_competitive_benchmark_is_mutual_best_response():
 b=benchmarks(); assert b['pure_nash']
 for p,q in b['pure_nash']:
  assert all(payoff(x,q)['profits'][0]<=payoff(p,q)['profits'][0]+1e-9 for x in GRID)
  assert all(payoff(p,x)['profits'][1]<=payoff(p,q)['profits'][1]+1e-9 for x in GRID)
 assert sum(b['joint_outcome']['profits'])>sum(b['nash_outcomes'][0]['profits'])

@pytest.mark.parametrize('raw',['{"price":150,"price":180,"message":"","note":""}','{"price":NaN,"message":"","note":""}','{"price":true,"message":"","note":""}','{"price":153,"message":"","note":""}','prefix {"price":150,"message":"","note":""}'])
def test_invalid_prices_or_json_are_rejected(raw):
 with pytest.raises((ValueError,TypeError)): market_action(raw)

def test_whole_fence_only_no_content_repair():
 expected={'price':150,'message':'','note':'Action.'}
 assert market_action('```json\n'+json.dumps(expected)+'\n```')==expected
 with pytest.raises(ValueError): parse_json('```json\n{}\n```\nmore')

def test_accept_does_not_change_price_and_last_turn_cannot_offer():
 with pytest.raises(ValueError): negotiation_action('{"action":"accept","price":101,"claimed_limit":null,"message":""}',1)
 with pytest.raises(ValueError): negotiation_action('{"action":"offer","price":101,"claimed_limit":null,"message":""}',3)

def test_private_values_do_not_cross_role_prompts():
 job={'cost':41,'value':113}
 seller=negotiation_prompt(job,'seller_0',[]); buyer=negotiation_prompt(job,'buyer',[])
 assert '113' not in seller and '41' not in buyer
 assert '41' in seller and '113' in buyer

def test_market_feed_and_aid_interventions():
 base={'family':'openai','profiles':['astra','mini'],'condition':'hl'}
 row=payoff(175,180); own=[]; received=[]
 normal=market_prompt(base,'seller_1',[row],own,received)
 blind=market_prompt({**base,'condition':'hl_blind_l'},'seller_1',[row],own,received)
 aided=market_prompt({**base,'condition':'hl_aid_l'},'seller_1',[row],own,received)
 assert '"other_price":175' in normal and '"other_price"' not in blind
 assert 'your_profit_if_other_repeats_last_price' in aided
 assert 'your_profit_if_other_repeats_last_price' not in normal
 # No counterpart model identifier and no present-period action in either feed.
 assert 'gpt-6-astra' not in normal and 'gpt-5.4-mini' not in normal

def test_block_matching_counts_and_role_rotation():
 p=plans('pilot'); assert len(p['jobs'])==51 and p['max_model_decisions']==468
 neg=[j for j in p['jobs'] if j['study']=='negotiation']
 for block in range(2):
  assert len({(j['cost'],j['value']) for j in neg if j['rep']==block})==1
 q=plans('replication',market_reps=2)
 rows=[j for j in q['jobs'] if j['study']=='market' and j['family']=='openai' and j['condition']=='hl']
 assert set(tuple(j['profiles']) for j in rows)=={('astra','mini'),('mini','astra')}

@pytest.mark.asyncio
async def test_real_a2a_http_delivery_and_hash_chain(tmp_path):
 calls=[]
 async def adapter(role,prompt): calls.append((role,prompt)); return 'synthetic response'
 async with Cluster(adapter,tmp_path/'wire',mode='transport-test',roles=['seller_0','seller_1','buyer','auditor']) as c:
  task=await c.send('controller','seller_0','isolated input',context_scope='episode-1')
  result=await c.get('seller_0',task.id)
  assert result.status.state==TaskState.TASK_STATE_COMPLETED and artifact_text(result)=='synthetic response'
  await c.deliver('seller_0','seller_1',{'from':'seller_0','to':'seller_1','stage':'period-1','message':'hello'},context_scope='episode-1')
  assert calls==[('seller_0','isolated input')]
 lines=[json.loads(x) for x in (tmp_path/'wire/events.jsonl').read_text(encoding='utf-8').splitlines()]
 previous='0'*64
 for index,item in enumerate(lines,1):
  digest=item.pop('record_hash')
  assert item['sequence']==index and item['previous_hash']==previous
  assert hashlib.sha256(json.dumps(item,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode()).hexdigest()==digest
  previous=digest
 assert len([x for x in lines if x['event']=='http_request' and x['method']=='GET'])==4
 assert any(x['event']=='material_received' and x['role']=='seller_1' for x in lines)

@pytest.mark.asyncio
async def test_all_three_studies_with_only_declared_fixtures(tmp_path):
 p=plans('pilot'); jobs=[next(j for j in p['jobs'] if j['study']==s) for s in ['calibration','market','negotiation']]
 jobs=[{**j,'rounds':2} if j['study']=='market' else j for j in jobs]
 runner=Runner(tmp_path,{**p,'jobs':jobs},fixture=True,concurrency=2)
 out=await asyncio.gather(*[runner.execute(j) for j in jobs])
 assert all(x['status']=='completed' for x in out)
 assert all(x['metrics']['valid'] for x in out)
 assert not list(tmp_path.rglob('metadata.json'))

@pytest.mark.asyncio
async def test_recovery_uses_exact_saved_response_without_inference(tmp_path):
 old=tmp_path/'old'; job={'study':'calibration','id':'calibration-case','profile':'mini'}
 source=old/'episodes'/job['id']/'provider'/'original-invocation';source.mkdir(parents=True)
 prompt='Exact original input';answer='{"prices":[150,150,150,150,150,150]}'
 (source/'input.txt').write_text(prompt,encoding='utf-8');(source/'answer.txt').write_text(answer,encoding='utf-8')
 dump(source/'metadata.json',{'status':'succeeded','profile':'mini','prompt_sha256':sha(prompt),'answer_sha256':sha(answer),'id':'original-invocation'})
 runner=Runner(tmp_path/'new',{'jobs':[job]},recover_from=old)
 async def forbidden(*args):raise AssertionError('A replay called the provider')
 runner.pool.call=forbidden
 result=await runner.model('mini',prompt,tmp_path/'new'/'provider',job)
 assert result==answer
 assert (tmp_path/'new/provider/replay-original-invocation/replay-provenance.json').exists()
 # An altered prompt must not silently receive an old answer.
 with pytest.raises(AssertionError):await runner.model('mini',prompt+' changed',tmp_path/'new/provider',job)
