#!/usr/bin/env bash
# bend-emit-pool.sh — bounded parallel emit pool for circuit-emission cells.
#
# Reads work items from $LUMBDA_QUEUE_DIR/*.lsp. For each item <tag>:
#   - skips if <tag>.ready or <tag>.done exists
#   - skips if <tag>.emitting marker exists (in-flight)
#   - throttles to MAX concurrent emits
#   - launches $LUMBDA_EMITTER_CMD <tag>.lsp, expects it to write <tag>.bin
#   - touches <tag>.ready when bin size > MIN_BIN_SIZE
#   - touches <tag>.done if emit fails or bin stays too small
#
# Exits when no .lsp without a marker remains AND no .emitting markers active.
# Touch $LUMBDA_QUEUE_DIR/STOP to halt new launches gracefully.
#
# Liveness:
#   - Writes $LUMBDA_QUEUE_DIR/pool.heartbeat (pid + timestamp) at top of every
#     loop iter via lib-heartbeat.sh::heartbeat_touch.
#   - On every exit path (STOP, drain timeout, signal trap, EXIT trap)
#     writes $LUMBDA_QUEUE_DIR/pool.exit-cause with reason + removes heartbeat.
#   - Supervisor reads pool.heartbeat to detect silent death.
#
# Usage: bend-emit-pool.sh [MAX [QUEUE_DIR [MIN_BIN_SIZE_BYTES]]]
#   MAX default 6
#   QUEUE_DIR default $LUMBDA_QUEUE_DIR
#   MIN_BIN_SIZE_BYTES default 268435456 (256 MB — sanity check only).
#
# Self-identifies as "bend-emit-pool" in cmdline for precise pkill targeting.
#
# Env vars consumed: see $LUMBDA_FACTORY_DIR/CONTRACT.md (LUMBDA_QUEUE_DIR,
# LUMBDA_DOMAIN_DIR, LUMBDA_REPO_DIR, LUMBDA_EMITTER_CMD, AUTO_PULL_MASTER,
# AUTO_PULL_TIMEOUT, DISK_MIN_FREE_GB, MEM_MIN_FREE_GB, POOL_MAX_IDLE_LOOPS,
# SLOW_MAX).
set -u

MAX="${1:-6}"
QUEUE_DIR="${2:-${LUMBDA_QUEUE_DIR:-/tmp/lumbda-queue}}"
MIN_BIN_SIZE="${3:-268435456}"

LUMBDA_DOMAIN_DIR="${LUMBDA_DOMAIN_DIR:-}"
LUMBDA_REPO_DIR="${LUMBDA_REPO_DIR:-$HOME/git/lumbda}"
LUMBDA_FACTORY_DIR="${LUMBDA_FACTORY_DIR:-$(dirname "$(readlink -f "$0")")}"
LUMBDA_EMITTER_CMD="${LUMBDA_EMITTER_CMD:?LUMBDA_EMITTER_CMD must be set by consumer (e.g. \"lumbda --fast \$LUMBDA_DOMAIN_DIR/lumbda/emit-stream.lsp\")}"
LIB_HEARTBEAT="${LIB_HEARTBEAT:-$LUMBDA_FACTORY_DIR/lib-heartbeat.sh}"
LIB_TIER="${LIB_TIER:-$LUMBDA_FACTORY_DIR/lib-tier.sh}"

mkdir -p "$QUEUE_DIR"

echo $$ > "$QUEUE_DIR/pool.pid"
trap 'rm -f "$QUEUE_DIR/pool.pid"' EXIT

# Self-tag in cmdline so pkill -f bend-emit-pool only matches THIS script,
# not other bash sessions that happen to mention factory paths.
exec -a bend-emit-pool bash -c '
QUEUE_DIR="'"$QUEUE_DIR"'"
MAX="'"$MAX"'"
MIN_BIN_SIZE="'"$MIN_BIN_SIZE"'"
LUMBDA_DOMAIN_DIR="'"$LUMBDA_DOMAIN_DIR"'"
LUMBDA_REPO_DIR="'"$LUMBDA_REPO_DIR"'"
LUMBDA_EMITTER_CMD="'"$LUMBDA_EMITTER_CMD"'"
LIB_HEARTBEAT="'"$LIB_HEARTBEAT"'"
LIB_TIER="'"$LIB_TIER"'"

# Source liveness library — provides heartbeat_touch / heartbeat_is_alive /
# heartbeat_exit_cause. Path resolved by parent shell before exec so inner
# shell does not need to derive its own dirname.
# shellcheck source=/dev/null
. "$LIB_HEARTBEAT"
# Source bin-size tier classifier — provides tier_of_size / tier_label_dlq.
# Pool uses these to route oversize bins (above TIER_LARGE_MAX) to DLQ at
# finalize time instead of letting them flow downstream to crush the
# dispatcher fleet.
# shellcheck source=/dev/null
. "$LIB_TIER"

echo "[$(date +%H:%M:%S)] bend-emit-pool start MAX=$MAX queue=$QUEUE_DIR"

# Exit-cause writers. Every exit path (drain, STOP, signal) records its
# reason via the library before falling off the loop. EXIT trap acts as
# safety net for unforeseen crashes; the loop body sets a more specific
# reason just before break + EXIT trap then forwards it.
POOL_EXIT_REASON=""
on_exit() {
    local r="${POOL_EXIT_REASON:-unknown-exit}"
    heartbeat_exit_cause "$QUEUE_DIR" "pool" "$r"
    echo "[$(date +%H:%M:%S)] bend-emit-pool exit reason=$r"
}
trap on_exit EXIT
trap "POOL_EXIT_REASON=signal-int;  exit 130" INT
trap "POOL_EXIT_REASON=signal-term; exit 143" TERM

# Pull origin/master before consuming a cell. Fail-soft: log a warning
# and return non-zero on failure (caller decides whether to keep emitting
# against checkout we have). Cheap (~1s fetch over LAN). Locked by
# .git/index.lock so concurrent pool restarts serialize cleanly.
pull_master_or_log() {
  [ "${AUTO_PULL_MASTER:-1}" = "1" ] || return 0
  local out rc
  out=$(timeout "${AUTO_PULL_TIMEOUT:-30}" git -C "$LUMBDA_REPO_DIR" pull --ff-only origin master 2>&1)
  rc=$?
  if [ "$rc" -ne 0 ]; then
    echo "[$(date +%H:%M:%S)] PULL-FAIL rc=$rc (substrate may be stale): $(echo "$out" | tr "\n" "|" | cut -c-200)"
    return "$rc"
  fi
  if ! echo "$out" | grep -q "Already up to date"; then
    echo "[$(date +%H:%M:%S)] PULL ok: $(echo "$out" | tr "\n" "|" | cut -c-200)"
  fi
  return 0
}

# Self-healing janitor: clear .emitting markers whose emitter PID is dead.
# Called before every slot-count check. Stale markers can accrue when emits
# get SIGKILLed (OOM, manual cleanup) before they can clean up their own
# marker.
#
# Subshell wrapper around emitter touches .ready (or .done) BEFORE removing
# its own .emitting marker. A brief window exists where emitter has exited
# (PID dead) but parent subshell has not finished tail cleanup. If this
# janitor fires in that window, it must NOT treat emit as failed —
# .ready/.done file proves completion. Likewise if a .bin already exists
# at full size, emit was real and .ready marker may already have been
# consumed by dispatcher.
#
# NOTE: comments inside bash -c body must NOT contain apostrophes — each
# closes outer single-quoted argument early + corrupts parse.
heal_stale_markers() {
  for em in "$QUEUE_DIR"/*.emitting; do
    [ ! -f "$em" ] && continue
    pid=$(cat "$em" 2>/dev/null)
    if [ -z "$pid" ] || ! kill -0 "$pid" 2>/dev/null; then
      tag=$(basename "$em" .emitting)
      # If ANY completion artifact already exists, the emit completed
      # before this janitor fired. Silently clear stale .emitting marker
      # — no HEAL log, no .done touch, no relaunch.
      if [ -f "$QUEUE_DIR/${tag}.ready" ] \
         || [ -f "$QUEUE_DIR/${tag}.done" ]; then
        rm -f "$em"
        continue
      fi
      # Orphan-bin recovery: if .bin exists at full size but no .ready/.done,
      # subshell wrote .bin then died (SIGTERM mid-cleanup) before touching
      # .ready. Promote: bin is real, dispatcher should consume it. Truncated
      # bins (< MIN_BIN_SIZE) treated as failure — rm + .done so cell does
      # not re-emit endlessly.
      if [ -f "$QUEUE_DIR/${tag}.bin" ]; then
        sz=$(stat -c %s "$QUEUE_DIR/${tag}.bin" 2>/dev/null || echo 0)
        magic=$(head -c 8 "$QUEUE_DIR/${tag}.bin" 2>/dev/null)
        if [ "$magic" != "QECCOPS1" ]; then
          # Header-rewrite never fired. See finalize() comment for the
          # SIGKILL/OOM mechanism.
          mkdir -p "$QUEUE_DIR/dlq"
          echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
          {
            echo "tier=zero-magic bin_bytes=$sz"
            echo "reason: emit-interrupted (heal-stale path) — bin header magic missing"
            echo "first-8-bytes-hex: $(head -c 8 $QUEUE_DIR/${tag}.bin 2>/dev/null | xxd -p -c 8)"
          } > "$QUEUE_DIR/dlq/${tag}.reason"
          echo "[$(date +%H:%M:%S)] DLQ orphan-bin-zero-magic $tag sz=$sz"
          rm -f "$QUEUE_DIR/${tag}.bin"
          touch "$QUEUE_DIR/${tag}.done.fail"
          touch "$QUEUE_DIR/${tag}.done"
        elif [ "$sz" -ge "$MIN_BIN_SIZE" ]; then
          touch "$QUEUE_DIR/${tag}.ready"
          echo "[$(date +%H:%M:%S)] RECOVER orphan-bin $tag ($(du -h $QUEUE_DIR/${tag}.bin | cut -f1))"
        else
          rm -f "$QUEUE_DIR/${tag}.bin"
          touch "$QUEUE_DIR/${tag}.done"
          echo "[$(date +%H:%M:%S)] HEAL orphan-bin-truncated $tag sz=$sz"
        fi
        rm -f "$em"
        continue
      fi
      echo "[$(date +%H:%M:%S)] HEAL stale $tag (pid=$pid dead)"
      rm -f "$em"
      touch "$QUEUE_DIR/${tag}.done"
    fi
  done
}

# Belt-and-suspenders: standalone scan for orphan .bin files whose .emitting
# marker is already gone (different death path than heal_stale_markers).
# Catches:
#   - pool subshell died AFTER .emitting was racy-cleaned but
#     BEFORE it touched .ready
#   - pool process murdered between unrelated janitor pass + .ready touch
#   - any .bin left behind by a manual operator intervention
#
# Same recovery rule as .emitting branch: full-size → .ready, truncated →
# rm + .done so cell does not re-emit forever. Skips bins already in flight
# (.inflight-N) or already complete (.ready / .done / .dispatch.log).
# Idempotent — safe to run every loop iter alongside heal_stale_markers.
heal_orphan_bins() {
  shopt -s nullglob
  for bin in "$QUEUE_DIR"/*.bin; do
    tag=$(basename "$bin" .bin)
    # Any in-progress or completion artifact = NOT orphan
    [ -f "$QUEUE_DIR/${tag}.ready" ] && continue
    [ -f "$QUEUE_DIR/${tag}.done" ] && continue
    [ -f "$QUEUE_DIR/${tag}.emitting" ] && continue
    [ -f "$QUEUE_DIR/${tag}.dispatch.log" ] && continue
    # 2026-06-14 defect fix: with shopt -s nullglob (set at function top
    # so the outer for loop tolerates empty *.bin), an unmatched
    # .inflight-* glob expands to NOTHING. Bare "ls >/dev/null" then
    # succeeds (lists CWD) + the if-branch incorrectly skips the orphan.
    # Compgen returns success only when pattern matches, immune to nullglob.
    if compgen -G "$QUEUE_DIR/${tag}.inflight-*" >/dev/null; then
      continue
    fi
    sz=$(stat -c %s "$bin" 2>/dev/null || echo 0)
    magic=$(head -c 8 "$bin" 2>/dev/null)
    if [ "$magic" != "QECCOPS1" ]; then
      # Same emit-interrupted class as finalize() + heal_stale_markers.
      mkdir -p "$QUEUE_DIR/dlq"
      echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
      {
        echo "tier=zero-magic bin_bytes=$sz"
        echo "reason: emit-interrupted (orphan-bin-scan path) — bin header magic missing"
        echo "first-8-bytes-hex: $(head -c 8 $bin 2>/dev/null | xxd -p -c 8)"
      } > "$QUEUE_DIR/dlq/${tag}.reason"
      echo "[$(date +%H:%M:%S)] DLQ orphan-bin-scan-zero-magic $tag sz=$sz"
      rm -f "$bin"
      touch "$QUEUE_DIR/${tag}.done.fail"
      touch "$QUEUE_DIR/${tag}.done"
    elif [ "$sz" -ge "$MIN_BIN_SIZE" ]; then
      touch "$QUEUE_DIR/${tag}.ready"
      echo "[$(date +%H:%M:%S)] RECOVER orphan-bin-scan $tag ($(du -h $bin | cut -f1))"
    else
      rm -f "$bin"
      touch "$QUEUE_DIR/${tag}.done"
      echo "[$(date +%H:%M:%S)] HEAL orphan-bin-scan-truncated $tag sz=$sz"
    fi
  done
  shopt -u nullglob
}

DISK_MIN_FREE_GB="${DISK_MIN_FREE_GB:-30}"
# Memory-free guard. Each emit RSS sits ~3-4 GiB at peak. With MAX=6 emits
# in flight + backend + dispatchers, total memory pressure can climb fast.
# Hold launches when free RAM falls below MEM_MIN_FREE_GB (default 8 GiB).
# Mirrors disk-free pattern: pool HOLDs rather than spawning an
# overcommit-OOM-kill cascade.
MEM_MIN_FREE_GB="${MEM_MIN_FREE_GB:-8}"
# Idle tolerance — pool stays alive while feeder may push more .lsp cells.
# Default 600 (x 5s = 50 min) — long enough to outlast feeder push cadence
# in normal operation. Exit only on STOP marker OR sustained idle past this
# bound.
POOL_MAX_IDLE_LOOPS="${POOL_MAX_IDLE_LOOPS:-600}"
# Substrate freshness — pull origin/master before emitting each cell so a
# backend never consumes a .lsp against stale substrate. Fail-soft: if pull
# fails (local mods, network), log + continue with checkout we have. Cell
# still emits — log entry tells operator "this host ran against potentially
# stale substrate".
AUTO_PULL_MASTER="${AUTO_PULL_MASTER:-1}"
AUTO_PULL_TIMEOUT="${AUTO_PULL_TIMEOUT:-30}"

# Startup orphan sweep — runs BEFORE main loop so any leftover bins from a
# SIGKILL-d previous pool (whose subshell wrappers died before touching
# .ready) get promoted immediately. In-loop heal also runs, but only after
# a full poll cycle (~5s + admission checks), during which the new pool
# could start spawning fresh emits + pushing past orphans. Catching them
# at startup closes that window.
echo "[$(date +%H:%M:%S)] startup orphan scan"
heal_stale_markers
heal_orphan_bins

idle_loops=0
while true; do
  # Heartbeat UNCONDITIONALLY at top of loop. Records both "I am alive"
  # mtime + "my pid is X" content. Supervisor on this host reads it every
  # poll cycle to decide whether to restart us.
  heartbeat_touch "$QUEUE_DIR" "pool"

  if [ -f "$QUEUE_DIR/STOP" ]; then
    echo "[$(date +%H:%M:%S)] STOP seen"
    POOL_EXIT_REASON="stop-marker"
    break
  fi
  pull_master_or_log || true
  heal_stale_markers
  heal_orphan_bins
  # Disk-free guard. Emits land 4-6 GiB bins; running near zero free disk
  # makes emitter spin on write retries instead of erroring out. Hold
  # launches until at least DISK_MIN_FREE_GB available.
  free_gb=$(df -BG --output=avail / 2>/dev/null | tail -1 | tr -dc 0-9)
  if [ "${free_gb:-0}" -lt "$DISK_MIN_FREE_GB" ]; then
    echo "[$(date +%H:%M:%S)] HOLD disk-free=${free_gb}G < ${DISK_MIN_FREE_GB}G, sleeping"
    sleep 30
    continue
  fi
  # Memory-free guard. MAX=6 concurrent emits can each climb to ~3-4 GiB
  # RSS; with backend + dispatchers also resident, sustained pressure
  # approaches host limits. Hold new launches until MemAvailable >=
  # MEM_MIN_FREE_GB so an OOM killer never picks a long-running emit
  # at the worst moment. Same backoff sleep as disk guard.
  free_mem_kb=$(grep ^MemAvailable: /proc/meminfo 2>/dev/null | tr -dc 0-9)
  free_mem_gb=$(( ${free_mem_kb:-0} / 1048576 ))
  if [ "$free_mem_gb" -lt "$MEM_MIN_FREE_GB" ]; then
    echo "[$(date +%H:%M:%S)] HOLD mem-free=${free_mem_gb}G < ${MEM_MIN_FREE_GB}G, sleeping"
    sleep 30
    continue
  fi
  # Read autoscaler advisory + apply dynamic MAX cap when RAM/swap pressure
  # detected. If advisory says slow_emit=1, cap concurrency at SLOW_MAX
  # (default = MAX/2). When pressure clears the advisory flips back to 0
  # and the cap restores. MEM_MIN_FREE_GB hold (above) is still the hard
  # floor — advisory acts as a graceful intermediate signal.
  effective_max="$MAX"
  if [ -f "$QUEUE_DIR/emit.advisory" ]; then
    slow=$(grep -E '^slow_emit=' "$QUEUE_DIR/emit.advisory" 2>/dev/null | tail -1 | cut -d= -f2)
    if [ "${slow:-0}" = "1" ]; then
      effective_max="${SLOW_MAX:-$(( MAX / 2 ))}"
      [ "$effective_max" -lt 1 ] && effective_max=1
    fi
  fi
  # Find next pending work item
  next=""
  for f in "$QUEUE_DIR"/*.lsp; do
    [ ! -f "$f" ] && continue
    tag=$(basename "$f" .lsp)
    [ -f "$QUEUE_DIR/${tag}.ready" ] && continue
    [ -f "$QUEUE_DIR/${tag}.done" ]  && continue
    # .done.fail = operator/caller marked cell as skip (incompatible config,
    # poison cell, retracted batch). Without this check pool re-emits the
    # cell every iter even after the operator explicitly excluded it.
    [ -f "$QUEUE_DIR/${tag}.done.fail" ] && continue
    [ -f "$QUEUE_DIR/${tag}.emitting" ] && continue
    # Dispatcher worker claims .ready by renaming to .inflight-N (atomic).
    # If worker dies mid-dispatch (backend crash, kill, OOM) .inflight-N
    # stays + no .done written. Treat .inflight-* as downstream-owned and
    # do not re-emit. Dispatcher reclaim path runs at bend-dispatcher
    # startup + heals stale .inflight-N back to .ready or .done.
    # NOTE: comments here must NOT contain apostrophes — each closes outer
    # single-quoted bash -c argument early + corrupts parse.
    if ls "$QUEUE_DIR/${tag}.inflight-"* >/dev/null 2>&1; then
      continue
    fi
    # Belt-and-suspenders against HEAL race: if a .bin exists at any size,
    # an emit has run. Dispatcher consumes .bin + may delete .ready when
    # done, leaving no other markers — without this check main loop would
    # relaunch the cell after dispatcher consumed it.
    [ -f "$QUEUE_DIR/${tag}.bin" ] && continue
    next="$f"
    break
  done
  if [ -z "$next" ]; then
    # No new items — wait for in-flight or for feeder to push more.
    running=$(ls "$QUEUE_DIR"/*.emitting 2>/dev/null | wc -l)
    if [ "$running" = "0" ]; then
      idle_loops=$((idle_loops + 1))
      if [ "$idle_loops" -ge "$POOL_MAX_IDLE_LOOPS" ]; then
        echo "[$(date +%H:%M:%S)] pool drained + idle past $POOL_MAX_IDLE_LOOPS loops"
        POOL_EXIT_REASON="drain-timeout"
        break
      fi
    fi
    sleep 5
    continue
  fi
  idle_loops=0
  # Throttle to MAX concurrent (self-healing each check). Heartbeat again
  # inside throttle wait — long emits can hold us here for minutes; without
  # re-touching heartbeat supervisor would see us as dead.
  while true; do
    heartbeat_touch "$QUEUE_DIR" "pool"
    heal_stale_markers
  heal_orphan_bins
    running=$(ls "$QUEUE_DIR"/*.emitting 2>/dev/null | wc -l)
    # Compare against effective_max (advisory-aware) rather than raw MAX
    # so RAM-pressure slow-down actually reduces concurrency. effective_max
    # computed once per main-loop iter against latest advisory.
    [ "$running" -lt "$effective_max" ] && break
    sleep 3
  done
  tag=$(basename "$next" .lsp)
  echo "[$(date +%H:%M:%S)] launch $tag"
  # Subshell: traps EXIT/HUP/TERM so partial emit still finalizes (.ready
  # or .done) when parent pool dies during hot-reload. SIGHUP explicitly
  # ignored — propagated from a dying parent must NOT kill the wrapper
  # mid-cleanup. SIGKILL on us still leaks (no trap fires), but startup
  # heal_orphan_bins promotes those bins on next pool start.
  (
    trap "" HUP
    bin="$QUEUE_DIR/${tag}.bin"
    finalize() {
      if [ -f "$bin" ]; then
        sz=$(stat -c %s "$bin" 2>/dev/null || echo 0)
        # Bin-magic check FIRST. Interrupted emits (OOM-killed mid-write,
        # SIGKILL on emitter hang) leave the 16-byte header reserve as
        # zero bytes — emit-stream writes 16 zeros, streams ops, then
        # seeks back to byte 0 to rewrite QECCOPS1 + n_ops at the end.
        # SIGKILL between those steps strands a file with zero magic.
        # Downstream backend chokes with opaque "bad magic" → bin-load-fail
        # → DLQ with no useful forensic info. Route here with actual cause
        # named so operators can trace back to upstream cell hang instead
        # of guessing at load_ops_bin internals.
        magic=$(head -c 8 "$bin" 2>/dev/null)
        if [ "$magic" != "QECCOPS1" ]; then
          mkdir -p "$QUEUE_DIR/dlq"
          echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
          {
            echo "tier=zero-magic bin_bytes=$sz"
            echo "reason: emit-interrupted — bin header magic missing or non-QECCOPS1 (likely SIGKILL or OOM mid-emit; emit-stream header rewrite never fired)"
            echo "first-8-bytes-hex: $(head -c 8 "$bin" 2>/dev/null | xxd -p -c 8)"
            echo "--- last 10 lines of emit.log ---"
            tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null
          } > "$QUEUE_DIR/dlq/${tag}.reason"
          echo "[$(date +%H:%M:%S)] DLQ $tag zero-magic sz=$sz ($(printf %.0f $((sz/1024/1024)))MiB) — emit-interrupted"
          rm -f "$bin"
          touch "$QUEUE_DIR/${tag}.done.fail"
          touch "$QUEUE_DIR/${tag}.done"
          rm -f "$QUEUE_DIR/${tag}.emitting"
          return 0
        fi
        # 2nd sub-class: GOOD magic + n_ops=0 — emit wrote QECCOPS1 magic +
        # streamed ops but SIGKILL hit between stream-end + finalize seek-
        # back-to-byte-8 that rewrites n_ops. Detect via n_ops=0
        # implausibility check: any in-tier bin should have n_ops > 0;
        # if not, finalize never fired.
        n_ops_le=$(dd if="$bin" bs=1 count=8 skip=8 2>/dev/null | xxd -p -c 8)
        if [ "$n_ops_le" = "0000000000000000" ]; then
          mkdir -p "$QUEUE_DIR/dlq"
          echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
          {
            echo "tier=zero-n_ops bin_bytes=$sz"
            echo "reason: emit-finalize-never-fired — QECCOPS1 magic present but n_ops field at byte 8 is zero (SIGKILL between stream-end + emit-stream seek-back; downstream load_ops_bin would fail length-mismatch)"
            echo "n_ops_le_hex: $n_ops_le"
            echo "--- last 10 lines of emit.log ---"
            tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null
          } > "$QUEUE_DIR/dlq/${tag}.reason"
          echo "[$(date +%H:%M:%S)] DLQ $tag zero-n_ops sz=$sz ($(printf %.0f $((sz/1024/1024)))MiB) — emit-finalize-never-fired"
          rm -f "$bin"
          touch "$QUEUE_DIR/${tag}.done.fail"
          touch "$QUEUE_DIR/${tag}.done"
          rm -f "$QUEUE_DIR/${tag}.emitting"
          return 0
        fi
        # Tier-classify the bin. Tier "below-min" / "above-max" → DLQ with
        # a typed reason. In-band tiers (micro/small/medium/large) get
        # promoted to .ready + dispatcher routes through tier-specific
        # admission. lib-tier.sh owns boundary policy.
        tier=$(tier_of_size "$sz")
        dlq_reason=$(tier_label_dlq "$tier")
        if [ -n "$dlq_reason" ]; then
          mkdir -p "$QUEUE_DIR/dlq"
          echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
          {
            echo "tier=$tier bin_bytes=$sz"
            echo "reason: $dlq_reason"
            echo "--- last 10 lines of emit.log ---"
            tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null
          } > "$QUEUE_DIR/dlq/${tag}.reason"
          echo "[$(date +%H:%M:%S)] DLQ $tag tier=$tier sz=$sz ($(printf %.0f $((sz/1024/1024)))MiB) — $dlq_reason"
          rm -f "$bin"
          touch "$QUEUE_DIR/${tag}.done.fail"
          touch "$QUEUE_DIR/${tag}.done"
        else
          # In-tier bin → optional domain-specific sidecar (vorticity, etc),
          # then mark .ready. Sidecar lands next to .bin. Read-only
          # post-process; emit substrate untouched. Fail-soft: sidecar
          # script errors log + continue — bin still promotes to .ready.
          # Dispatcher reads .bin, ignores sidecar; operators + analysis
          # tooling consume sidecar opportunistically.
          vort="$QUEUE_DIR/${tag}.vorticity.tsv"
          if [ -n "$LUMBDA_DOMAIN_DIR" ] && [ -x "$LUMBDA_DOMAIN_DIR/scripts/qubit-vorticity.py" ]; then
            if "$LUMBDA_DOMAIN_DIR/scripts/qubit-vorticity.py" "$bin" --out "$vort" \
                 > "$QUEUE_DIR/${tag}.vorticity.log" 2>&1; then
              :
            else
              echo "[$(date +%H:%M:%S)] vorticity $tag FAILED (see ${tag}.vorticity.log) — promoting bin anyway"
            fi
          fi
          touch "$QUEUE_DIR/${tag}.ready"
          echo "[$(date +%H:%M:%S)] ready $tag tier=$tier ($(du -h $bin | cut -f1))"
        fi
      else
        # No bin — emitter errored before writing (undefined flag, syntax
        # error, load failure, etc). Emit failures stay DETERMINISTIC —
        # retry will fail the same way. Route to DLQ for forensics, mark
        # .done.fail + .done so pool skips on next iter (no infinite
        # relaunch loop).
        mkdir -p "$QUEUE_DIR/dlq"
        echo 1 > "$QUEUE_DIR/dlq/${tag}.retries"
        tail -10 "$QUEUE_DIR/${tag}.emit.log" 2>/dev/null > "$QUEUE_DIR/dlq/${tag}.reason"
        echo "[$(date +%H:%M:%S)] DLQ $tag (emit error — see dlq/${tag}.reason)"
        touch "$QUEUE_DIR/${tag}.done.fail"
        touch "$QUEUE_DIR/${tag}.done"
      fi
      rm -f "$QUEUE_DIR/${tag}.emitting"
    }
    trap finalize EXIT
    trap "kill \$lpid 2>/dev/null; finalize; exit 130" INT
    trap "kill \$lpid 2>/dev/null; finalize; exit 143" TERM
    if [ -n "$LUMBDA_DOMAIN_DIR" ]; then
      cd "$LUMBDA_DOMAIN_DIR"
    fi
    PYTHONUNBUFFERED=1 ECDSA_OUT_BIN="$QUEUE_DIR/${tag}.bin" \
      $LUMBDA_EMITTER_CMD "$next" \
      > "$QUEUE_DIR/${tag}.emit.log" 2>&1 &
    lpid=$!
    echo "$lpid" > "$QUEUE_DIR/${tag}.emitting"
    wait "$lpid"
  ) &
done
wait
echo "[$(date +%H:%M:%S)] all emits done"
# Loop exited cleanly via STOP / drain; POOL_EXIT_REASON already set.
# EXIT trap calls heartbeat_exit_cause.
'
