[object Object]

← back to Wallco Ai

batch-regen: 2-design concurrency pool (DTD-C), deterministic-stdout id removes race so gen lock now opt-in; fix title col (no title in all_designs)

89e036bf92cd8acc2c9a09b8817587d7ede75d59 · 2026-06-01 12:40:15 -0700 · Steve Abrams

Files touched

Diff

commit 89e036bf92cd8acc2c9a09b8817587d7ede75d59
Author: Steve Abrams <steve@designerwallcoverings.com>
Date:   Mon Jun 1 12:40:15 2026 -0700

    batch-regen: 2-design concurrency pool (DTD-C), deterministic-stdout id removes race so gen lock now opt-in; fix title col (no title in all_designs)
---
 scripts/batch-regen-settlement.js | 133 ++++++++++++++++++++++++--------------
 scripts/regen-4gate-settlement.sh |  14 ++--
 2 files changed, 94 insertions(+), 53 deletions(-)

diff --git a/scripts/batch-regen-settlement.js b/scripts/batch-regen-settlement.js
index d88d3c4..9c265b0 100644
--- a/scripts/batch-regen-settlement.js
+++ b/scripts/batch-regen-settlement.js
@@ -30,11 +30,19 @@
  *   node scripts/batch-regen-settlement.js               # full worklist
  *   node scripts/batch-regen-settlement.js --limit 3     # smoke test first N
  *   node scripts/batch-regen-settlement.js --dry         # plan only, no gen
+ *
+ * Concurrency (DTD verdict C, 2026-06-01): runs CONCURRENCY designs in parallel
+ * (default 2). The id-grab race is already eliminated by reading the new id from
+ * generate_designs.js's deterministic stdout 'IDs:' line (NOT MAX(id)), so the
+ * gen lock is unnecessary for correctness — the worker no longer takes it. The
+ * 2-design cap is a pure concurrency budget: it cuts wall-clock ~2x vs serial
+ * while leaving Replicate/CPU headroom for the standing curator generator to
+ * interleave on its 30-min tick.
  */
 'use strict';
 const fs = require('fs');
 const path = require('path');
-const { spawnSync, execSync } = require('child_process');
+const { spawn, execSync } = require('child_process');
 
 const ROOT = path.join(__dirname, '..');
 const WORKLIST = process.env.WORKLIST || '/tmp/regen_worklist.json';
@@ -44,6 +52,7 @@ const WORKER = path.join(ROOT, 'scripts', 'regen-4gate-settlement.sh');
 const PORT = process.env.PORT || '9905';
 const MAX_ROLLS = parseInt(process.env.MAX_ROLLS || '20', 10);
 const COST_PER_ROLL = parseFloat(process.env.COST_PER_ROLL || '0.006'); // Replicate SDXL est.
+const CONCURRENCY = parseInt(process.env.CONCURRENCY || '2', 10); // DTD-C: 2 designs in flight
 
 const args = process.argv.slice(2);
 const LIMIT = args.includes('--limit') ? parseInt(args[args.indexOf('--limit') + 1], 10) : Infinity;
@@ -105,68 +114,94 @@ function promoteStaged(srcId, healId) {
   return j;
 }
 
-function main() {
+// Run ONE design through the worker (async). Resolves with its ledger record.
+function runDesign(item, idx) {
+  return new Promise((resolve) => {
+    const srcId = String(item.id);
+    const cq = conceptQuestion(item.category, item.prompt);
+    log(`[${idx}] START #${srcId} cat=${item.category} verdict=${item.verdict} — regen (cap ${MAX_ROLLS})`);
+    const T0 = Date.now();
+    const child = spawn('bash', [WORKER, srcId], {
+      cwd: ROOT,
+      env: { ...process.env, SRC_ID: srcId, CATEGORY: item.category, MAX_ROLLS: String(MAX_ROLLS), CONCEPT_Q: cq, GEN_BACKEND: process.env.GEN_BACKEND || 'replicate' },
+      stdio: ['ignore', 'inherit', 'inherit'],
+    });
+    const killT = setTimeout(() => { try { child.kill('SIGTERM'); } catch (_) {} }, 1000 * 60 * 120); // 120 min ceiling/design
+    child.on('exit', (code, signal) => {
+      clearTimeout(killT);
+      const dtMin = ((Date.now() - T0) / 60000).toFixed(1);
+      let rolls = 0;
+      try { rolls = parseInt(fs.readFileSync(`/tmp/regen-4gate-${srcId}.rolls`, 'utf8').trim(), 10) || 0; } catch (_) {}
+      let outcome, healId = null, reason = '';
+      const winnerFile = `/tmp/regen-4gate-${srcId}.winner`;
+      if (code === 0 && fs.existsSync(winnerFile)) {
+        healId = parseInt(fs.readFileSync(winnerFile, 'utf8').trim(), 10);
+        try {
+          promoteStaged(srcId, healId);
+          outcome = 'winner';
+          reason = `staged heal #${healId} (is_published=FALSE), original user_removed`;
+        } catch (e) { outcome = 'error'; reason = 'promote failed: ' + e.message; }
+      } else if (code === 2) {
+        outcome = 'abandoned';
+        reason = `hit ${MAX_ROLLS}-roll cap without settlement-OK winner (likely legally/structurally unwinnable)`;
+      } else {
+        outcome = 'error';
+        reason = `worker exit ${code}${signal ? ' signal=' + signal : ''}`;
+      }
+      resolve({ srcId, item, outcome, healId, rolls, dtMin: Number(dtMin), reason });
+    });
+  });
+}
+
+async function main() {
   if (!fs.existsSync(WORKLIST)) { console.error('worklist not found: ' + WORKLIST); process.exit(1); }
   const work = JSON.parse(fs.readFileSync(WORKLIST, 'utf8'));
   const done = alreadyDone();
 
-  let totalRolls = 0, winners = 0, abandoned = 0, errors = 0, processed = 0;
-  log(`=== batch-regen-settlement START · ${work.length} in worklist · MAX_ROLLS=${MAX_ROLLS} · cost/roll≈$${COST_PER_ROLL} · ${done.size} already done ===`);
-
+  // build the queue (skip non-integer ids + already-terminal designs + --limit)
+  const queue = [];
   for (const item of work) {
-    if (processed >= LIMIT) { log(`--limit ${LIMIT} reached, stopping`); break; }
     const srcId = String(item.id);
     if (!/^[0-9]+$/.test(srcId)) { log(`SKIP non-integer id "${srcId}"`); continue; }
-    if (done.has(srcId)) { log(`skip #${srcId} (already terminal in ledger)`); continue; }
-
-    const cq = conceptQuestion(item.category, item.prompt);
-    processed++;
-    log(`[${processed}] #${srcId} cat=${item.category} verdict=${item.verdict} — regen (cap ${MAX_ROLLS})`);
+    if (done.has(srcId)) continue;
+    queue.push(item);
+    if (queue.length >= LIMIT) break;
+  }
 
-    if (DRY) { log(`  DRY: would run worker with concept="${cq.slice(0, 60)}…"`); continue; }
+  let totalRolls = 0, winners = 0, abandoned = 0, errors = 0, processed = 0;
+  log(`=== batch-regen-settlement START · ${work.length} in worklist · ${queue.length} to process · ${done.size} already done · CONCURRENCY=${CONCURRENCY} · MAX_ROLLS=${MAX_ROLLS} · cost/roll≈$${COST_PER_ROLL} ===`);
 
-    const T0 = Date.now();
-    const r = spawnSync('bash', [WORKER, srcId], {
-      cwd: ROOT,
-      env: { ...process.env, SRC_ID: srcId, CATEGORY: item.category, MAX_ROLLS: String(MAX_ROLLS), CONCEPT_Q: cq, GEN_BACKEND: process.env.GEN_BACKEND || 'replicate' },
-      encoding: 'utf8',
-      stdio: ['ignore', 'inherit', 'inherit'],
-      timeout: 1000 * 60 * 90, // 90 min hard ceiling per design
-    });
-    const dtMin = ((Date.now() - T0) / 60000).toFixed(1);
+  if (DRY) {
+    queue.forEach((item, i) => log(`  DRY [${i + 1}] #${item.id} cat=${item.category} concept="${conceptQuestion(item.category, item.prompt).slice(0, 60)}…"`));
+    log(`=== DRY DONE · ${queue.length} planned ===`);
+    return;
+  }
 
-    // rolls actually spent (worker writes per-design roll count)
-    let rolls = 0;
-    try { rolls = parseInt(fs.readFileSync(`/tmp/regen-4gate-${srcId}.rolls`, 'utf8').trim(), 10) || 0; } catch (_) {}
-    totalRolls += rolls;
+  let cursor = 0;
+  function finish(rec) {
+    processed++;
+    totalRolls += rec.rolls;
+    if (rec.outcome === 'winner') winners++;
+    else if (rec.outcome === 'abandoned') abandoned++;
+    else errors++;
     const estCost = +(totalRolls * COST_PER_ROLL).toFixed(3);
-
-    let outcome, healId = null, reason = '';
-    const winnerFile = `/tmp/regen-4gate-${srcId}.winner`;
-    if (r.status === 0 && fs.existsSync(winnerFile)) {
-      healId = parseInt(fs.readFileSync(winnerFile, 'utf8').trim(), 10);
-      try {
-        promoteStaged(srcId, healId);
-        outcome = 'winner'; winners++;
-        reason = `staged heal #${healId} (is_published=FALSE), original user_removed`;
-      } catch (e) {
-        outcome = 'error'; errors++; reason = 'promote failed: ' + e.message;
-      }
-    } else if (r.status === 2) {
-      outcome = 'abandoned'; abandoned++;
-      reason = `hit ${MAX_ROLLS}-roll cap without settlement-OK winner (likely legally/structurally unwinnable)`;
-    } else {
-      outcome = 'error'; errors++;
-      reason = `worker exit ${r.status}${r.signal ? ' signal=' + r.signal : ''}`;
-    }
-
     ledgerAppend({
-      ts: new Date().toISOString(), src_id: Number(srcId), category: item.category,
-      original_verdict: item.verdict, outcome, heal_id: healId, rolls, dt_min: Number(dtMin),
-      cumulative_rolls: totalRolls, est_cost_usd: estCost, reason,
+      ts: new Date().toISOString(), src_id: Number(rec.srcId), category: rec.item.category,
+      original_verdict: rec.item.verdict, outcome: rec.outcome, heal_id: rec.healId,
+      rolls: rec.rolls, dt_min: rec.dtMin, cumulative_rolls: totalRolls, est_cost_usd: estCost, reason: rec.reason,
     });
-    log(`  → ${outcome.toUpperCase()} #${srcId}${healId ? ' heal=' + healId : ''} · rolls=${rolls} · ${dtMin}min · cum_rolls=${totalRolls} · est_cost≈$${estCost} · ${reason}`);
+    log(`  → ${rec.outcome.toUpperCase()} #${rec.srcId}${rec.healId ? ' heal=' + rec.healId : ''} · rolls=${rec.rolls} · ${rec.dtMin}min · [${processed}/${queue.length}] cum_rolls=${totalRolls} · est_cost≈$${estCost} · ${rec.reason}`);
+  }
+
+  // simple N-slot pool
+  async function worker() {
+    while (cursor < queue.length) {
+      const myIdx = cursor++;
+      const rec = await runDesign(queue[myIdx], myIdx + 1);
+      finish(rec);
+    }
   }
+  await Promise.all(Array.from({ length: Math.min(CONCURRENCY, queue.length) }, () => worker()));
 
   const estCost = +(totalRolls * COST_PER_ROLL).toFixed(3);
   log(`=== DONE · processed=${processed} winners=${winners} abandoned=${abandoned} errors=${errors} · total_rolls=${totalRolls} · est_cost≈$${estCost} ===`);
diff --git a/scripts/regen-4gate-settlement.sh b/scripts/regen-4gate-settlement.sh
index ca5e484..23fbf8b 100755
--- a/scripts/regen-4gate-settlement.sh
+++ b/scripts/regen-4gate-settlement.sh
@@ -72,7 +72,9 @@ prompt=$(psql -d dw_unified -tAc "SELECT prompt FROM all_designs WHERE id=$SRC_I
 if [ -z "$CATEGORY" ]; then
   CATEGORY=$(psql -d dw_unified -tAc "SELECT category FROM all_designs WHERE id=$SRC_ID")
 fi
-title=$(psql -d dw_unified -tAc "SELECT COALESCE(title, category || ' #' || id) FROM all_designs WHERE id=$SRC_ID")
+# all_designs (Mac2) has no `title` column — it's snapshot-derived. Build a
+# display title from category + id for the settlement check (cosmetic only).
+title="${CATEGORY:-design} #${SRC_ID}"
 
 # ── shared gen mutex (BLOCKING acquire, stale-PID recovery) ──────────────────
 # Mirrors scripts/with-lock.sh's mkdir mutex but BLOCKS (retries) instead of
@@ -105,10 +107,14 @@ echo "[concept] $CONCEPT_Q" | tee -a "$LOG"
 for n in $(seq 1 "$MAX_ROLLS"); do
   echo "$n" > "$ROLLS_FILE"
   T0=$(date +%s)
-  # ── locked gen call ──
-  gen_lock_acquire
+  # gen call. The id-grab race is eliminated by reading newid from the
+  # deterministic stdout "IDs:" line below (NOT MAX(id)), so no lock is needed
+  # for correctness. The batch driver caps concurrency to 2 designs; set
+  # GEN_LOCK=1 to additionally serialize gens against the standing curator's
+  # "wallco-gen" mutex (off by default — Replicate handles concurrency).
+  if [ "${GEN_LOCK:-0}" = "1" ]; then gen_lock_acquire; fi
   genout=$(node scripts/generate_designs.js --n 1 --kind seamless_tile --category "$CATEGORY" --prompts "$prompt" 2>&1)
-  gen_lock_release
+  if [ "${GEN_LOCK:-0}" = "1" ]; then gen_lock_release; fi
   dt=$(( $(date +%s) - T0 ))
   # newid from the deterministic stdout line: "Created X/N designs · IDs: a,b"
   newid=$(printf '%s\n' "$genout" | grep -oE 'IDs: [0-9,]+' | tail -1 | grep -oE '[0-9]+' | tail -1)

← d7523ba chore(deploy): exclude data/prompt-match-ratings.jsonl from  ·  back to Wallco Ai  ·  migration: add user_prompt_match/verdict/rated_at to all_des 25e596a →