import sqlite3,json,threading,tempfile
from pathlib import Path
class PayloadMismatch(ValueError): pass
SCHEMA='''
CREATE TABLE orders(id INTEGER PRIMARY KEY AUTOINCREMENT, account TEXT NOT NULL, sku TEXT NOT NULL, quantity INTEGER NOT NULL CHECK(quantity>0));
CREATE TABLE request_ledger(account TEXT NOT NULL, operation TEXT NOT NULL, request_key TEXT NOT NULL, payload_json TEXT NOT NULL, state TEXT NOT NULL, order_id INTEGER, PRIMARY KEY(account,operation,request_key), FOREIGN KEY(order_id) REFERENCES orders(id));
'''
def connect(path):
 db=sqlite3.connect(path,timeout=5,isolation_level=None);db.execute('PRAGMA foreign_keys=ON');return db
def create(path):
 db=connect(path);db.executescript(SCHEMA);db.close()
def submit(path,account,key,payload,fail_after_order=False):
 if set(payload)!={'sku','quantity'} or not isinstance(payload['sku'],str) or not payload['sku'] or type(payload['quantity']) is not int or payload['quantity']<=0: raise ValueError('invalid payload')
 canonical=json.dumps(payload,sort_keys=True,separators=(',',':'))
 db=connect(path)
 try:
  db.execute('BEGIN IMMEDIATE')
  old=db.execute('SELECT payload_json,state,order_id FROM request_ledger WHERE account=? AND operation=? AND request_key=?',(account,'create_order',key)).fetchone()
  if old:
   if old[0]!=canonical: raise PayloadMismatch('same key, changed payload')
   assert old[1]=='completed'
   db.execute('COMMIT');return {'order_id':old[2],'result':'replayed'}
  db.execute('INSERT INTO request_ledger VALUES(?,?,?,?,?,NULL)',(account,'create_order',key,canonical,'working'))
  order_id=db.execute('INSERT INTO orders(account,sku,quantity) VALUES(?,?,?)',(account,payload['sku'],payload['quantity'])).lastrowid
  if fail_after_order: raise RuntimeError('injected failure before ledger completion')
  db.execute('UPDATE request_ledger SET state=?,order_id=? WHERE account=? AND operation=? AND request_key=?',('completed',order_id,account,'create_order',key))
  db.execute('COMMIT');return {'order_id':order_id,'result':'created'}
 except BaseException:
  if db.in_transaction: db.execute('ROLLBACK')
  raise
 finally: db.close()
def counts(path):
 db=connect(path);out={'orders':db.execute('SELECT COUNT(*) FROM orders').fetchone()[0],'ledger':db.execute('SELECT COUNT(*) FROM request_ledger').fetchone()[0]};db.close();return out
payload={'sku':'FILTER-5','quantity':5}
results=[]
with tempfile.TemporaryDirectory(prefix='retry-pilot-') as directory:
 root=Path(directory);normal=root/'normal.db';create(normal)
 first=submit(normal,'A7','K1',payload) # Discarding this return models a lost response, not a network test.
 replay=submit(normal,'A7','K1',payload)
 assert first['order_id']==replay['order_id']==1 and counts(normal)['orders']==1
 results.append({'case':'return discarded, then same-key replay','observed':[first,replay,counts(normal)]})
 try: submit(normal,'A7','K1',{'sku':'FILTER-5','quantity':6});raise AssertionError('mismatch accepted')
 except PayloadMismatch: pass
 assert counts(normal)['orders']==1
 results.append({'case':'same key, changed quantity','observed':{'error':'PayloadMismatch','orders':1}})
 second=submit(normal,'A7','K2',payload);assert second['order_id']==2
 results.append({'case':'same payload, new deliberate key','observed':second})
 third=submit(normal,'A8','K1',payload);assert third['order_id']==3
 results.append({'case':'same key under another account','observed':third})
 concurrent=root/'concurrent.db';create(concurrent);barrier=threading.Barrier(2);out=[];errors=[]
 def worker():
  try: barrier.wait();out.append(submit(concurrent,'A7','K1',payload))
  except BaseException as error: errors.append(repr(error))
 threads=[threading.Thread(target=worker) for _ in range(2)]
 for thread in threads: thread.start()
 for thread in threads: thread.join()
 assert not errors and sorted(x['result'] for x in out)==['created','replayed'] and {x['order_id'] for x in out}=={1} and counts(concurrent)=={'orders':1,'ledger':1}
 results.append({'case':'two simultaneous threads, separate connections','observed':{'results':sorted(x['result'] for x in out),**counts(concurrent)}})
 rollback=root/'rollback.db';create(rollback)
 try: submit(rollback,'A7','K1',payload,True);raise AssertionError('failure not injected')
 except RuntimeError: pass
 assert counts(rollback)=={'orders':0,'ledger':0}
 after_failure=counts(rollback);recovered=submit(rollback,'A7','K1',payload)
 results.append({'case':'injected failure after order insert, before completion','observed':{'afterRollback':after_failure,'retry':recovered,**counts(rollback)}})
 forgotten=root/'forgotten.db';create(forgotten);submit(forgotten,'A7','K1',payload)
 db=connect(forgotten);db.execute('DELETE FROM request_ledger');db.close()
 forgotten_retry=submit(forgotten,'A7','K1',payload);assert counts(forgotten)['orders']==2
 results.append({'case':'negative control: deliberately forget key, retain order','observed':{**forgotten_retry,**counts(forgotten)}})
 naive=root/'naive.db';create(naive);db=connect(naive)
 # Controlled interleaving: two callers inspect absence before either writes.
 assert db.execute('SELECT 1 FROM request_ledger').fetchone() is None
 assert db.execute('SELECT 1 FROM request_ledger').fetchone() is None
 for _ in range(2): db.execute('INSERT INTO orders(account,sku,quantity) VALUES(?,?,?)',('A7','FILTER-5',5))
 db.execute('INSERT INTO request_ledger VALUES(?,?,?,?,?,?)',('A7','create_order','K1',json.dumps(payload,sort_keys=True,separators=(',',':')),'completed',1))
 try: db.execute('INSERT INTO request_ledger VALUES(?,?,?,?,?,?)',('A7','create_order','K1',json.dumps(payload,sort_keys=True,separators=(',',':')),'completed',2));raise AssertionError('duplicate ledger accepted')
 except sqlite3.IntegrityError: pass
 db.close();assert counts(naive)=={'orders':2,'ledger':1}
 results.append({'case':'negative control: effects committed before unique ledger','observed':counts(naive)})
report={'environment':{'python_sqlite_version':sqlite3.sqlite_version,'database':'temporary local SQLite, BEGIN IMMEDIATE serialises writers'},'limits':['No live supplier or network failure exercised.','No PostgreSQL lock behaviour or production throughput tested.','No lease expiry, multiworker async fulfilment or remote-side effect transaction modelled.'],'cases':results,'allAssertionsPassed':True}
Path('/tmp/retry_pilot_results.json').write_text(json.dumps(report,indent=2)+'\n');print(json.dumps(report,indent=2))