#!/usr/bin/env python3 """Run this complete fixture only in newly created owned /private/tmp roots.""" import csv,errno,json,os,platform,plistlib,shutil,subprocess,sys,uuid,zipfile from pathlib import Path import rename_packet as rp B=Path(__file__).resolve().parent def csvout(name,data): with (B/name).open('w',newline='',encoding='utf-8') as f: w=csv.DictWriter(f,fieldnames=list(data[0]));w.writeheader();w.writerows(data) def main(): commands=[];roots=[];cases=[];applied=[];mainruns=[] started=rp.datetime.datetime.now(rp.datetime.timezone.utc).isoformat() frozen=rp.PACKET_HASH def run(action,root=None,plan='rename-plan.csv'): args=[sys.executable,str(B/'rename_packet.py'),action]+([str(root)] if root else [])+['--plan',plan] p=subprocess.run(args,capture_output=True,text=True,timeout=15) commands.append(dict(argv=[Path(args[0]).name,'rename_packet.py',*args[2:]],exit=p.returncode,stdout=p.stdout,stderr=p.stderr)) assert p.returncode==0,p.stderr return json.loads(p.stdout) def fresh(label): root=Path('/private/tmp')/('fw1-'+label+'-'+uuid.uuid4().hex[:12]);run('init',root);roots.append(root);return root def state(root): fd=os.open(root/'files',os.O_RDONLY|os.O_DIRECTORY) try:return {n:rp.snapshot(fd,n) for n in sorted(os.listdir(fd))} finally:os.close(fd) def checkrow(result,id):return next(x for x in result['rows'] if x['source_id']==id) (B/'same-plan-with-blank-line.csv').write_bytes((B/'rename-plan.csv').read_bytes()+b'\n') assert rp.rows('same-plan-with-blank-line.csv')==rp.rows('rename-plan.csv') related=rp.rows('rename-plan.csv') related[0].update(proposed_name='2026-04-03__amber__airflow__rev-a__f01-revised.txt', approved_name='2026-04-03__amber__airflow__rev-a__f01-revised.txt') csvout('related-changed-plan.csv',related) try: for label in ('a','b'): root=fresh(label);forward=run('apply',root) assert [x['source_id'] for x in forward['rows'] if x['status']=='applied']==['F01','F02','F05','F07'] verified=[] for i in rp.INVENTORY: x=checkrow(forward,i['source_id']);name=x['target_name'] if x['status']=='applied' else i['old_name'];data=(root/'files'/name).read_bytes() assert rp.digest(data)==i['sha256'] and len(data)==int(i['bytes']) v=dict(source_id=i['source_id'],current_name=name,bytes=len(data),sha256=rp.digest(data));verified.append(v) applied.append(dict(copy=label,old_name=i['old_name'],approved_name=x['target_name'],status=x['status'],reason=x['reason'],**v)) assert len(state(root))==8 before=state(root);repeat=run('apply',root);assert sum(x['status']=='already_applied' for x in repeat['rows'])==4 and state(root)==before if label=='a': same_operation=run('apply',root,'same-plan-with-blank-line.csv') assert sum(x['status']=='already_applied' for x in same_operation['rows'])==4 and state(root)==before different_operation=run('apply',root,'related-changed-plan.csv') assert checkrow(different_operation,'F01')['status']=='blocked' and state(root)==before if label=='a': with zipfile.ZipFile(B/'renamed-copy.zip','w',zipfile.ZIP_DEFLATED) as z: for v in verified: info=zipfile.ZipInfo(v['current_name'],(2026,1,1,0,0,0));info.compress_type=zipfile.ZIP_DEFLATED;info.external_attr=0o100644<<16;z.writestr(info,(root/'files'/v['current_name']).read_bytes()) csvout('reversal-plan.csv',[dict(source_id=x['source_id'],from_name=x['target_name'],to_name=x['old_name'],sha256=next(i['sha256'] for i in rp.INVENTORY if i['source_id']==x['source_id']),eligibility='confirmed_forward_only_recheck_current_bytes_and_unoccupied_target') for x in forward['rows'] if x['status']=='applied']) reverse=run('reverse',root,'same-plan-with-blank-line.csv' if label=='a' else 'rename-plan.csv');assert sum(x['status']=='reversed' for x in reverse['rows'])==4 for i in rp.INVENTORY:assert rp.digest((root/'files'/i['old_name']).read_bytes())==i['sha256'] if label=='a': before=state(root);again=run('reverse',root) assert sum(x['status']=='already_reversed' for x in again['rows'])==4 and state(root)==before mainruns.append(dict(copy=label,verified_after_forward=verified,reverse_confirmed_ids=[x['source_id'] for x in reverse['rows'] if x['status']=='reversed'],exactOperationIdentifiedAcrossPlanBytes=label=='a')) root=fresh('wrong');before=state(root);x=run('apply',root,'wrong-plan.csv');assert state(root)==before and all(r['status']=='blocked' for r in x['rows']) cases.append(dict(case='authored_wrong_proposal',observed=x['rows'],before=before,after=state(root),all_files_unchanged=True)) sentinel=b'Later unrelated teaching file. Preserve these bytes.\n';target=rp.canonical(rp.INVENTORY[0]) root=fresh('occupied');(root/'files'/target).write_bytes(sentinel);before=state(root);x=run('apply',root) assert checkrow(x,'F01')['reason']=='target_exists' and (root/'files'/target).read_bytes()==sentinel assert rp.digest((root/'files'/'Scan001.txt').read_bytes())==rp.INVENTORY[0]['sha256'] cases.append(dict(case='existing_target',observed=[checkrow(x,'F01')],before=before,after=state(root),later_file_preserved=True)) root=fresh('changed');run('preview');p=root/'files'/'IMG002.txt';p.write_bytes(p.read_bytes()+b'Later authored edit.\n');before=state(root);x=run('apply',root) assert checkrow(x,'F02')['status']=='blocked' and state(root)['IMG002.txt']==before['IMG002.txt'] cases.append(dict(case='source_changed_after_preview',observed=[checkrow(x,'F02')],before=before,after=state(root),changed_bytes_preserved=True)) root=fresh('reverse-conflict');run('apply',root);(root/'files'/'Scan001.txt').write_bytes(sentinel);before=state(root);x=run('reverse',root) assert checkrow(x,'F01')['reason']=='target_exists' and sum(r['status']=='reversed' for r in x['rows'])==3 assert (root/'files'/'Scan001.txt').read_bytes()==sentinel and (root/'files'/target).is_file() cases.append(dict(case='reverse_target_occupied',observed=x['rows'],before=before,after=state(root),later_file_preserved=True,reversed_count=3,blocked_applied_count=1)) root=fresh('missing');fd=os.open(root/'files',os.O_RDONLY|os.O_DIRECTORY) try:rp.exclusive(fd,'Scan001.txt','held.txt') finally:os.close(fd) before=state(root);x=run('apply',root);assert checkrow(x,'F01')['reason']=='source_missing_changed_or_not_regular' cases.append(dict(case='source_missing',observed=[checkrow(x,'F01')],before=before,after=state(root))) for case,changes in [('duplicate',dict(proposed_name=target,approved_name=target)),('extension',dict(proposed_name='wrong.pdf',approved_name='wrong.pdf')),('outside',dict(proposed_name='../escape.txt',approved_name='../escape.txt'))]: plan=rp.rows('rename-plan.csv');plan[1].update(changes);name=case+'-plan.csv';csvout(name,plan) root=fresh(case);before=state(root);x=run('apply',root,name) assert checkrow(x,'F02')['status']=='blocked' and (root/'files'/'IMG002.txt').is_file() and not (root/'escape.txt').exists() if case=='duplicate':assert checkrow(x,'F01')['status']=='blocked' cases.append(dict(case=case,observed=[checkrow(x,id) for id in ('F01','F02')],before=before,after=state(root))) root=fresh('primitive');before=state(root);fd=os.open(root/'files',os.O_RDONLY|os.O_DIRECTORY) try: try:rp.exclusive(fd,'Scan001.txt','IMG002.txt');raise AssertionError('unexpected replacement') except OSError as e:assert e.errno==errno.EEXIST;observed_errno=e.errno finally:os.close(fd) after=state(root);assert before==after cases.append(dict(case='exclusive_syscall_occupied_target',errno=observed_errno,before=before,after=after,all_files_unchanged=True)) (root/'files'/'case-probe.txt').write_bytes(b'case');case_insensitive=(root/'files'/'CASE-PROBE.TXT').exists() root=fresh('pending');intent=dict(source_id='F01',plan_sha256=rp.digest((B/'rename-plan.csv').read_bytes()),phase='intent',intent_id='authored-pending-1') with (root/'journal.jsonl').open('a') as f:f.write(json.dumps(intent)+'\n') before=state(root);x=run('apply',root);assert checkrow(x,'F01')['status']=='unknown' and state(root)==before unknown_variants=[] def check_unknown(stage,plan,action='apply'): journal_before=(root/'journal.jsonl').read_bytes() result=run(action,root,plan) assert checkrow(result,'F01')['status']=='unknown' and state(root)==before assert result['unprocessed_ids']==[i['source_id'] for i in rp.INVENTORY[1:]] assert (root/'journal.jsonl').read_bytes()==journal_before unknown_variants.append(dict(stage=stage,action=action,plan=plan,result=result,filesAndJournalUnchanged=True)) for plan in ('same-plan-with-blank-line.csv','related-changed-plan.csv'): check_unknown('pending_intent',plan) with (root/'journal.jsonl').open('a') as f: f.write(json.dumps(dict(intent,phase='result',status='unknown'))+'\n') unknown_result=run('apply',root) assert checkrow(unknown_result,'F01')['status']=='unknown' and state(root)==before for plan in ('same-plan-with-blank-line.csv','related-changed-plan.csv'): check_unknown('unknown_result',plan) check_unknown('unknown_result','related-changed-plan.csv','reverse') cases.append(dict(case='authored_pending_intent_refusal_not_crash_test',observed=x['rows'],observed_after_authored_unknown_result=unknown_result['rows'],planVariants=unknown_variants,parsedBlankPlanRowsIdentical=True,originalPlanHash=rp.digest((B/'rename-plan.csv').read_bytes()),blankPlanHash=rp.digest((B/'same-plan-with-blank-line.csv').read_bytes()),relatedPlanHash=rp.digest((B/'related-changed-plan.csv').read_bytes()),unprocessed_ids=x['unprocessed_ids'],before=before,after=state(root),all_files_unchanged=True)) df=subprocess.run(['df','-P','/private/tmp'],capture_output=True,text=True,timeout=10) device=df.stdout.splitlines()[-1].split()[0] if df.returncode==0 else '/private/tmp' disk=subprocess.run(['/usr/sbin/diskutil','info','-plist',device],capture_output=True,timeout=10) fs=plistlib.loads(disk.stdout) if disk.returncode==0 else {} mount=subprocess.run(['/sbin/mount'],capture_output=True,text=True,timeout=10) selected=[line for line in mount.stdout.splitlines() if line.startswith(device+' on ')] filesystem={'type':fs.get('FilesystemType'),'diskutilExit':disk.returncode,'mountProbeExit':mount.returncode,'selectedMount':selected} if not filesystem['type'] and len(selected)==1: filesystem['type']=selected[0].split('(',1)[1].split(',',1)[0] assert rp.digest((B/'source-packet.zip').read_bytes())==frozen report=dict(startedAtUtc=started,finishedAtUtc=rp.datetime.datetime.now(rp.datetime.timezone.utc).isoformat(),runtime=dict(python=platform.python_version(),os=platform.platform(),filesystem=filesystem,caseInsensitiveProbe=case_insensitive),sameAuthorTwoFreshCopiesNotIndependentReview=True,mainRuns=mainruns,cases=cases,sourceArchiveUnchanged=True,commands=commands,createdRoots=[str(p) for p in roots],cloudOrModelRun=False) csvout('applied-results.csv',applied) conflicts=[dict(case=c['case'],source_id=x['source_id'],status=x['status'],reason=x['reason']) for c in cases for x in c.get('observed',[]) if x['status'] in ('blocked','unknown')] conflicts += [dict(case=v['stage']+'/'+v['action']+'/'+v['plan'],source_id=x['source_id'],status=x['status'],reason=x['reason']) for c in cases for v in c.get('planVariants',[]) for x in v['result']['rows'] if x['status']=='unknown'] csvout('conflicts.csv',conflicts) finally: for root in roots: assert root.parent==Path('/private/tmp') and root.name.startswith('fw1-');shutil.rmtree(root) report.update(cleanup='all_created_roots_removed',rootsAbsentAfterCleanup=all(not p.exists() for p in roots)) (B/'execution.json').write_text(json.dumps(report,separators=(',',':'))+'\n') print(json.dumps(dict(two_fresh_copies=len(mainruns),applied_per_copy=4,blocked_per_copy=4,cases=len(cases),commands=len(commands),source_archive_unchanged=True,cleanup=report['cleanup'],runtime=report['runtime']),indent=2)) if __name__=='__main__':main()