lumbda/factory/bend-dispatcher.sh

#!/usr/bin/env bash
# bend-dispatcher.sh — singleton dispatcher for a circuit-emission pipeline.
#
# Polls $LUMBDA_QUEUE_DIR/*.ready, dispatches each via $LUMBDA_BACKEND_CMD
# (or a custom bend-dispatch.py wrapper named in BEND_DISPATCH), logs result
# to $RESULTS_TSV, removes .bin to free disk, touches .done.
#
# Singleton: refuses to start if another instance holds
# $LUMBDA_QUEUE_DIR/dispatcher.lock.
#
# Exits when no .ready remains AND no .emitting marker AND $LUMBDA_QUEUE_DIR/STOP
# exists, OR after MAX_IDLE_LOOPS (default 60 = 5 min) of empty polling.
#
# Usage: bend-dispatcher.sh [QUEUE_DIR [RESULTS_TSV [BATCHES]]]
#   QUEUE_DIR default $LUMBDA_QUEUE_DIR
#   RESULTS_TSV default $QUEUE_DIR/results.tsv
#   BATCHES default 141
#
# Self-identifies as "bend-dispatcher" in cmdline for precise pkill targeting.
#
# Results TSV row format:
#   tag n_ops qubits avg_tof score status identity wall_s
#   config_root ops_hash trace_hash score_hash candidate_merkle_root
# Old readers ignore unknown trailing columns. Pre-fingerprint rows have
# these columns empty when migrated.
#
# Env vars consumed: see $LUMBDA_FACTORY_DIR/CONTRACT.md (LUMBDA_QUEUE_DIR,
# LUMBDA_BACKEND_CMD, LUMBDA_DOMAIN_DIR, BEND_DISPATCH, HASHES_PY).
set -u

QUEUE_DIR="${1:-${LUMBDA_QUEUE_DIR:-/tmp/lumbda-queue}}"
RESULTS_TSV="${2:-$QUEUE_DIR/results.tsv}"
BATCHES="${3:-141}"
MAX_IDLE_LOOPS="${MAX_IDLE_LOOPS:-60}"

# BEND_DISPATCH = python wrapper that talks to backend over a portal.
# When unset, consumer should set it via env (foxhop sets to
# $LUMBDA_DOMAIN_DIR/runs/lumbda-sweep-008/bend-dispatch.py).
# HASHES_PY = optional fingerprint helper; falls through to empty columns
# when missing.
LUMBDA_DOMAIN_DIR="${LUMBDA_DOMAIN_DIR:-}"
BEND_DISPATCH="${BEND_DISPATCH:-${LUMBDA_DOMAIN_DIR:+$LUMBDA_DOMAIN_DIR/runs/lumbda-sweep-008/bend-dispatch.py}}"
HASHES_PY="${HASHES_PY:-${LUMBDA_DOMAIN_DIR:+$LUMBDA_DOMAIN_DIR/scripts/ecdsa-hashes.py}}"

mkdir -p "$QUEUE_DIR"
LOCK="$QUEUE_DIR/dispatcher.lock"

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

exec -a bend-dispatcher bash -c '
QUEUE_DIR="'"$QUEUE_DIR"'"
RESULTS_TSV="'"$RESULTS_TSV"'"
BATCHES="'"$BATCHES"'"
MAX_IDLE_LOOPS="'"$MAX_IDLE_LOOPS"'"
BEND_DISPATCH="'"$BEND_DISPATCH"'"
HASHES_PY="'"$HASHES_PY"'"
LOCK="'"$LOCK"'"

# Singleton via flock
exec 9>"$LOCK"
if ! flock -n 9; then
  echo "[$(date +%H:%M:%S)] bend-dispatcher: another instance holds $LOCK, refusing to start"
  exit 1
fi
echo "[$(date +%H:%M:%S)] bend-dispatcher start queue=$QUEUE_DIR"
trap "echo \"[\$(date +%H:%M:%S)] bend-dispatcher exit\"; rm -f $LOCK" EXIT

# Header includes fingerprint columns. A fresh results.tsv gets the full
# 13-column header. A pre-fingerprint file with only the original 8 columns
# stays untouched (so readers keyed off the legacy header keep working) but
# every new row gets fingerprints appended after legacy fields.
if [ ! -f "$RESULTS_TSV" ]; then
  printf "tag\tn_ops\tqubits\tavg_tof\tscore\tstatus\tidentity\twall_s\tconfig_root\tops_hash\ttrace_hash\tscore_hash\tcandidate_merkle_root\n" > "$RESULTS_TSV"
fi

idle=0
while true; do
  ready=$(ls "$QUEUE_DIR"/*.ready 2>/dev/null | sort | head -1)
  if [ -z "$ready" ]; then
    # check stop conditions
    if [ -f "$QUEUE_DIR/STOP" ]; then
      n_emitting=$(ls "$QUEUE_DIR"/*.emitting 2>/dev/null | wc -l)
      [ "$n_emitting" = "0" ] && { echo "[$(date +%H:%M:%S)] STOP + no emits"; break; }
    fi
    idle=$((idle + 1))
    if [ "$idle" -ge "$MAX_IDLE_LOOPS" ]; then
      # nothing in flight, no ready, no new emits coming
      n_emitting=$(ls "$QUEUE_DIR"/*.emitting 2>/dev/null | wc -l)
      pending=$(ls "$QUEUE_DIR"/*.lsp 2>/dev/null | wc -l)
      n_done=$(ls "$QUEUE_DIR"/*.done 2>/dev/null | wc -l)
      n_ready_total=$((pending - n_done))
      if [ "$n_emitting" = "0" ] && [ "$n_ready_total" -le "0" ]; then
        echo "[$(date +%H:%M:%S)] idle drained"
        break
      fi
      idle=0
    fi
    sleep 5
    continue
  fi
  idle=0
  tag=$(basename "$ready" .ready)
  bin="$QUEUE_DIR/${tag}.bin"
  if [ ! -f "$bin" ]; then
    echo "[$(date +%H:%M:%S)] SKIP $tag bin missing"
    rm -f "$ready"
    touch "$QUEUE_DIR/${tag}.done"
    continue
  fi
  echo "[$(date +%H:%M:%S)] dispatch $tag ($(du -h $bin | cut -f1))"
  t0=$(date +%s)
  out=$(python3 "$BEND_DISPATCH" "$bin" "$BATCHES" 2>&1)
  t1=$(date +%s)
  wall=$((t1 - t0))

  # Dead-letter detection: dispatch returned non-parseable output (backend
  # crashed or empty response). Move bin + ready to DLQ for retry by
  # watchdog after backend comes back up. We dont mark .done here so the
  # cell stays out of completed state — only DLQ retries fully drain it.
  if ! echo "$out" | grep -q "^score = "; then
    mkdir -p "$QUEUE_DIR/dlq"
    retries=0
    if [ -f "$QUEUE_DIR/dlq/${tag}.retries" ]; then
      retries=$(cat "$QUEUE_DIR/dlq/${tag}.retries")
    fi
    retries=$((retries + 1))
    echo "[$(date +%H:%M:%S)] DLQ $tag (retry=$retries, wall=${wall}s, backend likely down)"
    echo "$out" | tail -10 > "$QUEUE_DIR/dlq/${tag}.reason"
    echo "$retries" > "$QUEUE_DIR/dlq/${tag}.retries"
    if [ "$retries" -lt 5 ]; then
      mv "$bin" "$QUEUE_DIR/dlq/${tag}.bin"
      rm -f "$ready"
      # Note: do NOT touch .done — cell still pending. Watchdog/reaper
      # moves bin back + re-touches .ready after backend stays healthy 60s+.
    else
      echo "[$(date +%H:%M:%S)] DLQ $tag exhausted retries — marking .done.fail"
      rm -f "$bin" "$ready"
      touch "$QUEUE_DIR/${tag}.done.fail"
      touch "$QUEUE_DIR/${tag}.done"
    fi
    continue
  fi
  # Keep last 5 lines: parsed= + score + trace_hash= + raw_portal_resp= +
  # trailing newline. Portal-resp blob lets us recompute fingerprints
  # offline if dispatcher TSV ever needs re-hashing.
  echo "$out" | tail -5 > "$QUEUE_DIR/${tag}.dispatch.log"
  nops=$(echo "$out"   | grep -oP "n_ops.: \K[0-9]+"     | head -1)
  qubits=$(echo "$out" | grep -oP "qubits.: \K[0-9]+"    | head -1)
  tof=$(echo "$out"    | grep -oP "avg_tof.: \K[0-9.]+"  | head -1)
  score=$(echo "$out"  | grep -oP "score = .* = \K\S+")
  status=$(echo "$out" | grep -oP "status.: .\K[A-Z]+"   | head -1)
  ident=$(echo "$out"  | grep -oP "identity.: .\K[a-z]+" | head -1)

  # ── Fingerprint columns ─────────────────────────────────────────────
  # cell .lsp drives config_root. Cells get staged into queue dir as
  # $QUEUE_DIR/${tag}.lsp by the feeder. When originating .lsp is absent
  # (e.g. probe path), config_root falls through to all-zero digest and
  # downstream readers can detect absence.
  cell="$QUEUE_DIR/${tag}.lsp"
  portal_resp=$(echo "$out" | grep -oP "^raw_portal_resp=\K.*" | tail -1)
  # Pipe portal response into HASHES_PY so every root and the
  # candidate_merkle_root land in a single JSON line. When HASHES_PY is
  # unset / missing, fall through with empty columns so legacy callers
  # still get the score row.
  score_json=$(printf "{\"tag\":\"%s\",\"n_ops\":\"%s\",\"qubits\":\"%s\",\"avg_tof\":\"%s\",\"score\":\"%s\",\"status\":\"%s\",\"identity\":\"%s\",\"wall_s\":\"%s\"}" \
    "$tag" "${nops:-}" "${qubits:-}" "${tof:-}" "${score:-}" "${status:-}" "${ident:-}" "$wall")
  hash_json="{}"
  if [ -n "$HASHES_PY" ] && [ -f "$HASHES_PY" ]; then
    hash_json=$(printf "%s" "$portal_resp" | \
      python3 "$HASHES_PY" all \
        --cell "$cell" \
        --bin "$bin" \
        --portal-stdin \
        --score-json "$score_json" 2>/dev/null || echo "{}")
  fi
  config_root=$(echo "$hash_json" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get(\"config_root\",\"\"))" 2>/dev/null)
  ops_hash=$(echo    "$hash_json" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get(\"ops_hash\",\"\"))" 2>/dev/null)
  trace_hash=$(echo  "$hash_json" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get(\"trace_hash\",\"\"))" 2>/dev/null)
  score_hash=$(echo  "$hash_json" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get(\"score_hash\",\"\"))" 2>/dev/null)
  merkle=$(echo      "$hash_json" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get(\"candidate_merkle_root\",\"\"))" 2>/dev/null)

  printf "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n" \
    "$tag" "${nops:-}" "${qubits:-}" "${tof:-}" "${score:-}" "${status:-}" "${ident:-}" "$wall" \
    "${config_root:-}" "${ops_hash:-}" "${trace_hash:-}" "${score_hash:-}" "${merkle:-}" \
    >> "$RESULTS_TSV"
  echo "  -> score=$score status=$status identity=$ident wall=${wall}s"
  echo "     merkle=${merkle:0:16}.. trace=${trace_hash:0:16}.. ops=${ops_hash:0:16}.. config=${config_root:0:16}.."
  rm -f "$bin" "$ready"
  touch "$QUEUE_DIR/${tag}.done"
done
'