← back to Dw Five Field Step0
Track field repairs by product and keep hourly drain resumable
7891e2a31c3b3f076044036cce6614c8f8e08026 · 2026-09-09 10:20:14 -0700 · Steve Abrams
Files touched
M .gitignoreM bulk-fivefield-exec.pyM drain.shA test_executor_progress.pyA test_worklist_progress.pyA verification/e2e-proof-before-TK11314.jsonM verification/e2e-proof.jsonA worklist_progress.py
Diff
commit 7891e2a31c3b3f076044036cce6614c8f8e08026
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Wed Sep 9 10:20:14 2026 -0700
Track field repairs by product and keep hourly drain resumable
---
.gitignore | 1 +
bulk-fivefield-exec.py | 29 ++++++---
drain.sh | 27 ++++----
test_executor_progress.py | 55 ++++++++++++++++
test_worklist_progress.py | 36 +++++++++++
verification/e2e-proof-before-TK11314.json | 17 +++++
verification/e2e-proof.json | 100 +++++++++++++++++++++++++----
worklist_progress.py | 63 ++++++++++++++++++
8 files changed, 293 insertions(+), 35 deletions(-)
diff --git a/.gitignore b/.gitignore
index 87910ab..76e41af 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,4 @@ dist/
build/
.next/
__pycache__/
+verification-live-worklist.json
diff --git a/bulk-fivefield-exec.py b/bulk-fivefield-exec.py
index 9276c1c..ffe92c6 100644
--- a/bulk-fivefield-exec.py
+++ b/bulk-fivefield-exec.py
@@ -33,6 +33,7 @@ DOES NOT mutate the worklist table. Audit JSON is the resume ledger.
"""
import json, urllib.request, urllib.error, subprocess, time, os, sys, argparse
from sku_guard import validate_dw_sku
+from worklist_progress import split_work, summarize
# Sentinel for Shopify's DAILY variant-creation 429. Inherits BaseException (NOT Exception) so the
@@ -453,15 +454,22 @@ def main():
# --- resume from audit file ---
results = []
- done = set()
if os.path.exists(RESULT):
try:
- results = json.load(open(RESULT))
- done = {r["dw_sku"] for r in results if r.get("status") in ("fixed", "skipped")}
- except Exception:
- results = []
- todo = [w for w in work if w["dw_sku"] not in done]
- print(f"=== bulk five-field: {total_work} pending, {len(done)} already done (audit), {len(todo)} remaining ===")
+ with open(RESULT) as audit_handle:
+ results = json.load(audit_handle)
+ if not isinstance(results, list):
+ raise ValueError('Repair audit must be a list')
+ except (OSError, ValueError) as error:
+ raise RuntimeError('Cannot read repair audit; refusing to replay completed work') from error
+ work, todo = split_work(work, results)
+ total_work = len(work)
+ print(f"=== bulk five-field: {total_work} product/repair tasks, {total_work-len(todo)} already done (audit), {len(todo)} remaining ===")
+ if not todo:
+ print('Variant worklist exhausted: processed=0 creates=0 remaining=0; no budget requested.')
+ if dv_enabled:
+ backfill_display_variant(dv_slot_max, dry_run=args.dry_run)
+ return
# --- budget sizing ---
if args.dry_run:
@@ -508,7 +516,7 @@ def main():
if created and created % 25 == 0:
fixed_so_far = sum(1 for r in results if r["status"] == "fixed")
print(f" … batch progress: {created}/{granted} creates this run · "
- f"{fixed_so_far} total fixed all-time · {total_work - fixed_so_far} remaining")
+ f"{fixed_so_far} total fixed all-time · {summarize(work, results)['remaining']} current tasks remaining")
time.sleep(0.4)
# --- refund unused grant (refund rule) ---
@@ -524,11 +532,12 @@ def main():
skipped = sum(1 for r in results if r["status"] == "skipped")
errored = sum(1 for r in results if r["status"] == "errored")
dryruns = sum(1 for r in results if r["status"] == "dryrun")
- remaining_work = total_work - fixed - skipped
+ progress = summarize(work, results)
+ remaining_work = progress['remaining']
print(f"\n=== RUN DONE ===")
print(f"this run: creates={created} processed={processed}")
print(f"all-time: fixed={fixed} skipped={skipped} errored={errored} dryrun={dryruns}")
- print(f"TOTAL DONE (fixed+skipped) = {fixed + skipped} / {total_work} · TOTAL REMAINING = {remaining_work}")
+ print(f"CURRENT TASKS DONE (fixed+skipped) = {progress['done']} / {progress['total']} · TOTAL REMAINING = {remaining_work}")
print(f"audit: {RESULT}")
if not args.dry_run:
json.dump(results, open(RESULT, "w"), indent=1)
diff --git a/drain.sh b/drain.sh
index 8d4d675..f606fbd 100644
--- a/drain.sh
+++ b/drain.sh
@@ -2,7 +2,7 @@
# Daily auto-drain of the 5-field auto-fixable worklist.
# Budget-aware (claims the day's leftover Shopify variant headroom via budget.cjs),
# idempotent + resumable (skips anything already in the result JSON), DELETE-nothing.
-# Self-disables when the worklist is fully drained. Singleton-guarded.
+# Keep polling for new work; the separate display pass shares this job. Singleton-guarded.
set -u
export PATH="/opt/homebrew/opt/postgresql@14/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"
SK="$HOME/Projects/dw-five-field-step0"
@@ -23,17 +23,16 @@ if ! mkdir "$LOCKDIR" 2>/dev/null; then
fi
trap 'rmdir "$LOCKDIR" 2>/dev/null' EXIT
-# executable total (worklist minus the held-no-cost reprice-zeros, minus
-# internal-only Schumacher — matches the executor's load_worklist() exclusion so
-# done/total stays consistent; Schumacher stays archived and off the storefront)
-TOTAL=$(psql -d dw_unified -tAc "select count(*) from bulk_fivefield_worklist where status='pending' and fix_type <> 'reprice-zero' and lower(coalesce(vendor,'')) <> 'schumacher'" 2>/dev/null || echo 0)
-DONE=$(node -e 'try{const r=require(process.argv[1]);console.log(r.filter(x=>x.status==="fixed"||x.status==="skipped").length)}catch(e){console.log(0)}' "$RESULT" 2>/dev/null || echo 0)
+# Count current product/repair identities against the same audit as the executor.
+# Never subtract historical event counts from current worklist row counts.
+COUNTS=$(python3 "$SK/worklist_progress.py" --audit "$RESULT" --counts) || {
+ echo "[$(ts)] cannot read current repair progress — refusing an unverified run" >>"$LOG"; exit 1
+}
+read TOTAL DONE REMAINING <<< "$COUNTS"
echo "[$(ts)] drain start — done=$DONE / total=$TOTAL" >>"$LOG"
-if [ "$TOTAL" -gt 0 ] && [ "$DONE" -ge "$TOTAL" ]; then
- echo "[$(ts)] DRAIN COMPLETE ($DONE/$TOTAL) — self-disabling launchd job" >>"$LOG"
- launchctl bootout "gui/$(id -u)/$LABEL" 2>/dev/null
- exit 0
+if [ "$REMAINING" -eq 0 ]; then
+ echo "[$(ts)] variant worklist complete ($DONE/$TOTAL); executor may still run the separately enabled display-variant pass" >>"$LOG"
fi
# run one SMALL per-slot batch so the day's 500 'backlog' budget spreads evenly
@@ -60,5 +59,9 @@ fi
echo "[$(ts)] display_variant backfill = ${DISPLAY_VARIANT_BACKFILL:-0} (slot_max=${DISPLAY_VARIANT_SLOT_MAX:-30})" >>"$LOG"
python3 "$EXE" --max "$SLOT_MAX" >>"$LOG" 2>&1
RC=$?
-DONE2=$(node -e 'try{const r=require(process.argv[1]);console.log(r.filter(x=>x.status==="fixed"||x.status==="skipped").length)}catch(e){console.log(0)}' "$RESULT" 2>/dev/null || echo 0)
-echo "[$(ts)] drain end rc=$RC — now done=$DONE2 / total=$TOTAL (this run +$((DONE2-DONE)))" >>"$LOG"
+COUNTS2=$(python3 "$SK/worklist_progress.py" --audit "$RESULT" --counts) || {
+ echo "[$(ts)] drain end rc=$RC — progress unavailable" >>"$LOG"; exit 1
+}
+read TOTAL2 DONE2 REMAINING2 <<< "$COUNTS2"
+echo "[$(ts)] drain end rc=$RC — now done=$DONE2 / total=$TOTAL2 remaining=$REMAINING2 (this run +$((DONE2-DONE)))" >>"$LOG"
+exit "$RC"
diff --git a/test_executor_progress.py b/test_executor_progress.py
new file mode 100644
index 0000000..6c9a5a5
--- /dev/null
+++ b/test_executor_progress.py
@@ -0,0 +1,55 @@
+"""Exercise the real executor main flow with remote/budget boundaries replaced."""
+import argparse
+import ast
+import json
+import os
+import pathlib
+import sys
+import tempfile
+import time
+import unittest
+from unittest.mock import patch
+from worklist_progress import split_work, summarize
+
+
+class ExecutorFlowTests(unittest.TestCase):
+ def run_main(self, work, audit, display=False):
+ events = []
+ with tempfile.TemporaryDirectory() as directory:
+ result = os.path.join(directory, 'result.json')
+ with open(result, 'w') as handle:
+ json.dump(audit, handle)
+ def process(row, results, dry_run=False):
+ events.append(('process', row['shopify_id'], row['fix_type'], dry_run))
+ results.append({'product_id': row['shopify_id'].rsplit('/', 1)[-1],
+ 'fix': row['fix_type'], 'status': 'dryrun'})
+ return 0
+ namespace = {'argparse': argparse, 'os': os, 'json': json, 'time': time,
+ 'OUTDIR': directory, 'RESULT': result, 'DOMAIN': 'test-store', 'API': 'test',
+ 'load_worklist': lambda: work, 'split_work': split_work, 'summarize': summarize,
+ 'process': process, 'DailyVariantLimit': type('DailyVariantLimit', (BaseException,), {}),
+ 'backfill_display_variant': lambda cap, dry_run: events.append(('display', cap, dry_run)),
+ 'budget_take': lambda *args: self.fail('dry replay requested a live budget')}
+ module = ast.parse(pathlib.Path('bulk-fivefield-exec.py').read_text())
+ main = next(node for node in module.body if isinstance(node, ast.FunctionDef) and node.name == 'main')
+ exec(compile(ast.Module(body=[main], type_ignores=[]), 'bulk-fivefield-exec.py', 'exec'), namespace)
+ with patch.object(sys, 'argv', ['executor', '--dry-run', '--max', '10']), patch.dict(
+ os.environ, {'DISPLAY_VARIANT_BACKFILL': '1' if display else '0'}):
+ namespace['main']()
+ with open(result) as handle:
+ self.assertEqual(json.load(handle), audit, 'dry replay changed persisted audit')
+ return events
+
+ def test_unrelated_blank_sku_skip_does_not_hide_a_repair(self):
+ events = self.run_main([{'shopify_id': 'gid://shopify/Product/2', 'dw_sku': None, 'fix_type': 'add-sample'}],
+ [{'product_id': '1', 'dw_sku': None, 'fix': 'add-sample', 'status': 'skipped'}])
+ self.assertEqual(events, [('process', 'gid://shopify/Product/2', 'add-sample', True)])
+
+ def test_empty_variant_queue_preserves_separate_display_pass_without_budget(self):
+ events = self.run_main([{'shopify_id': '1', 'fix_type': 'add-sample'}],
+ [{'product_id': '1', 'fix': 'add-sample', 'status': 'fixed'}], display=True)
+ self.assertEqual(events, [('display', 30, True)])
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/test_worklist_progress.py b/test_worklist_progress.py
new file mode 100644
index 0000000..f792b47
--- /dev/null
+++ b/test_worklist_progress.py
@@ -0,0 +1,36 @@
+import unittest
+from worklist_progress import split_work, summarize
+
+
+class ProgressTests(unittest.TestCase):
+ def test_shared_or_blank_sku_does_not_complete_other_products(self):
+ for sku in ('DW-SHARED', None):
+ work = [{'shopify_id': 'gid://shopify/Product/1', 'dw_sku': sku, 'fix_type': 'add-sample'},
+ {'shopify_id': 'gid://shopify/Product/2', 'dw_sku': sku, 'fix_type': 'add-sample'}]
+ audit = [{'product_id': '1', 'dw_sku': sku, 'fix': 'add-sample', 'status': 'fixed'}]
+ self.assertEqual(split_work(work, audit)[1], [work[1]])
+
+ def test_sample_completion_does_not_complete_roll_repair(self):
+ work = [{'shopify_id': '1', 'fix_type': 'add-sample'}, {'shopify_id': '1', 'fix_type': 'build-roll'}]
+ audit = [{'product_id': '1', 'fix': 'add-sample', 'status': 'skipped'}]
+ self.assertEqual(split_work(work, audit)[1], [work[1]])
+
+ def test_duplicates_and_unrelated_audit_do_not_distort_remaining(self):
+ row = {'shopify_id': '1', 'fix_type': 'add-sample'}
+ audit = [{'product_id': '2', 'fix': 'add-sample', 'status': 'fixed'}] * 20
+ self.assertEqual(summarize([row, row], audit), {'source_rows': 2, 'total': 1, 'done': 0, 'remaining': 1})
+
+ def test_errors_and_incomplete_history_cannot_complete_a_task(self):
+ row = {'shopify_id': '1', 'fix_type': 'add-sample'}
+ audit = [{'product_id': '1', 'fix': 'add-sample', 'status': 'errored'},
+ {'dw_sku': 'DW-1', 'fix': 'add-sample', 'status': 'fixed'}]
+ self.assertEqual(split_work([row], audit)[1], [row])
+
+ def test_completed_tasks_are_idempotent(self):
+ row = {'shopify_id': 'gid://shopify/Product/1', 'fix_type': 'add-sample'}
+ audit = [{'product_id': '1', 'fix': 'add-sample', 'status': 'fixed'}] * 2
+ self.assertEqual(summarize([row], audit)['remaining'], 0)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/verification/e2e-proof-before-TK11314.json b/verification/e2e-proof-before-TK11314.json
new file mode 100644
index 0000000..b6234d3
--- /dev/null
+++ b/verification/e2e-proof-before-TK11314.json
@@ -0,0 +1,17 @@
+{
+ "intent": "Block any five-field variant write whose SKU is not derived from a fetched live variant on that product.",
+ "risk_tier": "R1 isolated validation and pre-write integration; no Shopify or DB call",
+ "environment": "local Python unittest and static source boundary assertion",
+ "timestamp": "2026-08-29T06:42:00Z",
+ "ticket": "TK-10956-block-invented-skus-before-five-field-va",
+ "build_identity": "git parent b51daf4 plus owned Cycle 15 diff",
+ "checks": [
+ {"verdict":"PASS","boundary":"identity validation","command":"python3 -m unittest discover -s test -p 'test_*.py'","assertions":"sample and roll targets derive from exactly one fetched live variant by exact transformation; unrelated worklist values, suffix guesses, duplicate/conflicting identities, and malformed inputs fail"},
+ {"verdict":"PASS","boundary":"pre-write ordering","command":"static source assertion","assertions":"validate_dw_sku call and fail-closed return precede Shopify variant POST"},
+ {"verdict":"PASS","boundary":"syntax/diff","command":"python3 -m py_compile sku_guard.py bulk-fivefield-exec.py && git diff --check","assertions":"modules compile and diff is clean"}
+ ],
+ "negative_checks": ["self-authorizing worklist candidate", "unrelated sequential candidate", "blank/literal-null/coercible candidate", "Unicode or control characters", "missing or malformed live variant provenance", "wrong lane suffix", "Roll/Yard/Panel suffix inference", "duplicate or conflicting live identities"],
+ "side_effects": "none; no DB, Shopify, provider, customer-facing, schedule, restart, deploy, or send action",
+ "cleanup": "pycache files are ignored and removed after the test",
+ "verdict": "PASS for the local pre-write boundary"
+}
diff --git a/verification/e2e-proof.json b/verification/e2e-proof.json
index b6234d3..f7dd558 100644
--- a/verification/e2e-proof.json
+++ b/verification/e2e-proof.json
@@ -1,17 +1,91 @@
{
- "intent": "Block any five-field variant write whose SKU is not derived from a fetched live variant on that product.",
- "risk_tier": "R1 isolated validation and pre-write integration; no Shopify or DB call",
- "environment": "local Python unittest and static source boundary assertion",
- "timestamp": "2026-08-29T06:42:00Z",
- "ticket": "TK-10956-block-invented-skus-before-five-field-va",
- "build_identity": "git parent b51daf4 plus owned Cycle 15 diff",
+ "ticket": "TK-11314",
+ "timestamp": "2026-09-09T17:17:00.070133+00:00",
+ "risk_tier": "R4",
+ "authorization": "Steve: go do it (approved rollout and one-product canary)",
+ "environment": "Running hourly DW jobs on Mac; Shopify designer-laboratory-sandbox live store",
+ "verdict": "PASS for rollout/canary; image-dependent throughput BLOCKED by depleted Gemini prepaid credits",
+ "baseline": {
+ "product_id": "gid://shopify/Product/7867560886323",
+ "sku": "DWDX-220334",
+ "status_before": "DRAFT",
+ "ledger_before": {
+ "date": "2026-09-09",
+ "used": 0,
+ "file_absent": true
+ }
+ },
"checks": [
- {"verdict":"PASS","boundary":"identity validation","command":"python3 -m unittest discover -s test -p 'test_*.py'","assertions":"sample and roll targets derive from exactly one fetched live variant by exact transformation; unrelated worklist values, suffix guesses, duplicate/conflicting identities, and malformed inputs fail"},
- {"verdict":"PASS","boundary":"pre-write ordering","command":"static source assertion","assertions":"validate_dw_sku call and fail-closed return precede Shopify variant POST"},
- {"verdict":"PASS","boundary":"syntax/diff","command":"python3 -m py_compile sku_guard.py bulk-fivefield-exec.py && git diff --check","assertions":"modules compile and diff is clean"}
+ {
+ "check": "Reviewed runtime bytes installed under both existing worker locks",
+ "verdict": "PASS",
+ "files": 19
+ },
+ {
+ "check": "Installed regressions and guards",
+ "verdict": "PASS",
+ "tests": 45,
+ "notes": "34 activation,4 validator,7 repair; existing reintroduction and never-activate guard also passed"
+ },
+ {
+ "check": "Canary DRAFT to ACTIVE with correct publication",
+ "verdict": "PASS",
+ "product_id": "gid://shopify/Product/7867560886323",
+ "storefront": {
+ "url": "https://www.designerwallcoverings.com/products/cork-dove-dwdx-220334",
+ "status": 200,
+ "product_id_found": true
+ },
+ "googlePublished": false
+ },
+ {
+ "check": "Price/variant preservation and activation ledger",
+ "verdict": "PASS",
+ "notes": "Independent fresh query matches all before variants; ledger increased exactly0to1"
+ },
+ {
+ "check": "Independent verifier",
+ "verdict": "PASS",
+ "implementation": "Separate read-only Shopify2026-07 query + public storefrontHTTP, no activation-module imports",
+ "evidence": "/Users/macstudio3/Projects/dw-activation-debug-TK11314/verification/rollout/canary-verified.json"
+ },
+ {
+ "check": "Existing repair worker read-only operational path",
+ "verdict": "PASS",
+ "notes": "--max1 --dry-run reads20034 tasks,17445 completed,2589 unmatched; first blankDW SKU safely skipped,no Shopify write or budget debit"
+ },
+ {
+ "check": "Vision failure classification",
+ "verdict": "PASS",
+ "provider_result": {
+ "verdict": "HELD",
+ "tier": "vision",
+ "reason": "vision-http-429",
+ "diagnostics": {
+ "httpStatus": 429,
+ "status": "RESOURCE_EXHAUSTED",
+ "message": "Your prepayment credits are depleted. Please go to AI Studio at https://ai.studio/projects to manage your project and billing. Learn more at https://ai.google.dev/gemini-api/docs/billing#prepay. ",
+ "quota": [],
+ "retryAfterMs": 0
+ },
+ "retryAt": "2026-09-09T18:15:49.630Z",
+ "cost": 0
+ },
+ "notes": "One configured-key request; no product mutation or charge; confirmed depleted prepayment credits"
+ },
+ {
+ "check": "Image-review recovery",
+ "verdict": "BLOCKED",
+ "reason": "Gemini prepaid credits depleted; funding/credential changes outside approved rollout"
+ },
+ {
+ "check": "Source rollback",
+ "verdict": "PASS",
+ "evidence": "/Users/macstudio3/Projects/dw-activation-debug-TK11314/verification/rollback-rehearsal.json"
+ }
],
- "negative_checks": ["self-authorizing worklist candidate", "unrelated sequential candidate", "blank/literal-null/coercible candidate", "Unicode or control characters", "missing or malformed live variant provenance", "wrong lane suffix", "Roll/Yard/Panel suffix inference", "duplicate or conflicting live identities"],
- "side_effects": "none; no DB, Shopify, provider, customer-facing, schedule, restart, deploy, or send action",
- "cleanup": "pycache files are ignored and removed after the test",
- "verdict": "PASS for the local pre-write boundary"
+ "evidence_directory": "/Users/macstudio3/Projects/dw-activation-debug-TK11314/verification/rollout",
+ "rollback": "Exact pre-files in deployment-manifest backups. Source reversal rehearsed; retains one approved live canary product. No product rollback required.",
+ "schedule": "Existing minute25 activations21/hour,500/day; minute10 field repairs. No schedule changes.",
+ "project": "dw-five-field-step0"
}
diff --git a/worklist_progress.py b/worklist_progress.py
new file mode 100644
index 0000000..06cbc98
--- /dev/null
+++ b/worklist_progress.py
@@ -0,0 +1,63 @@
+"""Read-only, product-specific repair progress. Importing this module does no I/O."""
+import argparse
+import json
+import subprocess
+
+
+def task_key(row):
+ product = row.get('shopify_id') or row.get('product_id')
+ fix = row.get('fix_type') or row.get('fix')
+ if not product or not fix:
+ return None # An incomplete historical audit cannot complete another task.
+ return str(product).rsplit('/', 1)[-1], fix
+
+
+def split_work(work, audit):
+ completed = {task_key(row) for row in audit
+ if row.get('status') in ('fixed', 'skipped') and task_key(row)}
+ seen = set()
+ unique = []
+ for row in work:
+ key = task_key(row)
+ if key is None or key not in seen:
+ unique.append(row)
+ if key:
+ seen.add(key)
+ todo = [row for row in unique if task_key(row) not in completed]
+ return unique, todo
+
+
+def summarize(work, audit):
+ unique, todo = split_work(work, audit)
+ return {'source_rows': len(work), 'total': len(unique),
+ 'done': len(unique) - len(todo), 'remaining': len(todo)}
+
+
+def load_current_worklist():
+ query = """SELECT coalesce(json_agg(row_to_json(t)), '[]') FROM (
+ SELECT shopify_id, dw_sku, fix_type FROM bulk_fivefield_worklist
+ WHERE status='pending' AND lower(coalesce(vendor,'')) <> 'schumacher'
+ ) t"""
+ return json.loads(subprocess.check_output(
+ ['psql', '-X', '-d', 'dw_unified', '-At', '-c', query], text=True))
+
+
+def main():
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--audit', required=True)
+ parser.add_argument('--worklist-json')
+ parser.add_argument('--counts', action='store_true')
+ args = parser.parse_args()
+ with open(args.audit) as handle:
+ audit = json.load(handle)
+ if args.worklist_json:
+ with open(args.worklist_json) as handle:
+ work = json.load(handle)
+ else:
+ work = load_current_worklist()
+ result = summarize(work, audit)
+ print(f"{result['total']} {result['done']} {result['remaining']}" if args.counts else json.dumps(result))
+
+
+if __name__ == '__main__':
+ main()
← f2438d1 auto-data-snapshot: 2026-09-09T09:32:15 (1 data files) — out
·
back to Dw Five Field Step0
·
Record unlocked repair schedule and activation follow-throug 78b4bcd →