code/federation/agillm41_vast_side_cycle.sh

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

MAINLINE="${AGILLM41_MAINLINE:-/workspace/agillm41-mainline}"
SAVE_DIR="${AGILLM41_SAVE_DIR:-/workspace/agillm4_4090_ckpts_active}"
ROUND_ROOT="${AGILLM41_ROUND_ROOT:-/workspace/agillm41_side_rounds}"
GETH_HOST="${AGILLM41_GETH_HOST:-root@5.75.217.57}"
GETH_KEY="${AGILLM41_GETH_KEY:-/root/.ssh/agillm41_geth_ed25519}"
GETH_WORKER_ROOT="${AGILLM41_GETH_WORKER_ROOT:-/root/agillm41_worker}"
OPPORTUNISTIC_ROOT="${AGILLM41_OPPORTUNISTIC_ROOT:-/root/agillm41_opportunistic}"
INTERVAL_SEC="${AGILLM41_SIDE_CYCLE_SEC:-900}"
THREADS="${AGILLM41_SIDE_THREADS:-8}"
SMALL_NODE_THREADS="${AGILLM41_SMALL_NODE_THREADS:-2}"
WORKERS_SPEC="${AGILLM41_WORKERS:-geth:0,mcp:1,prime:2,communist-web:3,laptop-auto:0}"
KEEP_ROUNDS="${AGILLM41_SIDE_KEEP_ROUNDS:-2}"
LAST_SUCCESS_PATH="${AGILLM41_SIDE_LAST_SUCCESS:-/workspace/agillm41_side_cycle_last_success}"
LAST_FLUSH_REQUEST_PATH="${AGILLM41_SIDE_LAST_FLUSH_REQUEST:-/workspace/agillm41_side_cycle_last_flush_request}"
RUN_STATE_PATH="${AGILLM41_RUN_STATE_PATH:-$SAVE_DIR/run_state.json}"
MAX_CKPT_LAG_STEPS="${AGILLM41_SIDE_MAX_CKPT_LAG_STEPS:-2500}"
MAX_CKPT_AGE_SEC="${AGILLM41_SIDE_MAX_CKPT_AGE_SEC:-2700}"
FLUSH_MIN_INTERVAL_SEC="${AGILLM41_SIDE_FLUSH_MIN_INTERVAL_SEC:-1800}"
AUTO_FLUSH_STALE_CKPT="${AGILLM41_SIDE_AUTO_FLUSH:-1}"
SIDE_ATTN_BACKEND="${AGILLM41_SIDE_ATTN_BACKEND:-sdpa}"
case "$SIDE_ATTN_BACKEND" in
  sdpa|sublinear|manual) ;;
  *)
    printf 'invalid AGILLM41_SIDE_ATTN_BACKEND=%s (expected sdpa, sublinear, or manual)
' "$SIDE_ATTN_BACKEND" >&2
    exit 2
    ;;
esac
SIDE_ATTN_ARGS=(--attn-backend "$SIDE_ATTN_BACKEND")
SIDE_AR_PROB="${AGILLM41_SIDE_AR_PROB:-0.45}"
SIDE_SAT_PROB="${AGILLM41_SIDE_SAT_PROB:-0.40}"
SIDE_NAT_PROB="${AGILLM41_SIDE_NAT_PROB:-0.15}"
SIDE_AR_LOSS_TOKENS="${AGILLM41_SIDE_AR_LOSS_TOKENS:-128}"
SIDE_SAT_LOSS_TOKENS="${AGILLM41_SIDE_SAT_LOSS_TOKENS:-128}"
SIDE_NAT_LOSS_TOKENS="${AGILLM41_SIDE_NAT_LOSS_TOKENS:-128}"
SIDE_NAT_MASK_RATIO="${AGILLM41_SIDE_NAT_MASK_RATIO:-0.5}"
if [ "$SIDE_ATTN_BACKEND" = "sublinear" ]; then
  SIDE_ATTN_ARGS+=(
    --sublinear-window 128 --sublinear-stride 128 --sublinear-max-anchors 128 --sublinear-chunk 128
    --sublinear-sinks 4 --sublinear-recent-anchors 64
  )
fi

# Live AGILLM4.3 master config overrides
export AGILLM_MOE_SHARED_EXPERTS="${AGILLM_MOE_SHARED_EXPERTS:-1}"
export AGILLM_MOE_SHARED_MLP_MULT="${AGILLM_MOE_SHARED_MLP_MULT:-2}"

GPU_PACKAGE_WORKER="${AGILLM41_GPU_PACKAGE_WORKER:-vast-gpu}"
GPU_OPPORTUNISTIC_WORKER="${AGILLM41_GPU_OPPORTUNISTIC_WORKER:-vast-gpu-auto}"
LAPTOP_WORKERS_SPEC="${AGILLM41_LAPTOP_WORKERS:-laptop-cuda:0,laptop-cpu:1,laptop-igpu:2}"
LAPTOP_BATCH="${AGILLM41_LAPTOP_BATCH:-1}"
LAPTOP_BLOCK="${AGILLM41_LAPTOP_BLOCK:-128}"
LAPTOP_MAX_LAYERS="${AGILLM41_LAPTOP_MAX_LAYERS:-1}"

TARGET_WORKER=""
ONCE=0

while [ $# -gt 0 ]; do
  case "$1" in
    --once)
      ONCE=1
      shift
      ;;
    --worker)
      TARGET_WORKER="${2:?missing worker name}"
      shift 2
      ;;
    *)
      printf "unknown argument: %s\n" "$1" >&2
      exit 1
      ;;
  esac
done

GLOBAL_LOCK_PATH="${AGILLM41_SIDE_LOCK:-/tmp/agillm41_side_cycle.lock}"
LOCK_PATH="$GLOBAL_LOCK_PATH"
exec 9>"$GLOBAL_LOCK_PATH"
if ! flock -n 9; then
  printf '{"event":"side_cycle_lock_busy","worker":"%s","lock":"%s","at":"%s"}\n' "${TARGET_WORKER:-all}" "$GLOBAL_LOCK_PATH" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
  exit 0
fi

if [ -n "$TARGET_WORKER" ]; then
  WORKER_LOCK_PATH="/tmp/agillm41_side_cycle_${TARGET_WORKER}.lock"
  exec 8>"$WORKER_LOCK_PATH"
  if ! flock -n 8; then
    printf '{"event":"side_cycle_worker_lock_busy","worker":"%s","lock":"%s","at":"%s"}\n' "$TARGET_WORKER" "$WORKER_LOCK_PATH" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    exit 0
  fi
fi

latest_ckpt() {
  python - "$SAVE_DIR" <<'PY'
import json, sys
from pathlib import Path
save = Path(sys.argv[1])
latest = save / "latest.json"
if latest.exists():
    try:
        path = Path(json.loads(latest.read_text()).get("path", ""))
        if path.exists() and path.stat().st_size > 0:
            print(path)
            raise SystemExit
    except Exception:
        pass
matches = sorted(save.glob("pretrain_step*.pt"), key=lambda p: p.stat().st_mtime, reverse=True)
if not matches:
    raise SystemExit("no checkpoint found")
print(matches[0])
PY
}

copy_to_geth() {
  local out_dir="$1" base="$2" ckpt_path="$3"
  local ckpt_basename
  ckpt_basename="$(basename "$ckpt_path")"
  
  local cache_dir="$GETH_WORKER_ROOT/packages/shared_cache"
  local cache_file="$cache_dir/shared_frozen_${ckpt_basename}"
  local staging="$GETH_WORKER_ROOT/packages/.staging_$base"
  local attempt rc f fname local_size remote_size remote_sizes sizes_ok
  
  # Ensure cache directory exists on GETH
  ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" "mkdir -p '$cache_dir'" || true
  
  # Check if cached file on GETH matches local size
  local local_shared_size=0
  if [ -f "$out_dir/shared_frozen.pt" ]; then
    local_shared_size="$(stat -c %s "$out_dir/shared_frozen.pt")"
  fi
  
  local remote_cache_size
  remote_cache_size="$(ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
    "[ -f '$cache_file' ] && stat -c %s '$cache_file' || echo 'missing'" 2>/dev/null || echo 'error')"
    
  if [ "$local_shared_size" -gt 0 ] && [ "$remote_cache_size" != "$local_shared_size" ]; then
    # Cache miss! Copy shared_frozen.pt to GETH's cache
    printf '{"event":"shared_cache_miss_to_geth","ckpt":"%s","local_size":%s,"remote_size":"%s"}\n' "$ckpt_basename" "$local_shared_size" "$remote_cache_size"
    
    # Copy to temporary name first, then move atomically
    rc=0
    rsync -P --inplace -e "ssh -i \"$GETH_KEY\" -o BatchMode=yes -o StrictHostKeyChecking=no" \
      "$out_dir/shared_frozen.pt" \
      "$GETH_HOST:$cache_file.tmp" || rc=$?
      
    if [ "$rc" -eq 0 ]; then
      ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
        "mv -f '$cache_file.tmp' '$cache_file' && find '$cache_dir' -maxdepth 1 -type f -name 'shared_frozen_*' -printf '%T@ %p\n' 2>/dev/null | sort -rn | awk 'NR>2 {sub(/^[^ ]+ /, \"\"); print}' | xargs -r rm -f" || true
    else
      printf '{"event":"copy_shared_to_cache_failed","ckpt":"%s"}\n' "$ckpt_basename" >&2
      return 1
    fi
  fi
  
  # Prepare files to copy (exclude shared_frozen.pt because we will link/copy it locally on GETH)
  local files=()
  for f in "$out_dir/manifest.json" "$out_dir"/lease_*_block*_agillm4bench.pt; do
    [ -e "$f" ] && files+=("$f")
  done
  if [ "${#files[@]}" -eq 0 ]; then
    printf '{"event":"copy_to_geth_no_files","base":"%s","out_dir":"%s"}\n' "$base" "$out_dir" >&2
    return 1
  fi
  
  for attempt in 1 2 3; do
    rc=0
    # Create staging folder and populate shared_frozen.pt locally from cache on GETH (using ln or cp)
    ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
      "rm -rf '$staging' && mkdir -p '$staging' && if [ -f '$cache_file' ]; then ln -f '$cache_file' '$staging/shared_frozen.pt' || cp '$cache_file' '$staging/shared_frozen.pt'; fi" || rc=$?
      
    if [ "$rc" -eq 0 ]; then
      rsync -P --inplace -e "ssh -i \"$GETH_KEY\" -o BatchMode=yes -o StrictHostKeyChecking=no" \
        "${files[@]}" \
        "$GETH_HOST:$staging/" || rc=$?
    fi
    
    if [ "$rc" -eq 0 ]; then
      sizes_ok=1
      remote_sizes="$(ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
        "cd '$staging' && stat -c '%n %s' ./*" 2>/dev/null || true)"
        
      # Check shared_frozen.pt size
      remote_shared_size="$(printf '%s\n' "$remote_sizes" | awk '$1=="./shared_frozen.pt" {print $2; exit}')"
      if [ "$local_shared_size" -gt 0 ] && [ "$local_shared_size" != "${remote_shared_size:-}" ]; then
        printf '{"event":"copy_to_geth_size_mismatch","base":"%s","file":"shared_frozen.pt","local":%s,"remote":"%s","attempt":%s}\n' "$base" "$local_shared_size" "${remote_shared_size:-missing}" "$attempt" >&2
        sizes_ok=0
      fi
      
      # Check other files
      for f in "${files[@]}"; do
        fname="$(basename "$f")"
        local_size="$(stat -c %s "$f")"
        remote_size="$(printf '%s\n' "$remote_sizes" | awk -v n="./$fname" '$1==n {print $2; exit}')"
        if [ "$local_size" != "${remote_size:-}" ]; then
          printf '{"event":"copy_to_geth_size_mismatch","base":"%s","file":"%s","local":%s,"remote":"%s","attempt":%s}\n' "$base" "$fname" "$local_size" "${remote_size:-missing}" "$attempt" >&2
          sizes_ok=0
        fi
      done
      [ "$sizes_ok" -eq 1 ] || rc=1
    fi
    
    if [ "$rc" -eq 0 ]; then
      ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
        "mkdir -p '$GETH_WORKER_ROOT/packages/$base' && mv '$staging'/* '$GETH_WORKER_ROOT/packages/$base/' && rmdir '$staging'" || rc=$?
    fi
    
    if [ "$rc" -eq 0 ]; then
      return 0
    fi
    
    printf '{"event":"copy_to_geth_retry","base":"%s","attempt":%s,"at":"%s"}\n' "$base" "$attempt" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >&2
    if [ "$attempt" -lt 3 ]; then
      sleep 15
    fi
  done
  return 1
}

snapshot_manifest() {
  local out_dir="$1" label="$2" safe_label
  safe_label="${label//[^A-Za-z0-9_.-]/_}"
  if [ -s "$out_dir/manifest.json" ]; then
    cp "$out_dir/manifest.json" "$out_dir/manifest_${safe_label}.json"
  fi
}

refresh_manifest() {
  local out_dir="$1"
  python3 /workspace/agillm41_refresh_manifest.py "$out_dir" || true
}

mark_cycle_success() {
  date +%s > "$LAST_SUCCESS_PATH"
}

sleep_without_lock_in_child() {
  local seconds="$1"
  (exec 9>&-; sleep "$seconds")
}

snapshot_status_json() {
  python3 - "$SAVE_DIR" "$RUN_STATE_PATH" <<'PY'
import json, os, re, sys, time
save_dir, run_state_path = sys.argv[1], sys.argv[2]
latest_path = os.path.join(save_dir, 'latest.json')
now = time.time()

def load(path):
    try:
        with open(path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception:
        return {}

def age(path):
    try:
        return max(0, int(now - os.path.getmtime(path)))
    except Exception:
        return -1

latest = load(latest_path)
run_state = load(run_state_path)
ckpt_path = latest.get('path') or latest.get('checkpoint') or latest.get('ckpt_path') or ''
ckpt_step = latest.get('step') or latest.get('global_step') or 0
if not ckpt_step and ckpt_path:
    m = re.search(r'(?:step|pretrain_step)(\d+)', os.path.basename(str(ckpt_path)))
    if m:
        ckpt_step = int(m.group(1))
master_step = run_state.get('step') or run_state.get('global_step') or 0
status = {
    'master_step': int(master_step or 0),
    'ckpt_step': int(ckpt_step or 0),
    'delta_steps': int(master_step - ckpt_step) if master_step and ckpt_step else -1,
    'ckpt_age_sec': age(latest_path),
    'run_state_age_sec': age(run_state_path),
    'ckpt_path': ckpt_path,
}
print(json.dumps(status, separators=(',', ':')))
PY
}

json_field_int() {
  local json="$1" field="$2"
  python3 - "$json" "$field" <<'PY'
import json, sys
try:
    data = json.loads(sys.argv[1])
    val = data.get(sys.argv[2], -1)
    print(int(val if val is not None else -1))
except Exception:
    print(-1)
PY
}

request_fresh_checkpoint() {
  local status_json="$1" now last_req req_age trainer_pid
  now="$(date +%s)"
  last_req="$(cat "$LAST_FLUSH_REQUEST_PATH" 2>/dev/null || echo 0)"
  case "$last_req" in ''|*[!0-9]*) last_req=0;; esac
  req_age=$((now - last_req))
  if [ "$last_req" -gt 0 ] && [ "$req_age" -lt "$FLUSH_MIN_INTERVAL_SEC" ]; then
    printf '{"event":"side_cycle_waiting_for_fresh_checkpoint","reason":"flush_cooldown","cooldown_remaining_sec":%s,"snapshot":%s,"at":"%s"}
' "$((FLUSH_MIN_INTERVAL_SEC - req_age))" "$status_json" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    return 1
  fi
  if [ -e "$SAVE_DIR/FLUSH_NOW" ]; then
    printf '{"event":"side_cycle_waiting_for_fresh_checkpoint","reason":"flush_pending","snapshot":%s,"at":"%s"}
' "$status_json" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    return 1
  fi
  if [ "$AUTO_FLUSH_STALE_CKPT" != "1" ]; then
    printf '{"event":"side_cycle_stale_checkpoint_skip","reason":"auto_flush_disabled","snapshot":%s,"at":"%s"}
' "$status_json" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    return 1
  fi
  touch "$SAVE_DIR/FLUSH_NOW"
  trainer_pid="$(pgrep -f 'python.*agillm41\.py train' | head -1 || true)"
  if [ -n "$trainer_pid" ]; then
    kill -USR1 "$trainer_pid" 2>/dev/null || true
  fi
  date +%s > "$LAST_FLUSH_REQUEST_PATH"
  printf '{"event":"side_cycle_requested_fresh_checkpoint","trainer_pid":"%s","snapshot":%s,"at":"%s"}
' "${trainer_pid:-}" "$status_json" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
  return 1
}

snapshot_ready_or_request_flush() {
  local status_json delta ckpt_age
  status_json="$(snapshot_status_json)"
  delta="$(json_field_int "$status_json" delta_steps)"
  ckpt_age="$(json_field_int "$status_json" ckpt_age_sec)"
  if [ "$delta" -ne -1 ] && [ "$ckpt_age" -ge 0 ] && [ "$delta" -le "$MAX_CKPT_LAG_STEPS" ] && [ "$ckpt_age" -le "$MAX_CKPT_AGE_SEC" ]; then
    printf '{"event":"side_cycle_snapshot_fresh","snapshot":%s,"at":"%s"}
' "$status_json" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    return 0
  fi
  request_fresh_checkpoint "$status_json"
}

sleep_until_next_cycle() {
  local now last age wait_for
  now="$(date +%s)"
  last="$(cat "$LAST_SUCCESS_PATH" 2>/dev/null || echo 0)"
  case "$last" in ''|*[!0-9]*) last=0;; esac
  age=$((now - last))
  if [ "$last" -gt 0 ] && [ "$age" -lt "$INTERVAL_SEC" ]; then
    wait_for=$((INTERVAL_SEC - age))
    printf '{"event":"side_cycle_interval_guard_sleep","wait_sec":%s,"last_success":%s,"at":"%s"}\n' "$wait_for" "$last" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    sleep_without_lock_in_child "$wait_for"
  fi
}

prune_generated_artifacts() {
  find "$ROUND_ROOT" -maxdepth 1 -type d -name 'side_cycle_*' -printf '%T@ %p\n' 2>/dev/null \
    | sort -rn | awk -v keep="$KEEP_ROUNDS" 'NR>keep {sub(/^[^ ]+ /,""); print}' \
    | xargs -r rm -rf
  ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
    "find '$GETH_WORKER_ROOT/packages' -maxdepth 1 -type d -name 'side_cycle_*' -printf '%T@ %p\n' 2>/dev/null | sort -rn | awk -v keep='$KEEP_ROUNDS' 'NR>keep {sub(/^[^ ]+ /,\"\"); print}' | xargs -r rm -rf; find '$GETH_WORKER_ROOT/updates' -maxdepth 1 -type f -name 'side_cycle_*_update.pt' -mmin +120 -delete; find '$GETH_WORKER_ROOT/logs' -maxdepth 1 -type f -name 'side_cycle_*' -mmin +120 -delete; find '$OPPORTUNISTIC_ROOT/updates' -maxdepth 1 -type f -name 'laptop-auto_*.pt' -mmin +240 -delete"
  ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
    "for h in 10.0.1.20 10.0.1.30 10.0.1.1; do ssh -o BatchMode=yes -o StrictHostKeyChecking=no -o ConnectTimeout=5 \"\$h\" \"find '$GETH_WORKER_ROOT/packages' -maxdepth 1 -type d -name 'side_cycle_*' -printf '%T@ %p\n' 2>/dev/null | sort -rn | awk -v keep='$KEEP_ROUNDS' 'NR>keep {sub(/^[^ ]+ /,\\\"\\\"); print}' | xargs -r rm -rf; find '$GETH_WORKER_ROOT/updates' -maxdepth 1 -type f -name 'side_cycle_*_update.pt' -mmin +120 -delete; find '$GETH_WORKER_ROOT/logs' -maxdepth 1 -type f -name 'side_cycle_*' -mmin +120 -delete\" || true; done"
}

cycle_once() {
  local ckpt stamp
  snapshot_ready_or_request_flush || return 0
  ckpt="$(latest_ckpt)"
  stamp="$(date -u +%Y%m%dT%H%M%SZ)"
  local base="side_cycle_${stamp}"
  local out_dir="$ROUND_ROOT/$base"
  
  if [ -n "$TARGET_WORKER" ]; then
    base="${base}_${TARGET_WORKER}"
    out_dir="$ROUND_ROOT/$base"
    mkdir -p "$out_dir"
    cd "$MAINLINE"
    
    local workers="$TARGET_WORKER"
  else
    mkdir -p "$out_dir"
    cd "$MAINLINE"
    
    local workers="${AGILLM41_BATCH_WORKERS:-geth,mcp,prime,communist-web}"
    if [ "${AGILLM41_INCLUDE_OPPORTUNISTIC:-0}" = "1" ]; then
      local laptop_names=""
      for _lw in ${LAPTOP_WORKERS_SPEC//,/ }; do
        laptop_names="$laptop_names,${_lw%%:*}"
      done
      workers="$workers,${GPU_PACKAGE_WORKER}${laptop_names}"
    fi
  fi
  
  printf '{"event":"side_cycle_export_start","base":"%s","ckpt":"%s","workers":"%s","attn_backend":"%s"}\n' "$base" "$ckpt" "$workers" "$SIDE_ATTN_BACKEND"
  
  # Run batch planner and exporter. The exporter can return nonzero after
  # successfully writing packages if torch/Python finalizers trip, so treat a
  # complete manifest plus lease files as usable and continue dispatching.
  local export_rc=0
  python3 /workspace/agillm41_batch_planner.py \
    --no-daemon \
    --ckpt "$ckpt" \
    --out-dir "$out_dir" \
    --round "$base" \
    --workers "$workers" \
    --runtime agillm41.py \
    --source __default__ \
    "${SIDE_ATTN_ARGS[@]}" \
    --objective-mode stochastic --ar-prob "$SIDE_AR_PROB" --sat-prob "$SIDE_SAT_PROB" --nat-prob "$SIDE_NAT_PROB" \
    --ar-loss-tokens "$SIDE_AR_LOSS_TOKENS" --sat-loss-tokens "$SIDE_SAT_LOSS_TOKENS" --nat-loss-tokens "$SIDE_NAT_LOSS_TOKENS" \
    --nat-mask-ratio "$SIDE_NAT_MASK_RATIO" || export_rc=$?
  if [ "$export_rc" -ne 0 ]; then
    if [ -s "$out_dir/manifest.json" ] && find "$out_dir" -maxdepth 1 -type f -name 'lease_*_block*_agillm4bench.pt' | grep -q .; then
      printf '{"event":"side_cycle_export_nonzero_but_usable","base":"%s","ckpt":"%s","rc":%s,"at":"%s"}\n' "$base" "$ckpt" "$export_rc" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    else
      printf '{"event":"side_cycle_export_failed","base":"%s","ckpt":"%s","rc":%s,"at":"%s"}\n' "$base" "$ckpt" "$export_rc" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >&2
      rm -rf "$out_dir"
      return 1
    fi
  fi
  
  if [ ! -s "$out_dir/manifest.json" ]; then
    printf '{"event":"side_cycle_no_packages_exported","base":"%s"}\n' "$base"
    rm -rf "$out_dir"
    return 0
  fi
  
  # Copy packages to GETH in one go
  if ! copy_to_geth "$out_dir" "$base" "$ckpt"; then
    printf '{"event":"side_cycle_copy_failed","base":"%s","at":"%s"}\n' "$base" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >&2
    return 1
  fi
  
  # 1. Dispatch federated CPU workers
  local cpu_dispatch=0
  for w in geth mcp prime communist-web; do
    if [ -n "$(find "$out_dir" -maxdepth 1 -type f -name "lease_${w}_block*_agillm4bench.pt" | head -n 1)" ]; then
      cpu_dispatch=1
      break
    fi
  done
  
  if [ "$cpu_dispatch" -eq 1 ]; then
    printf '{"event":"dispatching_cpu_workers","base":"%s"}\n' "$base"
    ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
      "cd '$GETH_WORKER_ROOT' && AGILLM41_SIDE_THREADS='$THREADS' AGILLM41_SMALL_NODE_THREADS='$SMALL_NODE_THREADS' bash ./agillm41_dispatch_side_round.sh '$base'"
  fi
  
  # 2. Publish GPU opportunistic lease
  if [ -n "$(find "$out_dir" -maxdepth 1 -type f -name "lease_${GPU_PACKAGE_WORKER}_block*_agillm4bench.pt" | head -n 1)" ]; then
    printf '{"event":"publishing_gpu_lease","base":"%s"}\n' "$base"
    if ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
      "cd '$GETH_WORKER_ROOT' && /root/agillm3_geth_cpu/venv/bin/python code/agillm4_publish_opportunistic_lease.py --export-dir '$GETH_WORKER_ROOT/packages/$base' --root '$OPPORTUNISTIC_ROOT' --worker-id '$GPU_OPPORTUNISTIC_WORKER' --source-worker '$GPU_PACKAGE_WORKER'"; then
      printf '{"event":"gpu_lease_published","worker":"%s"}\n' "$GPU_OPPORTUNISTIC_WORKER"
    else
      printf '{"event":"gpu_lease_publish_failed","base":"%s"}\n' "$base" >&2
    fi
  fi
  
  # 3. Publish Laptop opportunistic leases
  for _lw in ${LAPTOP_WORKERS_SPEC//,/ }; do
    local _wid="${_lw%%:*}"
    if [ -n "$(find "$out_dir" -maxdepth 1 -type f -name "lease_${_wid}_block*_agillm4bench.pt" | head -n 1)" ]; then
      printf '{"event":"publishing_laptop_lease","worker":"%s","base":"%s"}\n' "$_wid" "$base"
      if ssh -i "$GETH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no "$GETH_HOST" \
        "cd '$GETH_WORKER_ROOT' && /root/agillm3_geth_cpu/venv/bin/python code/agillm4_publish_opportunistic_lease.py --export-dir '$GETH_WORKER_ROOT/packages/$base' --root '$OPPORTUNISTIC_ROOT' --worker-id '$_wid' --source-worker '$_wid'"; then
        printf '{"event":"laptop_lease_published","worker":"%s"}\n' "$_wid"
      else
        printf '{"event":"laptop_lease_publish_failed","worker":"%s","base":"%s"}\n' "$_wid" "$base" >&2
      fi
    fi
  done
  
  if [ -f /workspace/agillm41_hf_mirror.py ]; then
    HF_HUB_DISABLE_XET=1 nohup python3 /workspace/agillm41_hf_mirror.py --round-dir "$out_dir" --keep 2 >> /tmp/hf_mirror_auto.log 2>&1 &
  fi
  
  printf '{"event":"side_cycle_published","base":"%s","ckpt":"%s","at":"%s"}\n' "$base" "$ckpt" "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
  mark_cycle_success || true
  prune_generated_artifacts || true
}

if [ "$ONCE" -eq 1 ] || [ -n "$TARGET_WORKER" ]; then
  cycle_once
  exit 0
fi

while true; do
  sleep_until_next_cycle || true
  cycle_once || true
  sleep_without_lock_in_child "$INTERVAL_SEC"
done