from __future__ import annotations
import datetime as dt
import hashlib
import json
import os
import sqlite3
import uuid
from pathlib import Path

DB=Path('/home/agent/jarvis_runtime/finance-system/data/finance.sqlite3')
ROOT=Path('/home/agent/jarvis_runtime/finance-system/legacy_fk_repair')
PREFLIGHT=ROOT/'preflight-private-20260729T075746Z.json'
ACTOR='jarvis_legacy_fk_repair_authorized_by_MG'
MARKER='legacy_orphan_transaction_reference_neutralized'
REASON='Referenced legacy transaction no longer exists and no deterministic identity permits safe relinking.'

def open_exclusive(path:Path,mode=0o600):
 fd=os.open(path,os.O_CREAT|os.O_EXCL|os.O_WRONLY,mode); os.close(fd)

def backup(conn:sqlite3.Connection,path:Path):
 open_exclusive(path)
 dst=sqlite3.connect(path)
 try: conn.backup(dst)
 finally: dst.close()
 os.chmod(path,0o600)

def table_digest(conn,table,exclude=()):
 cols=[r[1] for r in conn.execute(f"PRAGMA table_info('{table}')") if r[1] not in exclude]
 h=hashlib.sha256(); h.update(table.encode())
 if not cols:return h.hexdigest()
 qcols=','.join('"'+c.replace('"','""')+'"' for c in cols)
 for row in conn.execute(f'SELECT {qcols} FROM "{table}" ORDER BY rowid'):
  h.update(json.dumps(list(row),ensure_ascii=False,default=str,separators=(',',':')).encode())
 return h.hexdigest()

def all_table_digests(conn):
 tables=[r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")]
 return {t:table_digest(conn,t) for t in tables}

def logical_digest(conn):
 h=hashlib.sha256()
 for t,d in all_table_digests(conn).items(): h.update(t.encode()); h.update(d.encode())
 return h.hexdigest()

def fk_findings(conn):
 return [tuple(r) for r in conn.execute('PRAGMA foreign_key_check')]

def repair(conn, expected):
 current=fk_findings(conn)
 expected_fk=[('broker_import_execution_plans',x['fk']['rowid'],'transactions',x['fk']['fkid']) for x in expected]
 if sorted(current,key=str)!=sorted(expected_fk,key=str): raise RuntimeError('current FK findings differ from frozen preflight expectations')
 rows=[]
 for x in expected:
  p=x['plan']; row=conn.execute('SELECT * FROM broker_import_execution_plans WHERE execution_plan_id=?',(p['execution_plan_id'],)).fetchone()
  if row is None or row['transaction_id']!=p['transaction_id']: raise RuntimeError('expected plan/transaction identity mismatch')
  if row['execution_status']!='imported': raise RuntimeError('execution status is not imported')
  if conn.execute('SELECT COUNT(*) FROM transactions WHERE transaction_id=?',(p['transaction_id'],)).fetchone()[0]!=0: raise RuntimeError('referenced transaction unexpectedly exists')
  if conn.execute('SELECT COUNT(*) FROM transactions WHERE row_hash=?',(row['source_row_hash'],)).fetchone()[0]!=0: raise RuntimeError('deterministic row-hash relink candidate exists')
  rows.append(dict(row))
 now=dt.datetime.now(dt.timezone.utc).isoformat()
 conn.execute('BEGIN IMMEDIATE')
 try:
  updated=0; audit_ids=[]
  for row in rows:
   cur=conn.execute("UPDATE broker_import_execution_plans SET transaction_id=NULL WHERE execution_plan_id=? AND transaction_id=? AND execution_status='imported'",(row['execution_plan_id'],row['transaction_id']))
   updated+=cur.rowcount
   aid=str(uuid.uuid4()); audit_ids.append(aid)
   conn.execute("""INSERT INTO audit_log(audit_id,timestamp,source,action,entity_type,entity_id,old_values_json,new_values_json,user_text_note,confirmed,confirmation_timestamp,created_by,quality_status,created_at) VALUES(?,?,?,?,?,?,?,?,?,1,?,?,?,?)""",
    (aid,now,'authorized_legacy_fk_repair',MARKER,'broker_import_execution_plan',row['execution_plan_id'],json.dumps({'transaction_id':row['transaction_id']},sort_keys=True),json.dumps({'transaction_id':None},sort_keys=True),REASON,now,ACTOR,'ok',now))
  if updated!=3: raise RuntimeError(f'expected exactly 3 updated rows, got {updated}')
  if fk_findings(conn): raise RuntimeError('foreign key findings remain before commit')
  if conn.execute('PRAGMA integrity_check').fetchone()[0]!='ok': raise RuntimeError('integrity failed before commit')
  conn.commit()
 except Exception:
  conn.rollback(); raise
 return rows,audit_ids,now

def replay_noop(conn,expected):
 conn.execute('BEGIN IMMEDIATE')
 try:
  changes=0
  for x in expected:
   p=x['plan']
   cur=conn.execute("UPDATE broker_import_execution_plans SET transaction_id=NULL WHERE execution_plan_id=? AND transaction_id=? AND execution_status='imported'",(p['execution_plan_id'],p['transaction_id']))
   changes+=cur.rowcount
   row=conn.execute('SELECT transaction_id,execution_status FROM broker_import_execution_plans WHERE execution_plan_id=?',(p['execution_plan_id'],)).fetchone()
   audits=conn.execute('SELECT COUNT(*) FROM audit_log WHERE action=? AND entity_type=? AND entity_id=?',(MARKER,'broker_import_execution_plan',p['execution_plan_id'])).fetchone()[0]
   if row is None or row['transaction_id'] is not None or row['execution_status']!='imported' or audits!=1: raise RuntimeError('idempotent repaired-state proof failed')
  if changes!=0: raise RuntimeError('replay was not a no-op')
  conn.rollback()
 except Exception:
  conn.rollback(); raise
 return changes

ROOT.mkdir(parents=True,exist_ok=True); os.chmod(ROOT,0o700)
stamp=dt.datetime.now(dt.timezone.utc).strftime('%Y%m%dT%H%M%SZ')
prebackup=ROOT/f'pre-repair-{stamp}.sqlite3'; postbackup=ROOT/f'post-repair-{stamp}.sqlite3'; restore=ROOT/f'post-repair-restore-{stamp}.sqlite3'; evidence_path=ROOT/f'repair-private-{stamp}.json'
pre=json.loads(PREFLIGHT.read_text())
expected=[x for x in pre['finding_details'] if x.get('plan')]
if len(expected)!=3: raise RuntimeError('frozen preflight does not contain exactly 3 expected plans')
conn=sqlite3.connect(DB); conn.row_factory=sqlite3.Row; conn.execute('PRAGMA foreign_keys=ON')
if conn.execute('PRAGMA foreign_keys').fetchone()[0]!=1: raise RuntimeError('foreign keys are not enforced')
backup(conn,prebackup)
pre_digests=all_table_digests(conn); pre_plan_except_tx=table_digest(conn,'broker_import_execution_plans',exclude=('transaction_id',)); pre_plan_count=conn.execute('SELECT COUNT(*) FROM broker_import_execution_plans').fetchone()[0]; pre_audit_count=conn.execute('SELECT COUNT(*) FROM audit_log').fetchone()[0]
pre_transaction_count=conn.execute('SELECT COUNT(*) FROM transactions').fetchone()[0]; pre_transaction_digest=table_digest(conn,'transactions')
rows,audit_ids,repair_time=repair(conn,expected)
post_digests=all_table_digests(conn); post_plan_except_tx=table_digest(conn,'broker_import_execution_plans',exclude=('transaction_id',)); post_plan_count=conn.execute('SELECT COUNT(*) FROM broker_import_execution_plans').fetchone()[0]; post_audit_count=conn.execute('SELECT COUNT(*) FROM audit_log').fetchone()[0]
post_transaction_count=conn.execute('SELECT COUNT(*) FROM transactions').fetchone()[0]; post_transaction_digest=table_digest(conn,'transactions')
noop_changes=replay_noop(conn,expected)
post_fk=fk_findings(conn); post_integrity=conn.execute('PRAGMA integrity_check').fetchone()[0]
backup(conn,postbackup); conn.close()
post=sqlite3.connect(postbackup); post.row_factory=sqlite3.Row
open_exclusive(restore); restored=sqlite3.connect(restore); post.backup(restored); restored.close(); os.chmod(restore,0o600)
r=sqlite3.connect(restore); r.row_factory=sqlite3.Row
restore_integrity=r.execute('PRAGMA integrity_check').fetchone()[0]; restore_fk=fk_findings(r); backup_restore_equal=(logical_digest(post)==logical_digest(r)); r.close(); post.close()
unchanged_tables=[t for t in pre_digests if pre_digests[t]==post_digests[t]]
changed_tables=[t for t in pre_digests if pre_digests[t]!=post_digests[t]]
allowed_changed=(set(changed_tables)=={'broker_import_execution_plans','audit_log'})
assert post_fk==[] and post_integrity=='ok' and restore_fk==[] and restore_integrity=='ok' and backup_restore_equal
assert pre_transaction_count==post_transaction_count and pre_transaction_digest==post_transaction_digest
assert pre_plan_count==post_plan_count and pre_plan_except_tx==post_plan_except_tx
assert post_audit_count-pre_audit_count==3 and allowed_changed and noop_changes==0
evidence={'timestamp':stamp,'actor':ACTOR,'marker':MARKER,'reason':REASON,'preflight_report':str(PREFLIGHT),'pre_backup':str(prebackup),'post_backup':str(postbackup),'restore_copy':str(restore),'file_modes':{str(p):oct(p.stat().st_mode&0o777) for p in (prebackup,postbackup,restore)},'expected_plan_and_old_transaction_ids':[{'execution_plan_id':x['plan']['execution_plan_id'],'old_transaction_id':x['plan']['transaction_id']} for x in expected],'audit_ids':audit_ids,'repair_time':repair_time,'post_fk_findings':post_fk,'post_integrity':post_integrity,'transaction_count_before':pre_transaction_count,'transaction_count_after':post_transaction_count,'transaction_digest_unchanged':pre_transaction_digest==post_transaction_digest,'plan_count_unchanged':pre_plan_count==post_plan_count,'plan_content_except_transaction_id_unchanged':pre_plan_except_tx==post_plan_except_tx,'audit_rows_added':post_audit_count-pre_audit_count,'changed_tables':changed_tables,'all_other_tables_unchanged':allowed_changed,'idempotent_replay_changes':noop_changes,'restore_integrity':restore_integrity,'restore_fk_findings':restore_fk,'backup_restore_logical_equal':backup_restore_equal}
open_exclusive(evidence_path)
with evidence_path.open('w') as f: json.dump(evidence,f,ensure_ascii=False,indent=2); os.chmod(evidence_path,0o600)
print(json.dumps({'private_evidence':str(evidence_path),'modes':evidence['file_modes']|{str(evidence_path):oct(evidence_path.stat().st_mode&0o777)},'neutralized_references':3,'audit_rows_added':3,'fk_findings_after':0,'integrity_after':post_integrity,'transactions_unchanged':True,'plan_content_except_reference_unchanged':True,'changed_tables':changed_tables,'all_other_tables_unchanged':allowed_changed,'idempotent_replay':'no-op','backup_restore_verified':True},indent=2))