`prepare.py`

119 lines
#!/usr/bin/env python3
"""Validate captured reads and prepare a non-executing, approval-gated plan."""
import collections, copy, datetime, difflib, hashlib, json, pathlib, subprocess

BASE=pathlib.Path(__file__).resolve().parent
V=BASE/'verification'
def read(name): return json.loads((V/name).read_text())
def save(name,data): (V/name).write_text(json.dumps(data,indent=2)+'\n')

def derive(prior,local,canonical,current,historical):
    expected={r['dw_sku']:r for r in prior['rows']}
    assert len(expected)==len(prior['rows'])==26
    assert local['read_only']==canonical['read_only']=='on'
    catalog=local['catalog']
    assert len(catalog)==5928
    assert len({r['mfr_sku'] for r in catalog})==5928
    local_by={r['dw_sku']:r for r in catalog}
    remote={r['dw_sku']:r for r in canonical['rows']}
    assert set(remote)==set(current)==set(expected)
    history={r['requested_id']:r['product'] for r in historical}
    assert len(history)==len(historical)==52
    rows=[]
    proposed={r['dw_sku']:r['mfr_sku'] for r in catalog}
    for sku,r in expected.items():
        old=r['snapshot_mac2_mfr_sku']; target=r['snapshot_kamatera_mfr_sku']
        assert remote[sku]['id']==local_by[sku]['id']==r['catalog_id']
        assert remote[sku]['mfr_sku']==target and local_by[sku]['mfr_sku']==old
        assert target!=old
        proposed[sku]=target
        variants=current[sku]
        assert len(variants)==3 and all(v['sku']==sku for v in variants)
        products=[v['product'] for v in variants]
        assert all(p['vendor']=='Carnegie' for p in products)
        assert all(not p['variants']['pageInfo']['hasNextPage'] for p in products)
        active=[p for p in products if p['status']=='ACTIVE']
        assert len(active)==1 and 'split-batch:carnegie-v2' in active[0]['tags']
        pilot_id='gid://shopify/Product/'+r['candidate_shopify_product_id']
        pilot=[p for p in products if p['id']==pilot_id]
        assert len(pilot)==1 and pilot[0]['status']=='ARCHIVED'
        assert history[pilot_id]['custom']==pilot[0]['custom']
        assert history[pilot_id]['dwc']==pilot[0]['dwc']
        assert history['gid://shopify/Product/'+r['earlier_generation_product_id']] is None
        plan=[]
        for generation,p in [('archived_pilot',pilot[0]),('active_v2',active[0])]:
            real=[v for v in p['variants']['nodes'] if 'sample' not in v['sku'].lower()]
            assert len(real)==1 and real[0]['sku']==sku, 'Grouped product cannot receive a scalar SKU'
            fields=[]
            for ns in ['custom','dwc']:
                field=p[ns]
                assert field and field['value']==old and field['type']=='single_line_text_field'
                fields.append(dict(namespace=ns,key='manufacturer_sku',id=field['id'],
                    type=field['type'],before=old,after=target,updated_at=field['updatedAt']))
            plan.append(dict(generation=generation,id=p['id'],status=p['status'],handle=p['handle'],
                variant_id=real[0]['id'],variant_sku=sku,metafields=fields))
        excluded=[p for p in products if p['id'] not in {q['id'] for q in plan}]
        assert len(excluded)==1 and excluded[0]['status']=='ARCHIVED'
        assert excluded[0]['custom'] is None and excluded[0]['dwc'] is None
        rows.append(dict(dw_sku=sku,catalog_id=r['catalog_id'],before=old,after=target,
            canonical_action='NOOP: already correct',local_updated_at=local_by[sku]['updated_at'],
            products=plan,excluded_grouped_product=excluded[0]['id']))
    assert len(set(proposed.values()))==5928, 'Restoration introduces identity collision'
    assert len({p['id'] for r in rows for p in r['products']})==52
    return dict(ticket='TK-11246',approval='REQUIRED — NOT EXECUTED',
        counts=dict(local_staging_rows=26,canonical_writes=0,active_products=26,archived_products=26,metafields=104),
        preconditions=['Re-read canonical/local rows and Shopify owner, exact variant SKU, status, metafield ID/type/value immediately before write.',
            'Abort entire scope on any identity, value, membership, uniqueness or status mismatch.',
            'Apply and verify suffix-preserving creation patch before any data repair.',
            'Create durable rollback prestate; local database transaction first, then one active-product canary, then bounded Shopify batches.',
            'Fresh API and DB readback must prove all targets plus unchanged exclusions; do not change SKU/title/handle/price/status/tags.',
            'Rollback only this repair using captured before values, conditional on current value equaling the proposed after value.'],
        historical_limit='Aug18 standalone write-ID log missing; pilot membership is corroborated by exact SKU, prior retitle IDs and metafield timestamps. Plan uses fresh observed defects, not assumed historical membership.',
        scope_limit='No claim about total current Carnegie storefront defects; broader stripping impact requires a separate read-only inventory.',rows=rows)

def main():
    args=[read(f) for f in ['prior-identity-report.json','current-catalog.json','canonical-catalog.json','current-by-sku.json','historical-products.json']]
    plan=derive(*args)
    save('repair-plan.json',plan)
    negatives=[]
    for label in ['wrong_vendor','wrong_metafield','extra_active','wrong_canonical','grouped_product']:
        a=copy.deepcopy(args); sku=next(iter(a[3])); variants=a[3][sku]
        p=next(v['product'] for v in variants if v['product']['status']=='ACTIVE')
        if label=='wrong_vendor':p['vendor']='Other'
        if label=='wrong_metafield':p['custom']['value']='different'
        if label=='extra_active':variants.append(copy.deepcopy(variants[-1]))
        if label=='wrong_canonical':a[2]['rows'][0]['mfr_sku']='wrong'
        if label=='grouped_product':p['variants']['nodes'].append(dict(id='other',sku='DWAG-OTHER'))
        try:derive(*a)
        except AssertionError:negatives.append(dict(case=label,verdict='PASS',rejected=True))
        else:raise AssertionError('Accepted unsafe fixture: '+label)
    sources=pathlib.Path.home()/'Projects/carnegie-split'
    patches=[]; checks=[]
    codes=[r['mfr_sku'] for r in args[1]['catalog']]+[r['after'] for r in plan['rows']]
    for name in ['rollout.mjs','rollout2.mjs','build-siltech-grain-v2-archived.mjs']:
        old=(sources/name).read_text()
        lines=[line for line in old.splitlines() if line.startswith('const cleanMfr')]
        assert len(lines)==1
        replacement='const cleanMfr = s => String(s); // Preserve category-bearing manufacturer identity (TK-11246).'
        new=old.replace(lines[0],replacement)
        patches.extend(difflib.unified_diff(old.splitlines(True),new.splitlines(True),fromfile='a/'+name,tofile='b/'+name))
        script="const vm=require('node:vm');const input="+json.dumps(dict(before=lines[0],after=replacement,codes=codes))+";const run=s=>vm.runInNewContext(s+'\\n;codes.map(cleanMfr)',{codes:input.codes});const before=run(input.before),after=run(input.after);if(after.some((x,i)=>x!==input.codes[i]))throw Error('Identity changed');console.log(JSON.stringify({tested:after.length,before_changed:before.filter((x,i)=>x!==input.codes[i]).length,after_changed:0}));"
        result=subprocess.run(['node','-'],input=script,text=True,capture_output=True,check=True)
        syntax=subprocess.run(['node','--check','--input-type=module'],input=new,text=True,capture_output=True,check=True)
        checks.append(dict(file=name,source_sha256=hashlib.sha256(old.encode()).hexdigest(),syntax='PASS',**json.loads(result.stdout)))
    (BASE/'preserve-manufacturer-identity.patch').write_text(''.join(patches))
    save('e2e-proof.json',dict(intent='Prepare exact read-only restoration evidence; no live repair performed',risk_tier='R3 read-only; proposed repair R4',
        timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(),environment='Local audit repo; Shopify read queries; read-only PostgreSQL local and Kamatera',
        commands=['python3 collect.py','ssh kamatera read-only SELECT (canonical-catalog.json)','python3 prepare.py','node isolated cleanMfr evaluation and module syntax check'],
        build_identity='Source digests per patch check; final git commit recorded on TK-11246',
        assertions=[dict(check='Exact26 canonical/local mapping and full-table postrepair uniqueness',verdict='PASS'),
            dict(check='52 product/104 namespace current prestate by exact SKU and generation',verdict='PASS'),
            dict(check='Missing historical generation excluded; missing node returns null',verdict='PASS'),
            dict(check='Original Aug18 standalone mutation membership log',verdict='SKIP',reason='Unavailable; only corroboration, no claim of recovered log'),
            dict(check='Production mutation and after-state verification',verdict='SKIP',reason='Approval required; no mutation executed')],
        negative_tests=negatives,patch_checks=checks,cleanup='No business writes or test entities; evidence retained locally.',
        result='PREPARATION VERIFIED; TICKET BLOCKED FOR APPROVAL',counts=plan['counts']))
    print(json.dumps(dict(counts=plan['counts'],negative_tests=len(negatives),patch_checks=checks),indent=2))

if __name__=='__main__':main()