lumbda/factory/bend-supervisor-dlq-runner.sh

#!/usr/bin/env bash
# bend-supervisor-dlq-runner.sh — auto-resolve loop for our DLQ.
#
# Polls $LUMBDA_QUEUE_DIR/dlq every $DLQ_POLL_S seconds. For each retry-
# eligible entry (has a .bin in dlq/), classify by reason file, decide:
#
#   AUTO-RETRY    → move .bin back to queue + touch .ready. Counters in
#                   $LUMBDA_QUEUE_DIR/dlq/<tag>.retries persist across passes.
#                   When retries reach AUTO_RETRY_MAX for an auto class,
#                   we ESCALATE instead.
#
#   ESCALATE      → move bin + reason + retries + copy .lsp into
#                   $LUMBDA_QUEUE_DIR/rdlq/. Cell now waits for operator
#                   to run a retry workflow.
#
#   DROP          → for cells whose .bin stays missing or 0-byte (emit truly
#                   failed; nothing to preserve). Touch .done.fail.
#
# State-stage classification drives WHAT we preserve when escalating:
#   stage=emit      → .lsp matters, .bin disposable (pool re-emits)
#   stage=dispatch  → .bin good, dispatcher infrastructure broke; retry
#                     preserves .bin
#   stage=sim       → .bin good, but backend failed (OOM / illegal-addr
#                     / etc). Operator decides: retry with different
#                     n_batches, or patch backend + rebuild, or quarantine.
#
# Designed to run as one supervisor-managed worker. Heartbeats to
# $LUMBDA_QUEUE_DIR/dlq-runner.heartbeat; supervisor restarts if stale.
#
# Exit signals:
#   - $LUMBDA_QUEUE_DIR/DLQ_RUNNER_STOP marker
#   - SIGTERM / SIGINT
#
# Env vars consumed: see $LUMBDA_FACTORY_DIR/CONTRACT.md (LUMBDA_QUEUE_DIR,
# DLQ_POLL_S, AUTO_RETRY_MAX, DLQ_RUNNER_LOG).
set -u

QUEUE_DIR="${LUMBDA_QUEUE_DIR:-${QUEUE_DIR:-/tmp/lumbda-queue}}"
DLQ_POLL_S="${DLQ_POLL_S:-30}"
AUTO_RETRY_MAX="${AUTO_RETRY_MAX:-3}"
LOG="${DLQ_RUNNER_LOG:-$QUEUE_DIR/dlq-runner.log}"
STOP_FILE="$QUEUE_DIR/DLQ_RUNNER_STOP"
RDLQ_DIR="$QUEUE_DIR/rdlq"

LUMBDA_FACTORY_DIR="${LUMBDA_FACTORY_DIR:-$(dirname "$(readlink -f "$0")")}"
LIB_HEARTBEAT="${LIB_HEARTBEAT:-$LUMBDA_FACTORY_DIR/lib-heartbeat.sh}"
# shellcheck source=/dev/null
. "$LIB_HEARTBEAT"

mkdir -p "$RDLQ_DIR"

log() {
    local ts msg
    ts=$(date +%H:%M:%S)
    msg="[$ts] $*"
    echo "$msg" | tee -a "$LOG"
}

# Classification table — single source of truth. Returns one of:
#   auto-retry   missing-bin  bisect-pool-race  cuda-error-transient  no-portal
#   escalate     cuda-oom     cuda-illegal-addr  bin-load-fail
#                memory-cap-refused  tier-classify  unknown
#
# classify_reason <reason-file>
#   echoes "<class> <stage> <salvage_bin>"
#     class       — one of named classes above
#     stage       — emit | dispatch | sim
#     salvage_bin — yes | no  (yes = preserve .bin on escalate; no = re-emit needed)
classify_reason() {
    local r="$1"
    if [ ! -f "$r" ]; then
        echo "unknown unknown no"; return
    fi
    # Empty or whitespace-only .reason files: classify as transient-empty-reason
    # directly. Auto-retry class. After AUTO_RETRY_MAX (3) escalates, but noise
    # stays contained AND operators see the right label.
    if [ ! -s "$r" ] || ! grep -qE '[^[:space:]]' "$r" 2>/dev/null; then
        echo "transient-empty-reason sim yes"; return
    fi
    # Specific (code XXX) matches FIRST — typed portals carry most reliable
    # signal. Generic "cuda-sim-error" / "cuda-error" catch-all comes last
    # so bin-load-fail / memory-cap-refused etc. don't get silently swallowed.
    if grep -q "cuda-oom\|cudaMalloc.*out of memory\|cudaErrorMemoryAllocation" "$r" 2>/dev/null; then
        echo "cuda-oom sim yes"; return
    fi
    if grep -q "cuda-illegal-addr\|cudaErrorIllegalAddress" "$r" 2>/dev/null; then
        echo "cuda-illegal-addr sim yes"; return
    fi
    if grep -q "memory-cap-refused" "$r" 2>/dev/null; then
        echo "memory-cap-refused sim yes"; return
    fi
    # Fast-fail (backend pre-flight cudaMemGetInfo). Bin is fine; needs
    # n_batches reduction or wait for less GPU contention.
    if grep -q "vram-budget-refused" "$r" 2>/dev/null; then
        echo "vram-budget-refused sim yes"; return
    fi
    # Structural reject (dispatcher pre-flight 2026-06-14): predicted VRAM
    # exceeds card capacity. Cell cannot ever fit. Bin is dead weight;
    # salvage=no since retry on same card is futile.
    if grep -q "vram-oversized" "$r" 2>/dev/null; then
        echo "vram-oversized sim no"; return
    fi
    if grep -q "bin-load-fail" "$r" 2>/dev/null; then
        echo "bin-load-fail emit no"; return
    fi
    # Emit-stage (pool DLQ'd at finalize)
    if grep -q "tier=zero-magic\|tier=zero-n_ops\|emit-interrupted" "$r" 2>/dev/null; then
        echo "emit-broken emit no"; return
    fi
    if grep -q "tier=above-max\|tier=below-min" "$r" 2>/dev/null; then
        echo "tier-classify emit no"; return
    fi
    # Generic typed cuda error catchall — fired when CUDA_CHECK surfaced an
    # error not matched by oom/illegal-addr/memory-cap.
    if grep -q "cuda-sim-error\|cuda-error" "$r" 2>/dev/null; then
        echo "cuda-error-transient sim yes"; return
    fi
    # Dispatch-stage (backend infrastructure)
    if grep -q "^missing:" "$r" 2>/dev/null; then
        echo "missing-bin dispatch yes"; return
    fi
    # BISECT-DIAG entries are NOT a step6 crash — they are pool finalize()
    # race vs heal_orphan_bins where .bin got consumed/deleted between
    # subshell-exit and finalize check. Cells actually completed emit. So
    # retry-emit is safe; the bin is gone.
    if grep -q "BISECT-DIAG" "$r" 2>/dev/null; then
        echo "bisect-pool-race emit no"; return
    fi
    # Generic no-portal — backend crashed without writing typed portal.
    if grep -q "no-portal" "$r" 2>/dev/null; then
        echo "no-portal sim yes"; return
    fi
    echo "unknown unknown no"
}

# Auto-retry classes — these resolve themselves with a re-try.
is_auto_class() {
    case "$1" in
        missing-bin|bisect-pool-race|cuda-error-transient|no-portal|transient-empty-reason) return 0 ;;
        *) return 1 ;;
    esac
}

# Escalate-on-first-fail classes — no point in auto-retrying.
is_escalate_class() {
    case "$1" in
        cuda-oom|cuda-illegal-addr|memory-cap-refused|vram-budget-refused|vram-oversized|bin-load-fail|tier-classify|emit-broken|unknown) return 0 ;;
        *) return 1 ;;
    esac
}

# Move <tag> from dlq/ back to queue/ for another dispatcher attempt.
auto_retry_one() {
    local tag="$1" reason_class="$2"
    local bin="$QUEUE_DIR/dlq/${tag}.bin"
    local ready="$QUEUE_DIR/${tag}.ready"
    if [ ! -f "$bin" ]; then
        # Bin missing in DLQ — nothing to retry. Drop to .done.fail so pool
        # stops re-emitting.
        touch "$QUEUE_DIR/${tag}.done.fail"
        touch "$QUEUE_DIR/${tag}.done"
        rm -f "$QUEUE_DIR/dlq/${tag}.reason" "$QUEUE_DIR/dlq/${tag}.retries"
        log "DROP $tag class=$reason_class (no .bin in dlq, marking done.fail)"
        return
    fi
    mv "$bin" "$QUEUE_DIR/${tag}.bin"
    touch "$ready"
    log "AUTO-RETRY $tag class=$reason_class -> .ready"
}

# Escalate <tag> from dlq/ to rdlq/. Preserves .bin if salvage_bin=yes,
# always copies .lsp source for operator to inspect or re-emit from.
escalate_one() {
    local tag="$1" reason_class="$2" stage="$3" salvage="$4"
    local dlq_bin="$QUEUE_DIR/dlq/${tag}.bin"
    local dlq_reason="$QUEUE_DIR/dlq/${tag}.reason"
    local dlq_retries="$QUEUE_DIR/dlq/${tag}.retries"
    local rdlq_bin="$RDLQ_DIR/${tag}.bin"
    local rdlq_reason="$RDLQ_DIR/${tag}.reason"
    local rdlq_retries="$RDLQ_DIR/${tag}.retries"
    local rdlq_stage="$RDLQ_DIR/${tag}.stage"
    local rdlq_class="$RDLQ_DIR/${tag}.class"
    local rdlq_lsp="$RDLQ_DIR/${tag}.lsp"
    local rdlq_first="$RDLQ_DIR/${tag}.first-seen"
    local src_lsp="$QUEUE_DIR/${tag}.lsp"
    local retries=0
    [ -f "$dlq_retries" ] && retries=$(cat "$dlq_retries" 2>/dev/null)
    [ -z "$retries" ] && retries=0

    if [ "$salvage" = "yes" ] && [ -f "$dlq_bin" ]; then
        mv "$dlq_bin" "$rdlq_bin"
    elif [ -f "$dlq_bin" ]; then
        # salvage=no — bin is suspect; remove so retry-reemit is the only path
        rm -f "$dlq_bin"
    fi
    # Copy .lsp if it still exists in queue. (Pool may have already
    # rm'd it once the cell hit .done.) If missing, escalation goes
    # forward without source — operator pulls from git.
    if [ -f "$src_lsp" ]; then
        cp "$src_lsp" "$rdlq_lsp"
    fi
    {
        echo "# rDLQ entry — escalated $(date -u +%FT%TZ)"
        echo "# class:  $reason_class"
        echo "# stage:  $stage"
        echo "# salvage_bin: $salvage"
        echo "# retries-at-escalation: $retries"
        echo ""
        echo "## Original DLQ reason:"
        if [ -f "$dlq_reason" ]; then cat "$dlq_reason"; fi
    } > "$rdlq_reason"
    echo "$retries" > "$rdlq_retries"
    echo "$stage" > "$rdlq_stage"
    echo "$reason_class" > "$rdlq_class"
    [ ! -f "$rdlq_first" ] && date -u +%FT%TZ > "$rdlq_first"
    rm -f "$dlq_bin" "$dlq_reason" "$dlq_retries"
    # Mark .done.fail so pool stops re-emitting while cell sits in rDLQ.
    # Re-emit path clears these.
    touch "$QUEUE_DIR/${tag}.done.fail"
    touch "$QUEUE_DIR/${tag}.done"
    log "ESCALATE $tag class=$reason_class stage=$stage salvage=$salvage retries=$retries -> rdlq/"
}

# One pass over the DLQ. Increments cell-local retry only on auto-retry
# decisions (so escalate paths don't inflate counter unfairly).
process_one_pass() {
    shopt -s nullglob
    local f tag class stage salvage parts
    for f in "$QUEUE_DIR"/dlq/*.reason; do
        [ -f "$f" ] || continue
        tag=$(basename "$f" .reason)
        parts=$(classify_reason "$f")
        class=$(echo "$parts" | awk '{print $1}')
        stage=$(echo "$parts" | awk '{print $2}')
        salvage=$(echo "$parts" | awk '{print $3}')

        local retries=0
        [ -f "$QUEUE_DIR/dlq/${tag}.retries" ] && retries=$(cat "$QUEUE_DIR/dlq/${tag}.retries" 2>/dev/null)
        [ -z "$retries" ] && retries=0

        if is_escalate_class "$class"; then
            escalate_one "$tag" "$class" "$stage" "$salvage"
            continue
        fi
        if is_auto_class "$class"; then
            if [ "$retries" -ge "$AUTO_RETRY_MAX" ]; then
                # Auto class exhausted retries — escalate.
                escalate_one "$tag" "$class" "$stage" "$salvage"
                continue
            fi
            retries=$((retries + 1))
            echo "$retries" > "$QUEUE_DIR/dlq/${tag}.retries"
            auto_retry_one "$tag" "$class"
            continue
        fi
        # Unknown classifier — escalate defensively (operator decides).
        escalate_one "$tag" "$class" "$stage" "$salvage"
    done
    shopt -u nullglob
}

EXIT_REASON=""
on_exit() {
    local r="${EXIT_REASON:-unknown-exit}"
    heartbeat_exit_cause "$QUEUE_DIR" "dlq-runner" "$r"
    echo "[$(date +%H:%M:%S)] bend-supervisor-dlq-runner exit reason=$r" | tee -a "$LOG"
}
trap on_exit EXIT
trap 'EXIT_REASON=signal-int;  exit 130' INT
trap 'EXIT_REASON=signal-term; exit 143' TERM

log "bend-supervisor-dlq-runner start queue=$QUEUE_DIR poll=${DLQ_POLL_S}s auto_max=$AUTO_RETRY_MAX"

while true; do
    heartbeat_touch "$QUEUE_DIR" "dlq-runner"
    if [ -f "$STOP_FILE" ]; then
        log "STOP marker — exiting"
        rm -f "$STOP_FILE"
        EXIT_REASON="stop-marker"
        break
    fi
    process_one_pass
    sleep "$DLQ_POLL_S"
done