import json, math, itertools
KIND="Constructed analytical model; not a measured benchmark"
def fig(i,title,xlabel,ylabel,series,caption,assumptions):
    return dict(id=i,title=title,x_label=xlabel,y_label=ylabel,series=series,caption=caption,assumptions=assumptions,kind=KIND)
def check(name,passed,cases,detail):
    assert passed, name
    return dict(name=name,passed=bool(passed),cases=cases,detail=detail)

EVENTS=[('a',1,2,10),('b',4,5,20),('c',8,25,30),('d',11,12,40),('e',18,19,50),('f',7,40,60)]
def accepted(lag):
    return [e for e in EVENTS if e[2]<=((e[1]//10)+1)*10+lag]
def frontier(a,b): return [min(x,y) for x,y in zip(a,b)]
def run():
    lags=[0,5,10,15,20,25,30,35]; counts=[len(accepted(l)) for l in lags]; totals=[sum(e[3] for e in accepted(l)) for l in lags]
    ticks=[0,5,10,15,20,25,30]; a=[0,5,10,15,20,25,30];b=[0,2,4,4,4,20,30];minimum=frontier(a,b)
    # At t=15 a pending event at time 8 remains possible on B, so max=15 cannot close [0,10).
    tests=[check('Retention monotonicity under fixed cutoff model',all(set(e[0] for e in accepted(lags[i]))<=set(e[0] for e in accepted(lags[i+1])) for i in range(len(lags)-1)),7,'Same six events and deterministic cutoff for every policy.'),check('Exact count and weighted-sum disagree as quality metrics',len(accepted(15))==5 and sum(e[3] for e in accepted(15))==150,1,'Five of six records is 83.33 percent; their value is 150 of 210.'),check('Min frontier safe against each input bound',all(w<=x and w<=y for w,x,y in zip(minimum,a,b)),7,'A scalar frontier represents inclusive lower bound on possible future timestamps.'),check('Max frontier counterexample',max(a[3],b[3])>=10 and b[3]<=8<10,1,'B can still produce timestamp 8 despite A reaching 15.'),check('Revision deltas telescope',sum([30,60-30,120-60])==120,1,'Window [0,10) initial total 30, then 60, then 120.')]
    return dict(slug='event-time-finality',checks=tests,figures=[fig('retention','Retention versus accepted event mass','Extra retention seconds after window end','Accepted share',[dict(name='Event count fraction',x=lags,y=[n/6 for n in counts]),dict(name='Value fraction',x=lags,y=[v/210 for v in totals])],'Six author-constructed records; cutoff is arrival <= window end + lag, not a simulation of a specific runner.',['Window width 10 seconds','No duplicate records','Final complete total 210']),fig('frontiers','One slow source limits combined progress','Observation time seconds','Event-time frontier seconds',[dict(name='Source A',x=ticks,y=a),dict(name='Source B',x=ticks,y=b),dict(name='Safe combined minimum',x=ticks,y=minimum)],'Constructed watermark trace. Progress stall is a correctness signal unless a justified idle-source contract excludes B.',['Both sources remain active','Future timestamps are bounded below by each input frontier','No processing delay inside the operator'])],result_summary=dict(total_records=6,total_value=210,lag0_count=len(accepted(0)),lag0_value=sum(e[3] for e in accepted(0)),lag15_count=len(accepted(15)),lag15_value=sum(e[3] for e in accepted(15)),lag30_count=len(accepted(30)),lag30_value=sum(e[3] for e in accepted(30)),window0_revisions=[30,60,120],frontier_at15=minimum[3]))

if __name__=="__main__":
    print(json.dumps(run(),indent=2))