"""R645：单写者JSON检查点重放，事件身份与有效载荷一起去重。"""
from decimal import Decimal as D
from pathlib import Path
from tempfile import TemporaryDirectory
from copy import deepcopy
import json,hashlib,os

def encoded(value):return json.dumps(value,sort_keys=True,separators=(',',':'),allow_nan=False).encode()
def digest(value):return hashlib.sha256(encoded(value)).hexdigest()
def save(path,state):
    envelope={'state':state,'sha256':digest(state)};temp=path.with_suffix('.tmp')
    with temp.open('wb') as f:f.write(encoded(envelope));f.flush();os.fsync(f.fileno())
    os.replace(temp,path)
def load(path):
    envelope=json.loads(path.read_text(encoding='utf-8'))
    if digest(envelope['state'])!=envelope['sha256']:raise ValueError('checkpoint checksum mismatch')
    return envelope['state']
def step(state,event):
    out=deepcopy(state)
    if not isinstance(event['id'],str) or not event['id'].strip():raise ValueError('invalid event identity')
    key=digest(event)
    if event['id'] in out['seen']:
        if out['seen'][event['id']]!=key:raise ValueError('conflicting duplicate')
        return out,'duplicate'
    if event['version']!=out['version']:raise ValueError('rule version unknown')
    if type(event['seq']) is not int or event['seq']!=out['seq']+1:raise ValueError('event sequence gap')
    if event['price'] is None:return out,'quote_unknown_stop'
    price=D(event['price']);fee=D(event['fee']);qty=event['qty']
    if not price.is_finite() or price<=0 or not fee.is_finite() or fee<0 or type(qty) is not int or qty<=0 or event['side'] not in {'buy','sell'}:raise ValueError('invalid execution')
    signed=qty if event['side']=='buy' else -qty;before=D(out['cash'])+out['shares']*price
    cash=D(out['cash'])-signed*price-fee;shares=out['shares']+signed
    if cash<0 or shares<0:raise ValueError('unfunded replay')
    out.update(cash=str(cash),shares=shares,seq=event['seq']);out['seen'][event['id']]=key
    assert cash+shares*price==before-fee
    return out,'applied'

initial={'version':'v1','seq':0,'cash':'1000','shares':0,'seen':{}}
events=[{'id':'E1','seq':1,'version':'v1','side':'buy','qty':2,'price':'100','fee':'1'},
        {'id':'E2','seq':2,'version':'v1','side':'sell','qty':1,'price':'110','fee':'1'},
        {'id':'E3','seq':3,'version':'v1','side':'buy','qty':1,'price':'105','fee':'1'}]
uninterrupted=deepcopy(initial)
for event in events:uninterrupted,_=step(uninterrupted,event)
with TemporaryDirectory() as tmp:
    path=Path(tmp)/'checkpoint.json';first,_=step(initial,events[0]);save(path,first)
    # 模拟提交后确认丢失：恢复后接收同一E1，不再扣款。
    recovered=load(path);statuses=[]
    for event in [events[0],events[1],events[1],events[2]]:
        recovered,status=step(recovered,event);statuses.append(status);save(path,recovered)
    assert load(path)==uninterrupted==recovered
    assert statuses==['duplicate','applied','duplicate','applied']
    known=deepcopy(first);unknown,status=step(known,dict(events[1],price=None))
    assert status=='quote_unknown_stop' and unknown==known
    for bad in [dict(events[1],version='v2'),events[2]]:
        try:step(first,bad)
        except ValueError:pass
        else:raise AssertionError('missing rule or event gap bridged')
    try:step(recovered,dict(events[1],price='111'))
    except ValueError:pass
    else:raise AssertionError('conflicting duplicate ignored')
    damaged=json.loads(path.read_text(encoding='utf-8'));damaged['state']['cash']='9999';path.write_text(json.dumps(damaged),encoding='utf-8')
    try:load(path)
    except ValueError:pass
    else:raise AssertionError('checkpoint mutation hidden')
assert recovered['cash']=='802' and recovered['shares']==2 and D(recovered['cash'])+2*D(110)==D(1022)
print('replay_statuses='+str(statuses)+' cash=802 shares=2 terminal_mark=110 NAV=1022')
print('duplicate_payload=checked unknown_quote=stop unknown_version=stop sequence_gap=stop checksum_mutation=rejected single_writer_only=True')