"""Synthetic p044 formula tests; no real market input or account return access."""
import math
import time
import io
import zipfile
import hashlib
import json
from pathlib import Path
import numpy as np
import pandas as pd
import pytest
from tradingagent.factor_lab import overnight_recency as s
from research import audit_overnight_recency as a
import run_overnight_recency as runner


def native(rows=165,names=('NASDAQ:A','NASDAQ:B','NASDAQ:C')):
    dates=pd.bdate_range('2020-01-01',periods=rows)
    rng=np.random.default_rng(431)
    close=pd.DataFrame(100*np.exp(rng.normal(0,.006,(rows,len(names))).cumsum(axis=0)),index=dates,columns=list(names))
    opening=close.shift(1)*np.exp(rng.normal(0,.004,close.shape))
    opening.iloc[0]=100.
    return {'open':opening,'close':close}


@pytest.mark.parametrize('n',[10,21,42,63,126])
def test_formula_transcript_scalar_components_ranks(n):
    data=native();dates=data['close'].index[-5:]
    production,parts=s.values(data,n,True)
    transcript,tp,tr=a.transcript_values(data,n,dates)
    independent,ip,ir=a.scalar_values(data,n,dates,True)
    assert s.exact_values(production.loc[dates],transcript.loc[dates]).all()
    for name in parts: assert s.exact_values(parts[name].loc[dates],tp[name].loc[dates]).all()
    ranks,other,masks=a.value_checks(production.loc[dates],independent.loc[dates],{k:v.loc[dates] for k,v in parts.items()},{k:v.loc[dates] for k,v in ip.items()})
    assert all(not v.to_numpy().any() for v in masks.values())
    assert s.exact_values(ranks,other).all()
    assert list(parts)==['S','T','L']
    assert tr.loc[dates].eq('valid').all().all() and ir.loc[dates].eq('valid').all().all()
    original=s.prior.values(data,n,'overnight')
    control=s.method_values(data,s.point(n,'cumulative'))
    assert s.exact_values(original,control).all()


def test_price_window_edges_and_missing_session():
    data=native(20);n=10;end=data['close'].index[-1];first=data['close'].index[-n-1]
    expected=s.values(data,n).loc[end]
    unchanged={k:v.copy() for k,v in data.items()};unchanged['open'].loc[first]=math.nan;unchanged['close'].loc[end]=math.nan
    assert s.exact_values(expected,s.values(unchanged,n).loc[end]).all()
    for field,date in [('close',first),('open',end),('open',data['close'].index[-5]),('close',data['close'].index[-5])]:
        broken={k:v.copy() for k,v in data.items()};broken[field].loc[date]=math.nan
        assert s.values(broken,n).loc[end].isna().all()
    assert s.values(data,10).iloc[:10].isna().all().all()
    broken={k:v.copy() for k,v in data.items()};broken['open'].iloc[-4]=math.nan
    assert a.scalar_values(broken,10).loc[end].isna().all()


def test_quotient_underflow_overflow_nonpositive():
    data=native(12);data['close'].iloc[:]=1.;data['open'].iloc[:]=1.
    for numerator,denominator in [(np.nextafter(0.,1.),1e308),(1e308,np.nextafter(0.,1.)),(0.,1.),(-1.,1.),(math.inf,1.)]:
        bad={k:v.copy() for k,v in data.items()};bad['open'].iloc[-1]=numerator;bad['close'].iloc[-2]=denominator
        assert s.values(bad,10).iloc[-1].isna().all()
        pd.testing.assert_frame_equal(s.daily_terms(bad)['reason'],a.scalar_daily_terms(bad)['reason'])


@pytest.mark.parametrize('values',[[-.02,.01,.01,.01,.05],[-.02,-.01,.01,.02],[-0.,0.,-0.,0.],[0.,-0.,-0.],[-1e-16,1e-16],[.01,.01],[-1000.,-1000.],[700.,700.],[1e308],[1e308,1e308],[1e308,-1e308]])
def test_weighted_component_arithmetic_and_defensive_extremes(values):
    # Magnitudes beyond legal price log quotients are defensive helper cases.
    logs=pd.DataFrame({'x':values});terms={'log':logs,'valid':logs.notna().astype('u1')}
    raw,parts=s.recency_from_terms(terms,len(values),True)
    transcript,tp=a.weighted_components(values,True)
    assert s.exact_values([raw.iloc[-1,0]],[transcript]).all()
    for name in parts:assert s.exact_values([parts[name].iloc[-1,0]],[tp[name]]).all()
    scalar,sp=a.weighted_components(values,False)
    if all(abs(v)<100 for v in values):
        assert s.within_tolerance([raw.iloc[-1,0]],[scalar]).all()
        for name in parts:assert s.within_tolerance([parts[name].iloc[-1,0]],[sp[name]]).all()
    assert math.isfinite(transcript)==math.isfinite(scalar)


def test_time_direction_equal_total_and_constant_actual_operations():
    for values in [[.01,.02,.03],[.03,.02,.01],[0.,0.,0.],[.01]*10,[-.01]*21]:
        logs=pd.DataFrame({'x':values})
        raw,parts=s.recency_from_terms({'log':logs,'valid':logs.notna().astype('u1')},len(values),True)
        weights=np.arange(1,len(values)+1,dtype=np.float64)
        total=np.sum(np.multiply(weights,np.asarray(values)),dtype=np.float64)
        doubled=np.multiply(np.float64(2),total);scaled=np.divide(doubled,np.float64(len(values)+1))
        for name,value in [('S',total),('T',doubled),('L',scaled)]:assert s.exact_values([parts[name].iloc[-1,0]],[value]).all()
        assert s.exact_values([raw.iloc[-1,0]],[np.expm1(scaled)]).all()
    forward=a.weighted_components([.01,.02,.03],True)[0]
    backward=a.weighted_components([.03,.02,.01],True)[0]
    assert forward>backward and math.fsum([.01,.02,.03])==math.fsum([.03,.02,.01])
    assert a.weighted_components([-1000.,-1000.],True)[0]==-1.


@pytest.mark.parametrize('axis',['duplicate_dates','reversed_dates','duplicate_symbols','misaligned'])
def test_invalid_calendar_axes_rejected(axis):
    data=native()
    if axis=='duplicate_dates':
        for f in data:data[f]=pd.concat([data[f].iloc[:1],data[f]])
    elif axis=='reversed_dates':data={f:v.iloc[::-1] for f,v in data.items()}
    elif axis=='duplicate_symbols':data={f:v.set_axis(['A','A','B'],axis=1) for f,v in data.items()}
    else:data['open']=data['open'].iloc[:-1]
    with pytest.raises(ValueError):s.values(data,10)


def test_permutation_causality_constant_and_negative_scores():
    data=native();expected=s.values(data,21)
    perm={k:v.iloc[:,::-1] for k,v in data.items()}
    pd.testing.assert_frame_equal(expected,s.values(perm,21).reindex(columns=expected.columns))
    pd.testing.assert_frame_equal(expected.iloc[:-20],s.values({k:v.iloc[:-20] for k,v in data.items()},21))
    data['close'].iloc[:]=100.;data['open'].iloc[:]=99.
    assert s.values(data,21).iloc[-1].lt(0).all()
    data['open'].iloc[:]=100.;score=s.values(data,21).iloc[-1]
    assert score.eq(0).all() and not np.signbit(score).any()


@pytest.mark.parametrize('values',[[1.,1.,2.],[2.,2.,1.]])
def test_raw_tolerance_cannot_hide_rank_tie_or_signed_zero_change(values):
    index=pd.DatetimeIndex(['2024-01-02']);columns=['A','B','C']
    left=pd.DataFrame([values],index=index,columns=columns)
    right=left.copy();right.iloc[0,0]=np.nextafter(left.iloc[0,0],np.inf)
    assert s.within_tolerance(left,right).all()
    _,_,masks=a.value_checks(left,right,{},{});assert masks['rank_difference'].to_numpy().any()
    right=left.copy();right.iloc[0,0]=math.nan
    assert a.value_checks(left,right,{},{})[2]['domain_difference'].to_numpy().any()
    left.iloc[0,0]=0.;right=left.copy();right.iloc[0,0]=-0.
    assert a.value_checks(left,right,{},{})[2]['sign_difference'].iloc[0,0]


def test_registered_branches_and_calendar_constants():
    assert len(s.input_points('train'))==10 and len(s.grid())==5
    assert len(s.expected_keys('train',s.grid()))==40
    for n in s.WINDOWS:
        assert len(s.input_points('validation',s.point(n)))==2
        assert len(s.expected_keys('validation',s.validation_points(s.point(n))))==(6 if n in (63,126) else 12)
    assert s.calendar_budget_metadata()['train']['sessions']==1516
    assert s.calendar_budget_metadata()['validation']['sessions']==1965
    for n in [0,True,1.0]:
        with pytest.raises(ValueError): s.values(native(),n)
    with pytest.raises(ValueError): s.point(11)


def longest_csv_frames():
    names=['NASDAQ:A%04d'%i for i in range(10)]
    assert all(len(name)==12 for name in names)
    stamp='2262-04-11 13:30:00+00:00';date='2262-04-11 00:00:00'
    extreme=-np.finfo(float).max
    return {
        'alignment':pd.DataFrame([{'field':'close','signal_date':date,'source_open':stamp,'symbol':names[0],
            'native_value':extreme,'execution_value':extreme,'exact_equal':False,'coverage':'execution_only'}]),
        'targets':pd.DataFrame([{'time':stamp,'signal_date':date,'top_score':extreme,'top_tie_symbols':names,
            'chosen_symbol':names[0],'weights':[1.]+[0.]*9}]),
        'quote_errors':pd.DataFrame([{'time':stamp,'symbol':names[0],'field':'close','reason':'invalid_execution_time_index'}]),
        'anchor_errors':pd.DataFrame([{'reason':'old_cumulative_exact_score_or_choice_mismatch',
            'date':date,'period':'validation_continuous','cost':'stress_'}]),
    }


def test_longest_legal_actual_csv_serialization_is_within_fixed_widths():
    import io
    observed={}
    for kind,frame in longest_csv_frames().items():
        payload=s.bounded_csv(frame,kind)
        assert payload==frame.to_csv(index=False).encode('ascii')
        observed[kind]=len(payload.splitlines(keepends=True)[1])
        assert observed[kind]<=s.CSV_WIDTHS[kind]
        decoded=pd.read_csv(io.BytesIO(payload),float_precision='round_trip')
        if kind=='targets': assert decoded.top_score.iloc[0]==frame.top_score.iloc[0]
        if kind=='alignment': assert decoded.native_value.iloc[0]==frame.native_value.iloc[0]
    # Exact serialized maxima, not a calculation based only on field widths.
    assert observed=={'alignment':136,'targets':300,'quote_errors':74,'anchor_errors':96}


@pytest.mark.parametrize('field,value',[
    ('source_open','2024-01-02 13:30:00.000000001+00:00'),
    ('source_open','2024-01-02 13:30:00-05:00'),
    ('signal_date','2024-01-02 01:00:00'),
    ('symbol','NASDAQ:TOOLONG'),
    ('symbol','NASDAQ:中'),
    ('native_value','-1.797693134862315700e+308'),
    ('coverage','common\nextra'),
])
def test_serialization_guards_reject_invalid_fields_before_any_write(field,value):
    frame=longest_csv_frames()['alignment'].astype(object)
    frame.loc[0,field]=value
    with pytest.raises((AssertionError,UnicodeEncodeError)): s.bounded_csv(frame,'alignment')


def test_failure_archive_is_terminal_separate_and_hashes_originals(tmp_path,monkeypatch):
    import run_overnight_recency as runner
    monkeypatch.setattr(runner,'AUDITS',tmp_path/'audits')
    reservations=[]
    monkeypatch.setattr(runner,'space_budget',lambda **kw:reservations.append(kw))
    monkeypatch.setattr(s,'ROOT',tmp_path);monkeypatch.setattr(s,'PREVIOUS',tmp_path/'p043')
    original=tmp_path/'frozen_input';original.write_bytes(b'unchanged input')
    s.write(s.PREVIOUS/'plan.json',{'settings':{'input_hashes':{'frozen_input':s.sha(original)}}})
    folder=tmp_path/'external';folder.mkdir();source=folder/'source';source.write_bytes(b'raw values')
    revision=s.failure_archive('train','quotes',folder,
        csv_frames={'all_errors.csv':(longest_csv_frames()['quote_errors'],'quote_errors')})
    record=s.read(revision/'record.json')
    assert record['result']=='terminal_fail'
    assert record['referenced_immutable_originals']=={str(original):s.sha(original),str(source):s.sha(source)}
    assert record['files']=={'all_errors.csv':s.sha(revision/'all_errors.csv')}
    assert reservations==[{'planned_extra':{'tests_revisions_logs':64*1024**2,'atomic_and_margin':32*1024**2}}]
    assert source.read_bytes()==b'raw values' and not (revision/'source').exists()
    with pytest.raises(AssertionError):s.bytes_once(source,b'changed source')
    inventory=runner.storage_inventory([('external_training',folder),('tests_revisions_logs',revision)])
    assert not inventory['hardlink_aliases']


def test_transcript_terminal_failure_keeps_full_arrays_and_manifest(tmp_path,monkeypatch):
    import run_overnight_recency as runner
    monkeypatch.setattr(runner,'AUDITS',tmp_path/'audits')
    monkeypatch.setattr(runner,'space_budget',lambda **kw:None)
    monkeypatch.setattr(s,'ROOT',tmp_path);monkeypatch.setattr(s,'PREVIOUS',tmp_path/'p043')
    s.write(s.PREVIOUS/'plan.json',{'settings':{'input_hashes':{}}})
    data=native(8);frame=data['open'].copy();frame.iloc[1,0]=-0.
    mask=frame.notna();mask.iloc[:]=False;mask.iloc[1,0]=True
    with pytest.raises(ArithmeticError):
        s.transcript_failure('train','full_source',data,{'production_raw':frame},{'raw_mismatch':mask})
    revision=next(runner.AUDITS.glob('p044_revision_transcript_*'))
    manifest=s.read(revision/'complete_manifest.json')
    assert manifest['result']=='terminal_fail' and len(manifest['files'])==4
    for name,entry in manifest['files'].items():
        assert s.sha(revision/name)==entry['sha256']
    arrays=s.read_arrays(revision/'transcript_failures/train_full_source.npz')
    assert s.exact_values(arrays['source_open'],data['open']).all()
    assert s.exact_values(arrays['production_raw'],frame).all()
    assert np.array_equal(arrays['raw_mismatch'],mask.to_numpy())


def test_storage_proof_covers_success_and_terminal_failure_without_market_reads(monkeypatch):
    import run_overnight_recency as runner
    def forbidden(*args,**kwargs):raise AssertionError('Market input/old implicit preflight access forbidden')
    monkeypatch.setattr(s,'inputs',forbidden);monkeypatch.setattr(s,'daily_inputs',forbidden)
    monkeypatch.setattr(s.prior,'calendar_budget_metadata',forbidden)
    budget=runner.static_storage_budget();MiB=1024**2
    assert budget['external_training_calculated_bound']<=64*MiB
    assert budget['final_calculated_bound']<=256*MiB
    assert not any(e['path'].startswith('@quote_failure') for e in budget['entries'])
    for bound in budget['failure_archive_bounds'].values():
        assert bound['calculated_upper']<=bound['reserved_upper']==64*MiB
        assert bound['terminal_transcripts_per_invocation']==1
    report=runner.space_budget(stage='budget')
    assert report['planned_extra_upper_bytes']['external_training']>0
    assert report['planned_extra_upper_bytes']['tests_revisions_logs']==64*MiB


def test_live_failure_tmp_is_atomic_even_with_terminal_record(tmp_path,monkeypatch):
    import run_overnight_recency as runner
    revision=tmp_path/'p044_revision_transcript_interrupted';revision.mkdir()
    (revision/'record.json').write_text('{"result":"terminal_fail"}\n')
    (revision/'pending.npz.tmp').write_bytes(b'interrupted')
    monkeypatch.setattr(runner,'storage_roots',lambda:[('tests_revisions_logs',revision)])
    report=runner.space_budget()
    assert report['inventory']['categories']['atomic_and_margin']['charged_bytes']>=4096
    assert report['inventory']['categories']['tests_revisions_logs']['charged_bytes']>=4096
    with pytest.raises(AssertionError,match='Category budget exceeded'):
        runner.space_budget(planned_extra={'atomic_and_margin':32*1024**2})
    # Preserve the deliberately interrupted fixture, but do not leave a live
    # .tmp for future independent test admission. This is an explicit test-only
    # rename after its classification has been exercised, never a run recovery.
    (revision/'pending.npz.tmp').rename(revision/'retained_interrupted_bytes')


def test_validation_without_training_lock_has_no_input_or_output(tmp_path,monkeypatch):
    import sys
    import run_overnight_recency as runner
    monkeypatch.setattr(s,'OUT',tmp_path/'unregistered');calls=[]
    for obj,name in [(runner,'space_budget'),(runner,'check_sealed'),(runner,'status'),(runner,'execution_inputs'),
        (s,'inputs'),(s,'daily_inputs'),(s,'write')]:
        monkeypatch.setattr(obj,name,lambda *args,**kwargs:calls.append('forbidden'))
    with pytest.raises(FileNotFoundError):runner.run_split('validation')
    with pytest.raises(FileNotFoundError):s.preflight('validation',s.input_points('validation',s.point(63)),tmp_path/'gate')
    monkeypatch.setattr(sys,'argv',['run_overnight_recency.py','validation'])
    with pytest.raises(FileNotFoundError):runner.main()
    assert not calls and not (tmp_path/'unregistered').exists() and not (tmp_path/'gate').exists()


def synthetic_market(monkeypatch,tmp_path,branch,compact_prices=False):
    import run_overnight_recency as runner
    train={'train_continuous':['2025-05-19','2025-05-23'],'train_2022_h2':['2025-05-19','2025-05-20'],
        'train_2023':['2025-05-21','2025-05-22'],'train_2024':['2025-05-23','2025-05-23']}
    validation={'validation_continuous':['2025-06-02','2025-06-05'],'validation_2025':['2025-06-02','2025-06-03'],'validation_2026_ytd':['2025-06-04','2025-06-05']}
    monkeypatch.setattr(s,'TRAIN',train);monkeypatch.setattr(s,'VALIDATION',validation)
    monkeypatch.setattr(s,'PREVIOUS',tmp_path/'previous_p043');s.PREVIOUS.mkdir()
    monkeypatch.setattr(s,'CONTROL_SOURCE',s.PREVIOUS/'inherited_p040');s.CONTROL_SOURCE.mkdir()
    monkeypatch.setattr(s,'OUT',tmp_path/'p044');s.OUT.mkdir()
    dates=s.clock.schedule('2024-10-21','2025-06-05').loc['2024-10-21':'2025-06-05'].index
    names=s.SYMBOLS+[f'SYNTH:{j}' for j in range(87)];train_start=dates.get_loc(pd.Timestamp('2025-05-19'))
    data={'close':pd.DataFrame(100.,index=dates,columns=names),'open':pd.DataFrame(index=dates,columns=names,dtype=float)}
    for j,name in enumerate(names):
        logs=np.full(len(dates),.03 if j==0 else .005 if j==1 else -.005-j*.0001)
        if branch==63 and j==0:
            logs[:]=-.001;logs[train_start-74:train_start-24]=.03
        if compact_prices:
            assert branch==10
            # Fault-injection fixtures use explicit cent prices, not rounded
            # factor output. All97 source windows remain lossless and present.
            price=103. if j==0 else 100.5 if j==1 else float(f'{99.50-j*.01:.2f}')
            data['open'][name]=price
        else:data['open'][name]=100.*np.exp(logs)
    index=s.pure.original.session_layout('2025-05-19','2025-06-05')[0];steps=np.arange(len(index));bars={}
    for j,name in enumerate(s.SYMBOLS):
        op=100.*(1.04 if j==0 else .99)**steps;cl=100.*(1.04 if j==0 else .99)**(steps+.9)
        bars[name]=pd.DataFrame({'open':op,'high':np.maximum(op,cl)*1.001,'low':np.minimum(op,cl)*.999,'close':cl},index=index)
    def inputs(cutoff):
        current={f:frame.loc[:cutoff].copy() for f,frame in data.items()};axis=current['close'].index
        metadata={'fetched_at':'2026-10-18T00:00:00Z','factor_panel_calendar':{'synthetic':True,
            'first':str(axis[0]),'last':str(axis[-1]),'sessions':len(axis)}}
        return {f:frame[s.SYMBOLS] for f,frame in current.items()},current,bars,metadata
    monkeypatch.setattr(s,'inputs',inputs);monkeypatch.setattr(runner,'execution_inputs',lambda cutoff:(bars,inputs(cutoff)[3]))
    opens=pd.DatetimeIndex(s.clock.schedule(str(dates[0].date()),str(dates[-1].date())).loc[dates,'open'])
    daily={};hashes={}
    for name in s.SYMBOLS:
        daily[name]=pd.DataFrame({f:frame[name].to_numpy() for f,frame in data.items()},index=opens)
        relative='data/lab/us_20261018/'+name.replace(':','_')+'_1D.pkl';path=tmp_path/'raw_daily'/relative
        path.parent.mkdir(parents=True,exist_ok=True);daily[name].to_pickle(path);hashes[relative]=s.sha(path)
    monkeypatch.setattr(s,'daily_inputs',lambda:(daily,hashes))
    return data,bars


def synthesize_prior(data,bars,readonly_links=False):
    import run_overnight_recency as runner
    base=s.CONTROL_SOURCE/'inherited_p024';base.mkdir()
    opportunity=pd.DataFrame([{'symbol':name,'trough_close':'2025-05-18T20:00:00Z','peak_close':'2025-06-06T20:00:00Z','opportunity_return':.2} for name in s.SYMBOLS])
    for period in s.TRAIN|s.VALIDATION:opportunity.to_csv(base/f'opportunities_{period}.csv',index=False)
    groups={'cumulative':[],'momentum':[],'recent':[]}
    for split,periods in [('train',s.TRAIN),('validation',s.VALIDATION)]:
        configs=[('momentum_n84','momentum',84)]+[(s.point(n,'cumulative')['factor'],'cumulative',n) for n in (s.WINDOWS if split=='train' else [63,126])]
        for factor,method,n in configs:
            raw=pd.DataFrame(0.,index=data['close'].index,columns=s.SYMBOLS) if method=='momentum' else s.method_values(data,s.point(n,method))[s.SYMBOLS]
            root=base if method=='momentum' else s.PREVIOUS if split=='validation' and n==126 else s.CONTROL_SOURCE
            for period,(begin,end) in periods.items():
                prepared=s.pure.prepare(bars,raw,begin,end);folder=root/'runs'/factor/period;folder.mkdir(parents=True)
                row={'factor':factor,'family':'momentum' if method=='momentum' else s.CONTROL,'split':split,'period':period,'continuous':{},'diagnostics':{}}
                if method=='momentum':row['window']=84
                else:row.update(n=n,method=method,start=begin,end=end,open_observations=n,close_observations=n,observation_span=n+1,execution_mode='top1_full_investment')
                for rate,key,cost,tag in [(.001,'metrics','base',''),(.002,'cost_sensitivity','stress','stress_')]:
                    result=s.pure.run_account(prepared,rate)
                    row[key],row['continuous'][cost],row['diagnostics'][cost],tagged=s.pure.measurement(result,opportunity);result['tagged_attribution']=tagged
                    for field,frame in result.items():frame.to_csv(folder/(tag+field+'.csv'),index=False)
                target=root/f'input_targets/{split}/{factor}/{period}.csv';target.parent.mkdir(parents=True,exist_ok=True)
                target.write_bytes(s.bounded_csv(s.pure.targets(prepared),'targets'))
                if method=='cumulative':
                    row['target_path']=str(target.relative_to(root))
                    row['files']={str(p.relative_to(root)):s.sha(p) for p in [*folder.glob('*.csv'),target]}
                row=s.clean(row);s.write(folder/'summary.json',row);groups['recent' if root==s.PREVIOUS else method].append(row)
    for name in ['plan.json','training_manifest.json','selection_lock.json','data_manifest.json','verification.json']:s.write(base/name,{'synthetic':True})
    (base/'astra_review.md').write_text('Synthetic fixture only\n');s.write(base/'results.json',groups['momentum'])
    s.write(base/'artifacts.json',{str(p.relative_to(base)):s.sha(p) for p in base.rglob('*') if p.is_file()})
    s.write(s.CONTROL_SOURCE/'results.json',groups['cumulative'])
    s.write(s.PREVIOUS/'results.json',groups['recent'])
    refs={name:s.sha(runner.reference_source(name)) for name in runner.reference_paths()}
    for name,digest in refs.items():
        # The fault-injection fixtures never mutate these originals or aliases.
        # They are constructed once as same-category read-only links; this is
        # not deduplication of an existing or recoverable test archive.
        copier=runner.link_exact if readonly_links else runner.copy_exact
        copier(runner.reference_source(name),s.OUT/name,digest)
    s.write(s.OUT/'reference_manifest.json',refs);s.write(s.OUT/'expected_reference_paths.json',sorted(refs))
    return refs


@pytest.mark.parametrize('branch',[10,63])
def test_full_synthetic_branches_recovery_and_cash_audit(tmp_path,monkeypatch,branch):
    import run_overnight_recency as runner
    data,bars=synthetic_market(monkeypatch,tmp_path,branch)
    monkeypatch.setattr(runner,'check_sealed',lambda:s.read(s.OUT/'plan.json'))
    monkeypatch.setattr(runner,'space_budget',lambda **kwargs:{'synthetic_fixture_only':True})
    monkeypatch.setattr(a,'independent_dependency_check',lambda:{'synthetic_fixture_only':True})
    # Only provenance/seal/storage admission are mocked here. Formula panels,
    # stages, selection, targets, fee accounts and the final audit are real code.
    refs=synthesize_prior(data,bars)
    path=a.frozen(s.study.SNAPSHOT/'history_1300.pkl');path.parent.mkdir(parents=True)
    pd.to_pickle({name:pd.DataFrame({field:frame[name] for field,frame in data.items()}) for name in data['close']},path)
    pd.DataFrame({'symbol':data['close'].columns}).to_csv(a.frozen(s.study.SNAPSHOT/'universe_NDX.csv'),index=False)
    for name,frame in bars.items():
        path=a.frozen(s.ROOT/'data/lab/us_20261018'/(name.replace(':','_')+'_240.pkl'));path.parent.mkdir(parents=True,exist_ok=True);frame.to_pickle(path)
    for name,frame in s.daily_inputs()[0].items():frame.to_pickle(a.frozen(s.ROOT/'data/lab/us_20261018'/(name.replace(':','_')+'_1D.pkl')))
    source=s.CONTROL_SOURCE/'results.json';runner.copy_exact(source,s.OUT/'provenance_p040/results.json',s.sha(source))
    recent=s.PREVIOUS/'results.json';runner.copy_exact(recent,s.OUT/'provenance_p043/results.json',s.sha(recent))
    settings={'source_hashes':{},'input_hashes':{str(p.relative_to(s.OUT/'inputs')):s.sha(p) for p in (s.OUT/'inputs').rglob('*') if p.is_file()},
        'reference_manifest':refs,'evidence_hashes':{},'previous':{'results.json':s.sha(recent)},'control_provenance':{'results.json':s.sha(source)},'previous_context':{}}
    s.write(s.OUT/'plan.json',{'registered':runner.now(),'synthetic':True,'settings':settings})
    for name in ['astra_plan.md','pre_execution_review.md']:(s.OUT/name).write_text('Synthetic transport only\n')
    report=s.preflight('train',s.input_points('train'),tmp_path/'pre');assert report['result']=='pass'
    for name,digest in report['files'].items():runner.copy_exact(tmp_path/'pre'/name,s.OUT/name,digest)
    s.write(s.OUT/'input_preflight.json',{'report':report});runner.check_preflight(report,'train',s.input_points('train'),s.OUT)
    original=s.write;failure={'name':'selection_lock.json'}
    def interrupted(path,value):
        if Path(path).name==failure['name']:failure['name']=None;raise RuntimeError('Injected interruption')
        return original(path,value)
    monkeypatch.setattr(s,'write',interrupted)
    with pytest.raises(RuntimeError,match='Injected interruption'):runner.run_split('train')
    first=s.sha(s.OUT/'training_results.json');failure['name']='train_lock.json'
    with pytest.raises(RuntimeError,match='Injected interruption'):runner.run_split('train')
    selection=s.sha(s.OUT/'selection_lock.json');runner.run_split('train')
    assert s.sha(s.OUT/'training_results.json')==first and s.sha(s.OUT/'selection_lock.json')==selection
    selected=s.read(s.OUT/'selection_lock.json')['selected'];assert selected['n']==branch
    assert selected==a.independent_selection(s.read(s.OUT/'training_results.json'))
    hashes={name:s.sha(s.OUT/name) for name in ['train_lock.json','selection_lock.json','training_manifest.json']}
    failure['name']='validation_start.json'
    with pytest.raises(RuntimeError,match='Injected interruption'):runner.run_split('validation')
    hashes['validation_preflight.json']=s.sha(s.OUT/'validation_preflight.json')
    gate=s.read(s.OUT/'validation_preflight.json')['report']
    for bad in [{**gate,'result':'fail'},{**gate,'checks':gate['checks'][:-1]},
        {**gate,'checks':gate['checks'][:-1]+[gate['checks'][0]]}]:
        with pytest.raises(AssertionError):runner.check_preflight(bad,'validation',s.input_points('validation',selected),s.OUT)
    failure['name']='artifacts.json'
    with pytest.raises(RuntimeError,match='Injected interruption'):runner.run_split('validation')
    hashes['validation_start.json']=s.sha(s.OUT/'validation_start.json');hashes['validation_lock.json']=s.sha(s.OUT/'validation_lock.json')
    original_status=runner.status;status_failure={'pending':True}
    def interrupted_status(state):
        if state=='historical_complete' and status_failure['pending']:
            status_failure['pending']=False;raise RuntimeError('Injected status interruption')
        return original_status(state)
    monkeypatch.setattr(runner,'status',interrupted_status)
    with pytest.raises(RuntimeError,match='status interruption'):runner.run_split('validation')
    artifact_hash=s.sha(s.OUT/'artifacts.json');runner.run_split('validation')
    assert s.sha(s.OUT/'artifacts.json')==artifact_hash
    for name,digest in hashes.items():assert s.sha(s.OUT/name)==digest
    stable={name:s.sha(s.OUT/name) for name in ['artifacts.json','status.json']}
    runner.run_split('train');runner.run_split('validation')
    for name,digest in stable.items():assert s.sha(s.OUT/name)==digest
    extra=s.OUT/'runs'/'unexpected.csv';extra.write_bytes(b'Unexpected account evidence')
    with pytest.raises(AssertionError):runner.check_run_scope(selected,True,True)
    extra.rename(tmp_path/'retained_unexpected.csv')
    temp=s.OUT/'runs'/'interrupted.csv.tmp';temp.write_bytes(b'Unfinished account evidence')
    with pytest.raises(AssertionError):runner.check_run_scope(selected,True,True)
    temp.rename(tmp_path/'retained_interrupted.csv')
    account=s.OUT/'runs'/s.point(10)['factor']/'train_continuous'/'daily.csv'
    missing=tmp_path/'retained_missing.csv';account.rename(missing)
    with pytest.raises(AssertionError):runner.check_run_scope(selected,True,True)
    missing.rename(account)
    a.main();verified=s.read(s.OUT/'verification.json')
    assert verified['result']=='pass' and verified['accounts']==(46 if branch in (63,126) else 52) and verified['inherited_accounts_csv_checked']==66
    assert verified['max_equity_gap']<1e-12
    if branch==63:
        # Validate126's independent saved-row provenance as well as its path.
        source=s.OUT/'inherited_p043/runs/overnight_momentum_complete_n126/validation_2025/summary.json'
        original_bytes=source.read_bytes();forged=s.read(source)
        forged['target_path']='inherited_p040/'+forged['target_path']
        source.write_bytes(s.encoded(forged))
        with monkeypatch.context() as patch:
            patch.setattr(a,'check_saved_choices',lambda *args:None)
            with pytest.raises(AssertionError):a.reference_connections(refs)
        (tmp_path/'retained_forged_126_row.json').write_bytes(source.read_bytes())
        source.write_bytes(original_bytes)
        assert s.sha(source)==refs[str(source.relative_to(s.OUT))]


def test_training_selection_rejects_any_invalid_required_metric():
    rows=[]
    for pt in s.grid():
        for period in s.TRAIN:
            metrics={'big_move_capital_capture':1.,'net_return':1.,'max_drawdown':-.1}
            rows.append({**pt,'split':'train','period':period,'metrics':dict(metrics),'cost_sensitivity':dict(metrics)})
    assert s.selection(rows)==a.independent_selection(rows) and s.selection(rows)['n']==10
    for row in rows:
        if row['n']==63 and row['period']=='train_continuous':
            row['metrics']['big_move_capital_capture']=2.;row['cost_sensitivity']['big_move_capital_capture']=1.1
    assert s.selection(rows)==a.independent_selection(rows) and s.selection(rows)['n']==63
    rows[-1]['cost_sensitivity']['net_return']=None
    with pytest.raises(AssertionError):s.selection(rows)
    with pytest.raises(AssertionError):a.independent_selection(rows)


def test_hash_scope_deduplicates_physical_aliases_and_detects_changes(tmp_path):
    a=tmp_path/'a'; b=tmp_path/'b'; a.write_bytes(b'original'); b.hardlink_to(a)
    time.sleep(1.05) # This case deliberately tests a stable pre-scope inode.
    check=runner.HashCheck(); digest=check.digest(a)
    assert check.digest(b)==digest and check.reads==1
    b.write_bytes(b'changed!')
    with pytest.raises(AssertionError): check.check(a,digest)
    assert check.reads==2 and check.digest(b)!=digest and check.reads==3


def test_hash_fresh_same_length_same_tick_write_is_never_cached(tmp_path):
    a=tmp_path/'a'; b=tmp_path/'b'; a.write_bytes(b'original'); b.hardlink_to(a)
    check=runner.HashCheck(); old=check.digest(a)
    assert check.digest(b)==old and check.reads==2
    b.write_bytes(b'changed!')
    with pytest.raises(AssertionError): check.check(a,old)
    assert check.reads==3 and check.digest(b)==s.sha(b) and check.reads==4


def test_old_inode_changed_then_second_same_tick_alias_write_stays_uncached(tmp_path):
    a=tmp_path/'a'; b=tmp_path/'b'; a.write_bytes(b'original'); b.hardlink_to(a)
    time.sleep(1.05)
    check=runner.HashCheck(); old=check.digest(a)
    assert check.digest(b)==old and check.reads==1
    a.write_bytes(b'changed!'); first=check.digest(a)
    assert first!=old and check.reads==2
    b.write_bytes(b'again!!!'); second=check.digest(b)
    assert second!=first and check.reads==3
    inode=(a.stat().st_dev,a.stat().st_ino)
    assert inode in check.uncacheable_inodes and check.digest(a)==second and check.reads==4


def test_npz_independent_bits_schema_order_and_tampering(tmp_path):
    values=np.array([0.,-0.,np.nan,np.inf,-np.inf,np.nextafter(0.,1.),np.finfo(float).max],dtype='<f8')
    arrays={'date':np.array([1,2],dtype='<i8'),'symbols':np.array([b'NASDAQ:B',b'NASDAQ:A'],dtype='S8'),'raw':values,'support':np.array([0,1],dtype='u1')}
    p=tmp_path/'fixture.npz';schema=s.put_arrays(p,arrays);decoded,_=a.decode_arrays(p,list(arrays))
    for n,array in arrays.items():assert decoded[n].tobytes()==array.tobytes()
    original=p.read_bytes();s.put_arrays(p,arrays);assert p.read_bytes()==original
    with pytest.raises(AssertionError):a.decode_arrays(p,list(arrays)[::-1])
    for bad in [np.array([{}],object),np.array([1.],dtype='f4'),np.array(['bad'])]:
        with pytest.raises(ValueError):s.put_arrays(tmp_path/'bad.npz',{'raw':bad})
    p.write_bytes(original[:-8]+b'corrupt!')
    with pytest.raises(AssertionError):a.decode_arrays(p)

    p.write_bytes(original)
    manifest=p.with_suffix('.schema.json');j=json.loads(manifest.read_text());j['members']['raw']['shape']=[8];manifest.write_text(json.dumps(j))
    with pytest.raises(AssertionError):a.decode_arrays(p)
    j['members']['raw']['shape']=[7]
    buffer=io.BytesIO()
    with zipfile.ZipFile(io.BytesIO(original)) as src,zipfile.ZipFile(buffer,'w') as dst:
        for n in src.namelist():dst.writestr(n,src.read(n))
        dst.writestr('raw.npy',src.read('raw.npy'))
    payload=buffer.getvalue();p.write_bytes(payload);j.update(file_bytes=len(payload),sha256=hashlib.sha256(payload).hexdigest());manifest.write_text(json.dumps(j))
    with pytest.raises(AssertionError):a.decode_arrays(p)


def test_clock_nanosecond_axis_and_invalid_source_schema(tmp_path,monkeypatch):
    index=pd.date_range('2025-01-02',periods=3).as_unit('ns')
    data={field:pd.DataFrame({'NASDAQ:A':[100.,101.,102.]},index=index) for field in ['open','close']}
    left=tmp_path/'ns.npz';right=tmp_path/'us.npz'
    s.put_arrays(left,s.prior.source_panel_arrays(data,index))
    micro={field:frame.set_axis(index.as_unit('us')) for field,frame in data.items()}
    s.put_arrays(right,s.prior.source_panel_arrays(micro,index.as_unit('us')))
    assert left.read_bytes()==right.read_bytes()
    shifted=pd.DatetimeIndex(index.asi8+1)
    fine={field:frame.set_axis(shifted) for field,frame in data.items()}
    assert np.array_equal(s.prior.source_panel_arrays(fine,shifted)['dates_ns'],index.asi8+1)
    assert not a.exact_utc_times(index,shifted)
    path=tmp_path/'wrong.npz';s.put_arrays(path,s.prior.source_panel_arrays(data,index),axes={'rows':'wrong'})
    monkeypatch.setattr(s,'OUT',tmp_path)
    with pytest.raises(AssertionError):
        a.check_scores(data,{'calendar':{'last':str(index[-1])},'layout':{'source_path':'wrong.npz'}},index,['NASDAQ:A'])


@pytest.mark.parametrize('mode',['domain','rank'])
def test_failed_full_grid_keeps_complete_domain_and_rank_evidence(tmp_path,monkeypatch,mode):
    data,bars=synthetic_market(monkeypatch,tmp_path,10,compact_prices=True)
    data['open']['SYNTH:1']=data['open']['SYNTH:0']
    monkeypatch.setattr(runner,'space_budget',lambda **kw:{'synthetic_fixture_only':True})
    synthesize_prior(data,bars,readonly_links=True);original=a.scalar_values
    def changed(native,n,dates=None,return_details=False,method='recency',terms=None):
        result=original(native,n,dates,return_details,method,terms)
        if n==10 and method=='recency':
            raw,parts,reasons=result;date=pd.Timestamp('2025-05-16')
            if mode=='domain':raw.loc[date,s.SYMBOLS[-1]]=np.nan
            else:raw.loc[date,'SYNTH:0']=np.nextafter(raw.loc[date,'SYNTH:0'],np.inf)
        return result
    monkeypatch.setattr(a,'scalar_values',changed)
    folder=tmp_path/'failed_preflight';report=s.preflight('train',s.input_points('train'),folder)
    assert report['result']=='fail' and len(report['checks'])==40 and len(report['scores'])==10
    score=next(r for r in report['scores'] if r['factor']==s.point(10)['factor'])
    arrays,_=a.decode_arrays(folder/score['path'])
    assert arrays['raw'].shape==arrays['independent_raw'].shape==(6,97)
    if mode=='domain':assert arrays['domain_difference'].sum()==1
    else:
        assert arrays['raw_outside'].sum()==0 and arrays['rank_difference'].sum()>0
        assert score['scalar_checks']['rank_difference']>0
    for name,digest in report['files'].items():assert s.sha(folder/name)==digest
    with pytest.raises(AssertionError):runner.check_preflight(report,'train',s.input_points('train'),folder)


def test_combined_failure_metadata_guard_preserves_oversized_components(tmp_path,monkeypatch):
    monkeypatch.setattr(runner,'AUDITS',tmp_path/'audits');monkeypatch.setattr(runner,'space_budget',lambda **kw:None)
    monkeypatch.setattr(s,'ROOT',tmp_path);monkeypatch.setattr(s,'PREVIOUS',tmp_path/'p043')
    s.write(s.PREVIOUS/'plan.json',{'settings':{'input_hashes':{}}})
    def oversized(folder,*args):
        # Artificial schema corruption tests aggregate admission, not a valid
        # producer schema (which is separately capped at32KiB).
        (folder/'oversized.schema.json').write_bytes(b' '*(2*1024**2))
        raise ArithmeticError('Injected malformed metadata')
    monkeypatch.setattr(s.prior,'transcript_failure',oversized)
    with pytest.raises(AssertionError,match='Combined failure metadata'):
        s.transcript_failure('train','metadata',{}, {}, {})
    folder=next(runner.AUDITS.glob('p044_revision_transcript_*'))
    assert (folder/'oversized.schema.json').stat().st_size==2*1024**2
    assert not (folder/'complete_manifest.json').exists()


def test_real_binding_registration_json_recovery_and_tamper(tmp_path,monkeypatch):
    assert runner.reference_paths()==a.independent_reference_paths()
    monkeypatch.setattr(s,'OUT',tmp_path/'registered');folder=tmp_path/'pre';folder.mkdir()
    points=s.input_points('train');files={};checks=[];scores=[];paired=[];samples=[]
    def fixture(name):
        path=folder/name;path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(b'Synthetic registration transport only\n')
        files[name]=s.sha(path);return name
    source=fixture('panels/train/source.npz');fixture(source.replace('.npz','.schema.json'))
    daily=fixture('panels/train/overnight_daily.npz');fixture(daily.replace('.npz','.schema.json'))
    alignment=fixture('daily_source/train.csv');quotes=fixture('daily_source/train_opening_quotes.csv')
    for pt in points:
        panel=fixture('panels/train/'+pt['factor']+'.npz');fixture(panel.replace('.npz','.schema.json'))
        scores.append({**pt,'path':panel,'native_symbols':97,'scalar_checks':{'raw_outside':0,'rank_difference':0}})
        for period in s.TRAIN:
            target=fixture(f'input_targets/train/{pt["factor"]}/{period}.csv')
            mapping=fixture(f'bar_mappings/train/{pt["factor"]}/{period}.csv');shared=f'bar_mappings/train/{period}.csv'
            if shared not in files:fixture(shared)
            checks.append({**pt,'period':period,'result':'pass','errors':[],'top1_exact':True,'target_path':target,
                'target_hash':files[target],'mapping_path':mapping,'mapping_hash':files[mapping],'shared_mapping_path':shared,'panel_path':panel})
    for n in s.WINDOWS:
        path=fixture(f'panels/train/paired_support_n{n}.npz');fixture(path.replace('.npz','.schema.json'));paired.append({'n':n,'path':path})
        samples.append({'n':n,'path':fixture(f'panels/train/window_samples_n{n}.json')})
    anchors={}
    for n in s.WINDOWS:
        factor=s.point(n,'cumulative')['factor']
        for period in s.TRAIN:
            names=[f'input_targets/train/{factor}/{period}.csv']+[f'runs/{factor}/{period}/{prefix}rebalances.csv' for prefix in ['', 'stress_']]
            for name in names:
                relative='inherited_p040/'+name
                anchors[relative]=s.sha(s.PREVIOUS/relative)
    report={'result':'pass','split':'train','points':points,'files':files,'checks':checks,'scores':scores,
        'native_symbols':s.SYMBOLS+[f'SYNTH:{i}' for i in range(87)],'execution_symbols':s.SYMBOLS,
        'layout':{'source_path':source},'daily_source_alignment':{'errors':[],'path':alignment},'opening_quotes':{'path':quotes},
        'daily_evidence':{'path':daily,'mismatch_counts':{'domain_difference':0}},
        'paired_support':paired,'window_samples':samples,'cumulative_anchor_files':anchors}
    pre=tmp_path/'pre.json';review=tmp_path/'review.md';review.write_text('Synthetic registration transport only; not scientific admission\n')
    monkeypatch.setattr(runner,'PREFLIGHT',pre);monkeypatch.setattr(runner,'PREFLIGHT_FOLDER',folder);monkeypatch.setattr(runner,'PRE_REVIEW',review)
    with runner.verification_scope('test_real_registration',save=False):
        bound=runner.binding(deep=True);s.write(pre,{'binding':bound,'report':report})
        assert [len(bound[key]) for key in ['source_hashes','input_hashes','reference_manifest','previous','control_provenance','previous_context','evidence_hashes']]==[278,31,477,7,7,8,2]
        original=runner.copy_exact;failure={'part':'provenance_p040'}
        def interrupted(source,target,digest):
            if failure['part'] and failure['part'] in target.parts:failure['part']=None;raise RuntimeError('Injected registration interruption')
            return original(source,target,digest)
        monkeypatch.setattr(runner,'copy_exact',interrupted)
        with pytest.raises(RuntimeError,match='registration interruption'):runner.register()
        plan=s.sha(s.OUT/'plan.json');failure['part']='input_preflight.json'
        with pytest.raises(RuntimeError,match='registration interruption'):runner.register()
        runner.register();assert s.sha(s.OUT/'plan.json')==plan;runner.check_sealed()
        with pytest.raises(FileNotFoundError):runner.run_split('validation')
        assert not (s.OUT/'validation_start.json').exists() and not (s.OUT/'runs').exists()
        for group,prefix in [('source_hashes','sources'),('input_hashes','inputs'),('reference_manifest','.')]:
            path=s.OUT/prefix/next(iter(bound[group]));payload=path.read_bytes();path.write_bytes(b'tampered')
            with pytest.raises(AssertionError):runner.check_sealed()
            evidence=tmp_path/('tampered_'+group);evidence.write_bytes(path.read_bytes());path.write_bytes(payload)
        path=s.OUT/checks[0]['target_path'];path.write_bytes(b'tampered gate fixture')
        with pytest.raises(AssertionError):runner.check_sealed()


@pytest.mark.parametrize('n,prefix',[(63,'inherited_p040/'),(126,'')])
def test_existing_validation_routes_and_reject_wrong_targets(tmp_path,monkeypatch,n,prefix):
    monkeypatch.setattr(s,'PREVIOUS',tmp_path/'p043')
    names=s.SYMBOLS;date=pd.Timestamp('2025-06-02');instant=pd.Timestamp('2025-06-03 13:30:00+00:00')
    raw=pd.DataFrame([np.arange(10,dtype=float)],index=[date],columns=names)
    targets=a.scalar_targets(raw,[instant],[date]);pt=s.point(n,'cumulative');period='2025'
    relative=prefix+f'input_targets/validation/{pt["factor"]}/{period}.csv'
    target=s.PREVIOUS/relative;target.parent.mkdir(parents=True,exist_ok=True);target.write_bytes(s.bounded_csv(targets,'targets'))
    rb=targets.copy();rb['scores']=[np.arange(10,dtype=float).tolist()]
    expected={relative:s.sha(target)}
    for tag in ['', 'stress_']:
        name=prefix+f'runs/{pt["factor"]}/{period}/{tag}rebalances.csv';path=s.PREVIOUS/name
        path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(rb.to_csv(index=False).encode());expected[name]=s.sha(path)
    assert s.control_reference_prefix('validation',n)==prefix
    assert s.control_anchor(raw,targets,'validation',pt,period)==(expected,[])
    wrong=targets.copy();wrong['chosen_symbol']=names[0]
    assert s.control_anchor(raw,wrong,'validation',pt,period)[1][0]['reason']=='old_cumulative_target_bytes_mismatch'
    # An apparently valid file at the wrong historical level must never rescue
    # a missing or corrupted source at the registered 63/126 route.
    other=('' if prefix else 'inherited_p040/')+relative.removeprefix(prefix)
    alternate=s.PREVIOUS/other;alternate.parent.mkdir(parents=True,exist_ok=True);alternate.write_bytes(target.read_bytes())
    target.rename(tmp_path/'preserved_original_target.csv')
    with pytest.raises(FileNotFoundError):s.control_anchor(raw,targets,'validation',pt,period)


def test_reference_sets_and_current_source_links_match_frozen_protocol():
    paths=runner.reference_paths();assert paths==a.independent_reference_paths() and len(paths)==477
    assert len([n for n in paths if n.endswith('/summary.json')])==33
    assert sum(n.startswith('inherited_p043/runs/overnight_momentum_complete_n126/') for n in paths)==39
    assert len(s.inherited_validation_keys())==18