#!/usr/bin/env zsh
# Captured r1-B serial loop; orchestration-only readiness/export correction.
# Parent owns restoration after every exit, including failure.
set -euo pipefail
E=${0:A:h:h}
M=$E/monitoring/queries.json

check_readiness() {
  local file=$1 gen=$2 epoch=$3 up fresh
  up=$(jq -r --arg g "$gen" '.identity|gsub("GENERATION";$g)' "$M")
  fresh=$(jq -r .collection_freshness "$M")
  jq -e --arg g "$gen" --argjson t "$epoch" --arg up "$up" --arg fresh "$fresh" '
    . as $p | (.queries.sample_age.response.data.result[0].value[1]|tonumber) as $age | .generation==$g and .evaluation_epoch==$t and
    (["up","scrape_count","scrape_min","sample_age","collection_freshness"] | all(. as $k |
      $p.queries[$k] as $q | ($q.response.data.result[0].value[1]|tonumber|isfinite) and $q.evaluation_epoch==$t and $q.response.status=="success" and
      ($q.response.data.result|length)==1 and
      ($q.response.data.result[0] as $r |
        (($r.value[0]-$t)|fabs)<=0.001 and
        (if $k=="collection_freshness" then true else $r.metric.registration_generation==$g end)) and
      ($q.query == (if $k=="up" then $up elif $k=="scrape_count" then "count_over_time("+$up+"[45s])"
        elif $k=="scrape_min" then "min_over_time("+$up+"[45s])"
        elif $k=="sample_age" then "time()-timestamp("+$up+")" else $fresh end)))) and
    (.queries.up.response.data.result[0].value[1]|tonumber)==1 and
    (.queries.scrape_min.response.data.result[0].value[1]|tonumber)==1 and
    (.queries.scrape_count.response.data.result[0].value[1]|tonumber)>=2 and
    $age>=0 and $age<=30 and
    (.queries.collection_freshness.response.data.result[0].value[1]|tonumber)==1
  ' "$file" >/dev/null
}

check_window() {
  local file=$1 expected_arm=$2 expected_rep=$3 now=$4
  local request_minutes=$(jq -r .request_cutoff_minutes "$E/formal/execution-plan.json")
  local wall_minutes=$(jq -r .window_minutes "$E/formal/execution-plan.json")
  jq -e --arg a "$expected_arm" --argjson r "$expected_rep" --argjson now "$now" --argjson request "$request_minutes" --argjson wall "$wall_minutes" '
    ([.window_start_epoch,.request_cutoff_epoch,.wall_deadline_epoch] | all(type=="number" and isfinite)) and
    (.window_start_epoch <= $now and $now < .request_cutoff_epoch and .request_cutoff_epoch < .wall_deadline_epoch) and
    (.request_cutoff_epoch-.window_start_epoch <= $request*60) and
    (.wall_deadline_epoch-.window_start_epoch <= $wall*60) and
    (.wall_deadline_epoch-.request_cutoff_epoch >= ($wall-$request)*60) and
    ((has("arm")|not) or .arm==$a) and ((has("repetition")|not) or .repetition==$r) and
    .audit==false and .target_requests==425 and .warmups==27
  ' "$file" >/dev/null
}
check_registry() {
  local record=$1 registry=$2 expected actual
  [[ -f $record && -f $registry ]]
  expected=$(awk -v selected="$registry" '$2==selected {print $1}' "$record")
  [[ $expected =~ '^[0-9a-f]{64}$' ]]
  actual=$(sha256sum "$registry" | cut -d' ' -f1)
  [[ $actual == $expected ]]
}
check_range() {
  local file=$1 key=$2 generation=$3
  jq -e --arg k "$key" --arg g "$generation" '
    .status=="success" and .data.resultType=="matrix" and (.data.result|type)=="array" and
    (.data.result|length)>0 and
    (if $k=="identity" or $k=="engine" then all(.data.result[];
      .metric.registration_generation==$g and
      (if $k=="identity" then .metric.__name__=="up" else (.metric.__name__|startswith("flare_")) end)) else true end)
  ' "$file" >/dev/null
}

case ${1:-} in
  --check-window) [[ $# == 5 ]]; check_window "$2" "$3" "$4" "$5"; exit;;
  --check-registry) [[ $# == 3 ]]; check_registry "$2" "$3"; exit;;
  --check-range) [[ $# == 4 ]]; check_range "$2" "$3" "$4"; exit;;
esac
if [[ ${1:-} == --check-readiness ]]; then
  [[ $# == 4 ]]
  check_readiness "$2" "$3" "$4"
  exit
fi
[[ $# == 1 ]] || { print -u2 'Usage: run-serial-lanes.zsh /absolute/preexisting/rN-arm'; exit 2; }
O=${1:A}
[[ -d $O && -f $O/window.json && ! -e $O/closeout.json && ${O:h} == ${E:A}/formal ]]
[[ ${O:t} =~ '^r[123]-[ab]$' ]]
arm=${O:t}; arm=${arm[-1]}
rep=${O:t}; rep=${rep[2]}
check_window "$O/window.json" "$arm" "$rep" "$(date +%s)"
[[ $(sha256sum "$E/runtime-pins.json" | cut -d' ' -f1) == $(jq -r .runtime_pins_sha256 "$E/formal/execution-plan.json") ]]
[[ $(sha256sum "$E/preformal-freeze.json" | cut -d' ' -f1) == $(jq -r .preformal_freeze_sha256 "$E/formal/execution-plan.json") ]]
for helper in run-phase.py replay.py decode.py oracles.py prospective.json; do
  [[ $(sha256sum "$E/harness/$helper" | cut -d' ' -f1) == $(jq -r --arg k "$helper" '.files[$k]' "$E/preformal-freeze.json") ]]
done
R=$(jq -r --arg a "$arm" '.registries[$a]' "$E/runtime-pins.json")
check_registry "$O/frozen-inputs.sha256" "$R"
W=OPERATOR_PATH_cc25b9dd30c4/anvil-serving
H=OPERATOR_PATH_601ba1a688c8/operator-home
PY=OPERATOR_PATH_9ba32d4067f1/python
model=local-inference-lab/GLM-5.3-Flash-NVFP4-MXFP8-CSF-QAD
cutoff=$(jq -r .request_cutoff_epoch "$O/window.json")
prom=$(jq -r .prometheus "$M")
lanes=(original-reverse original-forward small12-original fixed-work small12 pageheavy cold decode-c1 decode-c4)
for lane in $lanes; do [[ ! -e $O/$lane ]]; done
[[ ! -s $O/generations.txt ]]
touch "$O/generations.txt"
remaining() { local left=$((cutoff-$(date +%s))); ((left >= 75)); }
managed() { local limit=$1; shift; local left=$((cutoff-$(date +%s))); ((left > 0)); ((limit <= left)) || limit=$left; timeout "$limit" env ANVIL_SERVING_HOME="$H" "$PY" -m anvil_serving.cli "$@"; }
query_at() { local expression=$1 epoch=$2 output=$3; curl -fsS --max-time 15 --get "$prom/api/v1/query" --data-urlencode "query=$expression" --data-urlencode "time=$epoch" > "$output"; }
observe() {
  local n=$1 epoch=$(date +%s.%N) up fresh
  up=$(jq -r --arg g "$g" '.identity|gsub("GENERATION";$g)' "$M")
  fresh=$(jq -r .collection_freshness "$M")
  query_at "$up" "$epoch" "$d/monitor-up-$n.json"
  query_at "$fresh" "$epoch" "$d/monitor-freshness-$n.json"
  jq -nc --argjson n "$n" --argjson epoch "$epoch" --arg g "$g" '{sample:$n,epoch:$epoch,generation:$g}' >> "$d/monitoring-samples.jsonl"
}
ready() {
  local attempt epoch up fresh key expression
  up=$(jq -r --arg g "$g" '.identity|gsub("GENERATION";$g)' "$M")
  fresh=$(jq -r .collection_freshness "$M")
  # Preserve the captured 45-second collection catchup, then bounded retries.
  sleep 45
  for attempt in 1 2 3 4 5 6; do
    remaining
    epoch=$(date +%s.%N)
    local receipt=$d/readiness-attempt-$attempt.json
    jq -nc --arg g "$g" --argjson t "$epoch" '{schema:"flare-cache-fix-readiness/v1",generation:$g,evaluation_epoch:$t,kind:"actual pretraffic readiness guard",queries:{}}' > "$receipt"
    for key in up scrape_count scrape_min sample_age collection_freshness; do
      case $key in
        up) expression=$up;;
        scrape_count) expression="count_over_time($up[45s])";;
        scrape_min) expression="min_over_time($up[45s])";;
        sample_age) expression="time()-timestamp($up)";;
        collection_freshness) expression=$fresh;;
      esac
      local raw=$d/readiness-$attempt-$key.json
      query_at "$expression" "$epoch" "$raw"
      jq --arg k "$key" --arg q "$expression" --argjson t "$epoch" --slurpfile r "$raw" '.queries[$k]={query:$q,evaluation_epoch:$t,response:$r[0]}' "$receipt" > "$receipt.tmp"
      mv "$receipt.tmp" "$receipt"
    done
    if check_readiness "$receipt" "$g" "$epoch"; then cp "$receipt" "$d/readiness-proof.json"; return; fi
    ((attempt == 6)) || sleep 15
  done
  return 1
}
export_ranges() {
  local end=$(date +%s) key expression
  mkdir "$d/query-ranges"
  cp "$M" "$d/query-ranges/queries-manifest.json"
  jq -nc --arg g "$g" --argjson s "$lane_start" --argjson e "$end" '{generation:$g,start_epoch:$s,end_epoch:$e,keys:["engine","gpu_and_host","identity","collection_freshness"]}' > "$d/query-ranges/manifest.json"
  for key in engine gpu_and_host identity collection_freshness; do
    expression=$(jq -r --arg k "$key" --arg g "$g" '.[$k]|gsub("GENERATION";$g)' "$M")
    curl -fsS --max-time 30 --get "$prom/api/v1/query_range" --data-urlencode "query=$expression" --data-urlencode "start=$lane_start" --data-urlencode "end=$end" --data-urlencode "step=$(jq -r .step "$M")" > "$d/query-ranges/$key.json"
    check_range "$d/query-ranges/$key.json" "$key" "$g"
  done
}
run() {
  local lane=$1 sec=$2; shift 2
  check_window "$O/window.json" "$arm" "$rep" "$(date +%s)"
  check_registry "$O/frozen-inputs.sha256" "$R"
  remaining
  d=$O/$lane; mkdir "$d"; local lane_start=$(date +%s)
  cd "$W"
  managed 120 models recipes load "$model" --container glm53-flare-cache-fix --registry "$R" --gpu-device synthetic-id-14c581c315ea,synthetic-id-d013185ea964 --json --dry-run > "$d/load-preview.json" 2> "$d/load-preview.stderr"
  remaining
  managed 420 models recipes load "$model" --container glm53-flare-cache-fix --registry "$R" --gpu-device synthetic-id-14c581c315ea,synthetic-id-d013185ea964 --json --confirm > "$d/load.json" 2> "$d/load.stderr"
  g=''
  for i in 1 2 3 4 5 6; do
    remaining
    curl -fsS --max-time 15 "$(jq -r .engine_discovery "$M")" > "$d/engine-targets-$i.json"
    g=$(jq -r --arg x glm53-flare-47581cca-tp2-c4-mtp3 '..|objects|select(.model?==$x)|.registration_generation? // empty' "$d/engine-targets-$i.json" | head -1)
    [[ -n $g ]] && ! rg -Fxq "$g" "$O/generations.txt" && break
    sleep 15
  done
  [[ -n $g ]] && ! rg -Fxq "$g" "$O/generations.txt"
  print -r -- "$g" > "$d/registration-generation.txt"
  ready
  observe 1
  remaining
  timeout 240 env ANVIL_SERVING_HOME="$H" PYTHONPATH="$W" "$PY" "$E/harness/run-phase.py" --arm "$arm" --lane warmup --phase prime --output "$d/warmup" --seconds 180 --request-deadline-epoch "$cutoff" > "$d/warmup-command.json" 2>&1
  while (($# > 0)); do
    local ph=$1 lab=$2 it=$3; shift 3
    remaining
    local args=(--arm "$arm" --lane "$lane" --phase "$ph" --output "$d/$lab" --seconds "$sec" --request-deadline-epoch "$cutoff")
    [[ $it == 0 ]] || args+=(--iteration "$it")
    timeout $((sec+60)) env ANVIL_SERVING_HOME="$H" PYTHONPATH="$W" "$PY" "$E/harness/run-phase.py" "${args[@]}" > "$d/$lab-command.json" 2>&1
  done
  observe 2
  export_ranges
  # Unload remains managed; request cutoff does not block cleanup.
  timeout 120 env ANVIL_SERVING_HOME="$H" "$PY" -m anvil_serving.cli models recipes unload "$model" --container glm53-flare-cache-fix --registry "$R" --dry-run --json > "$d/unload-preview.json" 2>&1
  timeout 300 env ANVIL_SERVING_HOME="$H" "$PY" -m anvil_serving.cli models recipes unload "$model" --container glm53-flare-cache-fix --registry "$R" --confirm --json > "$d/unload.json" 2>&1
  print -r -- "$g" >> "$O/generations.txt"
  jq -nc --arg lane "$lane" --arg g "$g" --argjson now "$(date +%s)" --argjson cutoff "$cutoff" '{lane:$lane,generation:$g,completed_epoch:$now,remaining_request_seconds:($cutoff-$now)}' > "$d/lane-complete.json"
}
run original-reverse 1800 prime prime 0 measured measured 0
run original-forward 1800 prime prime 0 measured measured 0
run small12-original 1800 prime prime 0 measured measured-1 1 measured measured-2 2
run fixed-work 300 prime prime 0 measured measured 0
run small12 300 prime prime 0 measured measured-1 1 measured measured-2 2
run pageheavy 300 prime prime 0 measured measured 0
run cold 300 measured measured 0
run decode-c1 180 measured measured 0
run decode-c4 180 measured measured 0
