"""Independent value, scope, identity and validation checks for issues #139-#142."""
import hashlib
import json
import re
import sqlite3
from collections import defaultdict
from datetime import datetime
from decimal import Decimal
from pathlib import Path

LAB=Path(__file__).resolve().parents[1]
RELEASE=LAB/'releases/launcher-0.1.25'
EVIDENCE=LAB/'evidence/launcher-0.1.25-retest-2026-10-20'
TOLERANCE=Decimal('1E-12')
ZERO=Decimal('0')


def load(path):return json.loads(Path(path).read_text(encoding='utf-8-sig'))


def near(actual,expected):
    assert actual is not None,(actual,expected)
    assert abs(Decimal(str(actual))-expected)<TOLERANCE,(actual,expected)


def main():
    identity=load(EVIDENCE/'install-verification.json')
    proc=load(EVIDENCE/'process-identity.json')
    manifest=load(RELEASE/'runtime-launcher-manifest.json')
    assert identity['launcherVersion']==manifest['releaseVersion']=='0.1.25'
    assert manifest['source']['commit']=='db5db39aa36dc30b51d9db7d941e333c8d39e0d2'
    assert identity['githubAssetDigestsMatch']
    assert proc['listenerMatchesLauncher'] and proc['listeningPort']==18068
    assert str(RELEASE/'foggy-runtime-launcher-0.1.25.jar') in proc['commandLine']
    assert 'foggy-runtime-launcher-0.1.24.jar' not in proc['commandLine']
    assert hashlib.sha256((RELEASE/'foggy-runtime-launcher-0.1.25.jar').read_bytes()).hexdigest()==manifest['jar']['sha256']
    for name,digest in identity['modelHashes'].items():
        assert hashlib.sha256((RELEASE/'models'/name).read_bytes()).hexdigest()==digest
        assert hashlib.sha256((LAB/'models'/name).read_bytes()).hexdigest()==digest

    original=LAB.parent/'2026-10-13/open-data/chinook-v1.4.5/Chinook_Sqlite.sqlite'
    for p in [original,RELEASE/'data/Chinook.sqlite',LAB/'data/Chinook_Sqlite.sqlite']:
        assert hashlib.sha256(p.read_bytes()).hexdigest()==identity['dataSha256']
    primary=Path(load(EVIDENCE/'launcher-start.json')['sqlitePath'])
    with sqlite3.connect(primary.as_uri()+'?mode=ro&immutable=1',uri=True) as db:
        assert not db.execute("select name from sqlite_master where type='table' and lower(name)='invoice'").fetchall()
    with sqlite3.connect(original.as_uri()+'?mode=ro&immutable=1',uri=True) as db:
        invoices=list(db.execute('select InvoiceId, InvoiceDate, Total from Invoice'))
    totals=defaultdict(lambda:ZERO);counts=defaultdict(int)
    for _,date,value in invoices:totals[date[:7]]+=Decimal(str(value));counts[date[:7]]+=1
    cases={c['name']:c for c in load(EVIDENCE/'query-manifest.json')['cases']}
    assert len(cases)==17
    for issue,case_name in [(139,'07-time-yoy'),(140,'08-time-mom'),(141,'07-time-yoy'),(142,'10-lag-separate-calculations')]:
        body=load(EVIDENCE/f'issue-{issue}-before.json')['body']
        documented=[]
        for block in re.findall(r'```json\s*\n(.*?)\n```',body,re.S):
            try:documented.append(json.loads(block))
            except json.JSONDecodeError:pass
        assert load(cases[case_name]['payload']) in documented,(issue,case_name)
    for c in cases.values():
        raw=Path(c['payload']).read_bytes()
        assert hashlib.sha256(raw).hexdigest()==c['payloadSha256']
        if c['sourcePayload']:assert raw==Path(c['sourcePayload']).read_bytes()

    def rows(name):
        c=cases[name];assert c['validationSuccess'] and c['executionSuccess'],name
        assert load(c['validation'])['success']
        r=load(c['response']);assert r['success']
        assert not r['data'].get('hasNext') and not r['data'].get('pagination',{}).get('hasMore'),name
        return r['data']['items']

    def monthly(name,first='2012-01'):
        actual=rows(name);keys=sorted(k for k in totals if first<=k<'2013-01')
        assert [r['invoiceDate$yearMonth'] for r in actual]==keys,name
        for i,r in enumerate(actual):
            near(r['totalAmount'],totals[keys[i]])
            assert r['billCount']==counts[keys[i]]
        return actual,keys

    control,keys=monthly('01-monthly-control')
    assert len(control)==12 and sum((totals[k] for k in keys),ZERO)==Decimal('477.53')
    for name,first in [('03-lag-group-measure','2012-01'),('05-lag-group-alias','2012-01'),
        ('12-lag-control-without-explain','2012-01'),('13-lag-including-prior-december','2011-12')]:
        actual,keys=monthly(name,first)
        for i,r in enumerate(actual):
            if not i:assert r.get('previousAmount') is None
            else:near(r['previousAmount'],totals[keys[i-1]])
    assert 'explain' not in cases['12-lag-control-without-explain']
    ranked,_=monthly('06-rank-group')
    assert [r['previousAmount'] for r in ranked]==list(range(1,13))
    detail=rows('02-lag-detail')
    raw=sorted((i,Decimal(str(v))) for i,d,v in invoices if '2012-01-01'<=d<'2013-01-01')
    assert len(detail)==len(raw)==83
    for idx,(r,(ident,value)) in enumerate(zip(detail,raw)):
        assert r['invoiceId']==ident;near(r['invoiceAmount'],value)
        if idx:near(r['previousAmount'],raw[idx-1][1])
        else:assert r.get('previousAmount') is None

    # Original #142 negative input must fail before JDBC execution.
    reject=cases['04-lag-group-nested-sum']
    err=load(reject['validation'])['error']
    assert reject['validationSuccess'] is False and 'response' not in reject
    assert err['code']=='WINDOW_NESTED_AGGREGATE_UNSUPPORTED' and err['phase']=='query.validate'
    assert err['field']=='previousAmount' and err['path']=='calculatedFields[0].expression'
    assert err.get('suggestedNextAction') and 'alias' in err['suggestedNextAction'].lower()
    # The suggested alias-based repair actually works in case 05.
    assert cases['05-lag-group-alias']['executionSuccess']

    for name in ['10-lag-separate-calculations','11-lag-compound-no-sum']:
        actual,keys=monthly(name)
        for i,r in enumerate(actual):
            if not i:
                assert r.get('difference') is None
                if name.startswith('10-'):assert r.get('previousAmount') is None and r.get('changeRatio') is None
            else:
                prior=totals[keys[i-1]];diff=totals[keys[i]]-prior
                near(r['difference'],diff)
                if name.startswith('10-'):
                    near(r['previousAmount'],prior);near(r['changeRatio'],diff/prior)
    november=next(r for r in rows('10-lag-separate-calculations') if r['invoiceDate$yearMonth']=='2012-11')
    near(november['difference'],Decimal('-5.00'))

    def comparison(name):
        c=cases[name];p=load(c['payload']);tw=p['timeWindow']
        start,end=tw['value'];assert tw['range']=='[)'
        expected=sorted(k for k in totals if start[:7]<=k<end[:7])
        actual=rows(name)
        assert [f"{r['invoiceDate$year']:04}-{r['invoiceDate$month']:02}" for r in actual]==expected,name
        for r,k in zip(actual,expected):
            year,month=map(int,k.split('-'))
            if tw['comparison']=='yoy':prior_key=f'{year-1:04}-{month:02}'
            else:
                prior_key=f'{year-1:04}-12' if month==1 else f'{year:04}-{month-1:02}'
            near(r['invoiceAmount'],totals[k])
            if prior_key not in totals:
                assert all(r.get(f'invoiceAmount__{suffix}') is None for suffix in ['prior','diff','ratio'])
            else:
                prior=totals[prior_key];diff=totals[k]-prior
                near(r['invoiceAmount__prior'],prior);near(r['invoiceAmount__diff'],diff)
                near(r['invoiceAmount__ratio'],diff/prior)
            if 'invoiceDate$id' in r:assert r['invoiceDate$id']==k+'-01'
        trace=load(c['explain'])['data']['sqlTrace']
        sql=trace['finalPhysicalSql']
        assert 'WHERE' in sql.upper() and len(trace.get('parameters',[]))>=4,name
        return len(actual)

    comparative={name:comparison(name) for name in ['07-time-yoy','08-time-mom','09-mom-extra-day-key-changes-grain',
        '14-mom-cross-year','15-yoy-two-months','16-yoy-no-prior-data','17-mom-empty-period']}
    assert comparative['14-mom-cross-year']==2 and comparative['15-yoy-two-months']==2
    assert comparative['17-mom-empty-period']==0
    baseline=dict(monthly2012=[dict(month=k,amount=str(totals[k]),bills=counts[k]) for k in sorted(totals) if k.startswith('2012-')],
        november=november,comparativeRows=comparative,calculation='Read-only original Invoice rows, date-string grouping and Decimal.')
    (EVIDENCE/'baseline.json').write_text(json.dumps(baseline,ensure_ascii=False,indent=2),encoding='utf-8')
    report=dict(checkedAt=datetime.now().astimezone().isoformat(),success=True,launcherVersion='0.1.25',
        sourceCommit=manifest['source']['commit'],runtimeUrl='http://127.0.0.1:18068',namespace='chinook',dataSource='chinook-sqlite',
        processId=proc['pid'],model='ChinookInvoiceTimeQueryModel',cases=17,executedCorrectly=16,expectedValidationRejections=1,
        issueResults={
            '139':dict(result='Original issue no longer reproduces',evidenceCases=['07-time-yoy'],defaultDatabaseHasInvoice=False),
            '140':dict(result='Original issue and cross-year scenario pass',evidenceCases=['08-time-mom','09-mom-extra-day-key-changes-grain','14-mom-cross-year']),
            '141':dict(result='Date bounds and prior-period scope verified in SQL and actual rows',evidenceCases=['07-time-yoy','15-yoy-two-months','16-yoy-no-prior-data','17-mom-empty-period']),
            '142':dict(result='Correct dependency/compound results; nested aggregation rejected with field/path/action',evidenceCases=['04-lag-group-nested-sum','05-lag-group-alias','10-lag-separate-calculations','11-lag-compound-no-sum'])},
        validationDiagnostic=err,numericTolerance=str(TOLERANCE),originalPayloadsByteIdentical=13,
        actualNovemberDifference=november['difference'],actualNovemberRatio=november['changeRatio'],
        sourceModelsAndDataUnchanged=True,
        notVerified=['Other databases','Additional business-dimension partitions','Synthetic zero-valued prior month',
                     'Direct execute of an already validation-rejected payload','Production permissions','Console/FAP','Full release gate'],
        liveIssueReproductionsMatch=True,textSqlBenchmarkExecuted=False,cleanSessionPromptVerified=False)
    (EVIDENCE/'verification.json').write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding='utf-8')
    print(json.dumps(report,ensure_ascii=False,indent=2))


if __name__=='__main__':main()