#!/usr/bin/env python3 # Standalone synthetic fixture; no user files or external network. import copy,json initial=[{'id':f'R-{n}','version':1}for n in range(100,108)] current=copy.deepcopy(initial);pages=[copy.deepcopy(current[:3])] current.insert(0,{'id':'R-099','version':1});next(x for x in current if x['id']=='R-102')['version']=2 pages.extend([copy.deepcopy(current[3:6]),copy.deepcopy(current[6:9])]) ledger={};duplicates=[];conflicts=[] for p,items in enumerate(pages,1): for item in items: if item['id']in ledger: duplicates.append({'id':item['id'],'page':p}) if ledger[item['id']]!=item:conflicts.append({'id':item['id'],'old':ledger[item['id']],'new':item}) else:ledger[item['id']]=item missing=sorted({x['id']for x in current}-set(ledger));assert missing==['R-099'] and len(duplicates)==1 and len(conflicts)==1 chain={'start':{'items':initial[:3],'next':'middle-empty'},'middle-empty':{'items':[],'next':'last-a'},'last-a':{'items':initial[3:6],'next':'last-b'},'last-b':{'items':initial[6:],'next':''}} def collect(responses): token='start';seen=set();data=[];trace=[] while token: if token in seen:raise ValueError('repeated continuation token') seen.add(token);response=responses[token];data+=response['items'];trace.append({'token':token,'count':len(response['items']),'next':response['next']});token=response['next'] return data,trace snapshot,trace=collect(chain);assert snapshot==initial loopBlocked=False try:collect({'start':{'items':[],'next':'start'}}) except ValueError:loopBlocked=True assert loopBlocked result={'initial':initial,'currentAfterMutation':current,'offsetPages':pages,'offsetAudit':{'received':sum(map(len,pages)),'unique':len(ledger),'duplicates':duplicates,'conflicts':conflicts,'missingAgainstKnownCurrent':missing},'snapshotCursor':{'trace':trace,'ids':[r['id']for r in snapshot],'completeAgainstInitial':True},'loopBlocked':True,'scope':'Deterministic local response fixtures, no vendor API requests. Missing IDs are knowable only because this fixture exposes a reference inventory. Cursor alone does not imply provider snapshot consistency.'} import sys,json from datetime import datetime,timezone result={'experiment':'api-pagination-overlap-ledger','executedAt':datetime.now(timezone.utc).isoformat(),'runtime':sys.version.split()[0],'synthetic':True,**result} print(json.dumps(result,ensure_ascii=False,indent=2))