체크포인트와 새 Spot 노드 복구 시험

set -euo pipefail
cd "$ISAAC_REPO"
isaac submit --scenario stack --seed 271 --order blue,red,orange --recovery-test \
  > "$ISAAC_EVIDENCE/recovery-submit.json"
export ISAAC_RECOVERY_ID="$(jq -er '.id' "$ISAAC_EVIDENCE/recovery-submit.json")"
jq -e '.owner == "operator-validation" and .operatorRecoveryTest == true
       and .operatorTestHoldSeconds == 30 and (.plan.actions | length) >= 2' \
  "$ISAAC_EVIDENCE/recovery-submit.json"

ISAAC_CHECKPOINT_READY=0
for ((i=0; i<750; i++)); do
  isaac run --id "$ISAAC_RECOVERY_ID" > "$ISAAC_EVIDENCE/recovery-before.json"
  if jq -e '(.state == "queued" or .state == "starting" or .state == "running")
      and (.checkpointCompleted // 0) >= 1
      and .checkpointCompleted < (.plan.actions | length)' \
      "$ISAAC_EVIDENCE/recovery-before.json" >/dev/null; then
    ISAAC_CHECKPOINT_READY=1
    break
  fi
  case "$(jq -er '.state' "$ISAAC_EVIDENCE/recovery-before.json")" in
    succeeded|failed|cancelled) break ;;
  esac
  sleep 2
done
test "$ISAAC_CHECKPOINT_READY" -eq 1
jq '{id,state,attempts,workerId,checkpointCompleted,checkpointKey,gpu,readyEvidence}' \
  "$ISAAC_EVIDENCE/recovery-before.json"

isaac recover --id "$ISAAC_RECOVERY_ID"
cp ".local/isaac/results/$ISAAC_RECOVERY_ID-recovery.json" \
  "$ISAAC_EVIDENCE/recovery-request.json"
jq '{kind,actualSpotInterruptionInjected,runId,checkpointCompleted,
     oldNode,oldNodeClaim,oldPod,oldWorkerId,requestedAt}' \
  "$ISAAC_EVIDENCE/recovery-request.json"

"$ISAAC_PYTHON" - <<'PY'
import json, os
from pathlib import Path
from infra.isaac import deployment as d
from isaac.cache import download
from isaac.state import artifact_key
cfg = d.config()
aws = d.session(cfg)
evidence = Path(os.environ["ISAAC_EVIDENCE"])
request = json.loads((evidence / "recovery-request.json").read_text())
run_id = os.environ["ISAAC_RECOVERY_ID"]
assert request["runId"] == run_id and request["actualSpotInterruptionInjected"] is False
key = artifact_key(run_id, request["checkpointKey"])
path = evidence / "recovery-boundary.json"
digest = download(aws.client("s3"), d.outputs(aws, cfg)["ArtifactBucket"], key, path, 32 * 1024 * 1024)
checkpoint = json.loads(path.read_text())
assert checkpoint["completedActions"] == request["checkpointCompleted"]
assert all(o["released"] and o["supported"] and o["stable"]
           for o in checkpoint["boundary"]["objects"].values())
print(json.dumps({"path": str(path), "sha256": digest,
                  "completedActions": checkpoint["completedActions"]}))
PY

isaac_wait_run "$ISAAC_RECOVERY_ID" 1800
isaac run --id "$ISAAC_RECOVERY_ID" > "$ISAAC_EVIDENCE/recovery-after.json"
isaac artifact --id "$ISAAC_RECOVERY_ID" --kind replay \
  > "$ISAAC_EVIDENCE/recovery-replay-file.json"
isaac artifact --id "$ISAAC_RECOVERY_ID" --kind checkpoint \
  > "$ISAAC_EVIDENCE/recovery-checkpoint-file.json"
isaac artifact --id "$ISAAC_RECOVERY_ID" --kind preview \
  > "$ISAAC_EVIDENCE/recovery-preview-file.json"

"$ISAAC_PYTHON" - <<'PY'
from datetime import datetime
import json, os
from pathlib import Path
folder = Path(os.environ["ISAAC_EVIDENCE"])
load = lambda name: json.loads((folder / name).read_text())
before, after = load("recovery-before.json"), load("recovery-after.json")
requested, boundary = load("recovery-request.json"), load("recovery-boundary.json")
replay = json.loads(Path(load("recovery-replay-file.json")["path"]).read_text())
checkpoint = json.loads(Path(load("recovery-checkpoint-file.json")["path"]).read_text())
run_id = os.environ["ISAAC_RECOVERY_ID"]
assert before["id"] == after["id"] == requested["runId"] == replay["id"] == run_id
assert after["state"] == replay["status"] == "succeeded"
assert after["attempts"] > before["attempts"]
assert after["workerId"] != requested["oldWorkerId"]
assert after["gpu"]["node"] != requested["oldNode"]
assert after["readyEvidence"]["resumeCount"] >= 1
assert boundary["runId"] == checkpoint["runId"] == after["readyEvidence"]["runId"]
assert boundary["requestHash"] == checkpoint["requestHash"] == replay["provenance"]["requestHash"]
assert checkpoint["completedActions"] == len(after["plan"]["actions"])
assert replay["frames"][:len(boundary["frames"])] == boundary["frames"]
assert all(b["t"] > a["t"] for a, b in zip(replay["frames"], replay["frames"][1:]))
assert [s["objectId"] for s in replay["steps"]] == ["blue", "red", "orange"]
assert replay["metrics"]["success"] and replay["metrics"]["placed"] == 3
assert all(c["passed"] for c in replay["checks"])
assert any(c["name"] == "restored released milestone" and c["passed"] for c in replay["checks"])
ready_at = datetime.fromisoformat(after["readyEvidence"]["timestamp"].replace("Z", "+00:00")).timestamp()
seconds = ready_at - requested["requestedAt"]
assert seconds >= 0
report = {
    "runId": run_id, "oldNode": requested["oldNode"], "newNode": after["gpu"]["node"],
    "checkpointCompletedBeforeReplacement": boundary["completedActions"],
    "attempts": after["attempts"], "replacementRequestToRestoredReadySeconds": round(seconds, 3),
    "actualSpotInterruptionInjected": False, "passed": True,
}
(folder / "recovery-verification.json").write_text(json.dumps(report, indent=2) + "\n")
print(json.dumps(report, indent=2))
PY