import base64, hashlib, io, json, os, sqlite3, subprocess, zipfile
from pathlib import Path
from typing import Any
BASE=Path('/home/agent/jarvis_runtime/finance-system/quarantine/2026-08-26-current-import')
DB=Path(os.environ.get('PRODUCTION_CONFIRM_DB','/home/agent/jarvis_runtime/finance-system/data/finance.sqlite3'))
REPORT=BASE/'production_confirm_result_private.json'
# Load only the two owner-only approval keys from the already configured backend process
# when the controlled runner has not already exported them.
required_keys=('JARVIS_FINANCE_FINGERPRINT_KEY','JARVIS_FINANCE_OPERATOR_APPROVAL_KEY')
if not all(os.environ.get(key) for key in required_keys):
    pid=subprocess.check_output(['systemctl','--user','show','--property=MainPID','--value','finance-manager-backend.service'],text=True).strip()
    if pid and pid != '0':
        for item in Path(f'/proc/{pid}/environ').read_bytes().split(b'\0'):
            if item.startswith((b'JARVIS_FINANCE_FINGERPRINT_KEY=',b'JARVIS_FINANCE_OPERATOR_APPROVAL_KEY=')):
                key,value=item.split(b'=',1);os.environ[key.decode()]=value.decode()
if not all(os.environ.get(key) for key in required_keys):
    raise RuntimeError('owner approval keys are unavailable')
from jarvis_finance.storage.database import connect
from jarvis_finance.storage.migrations import get_schema_version
from jarvis_finance.services.household_import import preview_household_import,confirm_household_import
from jarvis_finance.services.cash_service import preview_cash_snapshot,confirm_cash_snapshot
from jarvis_finance.api.schemas.positions import CashSnapshotPreviewRequest,CashSnapshotConfirmRequest
from jarvis_finance.services.truewealth_service import preview_truewealth_import,confirm_truewealth_import
from jarvis_finance.services.postfinance_service import preview_postfinance_import,confirm_postfinance_import,confirm_known_cash_projection_correction
from jarvis_finance.services.wealth_cockpit import build_wealth_cockpit
from jarvis_finance.api.schemas.wealth_cockpit import WealthCockpitResponse

def scalar(value: Any) -> Any:
    if isinstance(value,bytes): return {'blob_sha256':hashlib.sha256(value).hexdigest(),'bytes':len(value)}
    return value

def table_names(conn: sqlite3.Connection) -> list[str]:
    return [str(r[0]) for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")]

def counts(conn: sqlite3.Connection) -> dict[str,int]:
    return {t:int(conn.execute(f'SELECT COUNT(*) FROM "{t}"').fetchone()[0]) for t in table_names(conn)}

def logical_digest(conn: sqlite3.Connection) -> str:
    h=hashlib.sha256()
    for table in table_names(conn):
        cols=[str(r[1]) for r in conn.execute(f'PRAGMA table_info("{table}")')]
        h.update(json.dumps([table,cols],separators=(',',':')).encode())
        order=','.join(f'"{c}"' for c in cols)
        for row in conn.execute(f'SELECT * FROM "{table}" ORDER BY {order}'):
            h.update(json.dumps([scalar(v) for v in row],ensure_ascii=False,separators=(',',':'),default=str).encode())
    return h.hexdigest()

def file_digests() -> dict[str,str]:
    names=json.loads((BASE/'source_manifest.json').read_text())
    return {str(x['name']):hashlib.sha256((BASE/str(x['name'])).read_bytes()).hexdigest() for x in names}

conn=connect(DB)
if get_schema_version(conn)!=51: raise RuntimeError('production schema must be 51 before confirm')
if conn.execute('PRAGMA integrity_check').fetchone()[0] != 'ok' or conn.execute('PRAGMA foreign_key_check').fetchall(): raise RuntimeError('pre-confirm integrity gate failed')
source_before=file_digests();before_counts=counts(conn);before_digest=logical_digest(conn)
melanie_before=[tuple(row) for row in conn.execute("""SELECT p.plan_item_id,p.monthly_amount_chf,p.is_active
    FROM budget_plan_items p JOIN budget_categories c ON c.category_id=p.category_id
    WHERE c.name='Lohn Melanie' ORDER BY p.plan_item_id""").fetchall()]
# Household import: only the explicitly identified August salary row is promoted from review.
payload=json.loads((BASE/'candidate_household_payload_private.json').read_text())
first=preview_household_import(conn,payload)
salary_rows=[r for r in first['rows'] if r['source_type']=='raiffeisen_bank' and r['transaction_date']=='2026-08-25' and r['signed_amount'] in ('9261.3','9261.30')]
if len(salary_rows)!=1: raise RuntimeError('expected exactly one August salary source row')
category=conn.execute("SELECT category_id FROM budget_categories WHERE name='Lohn Marcel' AND category_type='income' AND is_active=1").fetchone()
if not category: raise RuntimeError('active Lohn Marcel category missing')
payload['category_overrides']={salary_rows[0]['row_token']:category['category_id']};payload['confirmed_row_tokens']=[salary_rows[0]['row_token']]
hh_preview=preview_household_import(conn,payload)
hh_request={**payload,'preview_fingerprint':hh_preview['preview_fingerprint'],'baseline_fingerprint':hh_preview['baseline_fingerprint'],'confirm':True,'confirm_review_candidates':True}
hh=confirm_household_import(conn,hh_request)
# Dated snapshots only; no cashflow history is inferred.
snapshot_specs=[('AKB Zinsbereitstellungskonto','manual_balance','2026-08-26','2685.00'),('AKB Aargauer-Sparkonto','manual_balance','2026-08-26','60577.95'),('AKB Haushaltskonto','csv_anchor_balance','2026-08-26','11637.17'),('Raiffeisen Mitglieder-Privatkonto •••• 5632','csv_anchor_balance','2026-08-25','29062.37')]
snapshot_requests=[];snapshot_results=[]
for name,kind,day,amount in snapshot_specs:
    matches=conn.execute('SELECT account_id FROM accounts WHERE account_name=? AND is_active=1',(name,)).fetchall()
    if len(matches)!=1: raise RuntimeError(f'exactly one active snapshot account required: {name}')
    draft=CashSnapshotPreviewRequest(account_id=matches[0]['account_id'],snapshot_type=kind,balance_date=day,amount_chf=amount,note='Kontrollierter Quellenstand 26.08.2026')
    p=preview_cash_snapshot(conn,draft);req=CashSnapshotConfirmRequest(**draft.model_dump(),preview_id=p.preview_id,confirm=True)
    result=confirm_cash_snapshot(conn,req);snapshot_requests.append(req);snapshot_results.append({'label':name.split(' •')[0],'amount_chf':amount,'date':day,'entity_id':result.entity_id})
# True Wealth official statement and internal activities.
tw_file=BASE/'260826_Truewealth aktueller Stand Portfolio.pdf';tw_raw=tw_file.read_bytes();tw_req={'file_name':tw_file.name,'content_base64':base64.b64encode(tw_raw).decode()};tw_preview=preview_truewealth_import(conn,tw_req);tw=confirm_truewealth_import(conn,{**tw_req,'preview_id':tw_preview['preview_id'],'confirmation_id':tw_preview['confirmation_id'],'confirm':True})
# Append-only correction of legacy PostFinance cash projection, then official snapshot/activity import.
pf_correction=confirm_known_cash_projection_correction(conn,confirm=True)
pf_corr_replay=confirm_known_cash_projection_correction(conn,confirm=True)
if not pf_corr_replay['idempotent']: raise RuntimeError('PostFinance projection correction replay was not idempotent')
pf_statement=BASE/'260826_Postfinance_Kontoauszug.pdf';pf_overview=BASE/'260826_Postfinance aktueller Stand.pdf';buf=io.BytesIO()
with zipfile.ZipFile(buf,'w',zipfile.ZIP_DEFLATED) as archive: archive.writestr(pf_statement.name,pf_statement.read_bytes())
z=buf.getvalue();o=pf_overview.read_bytes();pf_req={'zip_file_name':'PostFinance_20260826.zip','overview_file_name':pf_overview.name,'zip_mime_type':'application/zip','overview_mime_type':'application/pdf','zip_content_base64':base64.b64encode(z).decode(),'overview_content_base64':base64.b64encode(o).decode(),'zip_size_bytes':len(z),'overview_size_bytes':len(o)}
pf_preview=preview_postfinance_import(conn,pf_req)
if not pf_preview['confirm_allowed'] or pf_preview['performance_available']: raise RuntimeError('PostFinance import/performance gates differ from approved preview')
pf=confirm_postfinance_import(conn,{**pf_req,'preview_id':pf_preview['preview_id'],'confirmation_id':pf_preview['confirmation_id'],'confirm':True})
first_counts=counts(conn);first_digest=logical_digest(conn)
# Exact replay: refresh generation-bound previews, then confirm all identical evidence again.
hh_replay=confirm_household_import(conn,hh_request)
for req,first_result in zip(snapshot_requests,snapshot_results,strict=True):
    replay=confirm_cash_snapshot(conn,req)
    if replay.entity_id != first_result['entity_id']: raise RuntimeError('snapshot replay changed entity identity')
tw_p2=preview_truewealth_import(conn,tw_req);tw_replay=confirm_truewealth_import(conn,{**tw_req,'preview_id':tw_p2['preview_id'],'confirmation_id':tw_p2['confirmation_id'],'confirm':True})
pf_p2=preview_postfinance_import(conn,pf_req);pf_replay=confirm_postfinance_import(conn,{**pf_req,'preview_id':pf_p2['preview_id'],'confirmation_id':pf_p2['confirmation_id'],'confirm':True})
hh_fresh=preview_household_import(conn,payload);replay_counts=counts(conn);replay_digest=logical_digest(conn)
source_after=file_digests()
if first_counts!=replay_counts or first_digest!=replay_digest: raise RuntimeError('replay changed productive database')
if source_before!=source_after: raise RuntimeError('source files changed during confirm')
if any(hh_fresh['expected_writes'].values()): raise RuntimeError('fresh household replay preview is not zero-write')
if not (hh_replay['idempotent'] and tw_replay['idempotent'] and pf_replay['idempotent'] and pf_corr_replay['idempotent']): raise RuntimeError('one replay did not report idempotency')
if conn.execute('PRAGMA integrity_check').fetchone()[0] != 'ok' or conn.execute('PRAGMA foreign_key_check').fetchall(): raise RuntimeError('post-confirm integrity gate failed')
cockpit=WealthCockpitResponse.model_validate(build_wealth_cockpit(conn)).model_dump()
after_counts=counts(conn);deltas={k:after_counts[k]-before_counts.get(k,0) for k in after_counts if after_counts[k]-before_counts.get(k,0)}
salary_count=int(conn.execute("SELECT COUNT(*) FROM budget_transactions WHERE transaction_date='2026-08-25' AND CAST(amount_chf AS NUMERIC)=9261.30 AND transaction_type='income' AND category_id=?",(category['category_id'],)).fetchone()[0])
melanie_after=[tuple(row) for row in conn.execute("""SELECT p.plan_item_id,p.monthly_amount_chf,p.is_active
    FROM budget_plan_items p JOIN budget_categories c ON c.category_id=p.category_id
    WHERE c.name='Lohn Melanie' ORDER BY p.plan_item_id""").fetchall()]
if salary_count != 1: raise RuntimeError('August salary was not confirmed exactly once')
if melanie_before != melanie_after: raise RuntimeError('Melanie plan values changed unexpectedly')
melanie_zero_count=sum(1 for _,amount,is_active in melanie_after if is_active and amount is None)
report={'schema':51,'source_files_unchanged':True,'before_digest':before_digest,'after_first_confirm_digest':first_digest,'replay_digest_unchanged':True,'household':{'preview_counts':hh_preview['counts'],'expected_writes':hh_preview['expected_writes'],'confirm':hh,'fresh_replay_expected_writes':hh_fresh['expected_writes'],'replay_idempotent':hh_replay['idempotent']},'snapshots':[{k:v for k,v in x.items() if k!='entity_id'} for x in snapshot_results],'truewealth':{'total_chf':tw_preview['source_total_chf'],'activity_count':tw_preview['activity_count'],'activity_counts':tw_preview['activity_counts'],'external_cashflows_complete':tw_preview['external_cashflows_complete'],'replay_idempotent':tw_replay['idempotent']},'postfinance':{'total_chf':pf_preview['total_chf'],'securities_chf':pf_preview['securities_chf'],'cash_chf':pf_preview['cash_chf'],'event_count':pf_preview['event_count'],'event_counts':pf_preview['event_counts'],'performance_available':pf_preview['performance_available'],'performance_blockers':pf_preview['performance_blocker_codes'],'replay_idempotent':pf_replay['idempotent'],'projection_correction_rows':pf_correction['corrected_rows']},'salary_9261_30_count':salary_count,'melanie_plan_values_remain_unset_count':melanie_zero_count,'table_deltas':deltas,'cockpit_source_rows':[{k:s.get(k) for k in ('key','as_of','value_basis','last_activity_day','last_confirmed_snapshot','coverage_from','coverage_to','coverage_status','new_rows','duplicate_rows','review_rows','performance_status','performance_blocker')} for s in cockpit['sources']],'integrity':'ok','fk_violations':0}
REPORT.write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding='utf-8');os.chmod(REPORT,0o600)
print(json.dumps({'schema':51,'household_new':hh_preview['counts']['candidates'],'household_duplicates':hh_preview['counts']['duplicates'],'household_review':hh_preview['counts']['review'],'salary_count':salary_count,'truewealth_activities':tw_preview['activity_count'],'postfinance_events':pf_preview['event_count'],'postfinance_performance':pf_preview['performance_available'],'replay_zero_delta':True,'integrity':'ok','fk':0,'report':str(REPORT)},ensure_ascii=False))
conn.close()