← back to Dw Repair Debug TK11314
Track repair completion by product and fix type and report current progress
ff27b0977d775a7519a027f427dd52843930ff90 · 2026-09-09 09:57:06 -0700 · Steve Abrams
Files touched
M .gitignoreM bulk-fivefield-exec.pyM drain.shA test_executor_progress.pyA test_worklist_progress.pyM verification/e2e-proof.jsonA worklist_progress.py
Diff
commit ff27b0977d775a7519a027f427dd52843930ff90
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Wed Sep 9 09:57:06 2026 -0700
Track repair completion by product and fix type and report current progress
---
.gitignore | 1 +
bulk-fivefield-exec.py | 29 ++++++++++++++-------
drain.sh | 27 ++++++++++---------
test_executor_progress.py | 55 +++++++++++++++++++++++++++++++++++++++
test_worklist_progress.py | 36 ++++++++++++++++++++++++++
verification/e2e-proof.json | 56 ++++++++++++++++++++++++++++++----------
worklist_progress.py | 63 +++++++++++++++++++++++++++++++++++++++++++++
7 files changed, 232 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.json b/verification/e2e-proof.json
index b6234d3..7e22abf 100644
--- a/verification/e2e-proof.json
+++ b/verification/e2e-proof.json
@@ -1,17 +1,47 @@
{
- "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-09T16:55:10.185732+00:00",
+ "environment": "Isolated Mac worktrees; live Shopify read-only snapshots; local dw_unified SELECTs",
+ "cleanup": "Retained local evidence and worktrees. No production code, product, credential, billing, schedule, or canonical database changes.",
+ "intent": "Correct product/repair completion identity and truthful remaining counts",
+ "risk_tier": "R1 isolated code + read-only database boundary",
+ "build": {
+ "branch": "fix/tk11314-repair-progress",
+ "base_commit": "f2438d10bcc68bcfa209598e7fc8b7664b82cd5b"
+ },
+ "verdict": "PASS for staged progress logic; no repair mutations/deployment",
+ "baseline": {
+ "worklist_rows": 20034,
+ "old_sku_only_remaining": 0,
+ "correct_product_fix_done": 17445,
+ "correct_product_fix_remaining": 2589,
+ "remaining_blank_dw_sku": 2589,
+ "remaining_nonblank_dw_sku": 0,
+ "remaining_in_current_1860": 41,
+ "currently_missing_sample_among_those_41": 0
+ },
+ "commands": [
+ "python3 -m unittest -v test_worklist_progress.py test_executor_progress.py",
+ "python3 worklist_progress.py --audit <live audit> --worklist-json verification-live-worklist.json",
+ "python3 -m py_compile worklist_progress.py bulk-fivefield-exec.py",
+ "zsh -n drain.sh"
+ ],
"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": "Seven progress/executor-flow tests",
+ "verdict": "PASS",
+ "evidence": "blank/shared SKU identities isolated,repair types separated,duplicate retries stable,unknown/error histories cannot complete task,dry-run audit unchanged,empty queue preserves display pass without budget"
+ },
+ {
+ "check": "Live DB/audit reconciliation",
+ "verdict": "PASS",
+ "evidence": "verification-live-worklist.json and live bulk-fivefield-result.json:20034=17445+2589; all2589 pending have blank DW SKU"
+ },
+ {
+ "check": "Shopify repair write",
+ "verdict": "SKIP",
+ "evidence": "Not run. Blank identifiers remain guarded; current failed drafts need verified source repairs."
+ }
],
- "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"
+ "limitations": "The2589 entries are identity problems, not2589 publish-ready products. None of the41 overlapping current drafts are currently missing their sample."
}
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 Repair Debug TK11314
·
(newest)