`execute.py`

192 lines
#!/usr/bin/env python3
"""TK-11246 approved bounded repair. Explicit phases, durable prestate, CAS."""
import argparse, copy, datetime, hashlib, json, os, pathlib, subprocess, time, urllib.request
import collect

BASE=pathlib.Path(__file__).resolve().parent
RUN=BASE/'execution-20260911'
RUN.mkdir(exist_ok=True)
PLAN=json.loads((BASE/'verification/repair-plan.json').read_text())
PRODUCTS={p['id']:(r,p) for r in PLAN['rows'] for p in r['products']}
EXCLUDED={r['excluded_grouped_product'] for r in PLAN['rows']}
FIELDS='''id title handle vendor status tags descriptionHtml productType
variants(first:100){pageInfo{hasNextPage} nodes{id sku title price compareAtPrice}}
metafields(first:250){pageInfo{hasNextPage} nodes{id namespace key type value compareDigest}}'''
SET='mutation Repair($m:[MetafieldsSetInput!]!){metafieldsSet(metafields:$m){metafields{id namespace key value compareDigest} userErrors{field message code}}}'
def now():return datetime.datetime.now(datetime.timezone.utc).isoformat()
def save(name,data):
    path=RUN/name
    with path.open('w') as f:
        json.dump(data,f,indent=2);f.write('\n');f.flush();os.fsync(f.fileno())
def read(name):return json.loads((RUN/name).read_text())
def product_baseline():
    return read('products-baseline.json')['products'] if (RUN/'products-baseline.json').exists() else read('preflight.json')['products']
def log(event,**data):
    with (RUN/'actions.jsonl').open('a') as f:
        f.write(json.dumps(dict(ts=now(),event=event,**data))+'\n');f.flush();os.fsync(f.fileno())
def sql(query,write=False):
    env=dict(os.environ,PGOPTIONS=f'-c default_transaction_read_only={"off" if write else "on"} -c statement_timeout=15000 -c lock_timeout=5000',PGCONNECT_TIMEOUT='3')
    r=subprocess.run(['psql','-X','-q','-d','dw_unified','-At','-v','ON_ERROR_STOP=1'],input=query,text=True,capture_output=True,env=env)
    if r.returncode:raise RuntimeError(r.stderr)
    return r.stdout.strip()
def canonical():
    q="SELECT json_build_object('read_only',current_setting('default_transaction_read_only'),'rows',(SELECT json_agg(t) FROM (SELECT id,dw_sku,mfr_sku FROM carnegie_catalog WHERE dw_sku BETWEEN 'DWAG-379296' AND 'DWAG-379321' ORDER BY dw_sku) t));"
    cmd="PGOPTIONS='-c default_transaction_read_only=on -c statement_timeout=5000' psql -d dw_unified -At -c "+"'"+q.replace("'","'\"'\"'")+"'"
    r=subprocess.run(['ssh','-o','BatchMode=yes','-o','ConnectTimeout=8','kamatera',cmd],capture_output=True,text=True,check=True)
    data=json.loads(r.stdout);assert data['read_only']=='on'
    assert {(r['id'],r['dw_sku'],r['mfr_sku']) for r in data['rows']}=={(r['catalog_id'],r['dw_sku'],r['after']) for r in PLAN['rows']}
    return data
def fetch_products(ids):
    output={}
    for i in range(0,len(ids),4):
        batch=ids[i:i+4]
        d=collect.gql('query Read($ids:[ID!]!){nodes(ids:$ids){... on Product{'+FIELDS+'}}}',{'ids':batch})
        assert len(d['nodes'])==len(batch)
        for pid,p in zip(batch,d['nodes']):
            assert p and p['id']==pid
            assert not p['variants']['pageInfo']['hasNextPage'] and not p['metafields']['pageInfo']['hasNextPage']
            output[pid]=p
        time.sleep(.4)
    return output
def mfmap(p):return {(m['namespace'],m['key']):m for m in p['metafields']['nodes']}
def stable(p):
    p=copy.deepcopy(p)
    p['metafields']['nodes']=[{k:v for k,v in m.items() if k!='compareDigest'} for m in p['metafields']['nodes'] if (m['namespace'],m['key']) not in {('custom','manufacturer_sku'),('dwc','manufacturer_sku')}]
    p['metafields']['nodes'].sort(key=lambda m:m['id'])
    p['variants']['nodes'].sort(key=lambda v:v['id'])
    p['tags'].sort()
    return p
def assert_product(pid,p,expected):
    r,planned=PRODUCTS[pid]
    assert p['vendor']=='Carnegie' and p['status']==planned['status'] and p['handle']==planned['handle']
    real=[v for v in p['variants']['nodes'] if 'sample' not in v['sku'].lower()]
    assert len(real)==1 and real[0]['sku']==r['dw_sku'] and real[0]['id']==planned['variant_id']
    m=mfmap(p)
    for field in planned['metafields']:
        live=m[(field['namespace'],field['key'])]
        assert live['id']==field['id'] and live['type']==field['type'] and live['value']==r[expected]
        assert live['compareDigest']
def db_state():
    return json.loads(sql("SELECT json_build_object('rows',(SELECT json_agg(t) FROM (SELECT * FROM carnegie_catalog WHERE dw_sku BETWEEN 'DWAG-379296' AND 'DWAG-379321' ORDER BY dw_sku) t),'count',(SELECT count(*) FROM carnegie_catalog),'distinct',(SELECT count(DISTINCT mfr_sku) FROM carnegie_catalog),'other_digest',(SELECT md5(string_agg(id::text||':'||dw_sku||':'||mfr_sku,'|' ORDER BY id)) FROM carnegie_catalog WHERE dw_sku NOT BETWEEN 'DWAG-379296' AND 'DWAG-379321'));"))
def preflight():
    assert not (RUN/'preflight.json').exists(), 'Do not overwrite durable prestate'
    c=canonical(); db=db_state()
    assert db['count']==db['distinct']==5928
    local={r['dw_sku']:r for r in db['rows']}
    assert len(local)==26
    for r in PLAN['rows']:
        assert local[r['dw_sku']]['id']==r['catalog_id'] and local[r['dw_sku']]['mfr_sku']==r['before']
    products=fetch_products(list(PRODUCTS)+list(EXCLUDED))
    for pid in PRODUCTS:assert_product(pid,products[pid],'before')
    for pid in EXCLUDED:
        assert products[pid]['status']=='ARCHIVED'
        assert not any(k in mfmap(products[pid]) for k in [('custom','manufacturer_sku'),('dwc','manufacturer_sku')])
    for r in PLAN['rows']:
        d=collect.gql('query Identity($q:String!){productVariants(first:100,query:$q){pageInfo{hasNextPage} nodes{sku product{id status}}}}',{'q':'sku:'+r['dw_sku']})['productVariants']
        assert not d['pageInfo']['hasNextPage']
        exact={v['product']['id'] for v in d['nodes'] if v['sku']==r['dw_sku']}
        assert exact=={p['id'] for p in r['products']}|EXCLUDED
        time.sleep(.25)
    save('preflight.json',dict(at=now(),canonical=c,database=db,products=products,approval='Steve: ungate and run',plan_sha256=hashlib.sha256((BASE/'verification/repair-plan.json').read_bytes()).hexdigest()))
    log('preflight_pass',rows=26,products=52,metafields=104);print('Preflight PASS: exact26/52/104; grouped generation excluded',flush=True)
def db_script(reverse=False,end='ROLLBACK'):
    # IDs and values come only from the approved immutable plan; quote literals.
    quote=lambda x:"'"+str(x).replace("'","''")+"'"
    values=','.join(f"({r['catalog_id']},{quote(r['dw_sku'])},{quote(r['after'] if reverse else r['before'])},{quote(r['before'] if reverse else r['after'])})" for r in PLAN['rows'])
    return f'''BEGIN;
LOCK TABLE carnegie_catalog IN SHARE ROW EXCLUSIVE MODE;
CREATE TEMP TABLE repair_expected(id int,sku text,before text,after text) ON COMMIT DROP;
INSERT INTO repair_expected VALUES {values};
DO $$ BEGIN
IF (SELECT count(*) FROM carnegie_catalog c JOIN repair_expected e ON c.id=e.id AND c.dw_sku=e.sku AND c.mfr_sku=e.before)<>26 THEN RAISE EXCEPTION 'prestate mismatch'; END IF;
END $$;
UPDATE carnegie_catalog c SET mfr_sku=e.after FROM repair_expected e WHERE c.id=e.id AND c.dw_sku=e.sku AND c.mfr_sku=e.before;
DO $$ BEGIN
IF (SELECT count(*) FROM carnegie_catalog c JOIN repair_expected e ON c.id=e.id AND c.dw_sku=e.sku AND c.mfr_sku=e.after)<>26 THEN RAISE EXCEPTION 'poststate mismatch'; END IF;
IF (SELECT count(*)=count(DISTINCT mfr_sku) FROM carnegie_catalog) IS NOT TRUE THEN RAISE EXCEPTION 'identity collision'; END IF;
END $$;
SELECT 'verified26';
{end};'''
def apply_db(rehearse=False):
    baseline=read('preflight.json')['database']
    assert db_state()==baseline, 'Database drift after preflight'
    canonical()
    script=db_script(end='ROLLBACK' if rehearse else 'COMMIT')
    # Rollback rehearsal includes the actual reverse update within same transaction.
    if rehearse:
        script=script.replace("SELECT 'verified26';", "UPDATE carnegie_catalog c SET mfr_sku=e.before FROM repair_expected e WHERE c.id=e.id AND c.dw_sku=e.sku AND c.mfr_sku=e.after;\nSELECT 'verified26-and-reversed';")
    (RUN/('database-rehearsal.sql' if rehearse else 'database-apply.sql')).write_text(script)
    out=sql(script,write=True)
    state=db_state();save('database-rehearsal-after.json' if rehearse else 'database-after.json',state)
    if rehearse:assert state==baseline
    else:
        assert state['count']==state['distinct']==5928 and state['other_digest']==baseline['other_digest']
        assert {r['mfr_sku'] for r in state['rows']}=={r['after'] for r in PLAN['rows']}
    (RUN/'database-rollback.sql').write_text(db_script(reverse=True,end='COMMIT'))
    log('database_rehearsal_pass' if rehearse else 'database_apply_pass',result=out);print(out,flush=True)
def mutate(fields,expect_rejection=False):
    assert len(fields)==2 and {f['namespace'] for f in fields}=={'custom','dwc'}
    assert all(f['ownerId'] in PRODUCTS and f['key']=='manufacturer_sku' and f['compareDigest'] for f in fields)
    assert len({f['ownerId'] for f in fields})==1
    assert all(f['value'] in {PRODUCTS[f['ownerId']][0]['before'],PRODUCTS[f['ownerId']][0]['after']} for f in fields)
    req=urllib.request.Request('https://designer-laboratory-sandbox.myshopify.com/admin/api/2026-07/graphql.json',data=json.dumps({'query':SET,'variables':{'m':fields}}).encode(),headers={'X-Shopify-Access-Token':collect.TOKEN,'Content-Type':'application/json'})
    with urllib.request.urlopen(req,timeout=35) as response:result=json.load(response)
    assert not result.get('errors'),result.get('errors')
    payload=result['data']['metafieldsSet']
    if expect_rejection:
        save('stale-digest-response.json',payload)
        assert payload['userErrors'] and not payload.get('metafields'), 'Stale digest was not rejected'
        return payload
    assert not payload['userErrors'],payload['userErrors']
    return payload
def repair_product(pid,reverse=False):
    baseline=product_baseline()[pid]
    r,plan=PRODUCTS[pid];fresh=fetch_products([pid])[pid]
    assert stable(fresh)==stable(baseline),'Unrelated product drift'
    m=mfmap(fresh)
    source,target=('after','before') if reverse else ('before','after')
    if all(m[(ns,'manufacturer_sku')]['value']==r[target] for ns in ['custom','dwc']):
        assert_product(pid,fresh,target);log('idempotent_skip',id=pid,reverse=reverse);return
    assert_product(pid,fresh,source)
    fields=[dict(ownerId=pid,namespace=ns,key='manufacturer_sku',type=m[(ns,'manufacturer_sku')]['type'],value=r[target],compareDigest=m[(ns,'manufacturer_sku')]['compareDigest']) for ns in ['custom','dwc']]
    log('mutation_intent',id=pid,fields=fields)
    result=mutate(fields);log('mutation_ack',id=pid,result=result)
    after=fetch_products([pid])[pid];assert_product(pid,after,target)
    assert stable(after)==stable(baseline)
    log('product_verified',id=pid,reverse=reverse);print('Verified '+pid,flush=True)
def shopify(phase):
    assert (RUN/'database-after.json').exists(), 'Database must be repaired first'
    for name in ['rollout.mjs','rollout2.mjs','build-siltech-grain-v2-archived.mjs']:
        assert 'const cleanMfr = s => String(s); // Preserve category-bearing manufacturer identity (TK-11246).' in (BASE.parent/'carnegie-split'/name).read_text()
    canary=next(pid for pid,(_,p) in PRODUCTS.items() if p['status']=='ACTIVE')
    if phase in ['canary','rollback-canary']:repair_product(canary,reverse=phase=='rollback-canary')
    elif phase=='stale-digest-test':
        r,p=PRODUCTS[canary];before=read('preflight.json')['products'][canary]
        m=mfmap(before)
        fields=[dict(ownerId=canary,namespace=ns,key='manufacturer_sku',type=m[(ns,'manufacturer_sku')]['type'],value=r['before'],compareDigest=m[(ns,'manufacturer_sku')]['compareDigest']) for ns in ['custom','dwc']]
        current=fetch_products([canary])[canary];assert_product(canary,current,'after')
        result=mutate(fields,expect_rejection=True)
        assert fetch_products([canary])[canary]==current
        save('stale-digest-test.json',dict(verdict='PASS',response=result,product_unchanged=True));log('stale_digest_rejected',id=canary)
    else:
        assert read('independent-canary.json')['verdict']=='PASS','Independent canary must pass'
        for pid in PRODUCTS:
            repair_product(pid);time.sleep(.7)
def main():
    p=argparse.ArgumentParser();p.add_argument('phase',choices=['preflight','inspect','rehearse','database','canary','rollback-canary','stale-digest-test','remaining']);a=p.parse_args()
    if (RUN/'completion.json').exists() and a.phase!='inspect':
        raise SystemExit('This approved repair is completed. Use verify_live.py for read-only checks; do not repeat completed writes.')
    if a.phase!='preflight':
        assert read('preflight.json')['plan_sha256']==hashlib.sha256((BASE/'verification/repair-plan.json').read_bytes()).hexdigest(), 'Approved plan drift'
    if a.phase=='preflight':preflight()
    elif a.phase=='inspect':
        baseline=read('preflight.json')['products'];fresh=fetch_products(list(PRODUCTS)+list(EXCLUDED));save('drift-current.json',fresh)
        for pid in PRODUCTS:
            before,after=stable(baseline[pid]),stable(fresh[pid])
            keys=[k for k in before if before[k]!=after[k]]
            if keys:
                print(json.dumps(dict(id=pid,changed={k:dict(before=before[k],after=after[k]) for k in keys})),flush=True)
    elif a.phase in ['rehearse','database']:apply_db(a.phase=='rehearse')
    else:shopify(a.phase)
if __name__=='__main__':main()