code/federation/agillm41_vast_side_update_puller.sh

#!/usr/bin/env bash
set -Eeuo pipefail

GETH_HOST="${AGILLM41_GETH_HOST:-root@5.75.217.57}"
GETH_KEY="${AGILLM41_GETH_KEY:-/root/.ssh/agillm41_geth_ed25519}"
REMOTE_UPDATE_DIRS="${AGILLM41_REMOTE_UPDATE_DIRS:-/root/agillm41_opportunistic/updates /root/agillm41_worker/updates}"
INCOMING_DIR="${AGILLM41_INCOMING_DIR:-/workspace/agillm41_side_updates/incoming}"
ACCEPTED_DIR="${AGILLM41_ACCEPTED_DIR:-/workspace/agillm41_side_updates/accepted}"
REJECTED_DIR="${AGILLM41_REJECTED_DIR:-/workspace/agillm41_side_updates/rejected}"
STATE_DIR="${AGILLM41_PULL_STATE_DIR:-/workspace/agillm41_side_updates/pulled}"
POLL_SEC="${AGILLM41_PULL_POLL_SEC:-15}"
MIN_BYTES="${AGILLM41_PULL_MIN_BYTES:-100000000}"
KEEP_ARCHIVE_FILES="${AGILLM41_KEEP_ARCHIVE_FILES:-8}"

mkdir -p "$INCOMING_DIR" "$ACCEPTED_DIR" "$REJECTED_DIR" "$STATE_DIR"

prune_archive_dir() {
  local dir="$1"
  find "$dir" -maxdepth 1 -type f -name '*.pt' -printf '%T@ %p\n' 2>/dev/null \
    | sort -rn | awk -v keep="$KEEP_ARCHIVE_FILES" 'NR>keep {sub(/^[^ ]+ /,""); print}' \
    | while IFS= read -r stale; do
        [ -n "$stale" ] || continue
        printf '{"event":"pruned_side_update_archive","path":"%s","at":"%s"}\n' "$stale" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
        rm -f -- "$stale"
      done
}

prune_local_archives() {
  prune_archive_dir "$ACCEPTED_DIR"
  prune_archive_dir "$REJECTED_DIR"
}

copy_once() {
  local remote_dir names name src dst marker safe_dir worker
  for remote_dir in $REMOTE_UPDATE_DIRS; do
    names="$(ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
      "find '$remote_dir' -maxdepth 1 -type f -name '*.pt' -size +$((MIN_BYTES - 1))c -printf '%f\n' 2>/dev/null | sort" || true)"
    safe_dir="$(printf '%s' "$remote_dir" | tr -c 'A-Za-z0-9_.-' '_')"
    while IFS= read -r name; do
      [ -n "$name" ] || continue
      marker="$STATE_DIR/${safe_dir}__$name.done"
      [ ! -e "$marker" ] || continue
      src="$remote_dir/$name"
      dst="$INCOMING_DIR/$name"
      if [ -e "$dst" ] || [ -e "$ACCEPTED_DIR/$name" ] || [ -e "$REJECTED_DIR/$name" ]; then
        printf '{"event":"side_update_already_seen","name":"%s","remote_dir":"%s","at":"%s"}\n' "$name" "$remote_dir" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" > "$marker"
        continue
      fi
      # disk headroom guard: a full disk corrupts master checkpoints (20260611 incident)
      if [ "$(df / | awk 'NR==2{gsub("%","",$5);print $5}')" -ge 80 ]; then
        printf '{"event":"side_update_pull_skipped_disk","at":"%s"}\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
        return 0
      fi
      if ! rsync -P --inplace -e "ssh -i \"$GETH_KEY\" -o BatchMode=yes -o StrictHostKeyChecking=no" "$GETH_HOST:$src" "$dst.tmp"; then
        printf '{"event":"side_update_pull_failed","name":"%s","at":"%s"}\n' "$name" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >&2
        rm -f "$dst.tmp"
        continue
      fi
      # durability guard: self-heal a deleted validator from its co-located backup
      # (uncommitted-file loss class; see promoter 03:18 incident). No-op when present.
      if [ ! -f /workspace/validate_side_update.py ] && [ -f /workspace/validate_side_update.py.bak_durable_20260616 ]; then
        cp /workspace/validate_side_update.py.bak_durable_20260616 /workspace/validate_side_update.py
        printf '{"event":"validator_self_healed_from_backup","at":"%s"}\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
      fi
      # sanity gate: never hand the trainer an update with NaN/inf/huge weights
      if ! python3 /workspace/validate_side_update.py "$dst.tmp"; then
        mv "$dst.tmp" "$REJECTED_DIR/$name"
        printf '{"event":"side_update_rejected_unsane","name":"%s","at":"%s"}\n' "$name" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" | tee "$marker"
        if [[ "$name" =~ ^side_cycle_[0-9]{8}T[0-9]{6}Z_([a-zA-Z0-9-]+)(_[a-zA-Z0-9-]+)?_b[0-9]+_update\.pt$ ]]; then
          worker="${BASH_REMATCH[1]}"
          if [[ "$worker" =~ ^(geth|mcp|prime|communist-web)$ ]]; then
            printf '{"event":"trigger_next_lease_after_rejection","worker":"%s","at":"%s"}\n' "$worker" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
            bash /workspace/agillm41_vast_side_cycle.sh --worker "$worker" &
          fi
        fi
        continue
      fi
      mv "$dst.tmp" "$dst"
      printf '{"event":"pulled_side_update","name":"%s","remote_dir":"%s","at":"%s","dst":"%s"}\n' "$name" "$remote_dir" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$dst" | tee "$marker"
      if [[ "$name" =~ ^side_cycle_[0-9]{8}T[0-9]{6}Z_([a-zA-Z0-9-]+)(_[a-zA-Z0-9-]+)?_b[0-9]+_update\.pt$ ]]; then
        worker="${BASH_REMATCH[1]}"
        if [[ "$worker" =~ ^(geth|mcp|prime|communist-web)$ ]]; then
          printf '{"event":"trigger_next_lease_after_pull","worker":"%s","at":"%s"}\n' "$worker" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
          bash /workspace/agillm41_vast_side_cycle.sh --worker "$worker" &
        fi
      fi
    done <<< "$names"
  done
}

if [ "${1:-}" = "--once" ]; then
  copy_once
  prune_local_archives || true
  exit 0
fi

while true; do
  copy_once || true
  prune_local_archives || true
  sleep "$POLL_SEC"
done