import pathlib,json,subprocess,threading,sqlite3,http.server,urllib.request,os,signal,selectors,hashlib,platform
root=pathlib.Path(__file__).resolve().parent;run=root/'run-data';run.mkdir(exist_ok=False);db=run/'service.sqlite'
with sqlite3.connect(db) as c:c.executescript('CREATE TABLE effects(id INTEGER PRIMARY KEY, operation_key TEXT, mode TEXT); CREATE TABLE dedup(operation_key TEXT PRIMARY KEY, receipt INTEGER);')
class Handler(http.server.BaseHTTPRequestHandler):
 def log_message(self,*args):pass
 def do_POST(self):
  key=json.loads(self.rfile.read(int(self.headers['Content-Length'])))['key'];mode=self.path.strip('/')
  with sqlite3.connect(db) as c:
   c.execute('BEGIN IMMEDIATE');old=c.execute('SELECT receipt FROM dedup WHERE operation_key=?',(key,)).fetchone() if mode=='idempotent' else None
   if old:receipt=old[0]
   else:
    receipt=c.execute('INSERT INTO effects(operation_key,mode) VALUES (?,?)',(key,mode)).lastrowid
    if mode=='idempotent':c.execute('INSERT INTO dedup VALUES (?,?)',(key,receipt))
  data=json.dumps({'effect_receipt':receipt}).encode();self.send_response(200);self.end_headers();self.wfile.write(data)
def count(key):
 with sqlite3.connect(db) as c:return c.execute('SELECT count(*) FROM effects WHERE operation_key=?',(key,)).fetchone()[0]
server=http.server.ThreadingHTTPServer(('127.0.0.1',0),Handler);threading.Thread(target=server.serve_forever,daemon=True).start();rows=[]
try:
 for phase in ['before_start','before_request','after_effect','after_checkpoint']:
  for mode in ['plain','idempotent']:
   for rep in range(5):
    key=f'{phase}-{mode}-{rep}';directory=run/key;url=f'http://127.0.0.1:{server.server_port}/{mode}'
    child=subprocess.Popen([str(root/'probe'),'write',str(directory),phase,url,key],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env={'PATH':'/usr/bin:/bin'})
    try:
     sel=selectors.DefaultSelector();sel.register(child.stdout,selectors.EVENT_READ);assert sel.select(15), 'barrier timeout';line=child.stdout.readline();assert line.strip()=='READY',(key,line,child.stderr.read() if child.poll() is not None else '')
     before=count(key);child.kill();child.wait(timeout=5);assert child.returncode==-signal.SIGKILL
    finally:
     if child.poll() is None:child.kill();child.wait()
    a=json.loads(subprocess.check_output([str(root/'probe'),'read',str(directory)],timeout=10));b=json.loads(subprocess.check_output([str(root/'probe'),'read',str(directory)],timeout=10));assert a==b
    retry=a['known_receipts']==0
    if retry:
     req=urllib.request.Request(url,json.dumps({'key':key}).encode(),{'Content-Type':'application/json'});urllib.request.urlopen(req,timeout=5).read()
    after=count(key);row={'case':key,'phase':phase,'service_mode':mode,'replicate':rep,'child_signal':'SIGKILL','effects_before_recovery':before,'recovered':a,'second_reopen':b,'retry_sent':retry,'effects_after_retry':after};rows.append(row)
    assert before==(1 if phase in ['after_effect','after_checkpoint'] else 0)
    assert a['known_receipts']==(1 if phase=='after_checkpoint' else 0)
    assert a['unknown_markers']==(1 if phase in ['before_request','after_effect'] else 0)
    assert after==(2 if phase=='after_effect' and mode=='plain' else 1)
finally:server.shutdown();server.server_close()
output={'source_revision':'77ac9b72409fc2b6afc5741fd2469c662758a1b4','environment':{'python':platform.python_version(),'platform':platform.platform(),'go':subprocess.check_output(['go','version'],text=True).strip()},'rows':rows}
(root/'results.json').write_text(json.dumps(output,indent=2)+'\n');print(json.dumps({'cases':len(rows),'duplicate_effect_cases':sum(r['effects_after_retry']>1 for r in rows),'reopens_stable':all(r['recovered']==r['second_reopen'] for r in rows)},indent=2))