← back to Carnegie Reprice
TK-11434: spec-complete Carnegie reprice executor (apply-tk11434.mjs)
12936580a7b149b4378648cdfeb5f962efa9e012 · 2026-09-11 08:35:00 -0700 · Steve Abrams
Successor to reprice-tk11434.mjs (the original, already Steve-fired live). Reads the
join-verified staging table carnegie_reprice_staging_tk11434 (source of truth, NOT the
buggy reprice.mjs REST pull). Adds: FULL-access token preference, DRY-RUN Σ-delta plan,
pre-flight >2%-drift abort, per-variant live-re-read + compare-and-set, batched
productVariantsBulkUpdate with THROTTLED backoff, --limit daily cap, end-of-apply
re-sweep, and a --rollback mode. SHARES the ledger + undo with reprice-tk11434.mjs so
resume + rollback stay unified and idempotent. DRY-RUN by default; --apply is gated.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CFntkQoFivTSkgwHjLY6ky
Files touched
Diff
commit 12936580a7b149b4378648cdfeb5f962efa9e012
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Fri Sep 11 08:35:00 2026 -0700
TK-11434: spec-complete Carnegie reprice executor (apply-tk11434.mjs)
Successor to reprice-tk11434.mjs (the original, already Steve-fired live). Reads the
join-verified staging table carnegie_reprice_staging_tk11434 (source of truth, NOT the
buggy reprice.mjs REST pull). Adds: FULL-access token preference, DRY-RUN Σ-delta plan,
pre-flight >2%-drift abort, per-variant live-re-read + compare-and-set, batched
productVariantsBulkUpdate with THROTTLED backoff, --limit daily cap, end-of-apply
re-sweep, and a --rollback mode. SHARES the ledger + undo with reprice-tk11434.mjs so
resume + rollback stay unified and idempotent. DRY-RUN by default; --apply is gated.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CFntkQoFivTSkgwHjLY6ky
---
apply-tk11434.mjs | 275 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 275 insertions(+)
diff --git a/apply-tk11434.mjs b/apply-tk11434.mjs
new file mode 100644
index 0000000..3034cac
--- /dev/null
+++ b/apply-tk11434.mjs
@@ -0,0 +1,275 @@
+// apply-tk11434.mjs — TK-11434 Carnegie at-cost reprice EXECUTOR (spec-complete).
+// ============================================================================
+// GATED: --apply / --rollback are customer-facing LIVE Shopify money-writes.
+// DRY-RUN by default (touches nothing on Shopify).
+//
+// CONTEXT (read before firing):
+// The live Carnegie v2 product set was CREATED at vendor cost (rollout*.mjs wrote
+// price:r.price with no markup — now fixed, commit ced1de9) and PUBLISHED by the
+// Sep-10 cutover, so 5,914 ACTIVE non-sample variants (99.76% of the line) sell at
+// cost (ratio ~1.0) vs the 1.5x floor. This fixer lifts each to proposed_retail.
+//
+// >>> A SIBLING EXECUTOR (reprice-tk11434.mjs) IS THE ORIGINAL and was Steve-fired
+// >>> live. THIS file is the spec-complete successor: it adds a pre-flight >2%
+// >>> drift-abort, prefers the FULL-access token, prints a Σ-delta plan + an
+// >>> end-of-run re-sweep, and SHARES the same ledger + undo so state and rollback
+// >>> are unified across both. Because it shares the ledger and uses
+// >>> LIVE-RE-READ + COMPARE-AND-SET, running it after (or even alongside) the
+// >>> original is SAFE and idempotent: variants already at proposed_retail are
+// >>> logged NOOP and never re-written.
+//
+// SOURCE OF TRUTH: dw_unified.carnegie_reprice_staging_tk11434 (needs_reprice rows,
+// staged + join-verified 2026-09-10). NEVER re-derive from the buggy reprice.mjs
+// (that pulled REST ?vendor=Carnegie with no status filter -> 11,822 rows + whole-
+// dollar rounding). proposed_retail = round(cost/0.65/0.85, 2) (~1.81x, TK-10686).
+//
+// SAFETY MODEL (mirrors the Steve-approved TK-11403 pattern):
+// * DRY-RUN default: prints count, sample rows, Σ delta (total uplift). Writes NOTHING.
+// * PRE-FLIGHT DRIFT GATE (--apply): live-reads a sample of the plan and classifies
+// each live price as at-staged-cost (repriceable) / at-proposed (already done)
+// / OTHER (real drift). Aborts the WHOLE run before any write if OTHER > 2%.
+// * PER-VARIANT COMPARE-AND-SET: only writes proposed_retail if the live price still
+// equals the staged cost baseline. If live == proposed -> NOOP. If live is
+// anything else -> SKIP (never clobber a value we didn't stage).
+// * BATCHED by product via productVariantsBulkUpdate; THROTTLED backoff; --limit=N
+// daily cap; RESUME-SAFE via the shared ledger (applied vids are skipped).
+// * SAMPLE ($4.25) variants are never in the plan and never touched.
+// * LEDGER: every intent/applied/noop/skip/failed row appended (reversible).
+// * ROLLBACK: node apply-tk11434.mjs --rollback [--apply] (restores prePrice via
+// compare-and-set) — equivalent to the sibling `node undo-tk11434.mjs`.
+//
+// Usage:
+// node apply-tk11434.mjs # DRY-RUN plan (count, samples, Σ delta) — writes nothing
+// node apply-tk11434.mjs --apply # GATED live reprice (Steve-approved only)
+// node apply-tk11434.mjs --apply --limit=1000 # first daily batch (Shopify variant cap)
+// node apply-tk11434.mjs verify # read-only re-sweep: how many still <1.5x
+// node apply-tk11434.mjs --rollback # DRY-RUN rollback plan
+// node apply-tk11434.mjs --rollback --apply # GATED restore of prePrice for applied variants
+import fs from 'node:fs';
+import { execFileSync } from 'node:child_process';
+
+const HOME = process.env.HOME;
+const ENV = `${HOME}/Projects/secrets-manager/.env`;
+const env = k => { const m = fs.readFileSync(ENV,'utf8').split('\n').find(l=>l.startsWith(k+'=')); return m ? m.slice(k.length+1).trim().replace(/^["']|["']$/g,'') : ''; };
+// Prefer the FULL-access token (write_products); fall back to the narrow admin token.
+const TOKEN_KEY = env('SHOPIFY_FULL_ACCESS_TOKEN') ? 'SHOPIFY_FULL_ACCESS_TOKEN' : 'SHOPIFY_ADMIN_TOKEN';
+const TOKEN = env(TOKEN_KEY);
+if (!TOKEN) { console.error('FATAL: no Shopify token in', ENV); process.exit(2); }
+const DOMAIN = 'designer-laboratory-sandbox.myshopify.com'; // the LIVE DW store (legacy "sandbox" name)
+const API = `https://${DOMAIN}/admin/api/2024-10/graphql.json`;
+const sleep = ms => new Promise(r=>setTimeout(r,ms));
+
+const ARGV = process.argv.slice(2);
+const APPLY = ARGV.includes('--apply');
+const VERIFY = ARGV[0] === 'verify';
+const ROLLBACK = ARGV.includes('--rollback');
+const LIMIT = Number((ARGV.find(a=>a.startsWith('--limit='))||'=0').split('=')[1]) || 0;
+const DRIFT_ABORT_PCT = 0.02; // >2% real-drift in the sample aborts the whole run
+const DRIFT_SAMPLE_MAX = 300; // pre-flight live-read sample size
+
+const DIR = new URL('.', import.meta.url).pathname;
+// SHARED ledger with reprice-tk11434.mjs / undo-tk11434.mjs so resume + rollback are unified.
+const LED = `${DIR}reprice-tk11434-ledger.jsonl`;
+const log = o => fs.appendFileSync(LED, JSON.stringify({ ts:new Date().toISOString(), tk:'TK-11434', src:'apply-tk11434', ...o }) + '\n');
+
+async function gql(query, variables={}) {
+ for (let a=0;a<6;a++){
+ const r = await fetch(API,{ method:'POST', headers:{'X-Shopify-Access-Token':TOKEN,'Content-Type':'application/json'},
+ body: JSON.stringify({query,variables}), signal: AbortSignal.timeout(45000) });
+ if (r.status===429 || r.status>=500){ await sleep(1500*(a+1)); continue; }
+ const j = await r.json();
+ if (j.errors && JSON.stringify(j.errors).includes('THROTTLED')){ await sleep(3000*(a+1)); continue; }
+ return j;
+ }
+ throw new Error('gql retries exhausted');
+}
+function psql(sql){
+ const out = execFileSync('psql', ['postgresql:///dw_unified?host=/tmp','-At','-F','|','-c',sql], { encoding:'utf8', maxBuffer:256*1024*1024 });
+ return out.trim() ? out.trim().split('\n').map(r=>r.split('|')) : [];
+}
+const money = n => Number(n).toFixed(2);
+
+// ---- plan from the staged, join-verified table (SOURCE OF TRUTH) ----
+const rows = psql(
+ `SELECT product_id, variant_id, variant_gid, dw_sku, handle, current_live_price, cost, proposed_retail
+ FROM carnegie_reprice_staging_tk11434
+ WHERE needs_reprice
+ ORDER BY product_id, variant_id`
+).map(([product_id,variant_id,variant_gid,dw_sku,handle,old,cost,prop]) => ({
+ product_id, variant_id, variant_gid: variant_gid || `gid://shopify/ProductVariant/${variant_id}`,
+ dw_sku, handle, old:+old, cost:+cost, prop:+prop
+}));
+const sumDelta = rows.reduce((s,r)=>s+(r.prop - r.old), 0);
+console.log(`[TK-11434] token=${TOKEN_KEY} store=${DOMAIN}`);
+console.log(`[plan] ${rows.length} needs_reprice variants | formula round(cost/0.65/0.85,2) (~1.81x)`);
+console.log(`[plan] Σ current(live)=$${money(rows.reduce((s,r)=>s+r.old,0))} Σ proposed=$${money(rows.reduce((s,r)=>s+r.prop,0))} Σ delta(uplift)=$${money(sumDelta)}`);
+console.log('[plan] sample:');
+for (const r of rows.slice(0,5)) console.log(` ${r.dw_sku} ${money(r.old)} -> ${money(r.prop)} (cost ${money(r.cost)}, +$${money(r.prop-r.old)})`);
+
+// ---- ledger state: applied (and not later undone) vids ----
+function ledgerApplied(){
+ const done = new Map(); // vid -> {prePrice,newPrice}
+ if (fs.existsSync(LED)) for (const l of fs.readFileSync(LED,'utf8').split('\n')){
+ if(!l.trim())continue; try{ const r=JSON.parse(l);
+ if (r.phase==='applied') done.set(String(r.vid), { prePrice:r.prePrice, newPrice:r.newPrice });
+ if (r.phase==='undone') done.delete(String(r.vid));
+ }catch{}
+ }
+ return done;
+}
+
+// ---- live price reader (batched by 100) ----
+async function livePrices(vids){
+ const map = {};
+ for (let i=0;i<vids.length;i+=100){
+ const ids = vids.slice(i,i+100).map(v=>`gid://shopify/ProductVariant/${v}`);
+ const j = await gql(`query($ids:[ID!]!){ nodes(ids:$ids){ ... on ProductVariant { id price product{ status } } } }`, { ids });
+ for (const n of (j.data?.nodes||[]).filter(Boolean)) map[n.id.split('/').pop()] = { price:+n.price, status:n.product?.status };
+ await sleep(250);
+ }
+ return map;
+}
+
+// ========================= VERIFY (read-only re-sweep) =========================
+async function reSweep(){
+ const live = await livePrices(rows.map(r=>r.variant_id));
+ let ge15=0, atCost=0, other=0, missing=0;
+ for (const r of rows){
+ const lv = live[r.variant_id];
+ if (!lv){ missing++; continue; }
+ if (r.cost>0 && lv.price/r.cost >= 1.5) ge15++;
+ else if (r.cost>0 && lv.price/r.cost < 1.5) atCost++;
+ else other++;
+ }
+ return { ge15, atCost, other, missing, total:rows.length };
+}
+if (VERIFY){
+ const s = await reSweep();
+ console.log(`[verify] ratio>=1.5x=${s.ge15} still<1.5x(at-cost)=${s.atCost} other=${s.other} missing=${s.missing} of ${s.total}`);
+ process.exit(s.atCost>0 ? 1 : 0);
+}
+
+// ============================= ROLLBACK ========================================
+if (ROLLBACK){
+ const done = ledgerApplied();
+ const targets = [...done.entries()].map(([vid,v])=>({ vid, ...v }));
+ console.log(`[rollback] ${targets.length} applied variant(s) in ledger to restore to prePrice`);
+ if (!targets.length) process.exit(0);
+ let restored=0, skip=0, failed=0;
+ const byV = await livePrices(targets.map(t=>t.vid));
+ const M = `mutation($pid:ID!,$id:ID!,$price:Money!){ productVariantsBulkUpdate(productId:$pid, variants:[{id:$id,price:$price}]){ productVariants{ id price } userErrors{ message } } }`;
+ for (const t of targets){
+ const gvid = `gid://shopify/ProductVariant/${t.vid}`;
+ const j = await gql(`query($id:ID!){ productVariant(id:$id){ id price product{ id } } }`, { id: gvid });
+ const v = j.data?.productVariant;
+ if (!v){ console.log('SKIP not-found', t.vid); skip++; continue; }
+ const cur = +v.price;
+ if (Math.abs(cur - Number(t.prePrice)) < 0.005){ console.log('ALREADY-restored', t.vid, cur); skip++; continue; }
+ if (Math.abs(cur - Number(t.newPrice)) >= 0.005){ console.log('SKIP moved-since', t.vid, 'live',cur,'expected',t.newPrice); log({phase:'undo-skip',vid:t.vid,liveNow:cur,expected:t.newPrice}); skip++; continue; }
+ if (!APPLY){ console.log(`DRY restore ${t.vid} ${money(cur)} -> ${money(t.prePrice)}`); continue; }
+ const r = await gql(M, { pid: v.product.id, id: gvid, price: money(t.prePrice) });
+ const ue = r.data?.productVariantsBulkUpdate?.userErrors || [];
+ if (ue.length || r.errors){ console.log('FAIL', t.vid, JSON.stringify(ue.length?ue:r.errors).slice(0,160)); log({phase:'undo-failed',vid:t.vid,err:ue.length?ue:r.errors}); failed++; await sleep(500); continue; }
+ log({ phase:'undone', vid:t.vid, restoredTo:t.prePrice, fromPrice:cur }); console.log(`OK restore ${t.vid} ${money(cur)} -> ${money(t.prePrice)}`);
+ restored++; await sleep(400);
+ }
+ console.log(`\n== ${APPLY?'RESTORED':'DRY-RUN would-restore'}=${restored} skip=${skip} failed=${failed} ==`);
+ if (!APPLY) console.log('DRY-RUN — nothing written. Add --apply to restore.');
+ process.exit(0);
+}
+
+// ============================= APPLY / DRY-RUN =================================
+const done = ledgerApplied();
+
+// ---- PRE-FLIGHT DRIFT GATE (only when about to write) ----
+if (APPLY){
+ const sampleVids = rows.filter(r=>!done.has(String(r.variant_id))).slice(0, DRIFT_SAMPLE_MAX).map(r=>r.variant_id);
+ if (sampleVids.length){
+ console.log(`[drift-gate] live-reading ${sampleVids.length}-variant sample...`);
+ const live = await livePrices(sampleVids);
+ let atCost=0, atProp=0, other=0, missing=0;
+ const byId = Object.fromEntries(rows.map(r=>[String(r.variant_id), r]));
+ for (const vid of sampleVids){
+ const lv = live[String(vid)]; const r = byId[String(vid)];
+ if (!lv){ missing++; continue; }
+ if (Math.abs(lv.price - r.old) < 0.005) atCost++;
+ else if (Math.abs(lv.price - r.prop) < 0.005) atProp++;
+ else other++;
+ }
+ const checked = sampleVids.length - missing;
+ const driftFrac = checked ? other/checked : 0;
+ console.log(`[drift-gate] at-staged-cost=${atCost} at-proposed(done)=${atProp} OTHER(drift)=${other} missing=${missing} drift=${(driftFrac*100).toFixed(2)}%`);
+ if (driftFrac > DRIFT_ABORT_PCT){
+ console.error(`[drift-gate] ABORT — ${(driftFrac*100).toFixed(2)}% of sample drifted off the staged baseline (> ${(DRIFT_ABORT_PCT*100)}%). Staging is stale; re-stage before writing. Nothing written.`);
+ log({ phase:'aborted', reason:'drift-gate', driftFrac, atCost, atProp, other, missing });
+ process.exit(3);
+ }
+ }
+}
+
+// ---- group plan by product for bulk update ----
+const byProduct = new Map();
+for (const r of rows){ if(!byProduct.has(r.product_id)) byProduct.set(r.product_id,[]); byProduct.get(r.product_id).push(r); }
+
+const M = `mutation($pid:ID!,$vars:[ProductVariantsBulkInput!]!){
+ productVariantsBulkUpdate(productId:$pid, variants:$vars){
+ productVariants{ id sku price } userErrors{ field message } } }`;
+
+let applied=0, noop=0, skip=0, failed=0, variantsSeen=0;
+for (const [pid, plan] of byProduct){
+ if (LIMIT && variantsSeen >= LIMIT) break;
+ const gpid = `gid://shopify/Product/${pid}`;
+ const j = await gql(`query($id:ID!){ product(id:$id){ id status variants(first:100){ nodes{ id sku price } } } }`, { id: gpid });
+ const p = j.data?.product;
+ if (!p){ for(const r of plan){ log({phase:'skip',pid,vid:r.variant_id,sku:r.dw_sku,why:'product not found'}); skip++; } continue; }
+ if (p.status !== 'ACTIVE'){ for(const r of plan){ log({phase:'skip',pid,vid:r.variant_id,sku:r.dw_sku,why:'status '+p.status}); skip++; } continue; }
+ const liveById = Object.fromEntries(p.variants.nodes.map(n=>[n.id.split('/').pop(), n]));
+
+ const vars = [];
+ for (const r of plan){
+ if (LIMIT && variantsSeen >= LIMIT) break;
+ variantsSeen++;
+ const vid = String(r.variant_id);
+ if (done.has(vid)){ skip++; continue; } // resume: already applied in ledger
+ const lv = liveById[vid];
+ if (!lv){ log({phase:'skip',pid,vid,sku:r.dw_sku,why:'variant not on product'}); skip++; continue; }
+ const cur = +lv.price;
+ if (Math.abs(cur - r.prop) < 0.005){ log({phase:'noop',pid,vid,sku:r.dw_sku,price:cur,why:'already at proposed'}); noop++; continue; }
+ if (Math.abs(cur - r.old) >= 0.005){ log({phase:'skip',pid,vid,sku:r.dw_sku,liveNow:cur,stagedOld:r.old,why:'live moved off staged baseline — not clobbering'}); skip++; continue; }
+ vars.push({ id: lv.id, price: money(r.prop), _r:r, _pre:cur });
+ }
+ if (!vars.length) continue;
+
+ for (const v of vars) log({ phase:'intent', pid, vid:v._r.variant_id, sku:v._r.dw_sku, prePrice:v._pre, toPrice:v.price, cost:v._r.cost, undo:'node apply-tk11434.mjs --rollback --apply' });
+
+ if (!APPLY){
+ for (const v of vars) console.log(`DRY ${v._r.dw_sku} ${money(v._pre)} -> ${v.price} (cost ${money(v._r.cost)})`);
+ applied += vars.length; // "would-apply" in dry mode
+ continue;
+ }
+
+ const r = await gql(M, { pid: gpid, vars: vars.map(v=>({ id:v.id, price:v.price })) });
+ const ue = r.data?.productVariantsBulkUpdate?.userErrors || [];
+ if (ue.length || r.errors){
+ for (const v of vars){ log({phase:'failed',pid,vid:v._r.variant_id,sku:v._r.dw_sku,err:ue.length?ue:r.errors}); failed++; }
+ console.log(`FAIL product ${pid}: ${JSON.stringify(ue.length?ue:r.errors).slice(0,200)}`);
+ await sleep(600); continue;
+ }
+ const got = Object.fromEntries((r.data.productVariantsBulkUpdate.productVariants||[]).map(n=>[n.id.split('/').pop(), n.price]));
+ for (const v of vars){
+ const np = got[v._r.variant_id];
+ log({ phase:'applied', pid, vid:v._r.variant_id, sku:v._r.dw_sku, prePrice:v._pre, newPrice:np });
+ console.log(`OK ${v._r.dw_sku} ${money(v._pre)} -> ${np}`);
+ applied++;
+ }
+ await sleep(500);
+}
+console.log(`\n== ${APPLY?'APPLIED':'DRY-RUN would-apply'}=${applied} noop=${noop} skip=${skip} failed=${failed} (variants scanned ${variantsSeen}${LIMIT?`/limit ${LIMIT}`:''}) ==`);
+
+if (APPLY){
+ console.log('[re-sweep] confirming ratio>=1.5x across the full plan...');
+ try { const s = await reSweep(); console.log(`[re-sweep] ratio>=1.5x=${s.ge15}/${s.total} still<1.5x=${s.atCost} other=${s.other} missing=${s.missing}`); }
+ catch(e){ console.log('[re-sweep] skipped:', e.message); }
+} else {
+ console.log('DRY-RUN — nothing written to Shopify. Add --apply (Steve-approved) to write.');
+}
← 086632c auto-data-snapshot: 2026-09-11T08:16:47 (2 data files) — rep
·
back to Carnegie Reprice
·
auto-data-snapshot: 2026-09-11T08:57:34 (2 data files) — rep 17bd730 →