"""Archive and remove explicitly confirmed fixtures; never certify live readiness.""" import argparse import contextlib from datetime import datetime,timezone import fcntl import hashlib import json import os from pathlib import Path import sqlite3 import stat import uuid KNOWN_NAMES={'BPA_TAX_FIRM_ANNUAL_AUDIT','TEST_VERIFICATION_PASS_NODE'} LEDGER_HASH='93922185a7c2b6fc76847a0a3da8eb9f4257caafb2d21f9a29132ac1f5c2378b' def regular_private(path): st=path.stat() if path.is_symlink() or not stat.S_ISREG(st.st_mode) or st.st_uid!=os.getuid() or st.st_mode & 0o077: raise ValueError('Unsafe file ownership/permissions: '+path.name) def tables(conn):return {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} @contextlib.contextmanager def lock(path): fd=os.open(path,os.O_RDWR|os.O_CREAT|os.O_NOFOLLOW,0o600) with os.fdopen(fd,'a+') as f: regular_private(Path(path)) fcntl.flock(f,fcntl.LOCK_EX|fcntl.LOCK_NB) yield def backup(conn,path): fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600);os.close(fd) destination=sqlite3.connect(path) try: conn.backup(destination) if [r[0] for r in destination.execute('PRAGMA integrity_check')]!=['ok']:raise ValueError('Backup integrity failed') finally:destination.close() def run(root,apply=False): root=Path(root).absolute() if not root.is_dir() or root.is_symlink():raise ValueError('Existing project root required') # Verify known filesystem inputs before any mutation. current=root/'luxzia_research.db';legacy=root/'lucxia_research.db' if current.exists() and legacy.exists():raise ValueError('Both research databases exist; explicit reconciliation needed') research=current if current.exists() else legacy if research.exists(): regular_private(research) check=sqlite3.connect(research.resolve().as_uri()+'?mode=ro',uri=True) try: if [r[0] for r in check.execute('PRAGMA integrity_check')]!=['ok']:raise ValueError('Research integrity check failed') finally:check.close() ledger=root/'generated/automated_irs_form_8949.json' if ledger.exists(): if ledger.is_symlink() or not ledger.is_file():raise ValueError('Unsafe ledger path') if hashlib.sha256(ledger.read_bytes()).hexdigest()!=LEDGER_HASH:raise ValueError('Ledger differs from confirmed sample; leave untouched for review') archive_parent=root/'test_data_archive' if archive_parent.is_symlink():raise ValueError('Unsafe archive directory') result={'mode':'APPLY' if apply else 'PREVIEW','notification_mode':'DRY_RUN_UNCHANGED','deployment_status':'UNVERIFIED','changes':[]} archive=root/'test_data_archive'/str(uuid.uuid4()) if apply:archive.mkdir(parents=True,mode=0o700) operation=root/'qzotic_pure_sandbox.db' if not operation.is_file():raise ValueError('Operational database not found') regular_private(operation) with lock(str(operation)+'.supervisor.lock'): conn=sqlite3.connect(operation.resolve().as_uri()+'?mode='+('rw' if apply else 'ro'),uri=True,isolation_level=None) try: conn.execute('PRAGMA foreign_keys=ON') rows=conn.execute('SELECT project_id,project_name,allocated_budget_cents FROM enterprise_portfolio').fetchall() # A new real record blocks this fixture-specific operation. if any(name not in KNOWN_NAMES for _,name,_ in rows):raise ValueError('Unconfirmed portfolio rows found; cleanup stopped') for _,name,amount in rows: expected=4500000 if name=='BPA_TAX_FIRM_ANNUAL_AUDIT' else 100000 if amount!=expected:raise ValueError('Confirmed sample budget changed; cleanup stopped') result['confirmed_portfolio_samples']=len(rows) if apply: snapshot=archive/'operational_before.sqlite3';backup(conn,snapshot) conn.execute('BEGIN IMMEDIATE') try: live=conn.execute('SELECT project_id,project_name,allocated_budget_cents FROM enterprise_portfolio').fetchall() if live!=rows:raise ValueError('Portfolio changed during preflight') names=[r[1] for r in rows] placeholders=','.join('?' for _ in names) present=tables(conn) if names: for table,column in (('notification_outbox','associated_project'),('remediation_lifecycle','associated_project'),('project_event_versions','project_name')): if table in present:conn.execute(f'DELETE FROM {table} WHERE {column} IN ({placeholders})',names) conn.execute(f'DELETE FROM enterprise_portfolio WHERE project_name IN ({placeholders})',names) # Retain journal, audit/history and monotonic sequence counters. if 'audit_events' in present:conn.execute('INSERT INTO audit_events(event_type,reference,recorded_at) VALUES(?,?,?)',('CONFIRMED_SAMPLE_PURGE',str(snapshot),datetime.now(timezone.utc).isoformat())) if conn.execute('PRAGMA foreign_key_check').fetchall():raise ValueError('Foreign-key check failed') conn.commit();result['changes'].append('Confirmed sample portfolio rows and their dependent action/outbox rows removed') except BaseException:conn.rollback();raise finally:conn.close() current=root/'luxzia_research.db';legacy=root/'lucxia_research.db' if current.exists() and legacy.exists():raise ValueError('Both research databases exist; explicit reconciliation needed') research=current if current.exists() else legacy if research.exists(): regular_private(research) with lock(str(research)+'.lucxia.lock'): c=sqlite3.connect(research.resolve().as_uri()+'?mode='+('rw' if apply else 'ro'),uri=True,isolation_level=None) try: present=tables(c) result['research_database']=research.name result['research_history_preserved']=True count=0 if 'lucxia_fictional_scenarios' in present: ids=[f'SCEN_20261101_{i:02d}' for i in range(1,6)] count=c.execute('SELECT COUNT(*) FROM lucxia_fictional_scenarios WHERE scenario_uid IN (?,?,?,?,?)',ids).fetchone()[0] if apply and count: backup(c,archive/'research_before.sqlite3');c.execute('BEGIN IMMEDIATE') try: c.execute('DELETE FROM lucxia_fictional_scenarios WHERE scenario_uid IN (?,?,?,?,?)',ids) if c.execute('PRAGMA foreign_key_check').fetchall():raise ValueError('Research foreign-key check failed') c.commit();result['changes'].append('Five explicitly identified fictional scenario fixtures removed, if present') except BaseException:c.rollback();raise result['fictional_scenario_fixtures']=count finally:c.close() ledger=root/'generated/automated_irs_form_8949.json' if ledger.exists(): if ledger.is_symlink() or not ledger.is_file():raise ValueError('Unsafe ledger path') raw=ledger.read_bytes() if hashlib.sha256(raw).hexdigest()!=LEDGER_HASH:raise ValueError('Ledger differs from confirmed sample; leave untouched for review') result['ledger_status']='CONFIRMED_SAMPLE_ARCHIVE_PENDING' if apply: frozen=archive/'sample_ledger.json' with frozen.open('xb') as f:f.write(raw);f.flush();os.fsync(f.fileno()) frozen.chmod(0o400) if frozen.read_bytes()!=raw:raise ValueError('Sample ledger archive verification failed') if ledger.read_bytes()!=raw:raise ValueError('Ledger changed during archiving') ledger.unlink();result['ledger_status']='SAMPLE_ARCHIVED_NO_LIVE_LEDGER_CREATED' result['changes'].append('Confirmed sample ledger archived read-only; operational ledger path removed') else:result['ledger_status']='NO_LEDGER_PRESENT' if apply: result['archive']=str(archive) (archive/'receipt.json').write_text(json.dumps(result,indent=2)+'\n');(archive/'receipt.json').chmod(0o600) return result if __name__=='__main__': p=argparse.ArgumentParser();p.add_argument('--root',default='/Users/bpa/ForesixaQuant');p.add_argument('--apply',action='store_true');args=p.parse_args() os.umask(0o077) try:print(json.dumps(run(args.root,args.apply),indent=2)) except Exception as exc: print(json.dumps({'status':'STOPPED','reason':str(exc),'notice':'Review retained snapshots/receipts. Separate databases and filesystem operations are not one atomic transaction; earlier stages may have completed.'},indent=2));raise SystemExit(2)